astrololo/services/presentation/tests/test_security.py

"""Logowanie do aplikacji i limit żądań (LOG-32/LOG-34).

To jest brama chroniąca oryginalne bazy interpretacyjne — bez niej każdy w sieci
mógł je wypompować przez `/significators` czy generator promptu. Testy pilnują,
że brama faktycznie zamyka, a nie tylko wygląda na zamkniętą.

Od LOG-34 bramą jest SESJA, nie HTTP Basic. Zmiana nie jest kosmetyczna: Basic
nie miał wylogowania, bo przeglądarka dosyłała hasło sama przy każdym żądaniu.
Teraz to serwer decyduje, czy dana przeglądarka jest w środku — i może to cofnąć.
"""
import pytest
from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from starlette.testclient import TestClient

from app import security

USER, PASSWORD = "astrololo", "haslo-testowe"


def _sesja(user: str = USER) -> dict[str, str]:
    """Nagłówek z ważną sesją — odpowiednik bycia zalogowanym."""
    from app import session

    return {"Cookie": f"{session.COOKIE}={security.issue_session(user)}"}


def _app() -> FastAPI:
    app = FastAPI()
    security.install(app)

    @app.get("/health")
    def health():
        return {"status": "ok"}

    @app.get("/significators", response_class=HTMLResponse)
    def significators():
        return "<p>treść z bazy interpretacyjnej</p>"

    return app


@pytest.fixture(autouse=True)
def _reset_rate_limit():
    security._hits.clear()
    yield
    security._hits.clear()


@pytest.fixture
def guarded(monkeypatch):
    monkeypatch.setenv("APP_PASSWORD", PASSWORD)
    monkeypatch.setenv("RATE_LIMIT_PER_MIN", "120")
    monkeypatch.setenv("SESSION_SECRET", "t" * 64)
    monkeypatch.setenv("COOKIES_INSECURE", "1")      # TestClient jedzie po http
    return TestClient(_app(), follow_redirects=False)


# ------------------------------------------------------------------ logowanie

def test_blocks_anonymous_access(guarded):
    r = guarded.get("/significators")
    assert r.status_code == 303
    assert "treść z bazy" not in r.text


def test_sends_people_to_the_login_screen_not_a_system_dialog(guarded):
    """Basic odpowiadał 401 z WWW-Authenticate, co otwierało okienko systemowe:
    nie dało się go opisać, ostylować ani z niego wyjść inaczej niż anulowaniem.
    Ekran logowania jest zwykłą stroną, więc może wyjaśnić, co się stało."""
    r = guarded.get("/significators")
    assert r.headers["location"].startswith("/logowanie")
    assert "WWW-Authenticate" not in r.headers, "okienko systemowe wróciło"


def test_remembers_where_someone_was_heading(guarded):
    """Po zalogowaniu ma wrócić tam, gdzie szedł — nie na stronę główną."""
    assert "dokad=/significators" in guarded.get("/significators").headers["location"]


def test_no_session_no_content(guarded):
    """Popsute albo podrobione ciasteczko jest tym samym, co jego brak."""
    from app import session

    # Same znaki ASCII — nagłówek Cookie nie przenosi innych.
    for zle in ("", "smiec", "a.b", "...", session.issue(USER, "zly-odcisk")):
        r = guarded.get("/significators", headers={"Cookie": f"{session.COOKIE}={zle}"})
        assert r.status_code == 303, f"wpuszczono z ciasteczkiem: {zle!r}"


def test_allows_a_valid_session(guarded):
    r = guarded.get("/significators", headers=_sesja())
    assert r.status_code == 200
    assert "treść z bazy" in r.text


def test_credentials_are_checked_by_verify(monkeypatch):
    """Samo sprawdzanie hasła — bez HTTP, bo to czysta funkcja."""
    monkeypatch.setenv("APP_PASSWORD", PASSWORD)
    monkeypatch.delenv("APP_USERS", raising=False)
    assert security.verify(USER, PASSWORD) is not None
    assert security.verify(USER, "zle") is None
    assert security.verify("obcy", PASSWORD) is None


def test_a_session_dies_when_the_password_changes(monkeypatch):
    """Odebranie dostępu przez zmianę hasła MUSI ubić otwarte sesje — inaczej
    odebranie dostępu nie odbiera dostępu aż do wygaśnięcia sesji."""
    monkeypatch.setenv("APP_PASSWORD", PASSWORD)
    monkeypatch.setenv("SESSION_SECRET", "t" * 64)
    monkeypatch.setenv("COOKIES_INSECURE", "1")
    monkeypatch.setenv("RATE_LIMIT_PER_MIN", "0")
    client = TestClient(_app(), follow_redirects=False)
    ciasteczko = _sesja()
    assert client.get("/significators", headers=ciasteczko).status_code == 200

    monkeypatch.setenv("APP_PASSWORD", "zupełnie-nowe-hasło")
    assert client.get("/significators", headers=ciasteczko).status_code == 303


def test_health_stays_public(guarded):
    assert guarded.get("/health").status_code == 200


def test_no_key_no_service(monkeypatch):
    """Fail-closed: usługa z kontami, ale bez klucza podpisu, nie umiałaby
    odróżnić ważnej sesji od podrobionej — więc nie wstaje."""
    from app import session

    monkeypatch.setenv("APP_PASSWORD", PASSWORD)
    monkeypatch.delenv("SESSION_SECRET", raising=False)
    with pytest.raises(session.SessionError):
        _app()


def test_open_when_password_unset(monkeypatch):
    """Brak hasła = zgodność wstecz; ochrona wyłączona (i ostrzegamy przy starcie).
    Klucz sesji nie jest wtedy potrzebny — nie ma czego podpisywać."""
    monkeypatch.delenv("APP_PASSWORD", raising=False)
    monkeypatch.delenv("SESSION_SECRET", raising=False)
    assert not security.auth_enabled()
    assert TestClient(_app()).get("/significators").status_code == 200


# --------------------------------------------------------------- limit żądań

def test_rate_limit_blocks_flood(monkeypatch):
    """Masowe odpytywanie to droga eksfiltracji nawet po zalogowaniu."""
    monkeypatch.setenv("APP_PASSWORD", PASSWORD)
    monkeypatch.setenv("RATE_LIMIT_PER_MIN", "5")
    monkeypatch.setenv("SESSION_SECRET", "t" * 64)
    client = TestClient(_app())
    auth = _sesja()
    codes = [client.get("/significators", headers=auth).status_code for _ in range(8)]
    assert codes[:5] == [200] * 5
    assert 429 in codes[5:]


def test_rate_limited_response_has_retry_after(monkeypatch):
    monkeypatch.setenv("APP_PASSWORD", PASSWORD)
    monkeypatch.setenv("RATE_LIMIT_PER_MIN", "1")
    monkeypatch.setenv("SESSION_SECRET", "t" * 64)
    client = TestClient(_app())
    auth = _sesja()
    client.get("/significators", headers=auth)
    r = client.get("/significators", headers=auth)
    assert r.status_code == 429 and r.headers.get("Retry-After") == "60"


def test_rate_limit_precedes_auth(monkeypatch):
    """Limit musi działać także dla niezalogowanych — inaczej zgadywanie hasła
    i sondowanie API jest darmowe."""
    monkeypatch.setenv("APP_PASSWORD", PASSWORD)
    monkeypatch.setenv("RATE_LIMIT_PER_MIN", "3")
    monkeypatch.setenv("SESSION_SECRET", "t" * 64)
    client = TestClient(_app(), follow_redirects=False)
    codes = [client.get("/significators").status_code for _ in range(6)]
    assert 429 in codes


def test_rate_limit_disabled_when_zero(monkeypatch):
    monkeypatch.setenv("RATE_LIMIT_PER_MIN", "0")
    monkeypatch.delenv("APP_PASSWORD", raising=False)
    client = TestClient(_app())
    assert all(client.get("/significators").status_code == 200 for _ in range(30))


# ------------------------------------------ adres klienta za odwrotnym proxy
#
# Po włączeniu TLS (PRE-16) aplikacja stoi za Ingressem, więc bezpośredni peer
# to zawsze POD proxy. Te testy pilnują obu stron kompromisu: żeby limit dalej
# rozróżniał ludzi, a jednocześnie żeby nagłówek nie stał się furtką do jego
# ominięcia.

class _Req:
    """Minimalny zamiennik Request — `client_ip` czyta tylko te dwa pola."""

    def __init__(self, peer: str | None, **headers: str):
        self.client = type("C", (), {"host": peer})() if peer else None
        self.headers = {k.replace("_", "-"): v for k, v in headers.items()}


def test_client_ip_ignores_headers_without_trust_proxy(monkeypatch):
    """Bez TRUST_PROXY nagłówek jest bezwartościowy — każdy może go dopisać."""
    monkeypatch.delenv("TRUST_PROXY", raising=False)
    req = _Req("10.42.0.7", x_forwarded_for="1.2.3.4", x_real_ip="5.6.7.8")
    assert security.client_ip(req) == "10.42.0.7"


def test_client_ip_takes_last_forwarded_entry(monkeypatch):
    """Ostatni wpis dopisał NASZ proxy; wcześniejsze mógł podstawić klient."""
    monkeypatch.setenv("TRUST_PROXY", "true")
    req = _Req("10.42.0.7", x_forwarded_for="1.1.1.1, 2.2.2.2, 192.168.1.50")
    assert security.client_ip(req) == "192.168.1.50"


def test_client_ip_falls_back_to_real_ip(monkeypatch):
    monkeypatch.setenv("TRUST_PROXY", "true")
    req = _Req("10.42.0.7", x_real_ip="192.168.1.50")
    assert security.client_ip(req) == "192.168.1.50"


def test_client_ip_falls_back_to_peer_when_headers_missing(monkeypatch):
    monkeypatch.setenv("TRUST_PROXY", "true")
    assert security.client_ip(_Req("10.42.0.7")) == "10.42.0.7"


def test_spoofed_forwarded_header_cannot_dodge_the_limit(monkeypatch):
    """Sedno sprawy: bez zaufania do proxy podszywanie się NIE resetuje wiadra.

    Gdyby limiter brał pierwszy lepszy `X-Forwarded-For`, wystarczyłoby zmieniać
    go co żądanie, żeby pompować bazę bez ograniczeń.
    """
    monkeypatch.delenv("TRUST_PROXY", raising=False)
    monkeypatch.setenv("RATE_LIMIT_PER_MIN", "3")
    monkeypatch.delenv("APP_PASSWORD", raising=False)
    client = TestClient(_app())
    codes = [client.get("/significators", headers={"X-Forwarded-For": f"9.9.9.{i}"}).status_code
             for i in range(6)]
    assert 429 in codes


def test_proxied_clients_get_separate_buckets(monkeypatch):
    """Za proxy dwie różne osoby nie mogą się nawzajem odcinać."""
    monkeypatch.setenv("TRUST_PROXY", "true")
    monkeypatch.setenv("RATE_LIMIT_PER_MIN", "3")
    monkeypatch.delenv("APP_PASSWORD", raising=False)
    client = TestClient(_app())
    first = [client.get("/significators", headers={"X-Forwarded-For": "192.168.1.50"}).status_code
             for _ in range(5)]
    second = client.get("/significators", headers={"X-Forwarded-For": "192.168.1.51"})
    assert 429 in first, "limit musi zadziałać dla pierwszego adresu"
    assert second.status_code == 200, "drugi adres ma własne wiadro"