98 lines
3.2 KiB
Python
98 lines
3.2 KiB
Python
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")
|