diff --git a/dockerfiles/hermes-hux-foundation/hux/budgets.py b/dockerfiles/hermes-hux-foundation/hux/budgets.py index 7858768a..31e8dabc 100644 --- a/dockerfiles/hermes-hux-foundation/hux/budgets.py +++ b/dockerfiles/hermes-hux-foundation/hux/budgets.py @@ -179,6 +179,10 @@ def gate(request: Request) -> Response: raise Invalid("argument_hash must be sha256:") external = bool(body.get("external", False)) conversation_id = run_conversation(request.store, run_id) + if policy.private_mode_denies(request.store, conversation_id, capability): + request.audit("gate.check", f"{run_id}:{capability}", outcome="deny", reason="private_mode") + policy.emit(request.store, request.identity, conversation_id, "side_effect.blocked", f"{capability} blocked: private mode", run_id=run_id) + return Response(200, {"proceed": False, "reason": "private_mode"}) state = budget_state(request.store, request.identity, run_id, conversation_id) if state["exhausted"]: # F6: an exhausted run releases nothing, whatever was approved request.audit("gate.check", f"{run_id}:{capability}", outcome="deny", reason="budget_exhausted") diff --git a/dockerfiles/hermes-hux-foundation/hux/policy.py b/dockerfiles/hermes-hux-foundation/hux/policy.py index eedeb5ff..156d16c9 100644 --- a/dockerfiles/hermes-hux-foundation/hux/policy.py +++ b/dockerfiles/hermes-hux-foundation/hux/policy.py @@ -290,6 +290,25 @@ def resolve_request(policy: dict[str, Any], capability: str, external: bool) -> return decision +PRIVATE_DENIED = frozenset({"network", "web_search", "send_message", "shell", "delegate"}) + + +def private_mode_denies(store: TenantStore, conversation_id: str | None, capability: str) -> bool: + """Private conversations never release web, shell, messaging or delegation. + + The mode catalog says so (``rules.MODE_CATALOG['private']['tools']``); + memory writes are already refused by the privacy state. The conversation + record's ``mode`` field is written only by the HUX-06 selection route. + """ + if capability not in PRIVATE_DENIED or not conversation_id: + return False + try: + record = store.get("conversations", conversation_id) + except Exception: + return False + return record.get("mode") == "private" + + def create_approval(request: Request) -> Response: """``POST /hux/v1/approvals``: the agent hook asks before a gated action.""" require_worker(request.identity, "approval requests") @@ -322,6 +341,8 @@ def create_approval(request: Request) -> Response: raise BudgetExhausted("run budget exhausted", state["exhausted"]) policy = effective_policy(request.store, request.identity, "conversation", conversation_id) decision = resolve_request(policy, capability, external) + if private_mode_denies(request.store, conversation_id, capability): + decision = "deny" stamp = now() record: dict[str, Any] = { "schema": "hux.approval.v1", "id": new_id("apr"), "run_id": run_id, "conversation_id": conversation_id, diff --git a/testing/tests/test_hermes_hux_policy_private_mode.py b/testing/tests/test_hermes_hux_policy_private_mode.py new file mode 100644 index 00000000..2c587161 --- /dev/null +++ b/testing/tests/test_hermes_hux_policy_private_mode.py @@ -0,0 +1,128 @@ +"""Private mode denies web, messaging, shell and delegation upstream. + +HUX-06/HUX-10: selecting the private friendly mode must change what the +agent may actually do — not merely relabel the conversation. The approval +resolver and the pre-side-effect gate both refuse the denied capability +set for a conversation whose stored mode is ``private``, whatever the +autonomy policy says. +""" + +from __future__ import annotations + +import json +import sys +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)) + +from hux import contracts, policy # noqa: E402 +from hux.server import build_router # noqa: E402 + +HEADERS = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef", "X-Hux-Surface": "chat", "X-Hux-Relay-Key": "rk"} +WORKER = {**HEADERS, "X-Hux-Surface": "worker", "X-Hux-Trust": "worker", "X-Hux-Relay-Key": "wk"} +ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"]) +HASH = "sha256:" + "ab" * 32 + + +def call(router, method, path, body=None, headers=None): + raw = b"" if body is None else json.dumps(body).encode() + response = router.dispatch(method, path, {**HEADERS, **(headers or {})}, raw) + return response.status, response.body + + +@pytest.fixture +def scoped(tmp_path): + router = build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_ROUTER_KEY": "rk", "HUX_WORKER_KEY": "wk"}) + _, project = call(router, "POST", "/hux/v1/projects", {"name": "P"}) + _, conversation = call(router, "POST", "/hux/v1/conversations", {"title": "C", "project_id": project["id"]}) + status, _ = call( + router, "POST", "/hux/v1/runs/run_9f/budget", + {"conversation_id": conversation["id"]}, WORKER, + ) + assert status == 200 + return router, project["id"], conversation["id"] + + +def _select_mode(router, project_id, conversation_id, mode): + status, body = call( + router, "PUT", + f"/hux/v1/projects/{project_id}/conversations/{conversation_id}/mode", + {"project_id": project_id, "mode": mode}, + {"If-Match": "0", "Idempotency-Key": f"mode-{mode}-0001"}, + ) + assert status == 200, body + + +def _approval(router, conversation_id, capability, external=False): + body = { + "run_id": "run_9f", "conversation_id": conversation_id, "capability": capability, + "request": {"summary": f"do {capability}", "risk": "low", "external": external, + "evidence": [{"kind": "tool_call", "id": "call-7", "hash": HASH}]}, + } + return call(router, "POST", "/hux/v1/approvals", body, WORKER) + + +def _gate(router, capability, external=False): + return call( + router, "POST", "/hux/v1/runs/run_9f/gate", + {"capability": capability, "argument_hash": HASH, "external": external, + "conversation_id": "ignored"}, + WORKER, + ) + + +def test_private_mode_denies_the_catalog_denied_capabilities(scoped): + router, project_id, conversation_id = scoped + status, _ = call(router, "PUT", "/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": "autonomous"}) + assert status == 200 + _select_mode(router, project_id, conversation_id, "private") + for capability in sorted(policy.PRIVATE_DENIED): + status, body = _approval(router, conversation_id, capability) + assert status == 201, body + assert body["status"] == "denied", (capability, body) + assert body["decision"]["by"] == {"type": "system", "id": "policy"} + # Reading files and writing artifacts stay governed by the normal matrix. + status, body = _approval(router, conversation_id, "read_files") + assert status == 201 and body["status"] == "approved" + + +def test_gate_refuses_private_mode_even_with_an_earlier_approval(scoped): + router, project_id, conversation_id = scoped + status, _ = call(router, "PUT", "/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": "autonomous"}) + assert status == 200 + status, approved = _approval(router, conversation_id, "network") + assert status == 201 and approved["status"] == "approved" + _select_mode(router, project_id, conversation_id, "private") + status, verdict = _gate(router, "network") + assert status == 200 + assert verdict == {"proceed": False, "reason": "private_mode"} + + +def test_non_private_modes_do_not_touch_the_matrix(scoped): + router, project_id, conversation_id = scoped + status, _ = call(router, "PUT", "/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": "autonomous"}) + assert status == 200 + _select_mode(router, project_id, conversation_id, "fast") + status, body = _approval(router, conversation_id, "network") + assert status == 201 and body["status"] == "approved" + status, verdict = _gate(router, "network") + assert status == 200 and verdict["proceed"] is True + + +def test_unbound_or_missing_conversations_fail_open_to_the_matrix(tmp_path): + router = build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_ROUTER_KEY": "rk", "HUX_WORKER_KEY": "wk"}) + status, _ = call( + router, "POST", "/hux/v1/runs/run_9f/budget", + {"conversation_id": "conv_0001abcd"}, WORKER, + ) + assert status == 200 + status, _ = call(router, "PUT", "/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": "autonomous"}) + assert status == 200 + # No stored conversation record: the private check never blocks. + status, body = _approval(router, "conv_0001abcd", "network") + assert status == 201 and body["status"] == "approved"