astrololo/services/astroklient-demo/app/security.py

"""Logowanie do astroklienta-demo (PRE-28).

OSOBNE KONTO, NIE WSPÓŁDZIELONE Z GŁÓWNĄ APLIKACJĄ. Demo pracuje na tej samej
warstwie danych co produkcja, więc kto ma do niego dostęp, czyta oryginalne bazy
interpretacyjne. Własny login i hasło pozwalają odciąć demo jedną zmienną
środowiskową, bez ruszania kont głównej aplikacji i bez zmiany hasła komukolwiek.

Mechanizm jest CELOWO prosty: jedno konto z konfiguracji, bez ekranu kont
i bez uprawnień. Astroklient ma dwie funkcje i obie są dla tego samego człowieka —
model uprawnień z PRE-27 nie miałby tu czego dzielić.

Hasło może być jawne albo jako hash scrypt (`scrypt$sól$hash`), tym samym formatem
co APP_USERS w głównej aplikacji — dzięki temu skrypt make_user.py działa na oba.
"""
from __future__ import annotations

import base64
import binascii
import hashlib
import hmac
import logging
import os
import time
from collections import deque

from fastapi import Request
from fastapi.responses import HTMLResponse, JSONResponse

log = logging.getLogger("astroklient-demo.security")
audit_log = logging.getLogger("astroklient-demo.audit")

_SCRYPT_N, _SCRYPT_R, _SCRYPT_P, _SCRYPT_LEN = 2**14, 8, 1, 32
_HASH_PREFIX = "scrypt$"

PUBLIC_PATHS = frozenset({"/health"})
PUBLIC_PREFIXES = ("/static/",)

_hits: dict[str, deque[float]] = {}
MAX_TRACKED_CLIENTS = 4096


def accounts() -> dict[str, str]:
    """Konta demo: `login:sekret`, po przecinku, ze zmiennej DEMO_USERS.

    WIELE KONT, bo każde ma WŁASNĄ, niewidoczną dla pozostałych pulę baz (PRE-29).
    Wspólne konto oznaczałoby wspólną pulę, czyli klientów oglądających nawzajem
    swoje wgrania.

    Sekret najlepiej jako hash scrypt (`scrypt$sól$hash`) — ten sam format i ten
    sam skrypt make_user.py, co w głównej aplikacji.

    Zgodność wstecz: pojedyncze DEMO_USER/DEMO_PASSWORD działa, gdy DEMO_USERS
    jest puste."""
    out: dict[str, str] = {}
    for entry in os.getenv("DEMO_USERS", "").split(","):
        entry = entry.strip()
        if not entry:
            continue
        user, sep, secret = entry.partition(":")
        if sep and user.strip() and secret.strip():
            out[user.strip()] = secret.strip()
    if out:
        return out
    single = os.getenv("DEMO_PASSWORD", "")
    return {os.getenv("DEMO_USER", "demo"): single} if single else {}


def rate_limit_per_min() -> int:
    return int(os.getenv("RATE_LIMIT_PER_MIN", "60"))


def trust_proxy() -> bool:
    return os.getenv("TRUST_PROXY", "").strip().lower() in {"1", "true", "yes", "on"}


def auth_enabled() -> bool:
    return bool(accounts())


def _secret_matches(spec: str, password: str) -> bool:
    if spec.startswith(_HASH_PREFIX):
        try:
            _, salt_hex, hash_hex = spec.split("$", 2)
            salt, expected = bytes.fromhex(salt_hex), bytes.fromhex(hash_hex)
        except ValueError:
            return False
        actual = hashlib.scrypt(password.encode("utf-8"), salt=salt,
                                n=_SCRYPT_N, r=_SCRYPT_R, p=_SCRYPT_P, dklen=_SCRYPT_LEN)
        return hmac.compare_digest(actual, expected)
    # BAJTY, nie tekst: compare_digest na stringach rzuca TypeError przy znakach
    # spoza ASCII, więc hasło z polskimi literami dawałoby 500 zamiast odmowy.
    return hmac.compare_digest(spec.encode("utf-8"), password.encode("utf-8"))


def authenticate(header: str | None) -> str | None:
    if not header or not header.lower().startswith("basic "):
        return None
    try:
        raw = base64.b64decode(header.split(" ", 1)[1]).decode("utf-8")
        user, _, password = raw.partition(":")
    except (binascii.Error, UnicodeDecodeError, IndexError):
        return None
    spec = accounts().get(user)
    if spec is None:
        return None
    # Nieznany login i złe hasło traktujemy tak samo — odmowa nie mówi, które
    # konto istnieje, a przy demo rozdawanym szeroko to realna informacja.
    return user if _secret_matches(spec, password) else None


def client_ip(request: Request) -> str:
    """Adres do rozliczania limitu. Nagłówkom wierzymy WYŁĄCZNIE za proxy —
    inaczej wystarczyłoby podstawić własny X-Forwarded-For, żeby ominąć limit."""
    peer = request.client.host if request.client else "?"
    if not trust_proxy():
        return peer
    forwarded = request.headers.get("x-forwarded-for", "")
    if forwarded:
        last = forwarded.rsplit(",", 1)[-1].strip()
        if last:
            return last
    return request.headers.get("x-real-ip", "").strip() or peer


def _rate_limited(client: str) -> bool:
    cap = rate_limit_per_min()
    if cap <= 0:
        return False
    now = time.monotonic()
    window = _hits.get(client)
    if window is None:
        if len(_hits) >= MAX_TRACKED_CLIENTS:
            _hits.clear()
        window = _hits[client] = deque()
    while window and now - window[0] > 60.0:
        window.popleft()
    if len(window) >= cap:
        return True
    window.append(now)
    return False


def _setup_audit_logging() -> None:
    """Własny handler na stdout. Domyślna konfiguracja uvicorna nie obsługuje
    naszych loggerów, więc wpisy INFO ginęły — dziennik istniał w kodzie, ale był
    pusty. Niewidoczny dziennik jest gorszy niż jego brak (błąd zastany w PRE-17)."""
    audit_log.setLevel(os.getenv("AUDIT_LEVEL", "INFO").upper())
    if not audit_log.handlers:
        handler = logging.StreamHandler()
        handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
        audit_log.addHandler(handler)
    audit_log.propagate = False


def audit_records(request: Request, count: int) -> None:
    try:
        request.state.audit_records = int(count)
    except (TypeError, ValueError):
        pass


def install(app) -> None:
    _setup_audit_logging()
    if not auth_enabled():
        log.warning(
            "UWAGA: brak kont (DEMO_USERS / DEMO_PASSWORD) — astroklient-demo stoi OTWARTY. "
            "Bez logowania nie ma też PULI: wszyscy pracowaliby na jednym, wspólnym "
            "zbiorze baz, widząc nawzajem swoje wgrania."
        )

    @app.middleware("http")
    async def _guard(request: Request, call_next):
        if request.url.path in PUBLIC_PATHS or request.url.path.startswith(PUBLIC_PREFIXES):
            return await call_next(request)

        started = time.monotonic()
        if _rate_limited(client_ip(request)):
            return JSONResponse({"detail": "Zbyt wiele żądań — spróbuj za chwilę."},
                                status_code=429, headers={"Retry-After": "60"})

        user = ""
        if auth_enabled():
            user = authenticate(request.headers.get("authorization")) or ""
            if not user:
                return HTMLResponse(
                    "<h1>401 — wymagane logowanie</h1>", status_code=401,
                    headers={"WWW-Authenticate": 'Basic realm="astroklient-demo"'})
        request.state.user = user

        response = await call_next(request)
        records = getattr(request.state, "audit_records", None)
        audit_log.info(
            "AUDYT user=%s ip=%s method=%s path=%s status=%s records=%s ms=%.0f",
            user or "-", client_ip(request), request.method, request.url.path,
            response.status_code, "-" if records is None else records,
            (time.monotonic() - started) * 1000)
        return response