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