From 1c66905b4a7a85f8f3429801654f7c64c27a7aa2 Mon Sep 17 00:00:00 2001 From: timmy Date: Wed, 26 Aug 2026 06:11:28 +0000 Subject: [PATCH] feat: fresh-authorize and audit Human Gate decisions --- src/main.py | 32 ++++++++ tests/test_human_gate_api.py | 151 +++++++++++++++++++++++++++++++++++ 2 files changed, 183 insertions(+) diff --git a/src/main.py b/src/main.py index 81e4c9f..a587b46 100644 --- a/src/main.py +++ b/src/main.py @@ -561,6 +561,7 @@ class QuietHoursPayload(BaseModel): StepUpAction = Literal[ + "decide_human_gate", "merge_pull", "delete_source_branch", "prepare_release_rollback", @@ -1884,11 +1885,35 @@ async def decide_human_gate( payload: HumanGateDecision, request: Request, idempotency_key: str = Header(alias="Idempotency-Key", min_length=1, max_length=128), + step_up_grant: str | None = Header( + default=None, alias="X-Step-Up-Grant", max_length=128 + ), ): + await _require_step_up( + request, + step_up_grant, + action="decide_human_gate", + target=gate_id, + ) login = await _human_gate_login(request) + journal = _security_event_store() + operation_id = None try: store = _human_gate_store() existed = await asyncio.to_thread(store.has_receipt_key, login, idempotency_key) + if not existed: + try: + operation_id = await asyncio.to_thread( + journal.reserve, + "human_gate_decision", + method=payload.decision, + target=gate_id, + ) + except SecurityEventStoreError as error: + raise HTTPException( + status_code=503, + detail="Security activity is temporarily unavailable", + ) from error receipt = await asyncio.to_thread( store.decide, login, @@ -1901,7 +1926,14 @@ async def decide_human_gate( idempotency_key=idempotency_key, ) except (GateValidationError, GateConflict, GateNotFound, sqlite3.Error) as error: + if operation_id is not None: + try: + await asyncio.to_thread(journal.discard, operation_id) + except SecurityEventStoreError: + pass raise _gate_error(error) from error + if operation_id is not None: + await asyncio.to_thread(journal.finalize, operation_id) return JSONResponse( receipt, status_code=200 if existed else 201, headers={"Cache-Control": "no-store"} ) diff --git a/tests/test_human_gate_api.py b/tests/test_human_gate_api.py index 1c2f0de..dd3d909 100644 --- a/tests/test_human_gate_api.py +++ b/tests/test_human_gate_api.py @@ -5,6 +5,7 @@ import pytest from src import main from src.human_gate_store import HumanGateStore +from src.security_event_store import SecurityEventStore, SecurityEventStoreError CANDIDATE = { @@ -68,6 +69,156 @@ async def test_decision_requires_revision_and_returns_durable_receipt(gate_api): assert all(response.headers["cache-control"] == "no-store" for response in (decided, repeated, receipt, stale)) +@pytest.mark.anyio +async def test_decision_requires_fresh_authorization_bound_to_the_exact_gate(monkeypatch, gate_api): + gate = gate_api.intake("1:timmy", CANDIDATE, idempotency_key="run-authorized") + authorization_calls = [] + + async def require_step_up(request, grant, *, action, target): + authorization_calls.append({"grant": grant, "action": action, "target": target}) + if grant != "one-time-grant": + raise main.HTTPException( + status_code=428, + detail={ + "detail": "Fresh authorization required", + "code": "step_up_required", + "action": action, + "target": target, + }, + ) + + monkeypatch.setattr(main, "_require_step_up", require_step_up) + payload = { + "expected_revision": gate["revision"], + "decision": "hold", + "reason": "Needs another browser run", + "override_reason": "", + "checklist": CHECKLIST, + } + transport = httpx.ASGITransport(app=main.app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + rejected = await client.post( + f"/api/v1/human-gates/{gate['id']}/decision", + json=payload, + headers={"Idempotency-Key": "decision-authorized"}, + ) + pending = await client.get(f"/api/v1/human-gates/{gate['id']}") + accepted = await client.post( + f"/api/v1/human-gates/{gate['id']}/decision", + json=payload, + headers={ + "Idempotency-Key": "decision-authorized", + "X-Step-Up-Grant": "one-time-grant", + }, + ) + + assert rejected.status_code == 428 + assert pending.json()["state"] == "pending" + assert accepted.status_code == 201 + assert authorization_calls == [ + {"grant": None, "action": "decide_human_gate", "target": gate["id"]}, + {"grant": "one-time-grant", "action": "decide_human_gate", "target": gate["id"]}, + ] + + +@pytest.mark.anyio +async def test_successful_decision_records_one_completed_privacy_safe_security_event( + monkeypatch, gate_api, tmp_path +): + gate = gate_api.intake("1:timmy", CANDIDATE, idempotency_key="run-audited") + journal_path = tmp_path / "security.sqlite3" + journal = SecurityEventStore(journal_path, clock=lambda: 200) + monkeypatch.setattr(main, "_security_event_store", lambda: journal) + private_reason = "Private launch context must not enter the security journal" + payload = { + "expected_revision": gate["revision"], + "decision": "hold", + "reason": private_reason, + "override_reason": "", + "checklist": CHECKLIST, + } + transport = httpx.ASGITransport(app=main.app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + decided = await client.post( + f"/api/v1/human-gates/{gate['id']}/decision", + json=payload, + headers={"Idempotency-Key": "decision-audited"}, + ) + + events = journal.list().events + assert decided.status_code == 201 + assert [ + {"kind": event.kind, "method": event.method, "target": event.target, "status": event.status} + for event in events + ] == [ + { + "kind": "human_gate_decision", + "method": "hold", + "target": gate["id"], + "status": "completed", + } + ] + assert private_reason.encode() not in journal_path.read_bytes() + + +@pytest.mark.anyio +async def test_decision_fails_closed_when_security_event_cannot_be_reserved( + monkeypatch, gate_api +): + gate = gate_api.intake("1:timmy", CANDIDATE, idempotency_key="run-no-journal") + + class UnavailableJournal: + def reserve(self, *args, **kwargs): + raise SecurityEventStoreError("private journal path") + + monkeypatch.setattr(main, "_security_event_store", UnavailableJournal) + payload = { + "expected_revision": gate["revision"], + "decision": "release", + "reason": "", + "override_reason": "", + "checklist": CHECKLIST, + } + transport = httpx.ASGITransport(app=main.app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + rejected = await client.post( + f"/api/v1/human-gates/{gate['id']}/decision", + json=payload, + headers={"Idempotency-Key": "decision-no-journal"}, + ) + pending = await client.get(f"/api/v1/human-gates/{gate['id']}") + + assert rejected.status_code == 503 + assert rejected.json() == {"detail": "Security activity is temporarily unavailable"} + assert pending.json()["state"] == "pending" + + +@pytest.mark.anyio +async def test_rejected_decision_discards_its_pending_security_event( + monkeypatch, gate_api, tmp_path +): + gate = gate_api.intake("1:timmy", CANDIDATE, idempotency_key="run-conflict") + journal = SecurityEventStore(tmp_path / "security.sqlite3", clock=lambda: 200) + monkeypatch.setattr(main, "_security_event_store", lambda: journal) + payload = { + "expected_revision": gate["revision"] + 1, + "decision": "hold", + "reason": "Outdated review", + "override_reason": "", + "checklist": CHECKLIST, + } + transport = httpx.ASGITransport(app=main.app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + rejected = await client.post( + f"/api/v1/human-gates/{gate['id']}/decision", + json=payload, + headers={"Idempotency-Key": "decision-conflict"}, + ) + + assert rejected.status_code == 409 + assert journal.list().events == [] + + @pytest.mark.anyio async def test_recycled_login_cannot_read_another_principal_gates(monkeypatch, tmp_path): store = HumanGateStore(tmp_path / "gates.sqlite3", clock=lambda: 100) -- 2.43.0