"""HUX-05 approval queue and the pre-side-effect gate. Security obligations exercised: SO-35 (only a human surface with router or relay trust decides; ``decision.by`` is the asserted user), SO-36 (``once`` is consumed exactly once), SO-37 (the gate must present the argument hash recorded at request time), SO-39 (external side effects always need an approval record), SO-40 (an exhausted budget blocks new approvals), SO-42 (pending approvals expire 24 h after ``requested_at``). """ from __future__ import annotations import importlib import json import sys from datetime import datetime, timedelta, timezone from pathlib import Path import pytest ROOT = Path(__file__).resolve().parents[2] FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation" if str(FOUNDATION) not in sys.path: sys.path.insert(0, str(FOUNDATION)) import hux # noqa: E402 from hux import audit, budgets, contracts, identity, policy, store # noqa: E402 from hux import events as hux_events # noqa: E402 from hux.server import build_router # noqa: E402 SCHEMAS = contracts.load_all() HEADERS = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef", "X-Hux-Surface": "chat"} OTHER = {**HEADERS, "X-Hux-Subject": "usr_fedcba9876543210"} WORKER = {**HEADERS, "X-Hux-Surface": "worker", "X-Hux-Trust": "worker", "X-Hux-Relay-Key": "wk"} RELAY = {**HEADERS, "X-Hux-Surface": "telegram", "X-Hux-Trust": "relay", "X-Hux-Relay-Key": "rk"} ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"]) T0 = datetime(2026, 8, 24, 12, 0, tzinfo=timezone.utc) CONV = "conv_0001abcd" HASH = "sha256:" + "ab" * 32 OTHER_HASH = "sha256:" + "cd" * 32 @pytest.fixture def router(tmp_path): return build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_WORKER_KEY": "wk", "HUX_RELAY_KEY": "rk"}) @pytest.fixture def frozen(monkeypatch): state = {"now": T0} monkeypatch.setattr(policy, "clock", lambda: state["now"]) return state @pytest.fixture def events(monkeypatch): """Stand-in for the events lane; records every emit call.""" seen: list[dict] = [] monkeypatch.setattr(hux_events, "emit", lambda store, identity, conversation_id, kind, summary, **extra: seen.append({"conversation_id": conversation_id, "kind": kind, "summary": summary, **extra})) return seen def call(router, method, path, body=None, headers=HEADERS): raw = b"" if body is None else json.dumps(body).encode() response = router.dispatch(method, path, headers, raw) return response.status, response.body def valid(record): problems = contracts.validate_record(record, SCHEMAS) assert problems == [], problems return record def request_body(capability="write_files", external=False, run_id="run_9f", **extra): evidence = [{"kind": "tool_call", "id": "call-7", "hash": HASH}] return {"run_id": run_id, "conversation_id": CONV, "capability": capability, "request": {"summary": f"do {capability}", "risk": "low", "external": external, "evidence": evidence, **extra}} def set_autonomy(router, level): status, body = call(router, "PUT", "/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": level}) assert status == 200, body # --- creation resolves against the matrix ------------------------------------------ def test_safe_policy_auto_allows_reads_and_queues_mutations(router, frozen, events): status, body = call(router, "POST", "/hux/v1/approvals", request_body("read_files"), WORKER) assert status == 201 and valid(body)["status"] == "approved" assert body["decision"] == {"choice": "once", "by": {"type": "system", "id": "policy"}, "at": policy.iso(T0)} status, body = call(router, "POST", "/hux/v1/approvals", request_body("write_files"), WORKER) assert status == 201 and valid(body)["status"] == "pending" and "decision" not in body assert body["expires_at"] == policy.iso(T0 + timedelta(hours=24)) status, body = call(router, "POST", "/hux/v1/approvals", request_body("network"), WORKER) assert status == 201 and valid(body)["status"] == "denied" and body["decision"]["choice"] == "deny" assert [e["kind"] for e in events] == ["approval.resolved", "approval.requested", "approval.resolved"] assert events[0]["evidence"] == [{"kind": "approval", "id": events[0]["evidence"][0]["id"]}] def test_external_side_effects_always_need_a_human_regardless_of_autonomy(router, events): set_autonomy(router, "autonomous") status, body = call(router, "POST", "/hux/v1/approvals", request_body("send_message"), WORKER) assert status == 201 and body["status"] == "approved" status, body = call(router, "POST", "/hux/v1/approvals", request_body("send_message", external=True), WORKER) assert status == 201 and body["status"] == "pending" for always_ask in ("deploy", "external_side_effect"): status, body = call(router, "POST", "/hux/v1/approvals", request_body(always_ask), WORKER) assert body["status"] == "pending", always_ask call(router, "PUT", "/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": "autonomous", "grants": [{"capability": "network", "decision": "deny"}]}) status, body = call(router, "POST", "/hux/v1/approvals", request_body("network", external=True), WORKER) assert body["status"] == "denied", "deny beats external ask" def test_idempotency_key_returns_the_original(router): key = {"Idempotency-Key": "run_9f:approval:call-7"} status, first = call(router, "POST", "/hux/v1/approvals", request_body(), {**WORKER, **key}) assert status == 201 and first["idempotency_key"] == "run_9f:approval:call-7" status, again = call(router, "POST", "/hux/v1/approvals", request_body("shell"), {**WORKER, **key}) assert (status, again) == (200, first) status, other = call(router, "POST", "/hux/v1/approvals", request_body(), {**WORKER, "Idempotency-Key": "run_9f:approval:call-8"}) assert status == 201 and other["id"] != first["id"] assert call(router, "GET", f"/hux/v1/approvals/{first['id']}")[1] == first @pytest.mark.parametrize("body,status", [ (None, 400), ([], 400), ({"conversation_id": "nope", "capability": "shell"}, 400), ({"conversation_id": CONV, "capability": "teleport"}, 400), ({"conversation_id": CONV, "capability": "shell", "run_id": "r"}, 400), ({"conversation_id": CONV, "capability": "shell", "run_id": "r", "request": {"summary": "", "risk": "low"}}, 400), ({"conversation_id": CONV, "capability": "shell", "run_id": "r", "request": {"summary": "x", "risk": "silly"}}, 400), ]) def test_create_rejects_bad_bodies(router, body, status): got, error = call(router, "POST", "/hux/v1/approvals", body, WORKER) assert got == status and valid(error)["code"] == "invalid" # --- deciding ----------------------------------------------------------------------- def pending(router, capability="write_files", **kw): status, body = call(router, "POST", "/hux/v1/approvals", request_body(capability, **kw), WORKER) assert status == 201 and body["status"] == "pending", body return body def test_only_humans_on_router_or_relay_decide(router, tmp_path): record = pending(router) status, error = call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "once"}, WORKER) assert (status, error["code"]) == (403, "forbidden") api = {**HEADERS, "X-Hux-Surface": "api"} assert call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "once"}, api)[0] == 403 status, body = call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "once"}, RELAY) assert status == 200 and valid(body)["status"] == "approved" assert body["decision"]["by"] == {"type": "user", "id": HEADERS["X-Hux-Subject"]} rows = audit.recent(store.TenantStore(tmp_path, identity.resolve(HEADERS, {}))) assert [(r["action"], r["outcome"]) for r in rows if r["action"] == "approvals.decide"] == [("approvals.decide", "deny")] * 2 + [("approvals.decide", "allow")] def test_terminal_states_never_change_again(router, events): record = pending(router) path = f"/hux/v1/approvals/{record['id']}" assert call(router, "POST", path, {"choice": "maybe"})[0] == 400 assert call(router, "POST", path, [])[0] == 400 status, body = call(router, "POST", path, {"choice": "deny"}) assert status == 200 and body["status"] == "denied" and body["decision"]["choice"] == "deny" status, error = call(router, "POST", path, {"choice": "once"}) assert (status, error["code"]) == (409, "conflict") assert call(router, "GET", path)[1] == body assert events[-1]["kind"] == "approval.resolved" and "denied" in events[-1]["summary"] def test_pending_approvals_expire_after_24h(router, frozen): record = pending(router) frozen["now"] = T0 + timedelta(hours=24) status, body = call(router, "GET", f"/hux/v1/approvals/{record['id']}") assert status == 200 and valid(body)["status"] == "expired" assert call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "once"})[0] == 409 assert call(router, "GET", "/hux/v1/approvals?status=expired")[1]["items"] == [body] assert call(router, "GET", "/hux/v1/approvals?status=pending")[1]["items"] == [] def test_session_and_always_create_grants_at_their_scope(router, frozen): record = pending(router, "shell") status, body = call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "session"}) assert status == 200 and body["decision"]["choice"] == "session" conv_policy = call(router, "GET", f"/hux/v1/policy?scope=conversation&scope_id={CONV}")[1] assert valid(conv_policy)["scope"] == {"level": "conversation", "scope_id": CONV} assert conv_policy["grants"] == [{"capability": "shell", "decision": "allow", "expires_at": policy.iso(T0 + policy.SESSION_TTL), "granted_by": {"type": "user", "id": HEADERS["X-Hux-Subject"]}}] status, body = call(router, "POST", "/hux/v1/approvals", request_body("shell"), WORKER) assert body["status"] == "approved", "later shell requests in this conversation auto-approve" assert call(router, "POST", "/hux/v1/approvals", request_body("write_files"), WORKER)[1]["status"] == "pending" record = pending(router, "write_files") call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "always"}) global_policy = call(router, "GET", "/hux/v1/policy")[1] assert [g["capability"] for g in global_policy["grants"]] == ["write_files"] assert global_policy["grants"][0]["expires_at"] == policy.iso(T0 + policy.ALWAYS_TTL) frozen["now"] = T0 + timedelta(hours=25) record = pending(router, "shell", run_id="run_other") call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "session"}) conv_policy = call(router, "GET", f"/hux/v1/policy?scope=conversation&scope_id={CONV}")[1] assert len(conv_policy["grants"]) == 1 and conv_policy["revision"] == 2, "a repeat grant replaces, never duplicates" def test_list_filters_and_orders_by_request_time(router, frozen): first = pending(router, "shell") frozen["now"] = T0 + timedelta(seconds=5) second = pending(router, "write_files") status, body = call(router, "GET", "/hux/v1/approvals") assert status == 200 and [i["id"] for i in body["items"]] == [first["id"], second["id"]] and body["next"] is None assert call(router, "GET", "/hux/v1/approvals?status=pending")[1]["items"] == body["items"] assert call(router, "GET", "/hux/v1/approvals?status=approved")[1]["items"] == [] assert call(router, "GET", "/hux/v1/approvals?status=bogus")[0] == 400 for item in body["items"]: valid(item) def test_second_subject_cannot_see_or_decide(router): record = pending(router) assert call(router, "GET", f"/hux/v1/approvals/{record['id']}", headers=OTHER)[0] == 404 assert call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "once"}, OTHER)[0] == 404 assert call(router, "GET", "/hux/v1/approvals", headers=OTHER)[1]["items"] == [] assert call(router, "GET", "/hux/v1/approvals/apr_doesnotexist")[0] == 404 assert call(router, "GET", "/hux/v1/approvals/..")[0] == 400 # --- budgets block new approvals ------------------------------------------------------- def test_exhausted_budget_blocks_new_approvals(router, events): call(router, "PUT", "/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": "safe", "budgets": {"tool_calls_per_run": 2}}) status, body = call(router, "POST", "/hux/v1/runs/run_9f/budget", {"conversation_id": CONV, "tool_calls": 2}, WORKER) assert status == 200 and valid(body)["exhausted"] == ["tool_calls_per_run"] status, error = call(router, "POST", "/hux/v1/approvals", request_body("read_files"), WORKER) assert (status, error["code"]) == (429, "budget_exhausted") and error["details"] == ["tool_calls_per_run"] assert [e["kind"] for e in events] == ["budget.exhausted", "budget.exhausted"] status, body = call(router, "POST", "/hux/v1/approvals", request_body("read_files", run_id="run_fresh"), WORKER) assert status == 201, "another run keeps its own budget" # --- gate -------------------------------------------------------------------------- def gate(router, run_id="run_9f", capability="write_files", argument_hash=HASH, external=False, **extra): return call(router, "POST", f"/hux/v1/runs/{run_id}/gate", {"capability": capability, "argument_hash": argument_hash, "external": external, **extra}, WORKER) def test_gate_blocks_before_approval_and_releases_once_exactly_once(router, events, tmp_path): status, body = gate(router) assert status == 200 and body == {"proceed": False, "reason": "no approval for this run and capability"} record = pending(router) assert gate(router)[1]["proceed"] is False and "pending" in gate(router)[1]["reason"] call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "once"}) status, wrong = gate(router, argument_hash=OTHER_HASH) assert wrong["proceed"] is False and "different arguments" in wrong["reason"], "TOCTOU (SO-37)" status, body = gate(router) assert body == {"proceed": True, "approval_id": record["id"], "reason": "released"} status, again = gate(router) assert again["proceed"] is False and "already consumed" in again["reason"], "once is once (SO-36)" kinds = [e["kind"] for e in events] assert kinds.count("side_effect.released") == 1 and kinds.count("side_effect.blocked") == 4, "the first block predates any approval, so no conversation is known" released = next(e for e in events if e["kind"] == "side_effect.released") assert released["evidence"] == [{"kind": "approval", "id": record["id"]}] and released["run_id"] == "run_9f" served = call(router, "GET", f"/hux/v1/approvals/{record['id']}")[1] assert valid(served) and "_consumed_at" not in served rows = audit.recent(store.TenantStore(tmp_path, identity.resolve(HEADERS, {}))) assert [r["outcome"] for r in rows if r["action"] == "gate.check"] == ["deny", "deny", "deny", "deny", "allow", "deny"] def test_gate_session_is_reusable_within_the_conversation(router): record = pending(router, "shell") call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "session"}) assert gate(router, capability="shell")[1]["proceed"] is True assert gate(router, capability="shell", argument_hash=OTHER_HASH)[1]["proceed"] is True assert gate(router, run_id="run_later", capability="shell", conversation_id=CONV)[1]["proceed"] is True assert gate(router, run_id="run_later", capability="shell")[1]["proceed"] is False assert gate(router, run_id="run_later", capability="shell", conversation_id="conv_elsewhere01")[1]["proceed"] is False assert gate(router, capability="write_files")[1]["proceed"] is False def test_gate_requires_external_approvals_for_external_effects_and_respects_expiry(router, frozen): record = pending(router, "send_message") call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "once"}) status, body = gate(router, capability="send_message", external=True) assert body["proceed"] is False and "not requested as external" in body["reason"] record = pending(router, "send_message", external=True) call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "once"}) frozen["now"] = T0 + timedelta(hours=25) status, body = gate(router, capability="send_message", external=True) assert body["proceed"] is False and "expired" in body["reason"] def test_gate_rejects_bad_bodies_and_other_tenants(router): assert gate(router, capability="teleport")[0] == 400 assert gate(router, argument_hash="md5:zz")[0] == 400 assert gate(router, conversation_id="nope")[0] == 400 assert call(router, "POST", "/hux/v1/runs/run_9f/gate", [], WORKER)[0] == 400 assert call(router, "POST", f"/hux/v1/runs/{'r' * 121}/gate", {"capability": "shell", "argument_hash": HASH}, WORKER)[0] == 400 record = pending(router) call(router, "POST", f"/hux/v1/approvals/{record['id']}", {"choice": "once"}) status, body = call(router, "POST", "/hux/v1/runs/run_9f/gate", {"capability": "write_files", "argument_hash": HASH}, OTHER) assert status == 200 and body["proceed"] is False def test_events_module_absence_is_tolerated(router, monkeypatch): monkeypatch.setitem(sys.modules, "hux.events", None) monkeypatch.delattr(hux, "events") with pytest.raises(ImportError): importlib.import_module("hux.events") assert call(router, "POST", "/hux/v1/approvals", request_body("read_files"), WORKER)[0] == 201 policy.emit(None, None, None, "x", "no conversation means no event") assert budgets.hashes_of({"request": {}}) == set()