diff --git a/scripts/task-plane/create_shadow_tasks.py b/scripts/task-plane/create_shadow_tasks.py index 38d1924..1363589 100644 --- a/scripts/task-plane/create_shadow_tasks.py +++ b/scripts/task-plane/create_shadow_tasks.py @@ -1,166 +1,178 @@ #!/usr/bin/env python3 -"""Create and complete the three authorized no-side-effect Shadow tasks.""" +"""Create one authenticated, no-side-effect Task Plane Shadow task. + +The client deliberately has no worker, Git, deployment, or credential-writing +capabilities. Its token is accepted only from a protected file reference or +stdin so that it cannot enter the process command line. +""" from __future__ import annotations import argparse +import hashlib import json +import os +import sys +import urllib.error +import urllib.request from pathlib import Path - -from app.task_plane.adapters.codex import CodexAdapter -from app.task_plane.adapters.hermes import HermesAdapter -from app.task_plane.adapters.openclaw import OpenClawAdapter -from app.task_plane.api import TaskPlaneService -from app.task_plane.policy import PolicyEngine -from app.task_plane.store import TaskPlaneStore +from typing import Any -TASKS = ( - ( - "AIOS-TASK-PLANE-OPENCLAW-ACCEPTANCE", - "platform.acceptance", - "openclaw", - "spiffe://aios/agent/openclaw/shadow-01", - ["test.execute", "runtime.shadow"], - OpenClawAdapter, - ), - ( - "AIOS-TASK-PLANE-CODEX-REVIEW", - "code.review", - "codex", - "spiffe://aios/agent/codex/shadow-01", - ["code.read", "review.security", "test.python"], - CodexAdapter, - ), - ( - "AIOS-TASK-PLANE-HERMES-ARCHITECTURE", - "architecture.review", - "hermes", - "spiffe://aios/agent/hermes/shadow-01", - ["architecture.review", "policy.review", "threat.model"], - HermesAdapter, - ), +MINIMUM_CLIENT_CAPABILITIES = ( + "task.creator", + "task.read", + "task.events.read", + "task.evidence.read", ) -def main() -> int: - parser = argparse.ArgumentParser() - parser.add_argument("--database", required=True) - parser.add_argument("--report", required=True) - args = parser.parse_args() +class ShadowClientError(RuntimeError): + """HTTP failure that intentionally carries no response body.""" - store = TaskPlaneStore(args.database) - service = TaskPlaneService( - store, policy=PolicyEngine(execution_enabled=True, max_risk="medium") + def __init__(self, status: int): + super().__init__(f"shadow task request failed with HTTP {status}") + self.status = status + + +def _request_json( + base_url: str, + path: str, + token: str, + *, + method: str = "GET", + payload: dict[str, Any] | None = None, + idempotency_key: str | None = None, +) -> tuple[int, dict[str, Any]]: + headers = { + "Accept": "application/json", + "Authorization": f"Bearer {token}", + } + data = None + if payload is not None: + headers["Content-Type"] = "application/json" + data = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + if idempotency_key: + headers["Idempotency-Key"] = idempotency_key + + request = urllib.request.Request( + base_url.rstrip("/") + path, + data=data, + headers=headers, + method=method, ) - union_capabilities = sorted({cap for item in TASKS for cap in item[4]}) - root = service.create_task( - { - "task_id": "AIOS-AGENT-TASK-CONTROL-PLANE-20260724-SHADOW", - "task_type": "platform.shadow-orchestration", - "required_capabilities": union_capabilities, - "risk": "medium", + try: + with urllib.request.urlopen(request, timeout=10) as response: + body = response.read() + parsed = json.loads(body.decode("utf-8")) if body else {} + return response.status, parsed + except urllib.error.HTTPError as error: + # Never read or print an error body: it could contain sensitive data. + raise ShadowClientError(error.code) from None + except urllib.error.URLError as error: + raise ShadowClientError(503) from error + + +def run_shadow_client( + *, + base_url: str, + token: str, + task_id: str, + idempotency_key: str, +) -> dict[str, Any]: + """Create and read one deterministic Shadow task without executing it.""" + create_status, _ = _request_json( + base_url, + "/tasks", + token, + method="POST", + idempotency_key=idempotency_key, + payload={ + "task_id": task_id, + "task_type": "platform.shadow", + "risk": "low", "approval_policy": "patch-only", - "max_delegation_depth": 1, - "budgets": {"attempts": 3}, "metadata": {"production_side_effects": False}, }, - idempotency_key="shadow-orchestration-v1", ) - - results = [] - for task_id, task_type, worker_type, identity, capabilities, adapter_type in TASKS: - service.register_worker( - { - "worker_id": f"{worker_type}-shadow-01", - "worker_type": worker_type, - "identity": identity, - "capabilities": capabilities, - "trust_zone": "isolated-shadow", - "max_concurrency": 1, - "heartbeat_interval_seconds": 15, - "supported_protocols": ["aios-task-v1"], - } - ) - child = service.delegate( - root["task_id"], - { - "task_id": task_id, - "task_type": task_type, - "required_capabilities": capabilities, - "risk": "low", - "approval_policy": "patch-only", - "max_delegation_depth": 0, - "budgets": {"attempts": 1}, - "evidence_requirements": ["shadow-report"], - "metadata": {"production_side_effects": False}, - "actor_identity": "spiffe://aios/owner/shadow-controller", - }, - ) - claim = service.claim( - task_id, - { - "worker_id": f"{worker_type}-shadow-01", - "actor_identity": identity, - "lease_seconds": 60, - }, - ) - service.progress( - task_id, - { - "claim_id": claim["claim_id"], - "actor_identity": identity, - "progress": 100, - "stage": "shadow-verification", - "message": "adapter contract verified without external execution", - }, - ) - adapter_result = adapter_type().execute( - store.get_task(task_id).task_plan, store.get_claim(claim["claim_id"]) - ) - completed = service.complete( - task_id, - { - "claim_id": claim["claim_id"], - "actor_identity": identity, - "status": "COMPLETED", - "outputs": [adapter_result], - "evidence": [ - { - "kind": "shadow-report", - "execution_mode": "shadow_adapter", - "production_side_effects": False, - } - ], - }, - ) - results.append( - { - "task_id": task_id, - "parent_task_id": child["parent_task_id"], - "task_status": store.get_task(task_id).status, - "worker_result": adapter_result, - "result_digest": completed["result_digest"], - } - ) - - report = { - "root_task_id": root["task_id"], - "execution_mode": "shadow_adapter", + task_status, task = _request_json(base_url, f"/tasks/{task_id}", token) + events_status, events = _request_json(base_url, f"/tasks/{task_id}/events", token) + evidence_status, evidence = _request_json(base_url, f"/tasks/{task_id}/evidence", token) + return { + "task_id": str(task.get("task_id", task_id)), + "status": str(task.get("status", "UNKNOWN")), + "event_count": len(events if isinstance(events, list) else events.get("events", [])), + "evidence_count": len(evidence if isinstance(evidence, list) else evidence.get("evidence", [])), + "http_statuses": { + "create": create_status, + "task": task_status, + "events": events_status, + "evidence": evidence_status, + }, "production_side_effects": False, - "tasks": results, - "event_chains_valid": all(store.verify_event_chain(item[0]) for item in TASKS), - "notification_delivery_status": sorted( - {row["delivery_status"] for row in store.list_outbox()} - ), } - report_path = Path(args.report) - report_path.parent.mkdir(parents=True, exist_ok=True) - report_path.write_text( - json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8" - ) - store.close() - print(f"shadow_report={report_path}") + + +def _reject_plaintext_token_argument(argv: list[str]) -> None: + if any(argument == "--token" or argument.startswith("--token=") for argument in argv): + raise SystemExit("plaintext token arguments are forbidden") + + +def _read_token(args: argparse.Namespace) -> str: + if args.token_stdin: + token = sys.stdin.buffer.read().decode("utf-8").strip() + else: + token_file = args.token_file or os.environ.get("AIOS_TASK_PLANE_TOKEN_FILE") + if not token_file: + raise SystemExit("runtime credential reference is required") + try: + token = Path(token_file).read_text(encoding="utf-8").strip() + except OSError: + raise SystemExit("runtime credential reference is unavailable") from None + if not token: + raise SystemExit("runtime credential reference is unavailable") + return token + + +def _write_report(path: Path, summary: dict[str, Any]) -> str: + encoded = (json.dumps(summary, sort_keys=True, separators=(",", ":")) + "\n").encode("utf-8") + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(encoded) + return hashlib.sha256(encoded).hexdigest() + + +def main(argv: list[str] | None = None) -> int: + argv = list(sys.argv[1:] if argv is None else argv) + _reject_plaintext_token_argument(argv) + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--base-url", default="http://127.0.0.1:18922") + token_group = parser.add_mutually_exclusive_group() + token_group.add_argument("--token-file", help="protected runtime credential reference") + token_group.add_argument("--token-stdin", action="store_true") + parser.add_argument("--task-id", default="AIOS-TASK-PLANE-SHADOW-CLIENT") + parser.add_argument("--idempotency-key", default="task-plane-shadow-client-v1") + parser.add_argument("--report", required=True, type=Path) + args = parser.parse_args(argv) + + token = _read_token(args) + try: + summary = run_shadow_client( + base_url=args.base_url, + token=token, + task_id=args.task_id, + idempotency_key=args.idempotency_key, + ) + except ShadowClientError as error: + print(f"http_status={error.status}", file=sys.stderr) + return error.status + + report_sha256 = _write_report(args.report, summary) + print(f"task_id={summary['task_id']}") + print(f"status={summary['status']}") + print(f"event_count={summary['event_count']}") + print(f"evidence_count={summary['evidence_count']}") + print("http_statuses=" + json.dumps(summary["http_statuses"], sort_keys=True)) + print(f"report_sha256={report_sha256}") return 0 diff --git a/tests/task_plane/public_interface/test_shadow_client_rbac.py b/tests/task_plane/public_interface/test_shadow_client_rbac.py new file mode 100644 index 0000000..5de705c --- /dev/null +++ b/tests/task_plane/public_interface/test_shadow_client_rbac.py @@ -0,0 +1,172 @@ +import contextlib +import importlib.util +import io +import json +import sys +import tempfile +import threading +import unittest +import urllib.error +import urllib.request +from http.server import ThreadingHTTPServer +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[0] + + +def _load_client(root: Path): + path = root / "scripts" / "task-plane" / "create_shadow_tasks.py" + spec = importlib.util.spec_from_file_location("shadow_client", path) + module = importlib.util.module_from_spec(spec) + assert spec and spec.loader + spec.loader.exec_module(module) + return module + + +class ShadowClientRBACPublicInterfaceTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.repo_root = Path.cwd() + sys.path.insert(0, str(cls.repo_root / "active-runtime")) + from app.task_plane.api import TaskPlaneRequestHandler, TaskPlaneService + from app.task_plane.policy import IdentityVerifier, PolicyEngine + from app.task_plane.store import TaskPlaneStore + + cls.RequestHandler = TaskPlaneRequestHandler + cls.TaskPlaneService = TaskPlaneService + cls.IdentityVerifier = IdentityVerifier + cls.PolicyEngine = PolicyEngine + cls.TaskPlaneStore = TaskPlaneStore + cls.client = _load_client(cls.repo_root) + + def _server(self, capabilities): + store = self.TaskPlaneStore(":memory:") + policy = self.PolicyEngine( + identity_verifier=self.IdentityVerifier({"shadow-client"}), + execution_enabled=False, + identity_capabilities={"shadow-client": set(capabilities)}, + ) + service = self.TaskPlaneService(store, policy=policy) + + class Handler(self.RequestHandler): + pass + + Handler.service = service + Handler.authenticator = lambda value: "shadow-client" if value == "test-token" else None + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + self.addCleanup(server.shutdown) + self.addCleanup(server.server_close) + self.addCleanup(thread.join, 2) + return f"http://127.0.0.1:{server.server_port}", store + + @staticmethod + def _request(base, path, token=None, body=None): + headers = {"Content-Type": "application/json"} + if token is not None: + headers["Authorization"] = f"Bearer {token}" + req = urllib.request.Request( + base + path, + data=None if body is None else json.dumps(body).encode(), + headers=headers, + method="POST" if body is not None else "GET", + ) + return urllib.request.urlopen(req, timeout=3) + + def test_missing_credential_returns_401(self): + base, _ = self._server({"task.creator", "task.read", "task.audit"}) + with self.assertRaises(urllib.error.HTTPError) as raised: + self._request(base, "/tasks") + self.assertEqual(raised.exception.code, 401) + + def test_missing_task_creator_returns_403(self): + base, _ = self._server({"task.read", "task.audit"}) + with self.assertRaises(self.client.ShadowClientError) as raised: + self.client.run_shadow_client( + base_url=base, + token="test-token", + task_id="SHADOW-RBAC-NO-CREATE", + idempotency_key="shadow-rbac-no-create-v1", + ) + self.assertEqual(raised.exception.status, 403) + + def test_minimum_client_is_idempotent_and_cannot_claim(self): + base, store = self._server({"task.creator", "task.read", "task.audit"}) + first = self.client.run_shadow_client( + base_url=base, + token="test-token", + task_id="SHADOW-RBAC-IDEMPOTENT", + idempotency_key="shadow-rbac-idempotent-v1", + ) + second = self.client.run_shadow_client( + base_url=base, + token="test-token", + task_id="SHADOW-RBAC-IDEMPOTENT", + idempotency_key="shadow-rbac-idempotent-v1", + ) + self.assertEqual(first["task_id"], second["task_id"]) + self.assertEqual(first["status"], "CREATED") + self.assertEqual(first["event_count"], 1) + self.assertEqual(first["evidence_count"], 0) + self.assertEqual(store.list_workers(), []) + claim_base, _ = self._server({"task.claim"}) + with self.assertRaises(urllib.error.HTTPError) as raised: + self._request( + claim_base, + "/tasks/SHADOW-RBAC-IDEMPOTENT/claim", + "test-token", + {"worker_id": "shadow-worker", "lease_seconds": 30}, + ) + self.assertEqual(raised.exception.code, 423) + + def test_forged_request_body_identity_is_rejected(self): + base, _ = self._server({"task.creator", "task.read", "task.audit"}) + with self.assertRaises(urllib.error.HTTPError) as raised: + self._request( + base, + "/tasks", + "test-token", + { + "task_id": "SHADOW-RBAC-FORGED", + "actor_identity": "administrator", + }, + ) + self.assertEqual(raised.exception.code, 400) + + def test_default_deny_get_rbac_and_redacted_report(self): + base, _ = self._server({"task.creator", "task.read", "task.audit"}) + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + token_file = root / "token" + report = root / "report.json" + token_file.write_text("test-token", encoding="utf-8") + stdout = io.StringIO() + with contextlib.redirect_stdout(stdout): + self.client.main( + [ + "--base-url", base, + "--token-file", str(token_file), + "--task-id", "SHADOW-RBAC-REPORT", + "--idempotency-key", "shadow-rbac-report-v1", + "--report", str(report), + ] + ) + output = stdout.getvalue() + self.assertNotIn("test-token", output) + self.assertNotIn("Authorization", output) + self.assertNotIn("test-token", report.read_text(encoding="utf-8")) + self.assertNotIn("Authorization", report.read_text(encoding="utf-8")) + self.assertIn("report_sha256=", output) + + denied_base, _ = self._server({"task.creator"}) + with self.assertRaises(urllib.error.HTTPError) as raised: + self._request(denied_base, "/tasks", "test-token") + self.assertEqual(raised.exception.code, 403) + + def test_plaintext_token_argument_is_rejected_without_echo(self): + stderr = io.StringIO() + with contextlib.redirect_stderr(stderr): + with self.assertRaises(SystemExit): + self.client.main(["--token", "sentinel-token"]) + self.assertNotIn("sentinel-token", stderr.getvalue())