diff --git a/docs/konta-i-audyt.md b/docs/konta-i-audyt.md new file mode 100644 index 0000000..c5a933a --- /dev/null +++ b/docs/konta-i-audyt.md @@ -0,0 +1,65 @@ +# Konta imienne i dziennik audytowy (PRE-17) + +Zamiast jednego wspólnego hasła: **konta imienne**, bo przy bazach o realnej +wartości handlowej trzeba wiedzieć **kto** sięgał do treści — i móc odciąć jedną +osobę bez zmiany hasła całej reszcie. + +## Zakładanie konta + +Hasło podajesz interaktywnie (nie trafia do historii powłoki ani do listy procesów); +na wyjściu dostajesz **hash**, nie hasło: + +```bash +cd services/presentation && python scripts/make_user.py alicja +``` + +Wynik wklejasz do `APP_USERS` (wpisy po przecinku): + +``` +APP_USERS='alicja:scrypt$…,bartek:scrypt$…' +``` + +Hash liczy `scrypt` ze stdlib — **bez nowych zależności**. Sekret ustawiasz jak +resztę (`kubectl create secret …`, `secretKeyRef`), nigdy w repo GitOps. + +## Odebranie dostępu jednej osobie + +Usuń jej wpis z `APP_USERS` i zrestartuj `presentation`. **Pozostali nie zmieniają +haseł** — to była główna bolączka wspólnego hasła. + +## Uwaga: wspólne hasło przestaje działać + +Gdy `APP_USERS` jest ustawione, stare `APP_PASSWORD` **nie działa** (aplikacja +zgłasza to ostrzeżeniem przy starcie). Celowo: działające obok kont wspólne hasło +byłoby tylnym wejściem bez śladu w dzienniku, czyli dokładnie problemem, który to +wymaganie zamyka. Po migracji usuń `APP_PASSWORD` z konfiguracji. + +Zgodność wstecz: dopóki `APP_USERS` **nie** jest ustawione, `APP_USER`/`APP_PASSWORD` +działa jak dotąd — aktualizacja nie wywraca istniejącego wdrożenia. + +## Dziennik audytowy + +Każde żądanie do chronionej ścieżki zostawia wpis na stdout (w k8s zbierany +standardowo): + +``` +2026-08-03 21:45:44 INFO AUDYT user=alicja ip=10.1.2.3 method=POST path=/interpret status=200 records=428 ms=1530 +``` + +| Pole | Znaczenie | +|---|---| +| `user` | kto (`-` przy nieudanym logowaniu — nie podpowiadamy, które konto istnieje) | +| `path`, `method`, `status` | co robił i z jakim skutkiem | +| `records` | **ile rekordów baz** oddaliśmy (`-` gdy żądanie nie dotyka baz) | +| `ms` | czas obsługi | + +`records` jest tu najważniejsze: pojedyncze zapytanie wygląda niewinnie, ale suma +pokazuje **powolne wypompowywanie bazy** przez osobę uprawnioną — czego żadne +uwierzytelnienie nie wykryje. Liczone m.in. dla wyszukiwarki sygnifikatorów, +raportu interpretacji i **eksportu do Excela** (ten wynosi najwięcej naraz). + +**W dzienniku nie ma treści** — ani rekordów, ani promptów, ani danych +urodzeniowych. Logi byłyby kolejnym nośnikiem wycieku; do wykrycia nadużycia +wystarczą metadane i liczby. + +Poziom sterujesz przez `AUDIT_LEVEL` (domyślnie `INFO`). diff --git a/services/presentation/app/main.py b/services/presentation/app/main.py index 58b9c0e..d3eb06d 100644 --- a/services/presentation/app/main.py +++ b/services/presentation/app/main.py @@ -76,6 +76,18 @@ def _llm_catalog() -> dict: return {"providers": {}, "defaults": {}} +def _report_records(report: dict | None) -> int: + """Ile rekordów baz niesie raport — do dziennika audytowego (PRE-17). + + Liczymy próbki we wszystkich fasetach: to one są treścią z baz. Sama liczba, + nigdy treść — dziennik ma pokazywać SKALĘ dostępu, nie powielać danych.""" + total = 0 + for o in (report or {}).get("objects") or []: + for f in o.get("facets") or []: + total += len(f.get("samples") or []) + return total + + def _logic_error(e: Exception) -> str: if isinstance(e, httpx.HTTPStatusError) and e.response.status_code == 404: return ( @@ -338,6 +350,8 @@ def significators_search( ctx: dict = {"form": form, "result": None, "error": None} try: ctx["result"] = logic.query(query=query, field=field, exact=exact, limit=limit) + # ile rekordów baz oddaliśmy — do dziennika audytowego (PRE-17), bez treści + security.audit_records(request, len((ctx["result"] or {}).get("rows") or [])) except httpx.HTTPError as e: ctx["error"] = _logic_error(e) return templates.TemplateResponse(request, "significators.html", ctx) @@ -383,6 +397,9 @@ def interpret_run( from app.report_export import report_to_xlsx report = logic.report(when_utc_iso=iso_utc, lat=lat, lon=lon, group=group) + # Eksport wynosi NAJWIĘCEJ treści baz naraz — tym bardziej ma zostawiać + # ślad w dzienniku (PRE-17): sama liczba rekordów, nigdy treść. + security.audit_records(request, _report_records(report)) return Response( content=report_to_xlsx(report), media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", @@ -400,6 +417,7 @@ def interpret_run( ) else: ctx["result"] = logic.report(when_utc_iso=iso_utc, lat=lat, lon=lon, group=group) + security.audit_records(request, _report_records(ctx["result"])) except httpx.HTTPError as e: ctx["error"] = _logic_error(e) except ValueError as e: diff --git a/services/presentation/app/security.py b/services/presentation/app/security.py index 0eddb25..59bff8f 100644 --- a/services/presentation/app/security.py +++ b/services/presentation/app/security.py @@ -5,22 +5,38 @@ na wielu ścieżkach (`/significators`, `/interpret`, generator promptu), więc 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. +Trzy mechanizmy: + * **HTTP Basic** — wejście do aplikacji; włącza się, gdy są jakiekolwiek konta. * **limit żądań** — hamuje masowe odpytywanie (eksfiltrację przez pętlę zapytań); rozliczany per adres klienta, a za odwrotnym proxy — po TRUST_PROXY=true — per adres z nagłówka, nie per adres proxy (patrz `client_ip`). + * **konta imienne + dziennik audytowy** (PRE-17) — patrz niżej. -Świadomie NIE logujemy treści żądań ani promptów — logi to kolejny nośnik wycieku. +KONTA IMIENNE (PRE-17). Jedno wspólne hasło nie mówi, KTO sięgał do baz, a +odebranie dostępu jednej osobie wymagało zmiany hasła wszystkim. `APP_USERS` +trzyma listę kont `login:sekret` — sekret najlepiej jako HASH scrypt (stdlib, +bez nowych zależności): `scrypt$$`, patrz scripts/make_user.py. +Odebranie dostępu = usunięcie wpisu. Gdy `APP_USERS` jest ustawione, wspólne +APP_PASSWORD PRZESTAJE działać — inaczej stare hasło cicho zostawiałoby tylne +wejście, czyli dokładnie problem, który to wymaganie zamyka. -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. +DZIENNIK AUDYTOWY (PRE-17). Każde żądanie do chronionej ścieżki zostawia wpis: +kto, kiedy, jaka ścieżka, status, ile rekordów (gdy handler to poda przez +`audit_records`). To jedyny sposób, by zauważyć POWOLNE wypompowywanie bazy przez +osobę uprawnioną — pojedyncze zapytanie wygląda niewinnie, suma już nie. +Świadomie NIE logujemy treści żądań, promptów ani rekordów — logi to kolejny +nośnik wycieku; liczby i metadane wystarczą do wykrycia nadużycia. + +UWAGA: bez kont 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 hashlib +import hmac import logging import os import secrets @@ -45,6 +61,54 @@ def app_password() -> str: return os.getenv("APP_PASSWORD", "") +# ── konta imienne (PRE-17) ─────────────────────────────────────────────── +# Parametry scrypt: rozsądny kompromis czas/pamięć dla logowania do aplikacji +# wewnętrznej. Zmiana wymaga przeliczenia haseł (hash niesie tylko sól). +_SCRYPT_N, _SCRYPT_R, _SCRYPT_P, _SCRYPT_LEN = 2**14, 8, 1, 32 +_HASH_PREFIX = "scrypt$" + + +def hash_password(password: str, salt: bytes | None = None) -> str: + """Hasło → wpis do APP_USERS: `scrypt$$`.""" + salt = salt or secrets.token_bytes(16) + dk = hashlib.scrypt(password.encode("utf-8"), salt=salt, + n=_SCRYPT_N, r=_SCRYPT_R, p=_SCRYPT_P, dklen=_SCRYPT_LEN) + return f"{_HASH_PREFIX}{salt.hex()}${dk.hex()}" + + +def accounts() -> dict[str, str]: + """Konta z `APP_USERS` (`login:sekret`, po przecinku). Gdy pusto — zgodność + wstecz: pojedyncze konto z APP_USER/APP_PASSWORD.""" + out: dict[str, str] = {} + for entry in os.getenv("APP_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 + return {app_user(): app_password()} if app_password() else {} + + +def _secret_matches(spec: str, password: str) -> bool: + """Sekret konta vs podane hasło. Hash scrypt albo (zgodność wstecz) jawne.""" + 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) + # Porównujemy BAJTY, nie tekst: `compare_digest` na stringach rzuca TypeError + # przy znakach spoza ASCII, więc hasło z polskimi literami wywracało logowanie + # błędem 500 zamiast zwykłej odmowy (błąd zastany, złapany testem). + return hmac.compare_digest(spec.encode("utf-8"), password.encode("utf-8")) + + def rate_limit_per_min() -> int: return int(os.getenv("RATE_LIMIT_PER_MIN", "120")) @@ -60,25 +124,30 @@ _hits: dict[str, deque[float]] = {} def auth_enabled() -> bool: - return bool(app_password()) + return bool(accounts()) def _is_public(path: str) -> bool: return path in PUBLIC_PATHS or path.startswith(PUBLIC_PREFIXES) -def _authorized(header: str | None) -> bool: +def authenticate(header: str | None) -> str | None: + """Nagłówek Basic → LOGIN zalogowanej osoby albo None. + + Zwracamy login (nie bool), bo dziennik audytowy musi wiedzieć KTO — to sedno + PRE-17. Nieznany login i złe hasło traktujemy tak samo (bez podpowiadania, + które konto istnieje).""" if not header or not header.lower().startswith("basic "): - return False + return None 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 + return None + spec = accounts().get(user) + if spec is None: + return None + return user if _secret_matches(spec, password) else None def client_ip(request: Request) -> str: @@ -124,12 +193,62 @@ def _rate_limited(client: str) -> bool: return False +# ── dziennik audytowy (PRE-17) ─────────────────────────────────────────── +audit_log = logging.getLogger("astrololo.audit") + + +def _setup_audit_logging() -> None: + """Gwarantuje, że wpisy audytowe FAKTYCZNIE wychodzą na stdout. + + Domyślna konfiguracja uvicorna nie obsługuje naszych loggerów, więc wpisy na + poziomie INFO ginęły — dziennik istniał w kodzie, ale był pusty (sprawdzone + na żywym serwerze). Niewidoczny dziennik jest gorszy niż jego brak: daje + złudzenie kontroli. Własny handler na stdout, bo w k8s to stamtąd zbierane + są logi. `propagate=False` — żeby wpis nie dublował się przez korzeń. + """ + 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: + """Handler zgłasza, ILE rekordów baz oddał — trafi do wpisu audytowego. + + Sama liczba, nigdy treść. Bez tego dziennik pokazywałby tylko „ktoś wchodził", + a nie „ktoś wyniósł 5000 rekordów" — a to drugie jest sygnałem nadużycia.""" + try: + request.state.audit_records = int(count) + except (TypeError, ValueError): + pass + + +def _audit(request: Request, user: str, status: int, ms: float) -> None: + 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, + status, "-" if records is None else records, ms, + ) + + def install(app) -> None: """Podpina ochronę pod wszystkie ścieżki poza /health i /static.""" + _setup_audit_logging() 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." + "UWAGA: brak kont (APP_USERS / APP_PASSWORD) — aplikacja jest OTWARTA " + "dla każdego, kto ma do niej dostęp sieciowy, wraz z treścią baz " + "interpretacyjnych." + ) + elif os.getenv("APP_USERS", "").strip() and app_password(): + # Gdyby wspólne hasło działało obok kont imiennych, zostawiałoby tylne + # wejście bez śladu w dzienniku — czyli problem, który PRE-17 zamyka. + log.warning( + "APP_USERS jest ustawione, więc wspólne APP_PASSWORD NIE działa " + "(konta imienne mają pierwszeństwo). Usuń APP_PASSWORD z konfiguracji." ) @app.middleware("http") @@ -137,15 +256,24 @@ def install(app) -> None: if _is_public(request.url.path): return await call_next(request) + started = time.monotonic() if _rate_limited(client_ip(request)): + _audit(request, "-", 429, (time.monotonic() - started) * 1000) 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( - "

401 — wymagane logowanie

", status_code=401, - headers={"WWW-Authenticate": 'Basic realm="astrololo"'}, - ) - return await call_next(request) + user = "" + if auth_enabled(): + user = authenticate(request.headers.get("authorization")) or "" + if not user: + _audit(request, "-", 401, (time.monotonic() - started) * 1000) + return HTMLResponse( + "

401 — wymagane logowanie

", status_code=401, + headers={"WWW-Authenticate": 'Basic realm="astrololo"'}, + ) + request.state.user = user + response = await call_next(request) + _audit(request, user, response.status_code, (time.monotonic() - started) * 1000) + return response diff --git a/services/presentation/scripts/make_user.py b/services/presentation/scripts/make_user.py new file mode 100644 index 0000000..e46a70c --- /dev/null +++ b/services/presentation/scripts/make_user.py @@ -0,0 +1,45 @@ +"""Generator wpisu do APP_USERS — konta imienne (PRE-17). + +Hasło podajesz interaktywnie (nie w argumencie), żeby nie zostało w historii +powłoki ani w liście procesów. Na wyjściu dostajesz gotowy wpis `login:hash` +do zmiennej APP_USERS — samo hasło nigdzie nie jest zapisywane. + + python scripts/make_user.py alicja + +Kilka kont sklejasz przecinkiem: + + APP_USERS='alicja:scrypt$…,bartek:scrypt$…' + +Odebranie dostępu jednej osobie = usunięcie jej wpisu (reszta nie zmienia haseł). +""" +import getpass +import sys +import pathlib + +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) + +from app.security import hash_password # noqa: E402 + + +def main() -> int: + if len(sys.argv) != 2: + print(__doc__) + return 2 + login = sys.argv[1].strip() + if not login or ":" in login or "," in login: + print("Login nie może być pusty ani zawierać ':' lub ',' (rozdzielają wpisy).") + return 2 + first = getpass.getpass(f"Hasło dla '{login}': ") + if not first: + print("Puste hasło — przerwane.") + return 2 + if first != getpass.getpass("Powtórz hasło: "): + print("Hasła się różnią — przerwane.") + return 2 + print("\nDopisz do APP_USERS (wpisy po przecinku):\n") + print(f"{login}:{hash_password(first)}\n") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/services/presentation/tests/test_accounts_audit.py b/services/presentation/tests/test_accounts_audit.py new file mode 100644 index 0000000..c0cb70f --- /dev/null +++ b/services/presentation/tests/test_accounts_audit.py @@ -0,0 +1,148 @@ +"""Konta imienne i dziennik audytowy (PRE-17). + +Kod ochronny — testujemy zachowanie, nie kształt źródła. Dwie rzeczy są tu +krytyczne i CICHE, gdy się zepsują: (1) odebranie dostępu jednej osobie musi +faktycznie działać, (2) dziennik musi mówić KTO i ILE, ale NIGDY nie powielać +treści baz (byłby wtedy kolejnym nośnikiem wycieku). +""" +import base64 +import logging + +import pytest + +from app import security + + +def _basic(user: str, password: str) -> dict: + raw = base64.b64encode(f"{user}:{password}".encode()).decode() + return {"Authorization": f"Basic {raw}"} + + +@pytest.fixture(autouse=True) +def _clean_env(monkeypatch): + for var in ("APP_USERS", "APP_PASSWORD", "APP_USER"): + monkeypatch.delenv(var, raising=False) + security._hits.clear() # limit żądań nie może przeciekać między testami + + +# ── hasła: hash zamiast jawnego tekstu ── + +def test_hash_roundtrip_and_salt_is_random(): + h = security.hash_password("tajne") + assert h.startswith("scrypt$") + assert security._secret_matches(h, "tajne") + assert not security._secret_matches(h, "inne") + assert security.hash_password("tajne") != h # inna sól za każdym razem + + +def test_broken_hash_denies_instead_of_crashing(): + assert security._secret_matches("scrypt$niepoprawny", "cokolwiek") is False + + +def test_non_ascii_password_is_handled_not_crashing(): + """REGRESJA (błąd zastany): `secrets.compare_digest` rzuca TypeError na znakach + spoza ASCII, więc hasło z polskimi literami wywracało logowanie błędem 500 + zamiast zwykłej odmowy. Porównujemy bajty.""" + assert security._secret_matches("zażółć", "zażółć") is True + assert security._secret_matches("zażółć", "inne") is False + assert security._secret_matches(security.hash_password("gęślą"), "gęślą") is True + + +# ── konta imienne ── + +def test_named_accounts_are_parsed(monkeypatch): + monkeypatch.setenv("APP_USERS", "alicja:tajne, bartek:inne") + assert set(security.accounts()) == {"alicja", "bartek"} + + +def test_login_with_hashed_password(monkeypatch): + monkeypatch.setenv("APP_USERS", f"alicja:{security.hash_password('tajne')}") + assert security.authenticate(_basic("alicja", "tajne")["Authorization"]) == "alicja" + assert security.authenticate(_basic("alicja", "złe")["Authorization"]) is None + + +def test_unknown_user_is_denied(monkeypatch): + monkeypatch.setenv("APP_USERS", "alicja:tajne") + assert security.authenticate(_basic("obcy", "tajne")["Authorization"]) is None + + +def test_revoking_one_account_leaves_others_untouched(monkeypatch): + """Sedno PRE-17: odcinamy jedną osobę BEZ zmiany hasła pozostałym.""" + monkeypatch.setenv("APP_USERS", "alicja:tajne,bartek:inne") + assert security.authenticate(_basic("bartek", "inne")["Authorization"]) == "bartek" + monkeypatch.setenv("APP_USERS", "alicja:tajne") # usunięto wpis Bartka + assert security.authenticate(_basic("bartek", "inne")["Authorization"]) is None + assert security.authenticate(_basic("alicja", "tajne")["Authorization"]) == "alicja" + + +def test_shared_password_still_works_alone(monkeypatch): + """Zgodność wstecz — dotychczasowe wdrożenie nie może paść po aktualizacji.""" + monkeypatch.setenv("APP_USER", "astrololo") + monkeypatch.setenv("APP_PASSWORD", "wspolne") + assert security.authenticate(_basic("astrololo", "wspolne")["Authorization"]) == "astrololo" + + +def test_named_accounts_disable_the_shared_password(monkeypatch): + """Gdy są konta imienne, wspólne hasło MUSI przestać działać — inaczej + zostawiałoby tylne wejście bez śladu w dzienniku.""" + monkeypatch.setenv("APP_USERS", "alicja:tajne") + monkeypatch.setenv("APP_USER", "astrololo") + monkeypatch.setenv("APP_PASSWORD", "wspolne") + assert security.authenticate(_basic("astrololo", "wspolne")["Authorization"]) is None + + +def test_no_accounts_means_auth_disabled(monkeypatch): + assert security.auth_enabled() is False + + +# ── dziennik audytowy ── + +def _client(): + from starlette.testclient import TestClient + + from app.main import app + + return TestClient(app) + + +def test_audit_logs_who_and_how_many(monkeypatch, caplog): + monkeypatch.setenv("APP_USERS", "alicja:tajne") + with caplog.at_level(logging.INFO, logger="astrololo.audit"): + _client().get("/synastry", headers=_basic("alicja", "tajne")) + entry = "\n".join(r.getMessage() for r in caplog.records) + assert "user=alicja" in entry and "path=/synastry" in entry and "status=200" in entry + + +def test_audit_marks_failed_login_without_naming_the_account(monkeypatch, caplog): + monkeypatch.setenv("APP_USERS", "alicja:tajne") + with caplog.at_level(logging.INFO, logger="astrololo.audit"): + _client().get("/synastry", headers=_basic("alicja", "złe")) + entry = "\n".join(r.getMessage() for r in caplog.records) + assert "status=401" in entry and "user=-" in entry + + +def test_audit_logger_actually_emits(monkeypatch): + """REGRESJA: wpisy szły na INFO, ale domyślna konfiguracja uvicorna nie + obsługuje naszych loggerów — dziennik istniał w kodzie i był PUSTY na żywym + serwerze. Niewidoczny dziennik jest gorszy niż jego brak (złudzenie kontroli), + więc pilnujemy, że ma własny handler i poziom pozwalający na INFO.""" + security._setup_audit_logging() + assert security.audit_log.handlers, "dziennik audytowy nie ma własnego handlera" + assert security.audit_log.isEnabledFor(logging.INFO) + + +def test_audit_never_carries_database_content(): + """Wpis niesie WYŁĄCZNIE metadane i liczby — logi nie mogą być nośnikiem wycieku.""" + src = (__import__("pathlib").Path(security.__file__)).read_text(encoding="utf-8") + fmt = src[src.index('"AUDYT'):src.index("ms=%.0f") + 10] + for forbidden in ("rows", "samples", "effect", "significator", "text", "body"): + assert forbidden not in fmt, f"format wpisu audytowego dotyka treści: {forbidden}" + + +def test_handlers_report_record_counts(): + """Bez liczby rekordów dziennik pokazywałby „ktoś wchodził", a nie „ktoś + wyniósł 5000 rekordów" — czyli nie wykryłby powolnego wypompowywania bazy.""" + import pathlib + + main = (pathlib.Path(security.__file__).parent / "main.py").read_text(encoding="utf-8") + assert main.count("security.audit_records(") >= 3 # sygnifikatory, raport, eksport