feat(bezpieczeństwo): konta imienne i dziennik audytowy (PRE-17)

Jedno wspólne hasło nie mówiło, KTO sięgał do baz, a odebranie dostępu jednej
osobie wymagało zmiany hasła wszystkim. Przy bazach o realnej wartości handlowej
to za mało.

KONTA IMIENNE: `APP_USERS='alicja:scrypt$…,bartek:scrypt$…'`. Hash liczy scrypt ze
STDLIB — zero nowych zależności; hasła nie ma w konfiguracji jawnie. Zakładanie
konta: scripts/make_user.py (hasło interaktywnie, nie w historii powłoki).
Odebranie dostępu = usunięcie wpisu, reszta nie zmienia haseł.

Gdy APP_USERS jest ustawione, wspólne APP_PASSWORD PRZESTAJE działać (ostrzeżenie
przy starcie) — działające obok kont byłoby tylnym wejściem bez śladu w dzienniku.
Dopóki APP_USERS nie jest ustawione, stary tryb działa jak dotąd (zgodność wstecz).

DZIENNIK AUDYTOWY: każde żądanie zostawia wpis „kto, skąd, co, status, ILE
rekordów, ile ms". Liczba rekordów jest tu sednem — pojedyncze zapytanie wygląda
niewinnie, suma pokazuje powolne wypompowywanie bazy przez osobę uprawnioną.
Liczona dla wyszukiwarki sygnifikatorów, raportu i eksportu do Excela (ten wynosi
najwięcej naraz). W logach NIE MA treści — ani rekordów, ani promptów.

Dwa realne błędy złapane po drodze:
- `secrets.compare_digest` rzuca TypeError na znakach spoza ASCII, więc hasło z
  polskimi literami wywracało logowanie błędem 500 zamiast odmowy (błąd ZASTANY,
  sprzed tej zmiany) — porównujemy teraz bajty;
- dziennik był PUSTY na żywym serwerze: domyślna konfiguracja uvicorna nie
  obsługuje naszych loggerów. Niewidoczny dziennik jest gorszy niż jego brak,
  więc audyt dostał własny handler na stdout. Oba przypadki mają testy regresji.

Instrukcja wdrożeniowa: docs/konta-i-audyt.md. Testy: +15. Prezentacja 243.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

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:

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$<sól_hex>$<hash_hex>`, 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$<sól_hex>$<hash_hex>`."""
+    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(
-                "<h1>401 — wymagane logowanie</h1>", 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(
+                    "<h1>401 — wymagane logowanie</h1>", 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

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())

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