astrololo/services/presentation/tests/test_slownik_zakazany.py

"""Zapora słownikowa: czego konto NIE MOŻE zobaczyć w żadnej swojej odpowiedzi.

Poprzednie testy pilnowały pojedynczych miejsc — tego przycisku, tamtego bloku.
Wyciek prawie nigdy nie siedzi w miejscu, o którym się pamięta: katalog modeli
jechał blokiem JSON, nazwy ekranów komentarzem w arkuszu stylów, mechanizm
kwarantanny zdaniem po polsku w komentarzu CSS, a nazwa gałęzi rozwojowej
w komunikacie o błędzie. Wspólne mają jedno — pewne SŁOWA nie mają prawa paść.

Więc test nie sprawdza miejsc, tylko przechodzi WSZYSTKO, co dane konto może
pobrać (strony i zasoby), i szuka słów zakazanych. Nowy ekran, nowy skrypt,
nowy komunikat wchodzą pod tę samą zaporę bez dopisywania czegokolwiek.

Dwie z trzech list biorą się wprost z katalogu funkcji, więc nie rozjadą się
z programem: dopisanie funkcji automatycznie dopisuje jej nazwę i jej adres do
tego, czego konto bez niej nie ma prawa zobaczyć.
"""
import re

import pytest

from app import accounts as store
from app import features

# ── persona ──────────────────────────────────────────────────────────────
# Osoba zatrudniona wyłącznie do wgrywania plików. Ma widzieć swój ekran i nic
# poza nim — ani co program robi, ani co będzie robił, ani że ktoś widzi więcej.
WGRYWACZ = ["files", "files_input"]

# ── słowa zakazane, których NIE DA SIĘ wyprowadzić z katalogu ────────────
# Katalog zna nazwy funkcji, ale nie zna dziedziny programu ani słownictwa jego
# mechanizmów. Te trzeba wypisać, i każde ma tu swój powód.
DZIEDZINA = [
    # (c) po co są wgrywane pliki
    "astrolog", "horoskop", "kosmogram", "zodiak", "efemeryd", "sygnifikator",
    "interpretacyjn", "aspektarian", "antyscj", "deklinacj",
    # „baza" mówi, że to nie są zwykłe pliki, tylko zbiory czegoś. Na ekranie
    # wgrywającego wszystko nazywa się plikiem; w częściach administracyjnych
    # słowo zostaje, bo tam jest właściwe (i tam ten test nie zagląda).
    "baz",
    # (b) że tekst pisze model językowy i u kogo
    "LLM", "anthropic", "openai", "gpt", "claude", "prompt", "token",
    # (e)/(f) mechanizmy i to, że istnieje ktoś, kto widzi więcej
    "administrator", "uprawnieni", "walidacj", "kwarantann", "wstrzyman",
    "zatwierdzi", "przypisan", "zakłada", "zakładał",
    # „konto" w obu przypadkach: pierwsza wersja listy miała samo „konto",
    # więc przepuszczała zdanie „Nie masz konta" — literówka, która kosztowałaby
    # cały sens testu, gdyby nie wyszła przy oglądaniu strony.
    "konto", "konta", "kontem",
    # (d) że produkt jest w budowie i ma ponumerowany backlog
    "roadmap", "przyszł", "planowan", "w budowie", "widok testowy",
    # topologia: gdzie pukać, żeby ominąć logowanie
    "warstwa logiczna", "warstwie logicznej", "openapi", "swagger",
    # architektura: „to jedna z warstw" mówi, że warstw jest więcej
    "layer", "presentation", "prezentacj",
    "localhost", "http://logic", "8001", "8002", "8003",
]
WZORCE = [
    (r"(?i)\b(PRE|LOG|DAN)-\d+\b", "identyfikator wymagania (ujawnia backlog)"),
    (r"(?i)\bfeat/[\w-]+", "nazwa gałęzi rozwojowej"),
    (r"(?i)\bmodel(u|em|e|i|ach|owi)?\b", "generowanie tekstu przez model"),
    (r"(?i)\bAI\b", "generowanie tekstu przez model"),
]


def _zakazane_dla(perms: set[str]) -> list[tuple[str, str]]:
    """(igła, powód) — wyprowadzone z katalogu plus lista dziedzinowa."""
    out = [(s, "słownik dziedziny/mechanizmu") for s in DZIEDZINA]
    for f in features.ALL:
        if f.key in perms:
            continue
        out.append((f.label, f"nazwa funkcji, której to konto nie ma ({f.key})"))
        if f.href and f.href != "/":
            out.append((f.href, f"adres ekranu, którego to konto nie ma ({f.key})"))
    # Nazwa zasobu też jest wyciekiem: `<script src=".../wheelzoom.js">` w źródle
    # mówi, że istnieje coś do powiększania, nawet jeśli sam plik zwróci 404.
    # Wykaz bierze się z mapy zasobów, więc nowy skrypt obejmuje sam z siebie.
    for nazwa, need in features.STATIC.items():
        if not features.can(frozenset(perms), need):
            out.append((nazwa, f"nazwa zasobu, którego to konto nie może pobrać"))
    return out


# Ekran logowania MUSI mówić o koncie — inaczej nie da się go opisać. To jedyne
# odstępstwo i jest wąskie: dotyczy tej jednej strony i tych dwóch słów.
WYJATKI = {"/logowanie": {"konto", "konta", "kontem"}}


def _przeskanuj(tresc: str, perms: set[str], gdzie: str) -> list[str]:
    wolno = WYJATKI.get(gdzie, set())
    grzechy = []
    for igla, powod in _zakazane_dla(perms):
        if igla in wolno:
            continue
        if re.search(rf"(?i)(?<![\w-]){re.escape(igla)}", tresc):
            grzechy.append(f"{gdzie}: „{igla}” — {powod}")
    for wzor, powod in WZORCE:
        m = re.search(wzor, tresc)
        if m:
            grzechy.append(f"{gdzie}: „{m.group(0)}” — {powod}")
    return grzechy


@pytest.fixture()
def env(tmp_path, monkeypatch):
    monkeypatch.setenv("ACCOUNTS_FILE", str(tmp_path / "accounts.json"))
    monkeypatch.setenv("APP_USER", "szef")
    monkeypatch.setenv("APP_PASSWORD", "tajne-szefa")
    monkeypatch.delenv("APP_USERS", raising=False)
    monkeypatch.setenv("RATE_LIMIT_PER_MIN", "0")
    monkeypatch.setenv("SESSION_SECRET", "t" * 64)
    monkeypatch.setenv("COOKIES_INSECURE", "1")


def _klient(monkeypatch):
    from starlette.testclient import TestClient

    from app.main import app, logic

    # Warstwa logiczna niedostępna CELOWO: ekran plików idzie wtedy ścieżką
    # błędu, a to właśnie komunikaty o błędach wypisywały nazwy tras, nazwę
    # gałęzi i wewnętrzny adres z portem.
    import httpx

    def pada(*a, **k):
        raise httpx.ConnectError("[Errno 111] connect to http://logic:8001/files failed")

    for nazwa in ("files_list", "report", "timeline", "prompt", "horoscope", "positions"):
        if hasattr(logic, nazwa):
            monkeypatch.setattr(logic, nazwa, pada, raising=False)
    return TestClient(app, follow_redirects=False)


def _co_moze_pobrac(perms: set[str]) -> list[str]:
    """Wszystko, co to konto ma prawo dostać: strony i zasoby."""
    strony = [p for (m, p), need in features.ROUTES.items()
              if m == "GET" and features.can(frozenset(perms), need)]
    zasoby = ["/static/base.css"] + [
        f"/static/{n}" for n, need in features.STATIC.items()
        if features.can(frozenset(perms), need)]
    return sorted(set(strony + zasoby))


def test_the_uploader_never_reads_a_forbidden_word(env, monkeypatch):
    store.create("wgrywacz", "x", WGRYWACZ)
    from app import security, session

    c = _klient(monkeypatch)
    naglowki = {"Cookie": f"{session.COOKIE}={security.issue_session('wgrywacz')}"}
    perms = set(WGRYWACZ)

    grzechy, obejrzane = [], 0
    for adres in _co_moze_pobrac(perms):
        r = c.get(adres, headers=naglowki)
        if r.status_code != 200:
            continue
        obejrzane += 1
        grzechy += _przeskanuj(r.text, perms, adres)
    assert obejrzane >= 3, f"test niczego nie obejrzał ({obejrzane}) — zapora byłaby pozorna"
    assert not grzechy, "\n" + "\n".join(grzechy)


def test_an_anonymous_visitor_never_reads_a_forbidden_word(env, monkeypatch):
    """Przed zalogowaniem widać ekran logowania i jeden arkusz. Nic więcej,
    i nic w nich nie może nazywać funkcji."""
    store.create("wgrywacz", "x", WGRYWACZ)
    c = _klient(monkeypatch)
    grzechy, obejrzane = [], 0
    for adres in ("/logowanie", "/health", "/static/base.css"):
        r = c.get(adres)
        assert r.status_code == 200, f"{adres}: {r.status_code}"
        obejrzane += 1
        grzechy += _przeskanuj(r.text, set(), adres)
    assert obejrzane == 3
    assert not grzechy, "\n" + "\n".join(grzechy)


def test_every_other_asset_is_denied_to_the_uploader(env, monkeypatch):
    """Zasób spoza jego zestawu ma być NIE DO ODRÓŻNIENIA od nieistniejącego."""
    store.create("wgrywacz", "x", WGRYWACZ)
    from app import security, session

    c = _klient(monkeypatch)
    naglowki = {"Cookie": f"{session.COOKIE}={security.issue_session('wgrywacz')}"}
    wolno = set(_co_moze_pobrac(set(WGRYWACZ)))
    nieistniejacy = c.get("/static/nie-ma-takiego-pliku.js", headers=naglowki)
    for nazwa in features.STATIC:
        adres = f"/static/{nazwa}"
        if adres in wolno:
            continue
        r = c.get(adres, headers=naglowki)
        assert r.status_code == nieistniejacy.status_code == 404, f"{adres}: {r.status_code}"
        assert r.text == nieistniejacy.text, f"{adres}: odpowiedź różni się od „nie ma takiego pliku”"


def test_the_catalogue_drives_the_list_so_it_cannot_go_stale(env):
    """Zapora ma się uczyć z katalogu, nie z pamięci tego, kto ją pisał."""
    zakazane = {i for i, _ in _zakazane_dla(set(WGRYWACZ))}
    assert "Sygnifikatory" in zakazane and "/interpret" in zakazane
    # a to, co konto MA, nie może trafić na listę zakazanych
    assert "Pliki" not in zakazane


def test_the_owner_sees_everything(env, monkeypatch):
    """Kontrola pozytywna: dla administratora te słowa MAJĄ się pojawiać.
    Bez niej zapora przechodziłaby także wtedy, gdyby program był pusty."""
    from app import security, session

    c = _klient(monkeypatch)
    naglowki = {"Cookie": f"{session.COOKIE}={security.issue_session('szef')}"}
    html = c.get("/accounts", headers=naglowki).text
    assert "Sygnifikatory" in html and "Generowanie tekstu przez model" in html