313 lines
13 KiB
Python
313 lines
13 KiB
Python
import sqlite3
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src import main
|
|
from src.human_gate_store import HumanGateStore
|
|
from src.security_event_store import SecurityEventStore, SecurityEventStoreError
|
|
|
|
|
|
CANDIDATE = {
|
|
"source": "release-bot", "project": "stackchain/dashboard", "candidate_hash": "abc123",
|
|
"title": "Release candidate", "priority": 7,
|
|
"artifacts": [{"name": "manifest", "url": "https://forge.example/manifest"}],
|
|
"links": [{"label": "pull", "url": "https://forge.example/pulls/1"}],
|
|
"checks": [{"name": "unit", "state": "success", "required": True}],
|
|
"score": {"value": 98, "provenance": "eval/v1"},
|
|
"provenance": {"run": "9"},
|
|
}
|
|
CHECKLIST = {"exact_hash": True, "artifacts_reviewed": True, "provenance_reviewed": True}
|
|
|
|
|
|
@pytest.fixture
|
|
def gate_api(monkeypatch, tmp_path):
|
|
ticks = iter(range(100, 120))
|
|
store = HumanGateStore(tmp_path / "gates.sqlite3", clock=lambda: next(ticks))
|
|
monkeypatch.setattr(main, "_human_gate_store", lambda: store, raising=False)
|
|
|
|
async def identity():
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
monkeypatch.setattr(main, "current_user", identity)
|
|
return store
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_intake_list_and_detail_are_account_bound_and_no_store(gate_api):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
created = await client.post("/api/v1/human-gates/intake", json=CANDIDATE, headers={"Idempotency-Key": "run-9"})
|
|
repeated = await client.post("/api/v1/human-gates/intake", json=CANDIDATE, headers={"Idempotency-Key": "run-9"})
|
|
listing = await client.get("/api/v1/human-gates")
|
|
detail = await client.get(f"/api/v1/human-gates/{created.json()['id']}")
|
|
|
|
assert created.status_code == 201
|
|
assert repeated.status_code == 200
|
|
assert repeated.json()["id"] == created.json()["id"]
|
|
assert listing.json()["pending_count"] == 1
|
|
assert detail.json()["candidate_hash"] == "abc123"
|
|
assert detail.json()["history"][0]["action"] == "intake"
|
|
assert all(response.headers["cache-control"] == "no-store" for response in (created, repeated, listing, detail))
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_decision_requires_revision_and_returns_durable_receipt(gate_api):
|
|
gate = gate_api.intake("1:timmy", CANDIDATE, idempotency_key="run-9")
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
payload = {"expected_revision": gate["revision"], "decision": "release", "reason": "", "override_reason": "", "checklist": CHECKLIST}
|
|
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-9"})
|
|
repeated = await client.post(f"/api/v1/human-gates/{gate['id']}/decision", json=payload, headers={"Idempotency-Key": "decision-9"})
|
|
receipt = await client.get(f"/api/v1/human-gate-receipts/{decided.json()['receipt_id']}")
|
|
stale = await client.post(f"/api/v1/human-gates/{gate['id']}/decision", json=payload, headers={"Idempotency-Key": "decision-10"})
|
|
|
|
assert decided.status_code == 201
|
|
assert repeated.status_code == 200
|
|
assert receipt.json() == decided.json()
|
|
assert stale.status_code == 409
|
|
assert all(response.headers["cache-control"] == "no-store" for response in (decided, repeated, receipt, stale))
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_decision_history_is_newest_first_receipt_linked_and_principal_bound(gate_api):
|
|
first = gate_api.intake("1:timmy", CANDIDATE, idempotency_key="history-first")
|
|
gate_api.decide(
|
|
"1:timmy", first["id"], expected_revision=1, decision="release", reason="",
|
|
override_reason="", checklist=CHECKLIST, idempotency_key="history-release",
|
|
)
|
|
second_candidate = {**CANDIDATE, "candidate_hash": "def456", "title": "New candidate"}
|
|
second = gate_api.intake("1:timmy", second_candidate, idempotency_key="history-second")
|
|
second_receipt = gate_api.decide(
|
|
"1:timmy", second["id"], expected_revision=1, decision="hold",
|
|
reason="Needs another mobile pass", override_reason="", checklist={},
|
|
idempotency_key="history-hold",
|
|
)
|
|
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
listing = await client.get("/api/v1/human-gates?state=all")
|
|
receipt = await client.get(
|
|
f"/api/v1/human-gate-receipts/{second_receipt['receipt_id']}"
|
|
)
|
|
|
|
assert listing.status_code == 200
|
|
assert listing.headers["cache-control"] == "no-store"
|
|
assert [item["state"] for item in listing.json()["items"]] == ["held", "released"]
|
|
assert listing.json()["items"][0]["receipt_id"] == second_receipt["receipt_id"]
|
|
assert receipt.json()["reason"] == "Needs another mobile pass"
|
|
assert gate_api.list("2:timmy", state="all")["items"] == []
|
|
with pytest.raises(LookupError):
|
|
gate_api.receipt("2:timmy", second_receipt["receipt_id"])
|
|
|
|
|
|
@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)
|
|
monkeypatch.setattr(main, "_human_gate_store", lambda: store, raising=False)
|
|
principal = {"id": 1, "login": "timmy"}
|
|
|
|
async def identity():
|
|
return principal.copy()
|
|
|
|
monkeypatch.setattr(main, "current_user", identity)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
created = await client.post(
|
|
"/api/v1/human-gates/intake", json=CANDIDATE,
|
|
headers={"Idempotency-Key": "principal-1"},
|
|
)
|
|
principal["id"] = 2
|
|
listing = await client.get("/api/v1/human-gates")
|
|
|
|
assert created.status_code == 201
|
|
assert listing.json()["pending_count"] == 0
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gate_store_failure_is_sanitized_no_store(monkeypatch):
|
|
def unavailable():
|
|
raise sqlite3.OperationalError("sensitive database path")
|
|
|
|
monkeypatch.setattr(main, "_human_gate_store", unavailable)
|
|
|
|
async def identity():
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
monkeypatch.setattr(main, "current_user", identity)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/api/v1/human-gates")
|
|
|
|
assert response.status_code == 503
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.json() == {"detail": "Human Gates are temporarily unavailable"}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_gate_mutations_require_idempotency_key_and_validate_override(gate_api):
|
|
failing = {**CANDIDATE, "candidate_hash": "fail123", "checks": [{"name": "browser", "state": "failure", "required": True}]}
|
|
gate = gate_api.intake("1:timmy", failing, idempotency_key="run-fail")
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
|
missing_key = await client.post("/api/v1/human-gates/intake", json=CANDIDATE)
|
|
no_override = await client.post(
|
|
f"/api/v1/human-gates/{gate['id']}/decision",
|
|
json={"expected_revision": 1, "decision": "release", "reason": "", "override_reason": "", "checklist": CHECKLIST},
|
|
headers={"Idempotency-Key": "decision-fail"},
|
|
)
|
|
|
|
assert missing_key.status_code == 422
|
|
assert no_override.status_code == 422
|
|
assert "override reason" in no_override.json()["detail"]
|