tests/test_security.py

"""Security-hardening tests.

Covers the 2026-06 hardening pass:
  * UUID validation of the JWT `sub` (the user_id chokepoint)
  * IP anonymization in run logging (data-minimization)
  * _user_pgrst_headers — per-user calls run AS THE USER (RLS) and fall back to
    the service key only when the anon key / token isn't available
  * per-IP rate limiting (sliding window, per (client, rule))
  * safe response security headers + frame-ancestors allowlist
  * removal of the /api/whoami debug endpoint

The dee.auth helpers are pure (flask + stdlib only) so they run everywhere,
including the minimal-deps CI. The server-level checks need dee.server, which
imports the ESM/torch stack; they're skipped automatically where that import
isn't available and run in full locally.
"""
import pytest
from flask import Flask, g

from dee import auth as A


# ─── UUID validation (the user_id chokepoint) ─────────────────────────
def test_is_uuid_accepts_canonical_uuid():
    assert A._is_uuid("3f8c1e2a-1b2c-4d5e-8f90-abcdef012345")


@pytest.mark.parametrize("bad", [
    "",
    "not-a-uuid",
    "3f8c1e2a1b2c4d5e8f90abcdef012345",        # no dashes
    "3f8c1e2a-1b2c-4d5e-8f90-abcdef01234",     # too short
    "zzzzzzzz-1b2c-4d5e-8f90-abcdef012345",    # non-hex
    "'; drop table users;--",
    "../../etc/passwd",
    "eq.attacker&select=*",                      # PostgREST filter injection shape
    None,
    12345,
])
def test_is_uuid_rejects_bad(bad):
    assert not A._is_uuid(bad)


# ─── IP anonymization (data-minimization) ─────────────────────────────
def test_anonymize_ipv4_zeros_last_octet():
    assert A._anonymize_ip("203.0.113.57") == "203.0.113.0"


def test_anonymize_ipv6_collapses_to_prefix():
    assert A._anonymize_ip("2001:db8:85a3:1:2:3:4:5") == "2001:db8:85a3::"


def test_anonymize_ip_collapses_hosts_in_same_subnet():
    # The host bits are gone — two hosts on one /24 become indistinguishable.
    assert A._anonymize_ip("10.0.0.1") == A._anonymize_ip("10.0.0.254") == "10.0.0.0"


@pytest.mark.parametrize("bad", ["", "   ", "garbage", "999.1.1.1", "1.2.3"])
def test_anonymize_ip_drops_unparseable(bad):
    assert A._anonymize_ip(bad) is None


# ─── _user_pgrst_headers: RLS-as-user vs service-key fallback ──────────
def _bare_app():
    return Flask(__name__)


def test_user_headers_use_user_jwt_when_signed_in(monkeypatch):
    monkeypatch.setattr(A, "SUPABASE_ANON_KEY", "anon-public-key")
    monkeypatch.setattr(A, "SUPABASE_SERVICE_KEY", "SERVICE-SECRET")
    app = _bare_app()
    with app.test_request_context("/"):
        g.auth = A.AuthContext(
            user_id="3f8c1e2a-1b2c-4d5e-8f90-abcdef012345",
            anonymous=False, token="USER.JWT.TOKEN",
        )
        h = A._user_pgrst_headers()
    assert h["apikey"] == "anon-public-key"           # public key, not service
    assert h["Authorization"] == "Bearer USER.JWT.TOKEN"
    # The service-role secret must NEVER ride along on a per-user call.
    assert "SERVICE-SECRET" not in " ".join(h.values())


def test_user_headers_use_service_when_anonymous(monkeypatch):
    monkeypatch.setattr(A, "SUPABASE_ANON_KEY", "anon-public-key")
    monkeypatch.setattr(A, "SUPABASE_SERVICE_KEY", "SERVICE-SECRET")
    app = _bare_app()
    with app.test_request_context("/"):
        g.auth = A.AuthContext(anonymous=True)        # no token
        h = A._user_pgrst_headers()
    assert h["apikey"] == "SERVICE-SECRET"


def test_user_headers_fall_back_when_anon_key_unset(monkeypatch):
    # Reversible rollout: with the anon key unset, behaviour == previous (service).
    monkeypatch.setattr(A, "SUPABASE_ANON_KEY", "")
    monkeypatch.setattr(A, "SUPABASE_SERVICE_KEY", "SERVICE-SECRET")
    app = _bare_app()
    with app.test_request_context("/"):
        g.auth = A.AuthContext(
            user_id="3f8c1e2a-1b2c-4d5e-8f90-abcdef012345",
            anonymous=False, token="USER.JWT.TOKEN",
        )
        h = A._user_pgrst_headers()
    assert h["apikey"] == "SERVICE-SECRET"


def test_user_headers_outside_request_context_use_service(monkeypatch):
    # Background-thread style call (no flask g) must fall back to service.
    monkeypatch.setattr(A, "SUPABASE_ANON_KEY", "anon-public-key")
    monkeypatch.setattr(A, "SUPABASE_SERVICE_KEY", "SERVICE-SECRET")
    h = A._user_pgrst_headers()
    assert h["apikey"] == "SERVICE-SECRET"


def test_user_headers_merge_extra(monkeypatch):
    monkeypatch.setattr(A, "SUPABASE_ANON_KEY", "")
    monkeypatch.setattr(A, "SUPABASE_SERVICE_KEY", "SERVICE-SECRET")
    h = A._user_pgrst_headers({"Prefer": "return=minimal"})
    assert h["Prefer"] == "return=minimal"
    assert h["Accept"] == "application/json"


# ─── Server-level checks (need dee.server / ESM stack) ────────────────
try:
    from dee import server as _srv
except Exception:  # noqa: BLE001 — torch/transformers absent (e.g. CI)
    _srv = None

requires_server = pytest.mark.skipif(
    _srv is None,
    reason="dee.server (ESM/torch stack) not importable in this environment",
)


@requires_server
def test_whoami_endpoint_removed():
    c = _srv.create_app().test_client()
    assert c.get("/api/whoami").status_code == 404


@requires_server
def test_security_headers_present():
    c = _srv.create_app().test_client()
    r = c.get("/")
    assert r.headers.get("X-Content-Type-Options") == "nosniff"
    assert r.headers.get("Referrer-Policy") == "strict-origin-when-cross-origin"
    assert r.headers.get("Strict-Transport-Security")
    assert "geolocation=()" in r.headers.get("Permissions-Policy", "")
    csp = r.headers.get("Content-Security-Policy", "")
    assert "frame-ancestors" in csp
    assert "turingdna.com" in csp


@requires_server
def test_rate_limit_rule_matching():
    # Most-specific prefix wins; non-API paths are never limited.
    assert _srv._rate_limit_rule("/api/run")[0] == "/api/run"
    assert _srv._rate_limit_rule("/api/crispr/design")[0] == "/api/crispr"
    assert _srv._rate_limit_rule("/api/admin/priors/de")[0] == "/api"  # falls to default
    assert _srv._rate_limit_rule("/static/app.js") is None
    assert _srv._rate_limit_rule("/") is None


@requires_server
def test_rate_limit_pure_function_trips():
    _srv._RL_HITS.clear()
    key, path = "1.2.3.4", "/api/run"   # rule (20, 60)
    tripped = False
    for _ in range(25):
        limited, retry = _srv._check_rate_limit(path, key)
        if limited:
            tripped = True
            assert retry >= 1
            break
    assert tripped
    _srv._RL_HITS.clear()


@requires_server
def test_rate_limit_buckets_are_independent_per_rule():
    # Hammering one rule must not exhaust another's budget for the same client.
    _srv._RL_HITS.clear()
    key = "5.6.7.8"
    for _ in range(60):
        _srv._check_rate_limit("/api/plasmid/import", key)  # rule (60,60)
    limited, _ = _srv._check_rate_limit("/api/run", key)    # different bucket
    assert not limited
    _srv._RL_HITS.clear()


@requires_server
def test_rate_limit_returns_429_over_http():
    app = _srv.create_app()
    c = app.test_client()
    _srv._RL_HITS.clear()
    saw_429 = False
    for _ in range(40):
        r = c.post("/api/run", json={})
        if r.status_code == 429:
            saw_429 = True
            assert r.headers.get("Retry-After")
            break
    assert saw_429
    _srv._RL_HITS.clear()