import hashlib import hmac import re import secrets from pathlib import Path from fastapi import HTTPException, Request from itsdangerous import BadSignature, TimestampSigner from app.config import get_settings COOKIE = "fb_session" MAX_AGE = 60 * 60 * 12 # 12 h # mtime-Cache fuer den aus der (ggf. bind-gemounteten) .env gelesenen # GUI-Passwort-Hash: {Pfad: (mtime, hash)}. Vermeidet, die Datei bei jedem # Login/Request neu zu parsen, invalidiert sich aber automatisch, sobald # services.admin.change_password() die Datei in-place neu schreibt (neue # mtime). _hash_cache: dict[str, tuple[float, str]] = {} _ENV_LINE_RE = re.compile(r"^([A-Za-z_][A-Za-z0-9_]*)='([^']*)'\s*$") def _parse_env_file(env_file: Path) -> dict[str, str]: """Parst single-quoted KEY='value'-Zeilen wie sie create_pod_finance.sh schreibt. Zeilen in anderer Form (Kommentare, unquoted, leer) werden ignoriert statt einen Fehler zu werfen.""" values: dict[str, str] = {} for line in env_file.read_text().splitlines(): m = _ENV_LINE_RE.match(line) if m: values[m.group(1)] = m.group(2) return values def current_password_hash() -> str: """Liest FB_GUI_PASSWORD_HASH zur Laufzeit aus der .env (Bind-Mount, siehe KRITISCH-Hinweis Global Constraints Ausbaustufe 4: Einzeldatei-Mount folgt dem Inode, die Datei wird von services.admin.change_password() in-place ueberschrieben). Existiert die Datei nicht (z.B. lokale Entwicklung ohne Pod), wird auf die Umgebungsvariable zurueckgefallen.""" settings = get_settings() env_file = settings.env_file if env_file.exists(): mtime = env_file.stat().st_mtime key = str(env_file) cached = _hash_cache.get(key) if cached is not None and cached[0] == mtime: return cached[1] value = _parse_env_file(env_file).get("FB_GUI_PASSWORD_HASH") if value is not None: _hash_cache[key] = (mtime, value) return value return settings.gui_password_hash def hash_password(pw: str, salt: str | None = None) -> str: salt = salt or secrets.token_hex(16) digest = hashlib.pbkdf2_hmac("sha256", pw.encode(), salt.encode(), 200_000).hex() return f"{salt}${digest}" def verify_password(pw: str, stored: str) -> bool: try: salt, digest = stored.split("$", 1) except ValueError: return False return hmac.compare_digest(hash_password(pw, salt).split("$", 1)[1], digest) def _signer() -> TimestampSigner: return TimestampSigner(get_settings().session_secret) def make_session_token() -> str: return _signer().sign(b"gui").decode() def session_valid(token: str | None) -> bool: if not token: return False try: _signer().unsign(token, max_age=MAX_AGE) return True except BadSignature: return False def require_auth(request: Request) -> None: settings = get_settings() header = request.headers.get("authorization", "") if settings.api_key and hmac.compare_digest(header, f"Bearer {settings.api_key}"): return if session_valid(request.cookies.get(COOKIE)): return raise HTTPException(status_code=401, detail="Nicht angemeldet")