astrololo/services/presentation/app/security.py

"""Ochrona dostępu do aplikacji (LOG-32).

Rdzeniem produktu są oryginalne bazy interpretacyjne. Aplikacja podaje ich treść
na wielu ścieżkach (`/significators`, `/interpret`, generator promptu), więc BRAK
uwierzytelnienia oznacza, że każdy w sieci może je wypompować — bez udziału
jakiegokolwiek modelu językowego. Ten moduł zamyka tę drogę.

Dwa mechanizmy:
  * **HTTP Basic** — wejście do aplikacji; włącza się, gdy ustawiono APP_PASSWORD.
  * **limit żądań** — hamuje masowe odpytywanie (eksfiltrację przez pętlę zapytań).

Świadomie NIE logujemy treści żądań ani promptów — logi to kolejny nośnik wycieku.

UWAGA: bez APP_PASSWORD ochrona jest WYŁĄCZONA (zgodność wstecz i wygoda dev).
Wtedy przy starcie leci głośne ostrzeżenie — żeby nikt nie wdrożył tego w
przekonaniu, że jest chroniony.
"""
from __future__ import annotations

import base64
import binascii
import logging
import os
import secrets
import time
from collections import deque

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

log = logging.getLogger("astrololo.security")

MAX_TRACKED_CLIENTS_DEFAULT = 4096
MAX_TRACKED_CLIENTS = MAX_TRACKED_CLIENTS_DEFAULT   # zabezpieczenie przed puchnięciem pamięci


# konfiguracja czytana leniwie — testy i restart mogą ją zmienić bez importu modułu
def app_user() -> str:
    return os.getenv("APP_USER", "astrololo")


def app_password() -> str:
    return os.getenv("APP_PASSWORD", "")


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

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

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


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


def _is_public(path: str) -> bool:
    return path in PUBLIC_PATHS or path.startswith(PUBLIC_PREFIXES)


def _authorized(header: str | None) -> bool:
    if not header or not header.lower().startswith("basic "):
        return False
    try:
        raw = base64.b64decode(header.split(" ", 1)[1]).decode("utf-8")
        user, _, password = raw.partition(":")
    except (binascii.Error, UnicodeDecodeError, IndexError):
        return False
    # porównanie odporne na atak czasowy; oba pola muszą się zgadzać
    ok_user = secrets.compare_digest(user, app_user())
    ok_pass = secrets.compare_digest(password, app_password())
    return ok_user and ok_pass


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()               # prosty reset zamiast nieograniczonego wzrostu
        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 install(app) -> None:
    """Podpina ochronę pod wszystkie ścieżki poza /health i /static."""
    if not auth_enabled():
        log.warning(
            "UWAGA: APP_PASSWORD nie ustawione — aplikacja jest OTWARTA dla każdego, "
            "kto ma do niej dostęp sieciowy, wraz z treścią baz interpretacyjnych."
        )

    @app.middleware("http")
    async def _guard(request: Request, call_next):
        if _is_public(request.url.path):
            return await call_next(request)

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

        if auth_enabled() and not _authorized(request.headers.get("authorization")):
            return HTMLResponse(
                "<h1>401 — wymagane logowanie</h1>", status_code=401,
                headers={"WWW-Authenticate": 'Basic realm="astrololo"'},
            )
        return await call_next(request)