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, RedirectResponse

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$"

# /logowanie i /wyloguj poza bramką: inaczej żeby się zalogować, trzeba by być
# zalogowanym, a wyjście nie działałoby po wygaśnięciu sesji.
PUBLIC_PATHS = frozenset({"/health", "/logowanie", "/wyloguj"})
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 verify(login: str, password: str) -> str | None:
    """Sprawdza login i hasło; zwraca login albo 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."""
    spec = accounts().get(login)
    if spec is None:
        return None
    return login if _secret_matches(spec, password) else None


def issue_session(login: str) -> str:
    """Zawartość ciasteczka dla właśnie zalogowanego konta.

    W ciasteczku ląduje ODCISK poświadczenia, więc zmiana hasła albo usunięcie
    konta z DEMO_USERS unieważnia jego sesje samo z siebie. To jest tu jedyny
    sposób zdalnego unieważnienia: demo nie ma gdzie trwale zapisywać stanu,
    bo nie ma własnego wolumenu — a stan sesji na udziale z bazami klientów
    byłby pomieszaniem dwóch zupełnie różnych rzeczy."""
    from app import session

    return session.issue(login, session.fingerprint(accounts().get(login) or ""), 0)


def principal(request: Request) -> str | None:
    """Ciasteczko → login zalogowanej osoby albo None."""
    from app import session

    dane = session.read(request.cookies.get(session.COOKIE))
    if dane is None:
        return None
    spec = accounts().get(dane["login"])
    if spec is None or session.fingerprint(spec) != dane["odcisk"]:
        return None
    return dane["login"]


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 cookies_secure() -> bool:
    """Ciasteczko tylko dla https. Wyłączane wyłącznie w pracy lokalnej po http —
    inaczej przeglądarka po prostu go nie zapamięta i logowanie „nie działa"
    bez żadnego komunikatu."""
    return os.getenv("COOKIES_INSECURE", "").strip().lower() not in {"1", "true", "yes", "on"}


def audit_login(request: Request, login: str, udane: bool) -> None:
    """Zdarzenie logowania. Nieudane próby są ważniejsze od udanych: pojedyncza
    nic nie znaczy, seria pod jednym adresem to ślad zgadywania haseł. Login tak,
    hasło NIGDY."""
    audit_log.info("AUDYT zdarzenie=%s user=%s ip=%s",
                   "logowanie-ok" if udane else "logowanie-odmowa",
                   login or "-", client_ip(request))


def audit_logout(request: Request) -> None:
    audit_log.info("AUDYT zdarzenie=wylogowanie user=%s ip=%s",
                   getattr(request.state, "user", "") or "-", client_ip(request))


def install(app) -> None:
    from app import session

    _setup_audit_logging()

    # Fail-closed: konta bez klucza podpisu to sesje nie do odróżnienia od
    # podrobionych. Lepiej, żeby usługa nie wstała.
    if auth_enabled() and not session.available():
        raise session.SessionError(
            "Konta demo są skonfigurowane, ale brakuje SESSION_SECRET — bez klucza "
            "podpisu logowanie nie miałoby czym się bronić. openssl rand -hex 32")

    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, odswiez = "", None
        if auth_enabled():
            user = principal(request) or ""
            if not user:
                from urllib.parse import quote

                cel = request.url.path
                dokad = f"?dokad={quote(cel, safe='/')}" if request.method == "GET" and cel != "/" else ""
                return RedirectResponse(f"/logowanie{dokad}", status_code=303)
            from app import session as _s

            dane = _s.read(request.cookies.get(_s.COOKIE))
            odswiez = _s.refresh(dane) if dane else None
        request.state.user = user

        response = await call_next(request)
        if odswiez:
            from app import session as _s

            response.set_cookie(_s.COOKIE, odswiez, max_age=_s.max_age(),
                                **_s.cookie_params(secure=cookies_secure()))
        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