"""Application CI: exercise HTTP clients and assert protected filesystem effects. Build the Rust binaries first. This runner owns temporary state and subprocesses; it is optional verification tooling, not a prerequisite for using the apps. """ from contextlib import contextmanager import concurrent.futures import json import os from pathlib import Path import subprocess import tempfile import time import urllib.error import urllib.request BIN = Path(os.environ.get("CHIO_APP_BIN_DIR", "target/debug")).resolve() @contextmanager def host(binary, port, state=None, **extra): with tempfile.TemporaryDirectory(prefix="chio-application-") as temporary: directory = Path(state or temporary) with (directory / "host.log").open("a") as log: process = subprocess.Popen([str(BIN / binary)], env={**os.environ, "CHIO_APP_STATE": str(directory), **extra}, stdout=log, stderr=log) try: for _ in range(200): if process.poll() is not None: raise AssertionError((directory / "host.log").read_text()) try: urllib.request.urlopen(f"http://127.0.0.1:{port}/", timeout=.2).close() break except (OSError, urllib.error.URLError): time.sleep(.1) else: raise AssertionError("Application did not start") yield directory finally: process.terminate() try: process.wait(timeout=20) except subprocess.TimeoutExpired: process.kill() process.wait() raise AssertionError("host required forced shutdown") assert process.returncode == 0, (process.returncode, (directory / "host.log").read_text()) def api(port, token, path, body=None, expected=200): request = urllib.request.Request(f"http://127.0.0.1:{port}{path}", data=None if body is None else json.dumps(body).encode(), headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"}) try: response = urllib.request.urlopen(request, timeout=15) except urllib.error.HTTPError as error: response = error with response: content = response.read().decode() assert response.status == expected, (path, response.status, content) if path == "/export": return [json.loads(line) for line in content.splitlines()] try: return json.loads(content) except json.JSONDecodeError: assert expected >= 400, content return {"error": content} def publishing(): with host("publishing-desk", 4310, APPROVAL_TTL_SECONDS="2") as state: agent = (state / "agent-token.txt").read_text() reviewer = (state / "reviewer-token.txt").read_text() def proposal(text): return api(4310, agent, "/proposals", {"title": "CI publication", "text": text})["approval_id"] def decision(id, choice, **changes): preview = api(4310, reviewer, f"/proposals/{id}") return {"decision": choice, "parameter_hash": preview["pending"]["parameter_hash"], **changes} first = proposal("This exact text is approved.") assert not list((state / "published").iterdir()) body = decision(first, "approve") api(4310, agent, f"/proposals/{first}/decision", body, 401) api(4310, reviewer, f"/proposals/{first}/decision", {**body, "parameter_hash": "changed"}, 409) api(4310, reviewer, f"/proposals/{first}/decision", {**body, "arguments": {"text": "changed"}}, 422) allowed = api(4310, reviewer, f"/proposals/{first}/decision", body) assert allowed["verdict"] == "Allow", allowed assert json.loads((state / "published" / f"{first}.json").read_text())["text"] == "This exact text is approved." api(4310, reviewer, f"/proposals/{first}/decision", body, 409) denied_id = proposal("This must not be published.") denied = api(4310, reviewer, f"/proposals/{denied_id}/decision", decision(denied_id, "reject")) assert denied["verdict"] == "Deny" and denied["receipt_id"] != allowed["receipt_id"] assert not (state / "published" / f"{denied_id}.json").exists() expired = proposal("This review expires.") expired_decision = decision(expired, "approve") time.sleep(2.1) api(4310, reviewer, f"/proposals/{expired}/decision", expired_decision, 409) assert not (state / "published" / f"{expired}.json").exists() old_key = (state / "kernel-public-key.txt").read_text() revoked = api(4310, reviewer, "/publisher/revoke", {}) revoked_id = proposal("This approval has revoked execution authority.") refused = api(4310, reviewer, f"/proposals/{revoked_id}/decision", decision(revoked_id, "approve")) assert refused["verdict"] == "Deny" and not (state / "published" / f"{revoked_id}.json").exists() rotated = api(4310, reviewer, "/publisher/rotate", {}) assert rotated["old_capability_id"] == revoked["revoked_capability_id"] assert rotated["capability_id"] != rotated["old_capability_id"] and rotated["kernel_key"] == old_key new_id = proposal("This publication uses the new agent identity.") renewed = api(4310, reviewer, f"/proposals/{new_id}/decision", decision(new_id, "approve")) assert renewed["verdict"] == "Allow" and (state / "published" / f"{new_id}.json").exists() print("PASS publishing: exact approved effect; rejection, expiry, mutation, unauthorized reviewer, and replay produce no publication") print("PASS rotation: revoked authority cannot publish; fresh agent authority can; historical kernel verification key is unchanged") def tenants(): with host("customer-service", 4311) as state: tokens = {name: (state / f"{name}-token.txt").read_text() for name in ("alice", "bob", "operator")} alice = api(4311, tokens["alice"], "/notes", {"text": "Alice owns this"}) bob = api(4311, tokens["bob"], "/notes", {"text": "Bob owns this"}) assert alice["verdict"] == bob["verdict"] == "Allow", (alice, bob) assert alice["receipt"]["tenant_id"] == "customer-a" and bob["receipt"]["tenant_id"] == "customer-b" api(4311, tokens["alice"], "/receipts/" + bob["receipt_id"], expected=404) assert api(4311, tokens["bob"], "/receipts/" + bob["receipt_id"])["id"] == bob["receipt_id"] api(4311, tokens["alice"], "/receipts?tenant=customer-b", expected=409) api(4311, tokens["alice"], "/notes", {"text": "spoof", "tenant": "customer-b"}, 422) assert {row["tenant_id"] for row in api(4311, tokens["alice"], "/export")} == {"customer-a"} assert {row["tenant_id"] for row in api(4311, tokens["bob"], "/export")} == {"customer-b"} assert api(4311, tokens["operator"], "/receipts")["total"] == 2 assert api(4311, tokens["alice"], "/notes")["notes"][0]["text"] == "Alice owns this" assert api(4311, tokens["bob"], "/notes")["notes"][0]["text"] == "Bob owns this" print("PASS tenancy: authenticated execution, isolated resources, exact receipt reads, scoped exports, and separate administrative access") def terminal(token, id): for _ in range(160): result = api(4312, token, f"/reports/{id}") if result["result"] is not None: return result time.sleep(.1) raise AssertionError("Report never reached a terminal outcome") def reports(): for mode, delay in [("refuse", None), ("complete", 0), ("complete", 1.4), ("fail", None), ("uncertain", None)]: with host("report-service", 4312) as state: token = (state / "client-token.txt").read_text() api(4312, token, "/reports", {"text": "", "mode": mode}, 400) id = api(4312, token, "/reports", {"text": "A metered report for a customer", "mode": mode})["report_id"] if delay is not None: time.sleep(delay) api(4312, token, f"/reports/{id}/cancel", {}) result = terminal(token, id) if delay == 0: assert result["result"].get("status") == "cancelled_before_dispatch", result assert api(4312, token, "/budget")["available"] == 50 else: assert result["result"]["receipt_id"], result if mode == "refuse" or delay == 0: assert not result["started"] and not result["published"], result elif mode == "uncertain": assert result["published"], result else: assert result["started"] and not result["published"], result print("PASS report", mode, "cancel_delay=", delay, "verdict=", result["result"].get("verdict"), "published=", result["published"]) with tempfile.TemporaryDirectory(prefix="chio-shared-budget-") as persistent: with host("report-service", 4312, persistent) as state: token = (state / "client-token.txt").read_text() with concurrent.futures.ThreadPoolExecutor(max_workers=3) as workers: ids = list(workers.map(lambda _: api(4312, token, "/reports", {"text": "Concurrent report", "mode": "complete"})["report_id"], range(3))) results = [terminal(token, id) for id in ids] assert sum(result["published"] for result in results) == 2, results assert len({result["result"]["receipt_id"] for result in results}) == 3 before = api(4312, token, "/budget") assert before["spent"] == 40 and before["available"] == 10, before with host("report-service", 4312, persistent): after = api(4312, token, "/budget") assert after == before, (before, after) id = api(4312, token, "/reports", {"text": "After restart", "mode": "complete"})["report_id"] assert not terminal(token, id)["started"] print("PASS shared allowance: three concurrent workers, two actual publications, exact receipts, exhaustion, and unchanged balance after restart") def shutdown_active_report(): with tempfile.TemporaryDirectory(prefix="chio-report-shutdown-") as persistent: with host("report-service", 4312, persistent) as state: token = (state / "client-token.txt").read_text() id = api(4312, token, "/reports", {"text": "Stop this owned report", "mode": "complete"})["report_id"] for _ in range(100): if api(4312, token, f"/reports/{id}")["started"]: break time.sleep(.05) else: raise AssertionError("report never entered execution") # Exiting the owned host sends SIGTERM while a worker is active. result = json.loads((Path(persistent) / "jobs" / f"{id}.result.json").read_text()) assert result.get("receipt_id"), result assert not (Path(persistent) / "reports" / f"{id}.md").exists() print("PASS SIGTERM: active report cancelled, final receipt retained, no publication, host joined cleanly") if __name__ == "__main__": publishing() tenants() reports() shutdown_active_report()