""" server-26#64 — a headless, attributable, total AI-flag flip. Three things are held here: * ``require_agent_key_or_admin`` is a DISTINCT principal. It takes the agent service key or a Firebase admin token and refuses the Discord bot's ``service_key``, so an audit entry can name who flipped the switch. * ``set_flags`` writes an ``audit_log`` entry carrying before/after values, and an audit failure can neither lose the flag write nor 500 the route. * ``cascade=True`` clears per-system ``ai_flags`` overrides for the keys being set, so a flip cannot half-apply — discovered by scanning for documents that carry the map, never a hardcoded system-id list. The dependency is exercised directly rather than through TestClient: these are assertions about the credential check, and routing them through the ASGI stack would only add ways for the test to pass for the wrong reason. """ import pytest from unittest.mock import AsyncMock, patch from fastapi import HTTPException from fastapi.security import HTTPAuthorizationCredentials from app.config import settings from app.internal import auth, feature_flags from app.routers import admin AGENT_KEY = "agent-key-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" BOT_KEY = "bot-key-bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" def _creds(token: str) -> HTTPAuthorizationCredentials: return HTTPAuthorizationCredentials(scheme="Bearer", credentials=token) @pytest.fixture def keys(monkeypatch): """Both keys configured and different — the production shape.""" monkeypatch.setattr(settings, "agent_service_key", AGENT_KEY, raising=False) monkeypatch.setattr(settings, "service_key", BOT_KEY, raising=False) @pytest.fixture(autouse=True) def _clear_flag_cache(): """feature_flags keeps module-level cache state; don't leak it across tests.""" feature_flags._cache = {} feature_flags._cache_ts = 0.0 yield feature_flags._cache = {} feature_flags._cache_ts = 0.0 # ── Item 1: the credential ──────────────────────────────────────────────────── @pytest.mark.asyncio async def test_agent_key_is_accepted_and_identifies_itself(keys): principal = await auth.require_agent_key_or_admin(_creds(AGENT_KEY)) assert principal["principal"] == "agent" # The caller must be able to tell the agent from a human admin, or the # audit entry in item 3 cannot name the actor. assert auth.describe_actor(principal) == ( auth.AGENT_PRINCIPAL_UID, auth.AGENT_PRINCIPAL_EMAIL, ) @pytest.mark.asyncio async def test_discord_bot_service_key_is_rejected(keys): """The whole point of a second key: the bot's key must not open this door. It falls through to the Firebase branch and fails there, so the bot gets a 401 rather than an unattributable flag flip. """ with patch.object(auth.firebase_auth, "verify_id_token", side_effect=Exception("not a token")): with pytest.raises(HTTPException) as exc: await auth.require_agent_key_or_admin(_creds(BOT_KEY)) assert exc.value.status_code == 401 @pytest.mark.asyncio async def test_unset_agent_key_cannot_be_bypassed(monkeypatch): """An unconfigured key must match nothing — especially not an empty string. ``secrets.compare_digest("", "")`` is a match, so the guard has to be on the key being configured, not on a ``or ""`` fallback. """ monkeypatch.setattr(settings, "agent_service_key", None, raising=False) with patch.object(auth.firebase_auth, "verify_id_token", side_effect=Exception("not a token")): for token in ("", " ", "None", "null", AGENT_KEY): with pytest.raises(HTTPException) as exc: await auth.require_agent_key_or_admin(_creds(token)) assert exc.value.status_code == 401, token @pytest.mark.asyncio async def test_empty_string_agent_key_cannot_be_bypassed(monkeypatch): """Same guarantee for a key set to "" by an empty env var.""" monkeypatch.setattr(settings, "agent_service_key", "", raising=False) with patch.object(auth.firebase_auth, "verify_id_token", side_effect=Exception("not a token")): with pytest.raises(HTTPException) as exc: await auth.require_agent_key_or_admin(_creds("")) assert exc.value.status_code == 401 @pytest.mark.asyncio async def test_firebase_admin_token_still_works(keys): decoded = {"uid": "u-1", "email": "admin@example.com", "role": "admin"} with patch.object(auth.firebase_auth, "verify_id_token", return_value=decoded): principal = await auth.require_agent_key_or_admin(_creds("firebase-id-token")) assert principal == decoded assert auth.describe_actor(principal) == ("u-1", "admin@example.com") @pytest.mark.asyncio async def test_non_admin_firebase_token_is_forbidden(keys): decoded = {"uid": "u-2", "email": "viewer@example.com", "role": "viewer"} with patch.object(auth.firebase_auth, "verify_id_token", return_value=decoded): with pytest.raises(HTTPException) as exc: await auth.require_agent_key_or_admin(_creds("firebase-id-token")) assert exc.value.status_code == 403 @pytest.mark.asyncio async def test_missing_credentials_is_401(keys): with pytest.raises(HTTPException) as exc: await auth.require_agent_key_or_admin(None) assert exc.value.status_code == 401 def test_features_routes_use_the_agent_dependency_and_others_do_not(): """Guards the wiring: only /admin/features moved off require_admin_token.""" def deps(path, method): for r in admin.router.routes: if r.path == path and method in r.methods: return {d.call for d in r.dependant.dependencies} raise AssertionError(f"no route {method} {path}") assert auth.require_agent_key_or_admin in deps("/admin/features", "GET") assert auth.require_agent_key_or_admin in deps("/admin/features", "PUT") assert auth.require_admin_token in deps("/admin/audit", "GET") assert auth.require_admin_token in deps("/admin/debug/correlation", "GET") # ── Items 3 and 4: set_flags audits, and cascades on request ────────────────── def _fstore_mock(stored: dict, systems: list[dict], updates_sink: list): """A Firestore stand-in for feature_flags: one config doc, N system docs.""" mock = AsyncMock() async def doc_get(collection, doc_id): return dict(stored) if collection == "config" else None async def doc_set(collection, doc_id, data, merge=True): stored.update(data) async def collection_list(collection, **filters): return systems if collection == "systems" else [] async def doc_update(collection, doc_id, data): updates_sink.append((collection, doc_id, data)) mock.doc_get = AsyncMock(side_effect=doc_get) mock.doc_set = AsyncMock(side_effect=doc_set) mock.collection_list = AsyncMock(side_effect=collection_list) mock.doc_update = AsyncMock(side_effect=doc_update) return mock def _systems(): return [ # Two systems carry overrides today; the ids are irrelevant to the # helper and must stay that way. {"system_id": "sys-a", "ai_flags": {"stt_enabled": False, "correlation_enabled": False}}, {"system_id": "sys-b", "ai_flags": {"stt_enabled": False}}, # Carries the map but not the key being flipped — must be left alone. {"system_id": "sys-c", "ai_flags": {"summaries_enabled": False}}, # No overrides at all: already inherits, nothing to cascade to. {"system_id": "sys-d"}, {"system_id": "sys-e", "ai_flags": {}}, ] async def _run_set_flags(updates, *, stored=None, systems=None, cascade=False, actor=None, audit_side_effect=None): stored = stored if stored is not None else {"stt_enabled": True, "correlation_enabled": True} systems = systems if systems is not None else _systems() updates_sink: list = [] audit_mock = AsyncMock(side_effect=audit_side_effect) with patch.object(feature_flags, "fstore", _fstore_mock(stored, systems, updates_sink)), \ patch("app.internal.audit.write_audit", new=audit_mock): result = await feature_flags.set_flags(updates, actor=actor, cascade=cascade) return result, stored, updates_sink, audit_mock @pytest.mark.asyncio async def test_set_flags_is_backward_compatible_without_actor_or_cascade(): """Existing call shape — set_flags({...}) — must keep working.""" result, stored, updates_sink, audit_mock = await _run_set_flags({"stt_enabled": False}) assert result["stt_enabled"] is False assert stored["stt_enabled"] is False assert updates_sink == [] # no cascade unless asked assert audit_mock.await_count == 1 # but still audited @pytest.mark.asyncio async def test_audit_records_before_and_after_and_the_actor(): _, _, _, audit_mock = await _run_set_flags( {"stt_enabled": False}, actor=(auth.AGENT_PRINCIPAL_UID, auth.AGENT_PRINCIPAL_EMAIL), ) kwargs = audit_mock.await_args.kwargs assert kwargs["action"] == "feature_flags.update" assert kwargs["actor_uid"] == auth.AGENT_PRINCIPAL_UID assert kwargs["actor_email"] == auth.AGENT_PRINCIPAL_EMAIL details = kwargs["details"] assert details["changed"]["stt_enabled"] == {"from": True, "to": False} assert details["before"]["stt_enabled"] is True assert details["after"]["stt_enabled"] is False @pytest.mark.asyncio async def test_audit_failure_neither_loses_the_write_nor_raises(): """audit_log is a record OF the write, never a precondition for it.""" result, stored, _, audit_mock = await _run_set_flags( {"stt_enabled": False}, audit_side_effect=RuntimeError("firestore down"), ) assert audit_mock.await_count == 1 assert stored["stt_enabled"] is False # flag write survived assert result["stt_enabled"] is False # and the route returns normally @pytest.mark.asyncio async def test_cascade_clears_matching_system_overrides_at_both_levels(): result, stored, updates_sink, audit_mock = await _run_set_flags( {"stt_enabled": True}, cascade=True, ) # Global level. assert stored["stt_enabled"] is True assert result["stt_enabled"] is True # System level: only the two documents whose ai_flags carry stt_enabled. written = {sid: data["ai_flags"] for _, sid, data in updates_sink} assert set(written) == {"sys-a", "sys-b"} # The flipped key is removed so the system inherits; unrelated overrides stay. assert written["sys-a"] == {"correlation_enabled": False} assert written["sys-b"] == {} # And the cascade is recorded, per system, in the audit entry. cascaded = audit_mock.await_args.kwargs["details"]["cascaded_systems"] assert {c["system_id"] for c in cascaded} == {"sys-a", "sys-b"} assert cascaded[0]["cleared_overrides"] == {"stt_enabled": False} @pytest.mark.asyncio async def test_cascade_finds_systems_by_shape_not_by_hardcoded_id(): """A newly added system carrying an override must not defeat a flip.""" systems = _systems() + [{"system_id": "sys-new", "ai_flags": {"stt_enabled": False}}] _, _, updates_sink, _ = await _run_set_flags( {"stt_enabled": True}, systems=systems, cascade=True, ) assert "sys-new" in {sid for _, sid, _ in updates_sink} @pytest.mark.asyncio async def test_cascade_off_leaves_every_system_override_intact(): """The default path must not silently erase a deliberate per-system value.""" _, _, updates_sink, _ = await _run_set_flags({"stt_enabled": True}, cascade=False) assert updates_sink == [] @pytest.mark.asyncio async def test_cascade_error_on_one_system_does_not_stop_the_others(): systems = _systems() stored = {"stt_enabled": True, "correlation_enabled": True} updates_sink: list = [] fs = _fstore_mock(stored, systems, updates_sink) real_update = fs.doc_update.side_effect async def flaky(collection, doc_id, data): if doc_id == "sys-a": raise RuntimeError("write conflict") return await real_update(collection, doc_id, data) fs.doc_update = AsyncMock(side_effect=flaky) audit_mock = AsyncMock() with patch.object(feature_flags, "fstore", fs), \ patch("app.internal.audit.write_audit", new=audit_mock): await feature_flags.set_flags({"stt_enabled": True}, cascade=True) assert [sid for _, sid, _ in updates_sink] == ["sys-b"] details = audit_mock.await_args.kwargs["details"] assert [e["system_id"] for e in details["cascade_errors"]] == ["sys-a"] @pytest.mark.asyncio async def test_unrecognised_keys_still_raise(): with pytest.raises(ValueError): await _run_set_flags({"not_a_flag": True})