Zugangsdaten und Proxy-Handling ohne Keepass
This commit is contained in:
C Maier
+66
-41
@@ -1,15 +1,21 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Gemeinsame Helfer für die INWX-Skripte.
|
||||
|
||||
Kapselt das Laden der Zugangsdaten aus KeePassXC sowie den Login mit
|
||||
Session-Caching: Nach dem ersten Login wird das Session-Cookie von INWX
|
||||
lokal zwischengespeichert. Folgeaufrufe nutzen diese Session wieder und
|
||||
benötigen daher weder das KeePass-Master-Passwort noch einen neuen Login,
|
||||
solange die Session bei INWX noch gültig ist.
|
||||
Kapselt das Laden der Zugangsdaten sowie den Login mit Session-Caching:
|
||||
Nach dem ersten Login wird das Session-Cookie von INWX lokal zwischen-
|
||||
gespeichert. Folgeaufrufe nutzen diese Session wieder und benötigen daher
|
||||
keinen neuen Login, solange die Session bei INWX noch gültig ist.
|
||||
|
||||
Zugangsdaten werden in dieser Reihenfolge gesucht:
|
||||
1. Umgebungsvariablen INWX_USER / INWX_PASSWORD
|
||||
(z.B. gesetzt über eine systemd EnvironmentFile=)
|
||||
2. Credentials-Datei CREDS_FILE_PATH im Format:
|
||||
username=dein_inwx_benutzername
|
||||
password=dein_inwx_passwort
|
||||
Die Datei muss chmod 600 und dem ausführenden Benutzer gehören.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
import requests
|
||||
@@ -23,50 +29,64 @@ except ImportError:
|
||||
# --- KONFIGURATION ---
|
||||
API_URL = 'https://api.domrobot.com'
|
||||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
KP_DB_PATH = os.path.join(BASE_DIR, "hosting.kdbx")
|
||||
KP_ENTRY_NAME = "inwx"
|
||||
# HTTP-Proxy für die Verbindung zur INWX-API (optional).
|
||||
# Format: http://proxyhost:port oder mit Auth: http://user:passwort@proxyhost:port
|
||||
# Kann per Umgebungsvariable INWX_PROXY gesetzt werden.
|
||||
PROXY_URL = os.environ.get("INWX_PROXY")
|
||||
# Datei mit den INWX-Zugangsdaten (username=... / password=...).
|
||||
# WICHTIG: chmod 600, nur für den ausführenden Benutzer lesbar!
|
||||
CREDS_FILE_PATH = os.path.join(BASE_DIR, ".inwx_credentials")
|
||||
# Datei mit der zwischengespeicherten Session (enthält nur ein kurzlebiges
|
||||
# Session-Cookie, niemals das Passwort).
|
||||
SESSION_CACHE_PATH = os.path.join(BASE_DIR, ".inwx_session")
|
||||
|
||||
|
||||
def get_creds():
|
||||
"""Holt Benutzername und Passwort aus KeePassXC.
|
||||
|
||||
Das KeePass-Master-Passwort wird interaktiv abgefragt. Ist die
|
||||
Umgebungsvariable KP_PW gesetzt, wird deren Wert stattdessen automatisch
|
||||
an keepassxc-cli übergeben (praktisch in nicht-interaktiven Umgebungen).
|
||||
"""
|
||||
def _check_perms(path):
|
||||
"""Warnt, falls eine Secrets-Datei für Gruppe/Andere lesbar ist."""
|
||||
try:
|
||||
# -s sorgt dafür, dass das Passwort nicht als 'PROTECTED' ausgegeben wird
|
||||
cmd = ["keepassxc-cli", "show", "-s", KP_DB_PATH, KP_ENTRY_NAME]
|
||||
mode = os.stat(path).st_mode
|
||||
if mode & 0o077:
|
||||
print(f"⚠️ {path} ist für andere Benutzer les-/schreibbar! "
|
||||
f"Bitte 'chmod 600 {path}' ausführen.")
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
master_pw = os.environ.get("KP_PW")
|
||||
if master_pw:
|
||||
# Master-Passwort über stdin an keepassxc-cli durchreichen.
|
||||
result = subprocess.run(
|
||||
cmd, input=master_pw + "\n",
|
||||
capture_output=True, text=True,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
print(f"❌ KeePass-Fehler: {result.stderr.strip()}")
|
||||
return None, None
|
||||
output = result.stdout
|
||||
else:
|
||||
# Kein KP_PW gesetzt -> keepassxc-cli fragt interaktiv nach.
|
||||
output = subprocess.check_output(cmd, text=True)
|
||||
|
||||
def get_creds():
|
||||
"""Holt Benutzername und Passwort für die INWX-API.
|
||||
|
||||
Siehe Modul-Docstring für die Reihenfolge der Quellen (ENV, dann Datei).
|
||||
"""
|
||||
env_user = os.environ.get("INWX_USER")
|
||||
env_password = os.environ.get("INWX_PASSWORD")
|
||||
if env_user and env_password:
|
||||
return env_user, env_password
|
||||
|
||||
if not os.path.exists(CREDS_FILE_PATH):
|
||||
print(f"❌ Keine Zugangsdaten gefunden: weder INWX_USER/INWX_PASSWORD "
|
||||
f"gesetzt, noch {CREDS_FILE_PATH} vorhanden.")
|
||||
return None, None
|
||||
|
||||
_check_perms(CREDS_FILE_PATH)
|
||||
|
||||
try:
|
||||
creds = {}
|
||||
for line in output.splitlines():
|
||||
if ":" in line:
|
||||
k, v = line.split(":", 1)
|
||||
with open(CREDS_FILE_PATH, "r") as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#") or "=" not in line:
|
||||
continue
|
||||
k, v = line.split("=", 1)
|
||||
creds[k.strip().lower()] = v.strip()
|
||||
|
||||
user = creds.get("username") or creds.get("benutzername") or creds.get("user")
|
||||
password = creds.get("password") or creds.get("passwort")
|
||||
user = creds.get("username") or creds.get("user")
|
||||
password = creds.get("password")
|
||||
if not user or not password:
|
||||
print(f"❌ {CREDS_FILE_PATH} enthält keine gültigen username=/password=-Zeilen.")
|
||||
return None, None
|
||||
return user, password
|
||||
except Exception as e:
|
||||
print(f"❌ KeePass-Fehler: {e}")
|
||||
print(f"❌ Fehler beim Lesen von {CREDS_FILE_PATH}: {e}")
|
||||
return None, None
|
||||
|
||||
|
||||
@@ -107,7 +127,7 @@ def _clear_cached_session():
|
||||
|
||||
|
||||
def _do_login(api):
|
||||
"""Frischer Login via KeePass-Zugangsdaten; speichert die Session. True bei Erfolg."""
|
||||
"""Frischer Login mit den konfigurierten Zugangsdaten; speichert die Session. True bei Erfolg."""
|
||||
user, password = get_creds()
|
||||
if not user or not password:
|
||||
return False
|
||||
@@ -129,9 +149,14 @@ def get_client():
|
||||
Nutzt eine zwischengespeicherte Session, falls vorhanden. Ob diese noch
|
||||
gültig ist, wird erst beim ersten echten Aufruf via call_api() geprüft –
|
||||
so sparen wir uns eine zusätzliche Anfrage. Ist keine Session vorhanden,
|
||||
wird sofort via KeePass eingeloggt.
|
||||
wird sofort neu eingeloggt.
|
||||
|
||||
Ist PROXY_URL gesetzt, laufen alle Requests der Session (Login und
|
||||
API-Aufrufe) über diesen HTTP-Proxy.
|
||||
"""
|
||||
api = ApiClient(api_url=API_URL)
|
||||
if PROXY_URL:
|
||||
api.api_session.proxies = {"http": PROXY_URL, "https": PROXY_URL}
|
||||
if _load_cached_cookies(api):
|
||||
print("ℹ️ Verwende zwischengespeicherte Session.")
|
||||
return api
|
||||
@@ -159,8 +184,8 @@ def call_api(api, method, params=None):
|
||||
"""Wie ApiClient.call_api, meldet sich aber bei abgelaufener Session neu an.
|
||||
|
||||
Deutet der Rückgabecode auf eine fehlende/abgelaufene Session hin (siehe
|
||||
_is_auth_error), wird die zwischengespeicherte Session verworfen, via
|
||||
KeePass neu eingeloggt und der Aufruf einmal wiederholt.
|
||||
_is_auth_error), wird die zwischengespeicherte Session verworfen, neu
|
||||
eingeloggt und der Aufruf einmal wiederholt.
|
||||
"""
|
||||
params = params or {}
|
||||
res = api.call_api(method, params)
|
||||
|
||||
Reference in New Issue
Block a user