astrololo/services/presentation/tests/test_security.py

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

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ą.
"""
import base64

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 _basic(user: str, password: str) -> dict[str, str]:
    raw = base64.b64encode(f"{user}:{password}".encode()).decode()
    return {"Authorization": f"Basic {raw}"}


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")
    return TestClient(_app())


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

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


def test_challenges_with_basic_realm(guarded):
    """Bez nagłówka WWW-Authenticate przeglądarka nie pokaże okna logowania."""
    assert "Basic" in guarded.get("/significators").headers.get("WWW-Authenticate", "")


def test_rejects_wrong_password(guarded):
    assert guarded.get("/significators", headers=_basic(USER, "zle")).status_code == 401


def test_rejects_wrong_user(guarded):
    assert guarded.get("/significators", headers=_basic("obcy", PASSWORD)).status_code == 401


def test_rejects_malformed_header(guarded):
    for bad in ("Basic !!!niebase64!!!", "Bearer cokolwiek", "", "Basic"):
        assert guarded.get("/significators", headers={"Authorization": bad}).status_code == 401


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


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


def test_open_when_password_unset(monkeypatch):
    """Brak hasła = zgodność wstecz; ochrona wyłączona (i ostrzegamy przy starcie)."""
    monkeypatch.delenv("APP_PASSWORD", 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")
    client = TestClient(_app())
    auth = _basic(USER, PASSWORD)
    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")
    client = TestClient(_app())
    auth = _basic(USER, PASSWORD)
    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")
    client = TestClient(_app())
    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))