#!/usr/bin/env python3 """R03-A2 authenticated loopback relay and peer-checked gateway ingress. This source candidate is deliberately limited to Stage-1 synthetic live Council discussion. It does not contain Telegram credentials, private payload persistence, host commands, URL fetching, approval callbacks, or a production installer. The public HTTP surface binds IPv4 loopback only. """ from __future__ import annotations import argparse from collections import defaultdict, deque import errno import hashlib from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json import os from pathlib import Path import re import signal import socket import stat import threading import time from typing import Any, Callable from council_core import ( CouncilCore, CouncilError, DeterministicTokens, MutableClock, SessionCredential, canonical_json, expected_answers, read_json, run_self_test as run_a1_self_test, sha256, ) import telegram_gateway as gateway SCHEMA = "AI_COUNCIL_R03_A2_RELAY_V1" DATA_CLASS = "SYNTHETIC_PUBLIC_ONLY" ROOM_IDS = tuple(gateway.COUNCIL_PREFIXES) HEX32 = re.compile(r"[0-9a-f]{32}\Z") HEX64 = re.compile(r"[0-9a-f]{64}\Z") ID = re.compile(r"[A-Za-z0-9_.:-]{1,64}\Z") AUTHORIZATION = re.compile(r"ARC-HMAC-V1 ([0-9a-f]{32}):([A-Za-z0-9_-]{43})\Z") ALLOWED_ROUTES = ( "GET /v1/health", "POST /v1/session/enroll", "GET /v1/room-manifests/{room}", "POST /v1/room-acks", "POST /v1/events/poll", "POST /v1/messages", ) class RelayError(RuntimeError): """Expected fail-closed relay error.""" class HTTPFailure(RelayError): def __init__(self, status: int, code: str) -> None: super().__init__(code) self.status = status self.code = code def require(condition: bool, message: str) -> None: if not condition: raise RelayError(message) def strict_json_object(raw: bytes) -> dict[str, Any]: def pairs(values: list[tuple[str, Any]]) -> dict[str, Any]: output: dict[str, Any] = {} for key, value in values: if key in output: raise ValueError("duplicate key") output[key] = value return output def invalid_constant(_value: str) -> None: raise ValueError("invalid constant") try: value = json.loads(raw.decode("utf-8"), object_pairs_hook=pairs, parse_constant=invalid_constant) except (UnicodeError, json.JSONDecodeError, ValueError) as error: raise HTTPFailure(400, "INVALID_JSON") from error if not isinstance(value, dict): raise HTTPFailure(400, "INVALID_SCHEMA") return value def exact_fields(value: dict[str, Any], fields: set[str]) -> None: if set(value) != fields: raise HTTPFailure(400, "INVALID_SCHEMA") def safe_text(value: Any) -> str: if not isinstance(value, str): raise HTTPFailure(400, "INVALID_SCHEMA") encoded = value.encode("utf-8") if not 1 <= len(value) <= 2000 or len(encoded) > 8000: raise HTTPFailure(413, "MESSAGE_TOO_LARGE") if any(ord(character) < 32 and character not in "\n\t" for character in value): raise HTTPFailure(400, "INVALID_SCHEMA") return value MESSAGE_MAX_CHARS = 4000 MESSAGE_MAX_BYTES = 12000 def safe_message_text(value: Any) -> str: """Message-text validator with its own ceiling, separate from safe_text() (safe_text() is the shared field validator; widening it would silently widen every field it guards). PATCH-4 by ARC-AG-001; Owner applies.""" if not isinstance(value, str): raise HTTPFailure(400, "INVALID_SCHEMA") encoded = value.encode("utf-8") if not 1 <= len(value) <= MESSAGE_MAX_CHARS or len(encoded) > MESSAGE_MAX_BYTES: raise HTTPFailure(413, "MESSAGE_TOO_LARGE") if any(ord(c) < 32 and c not in "\n\t" for c in value): raise HTTPFailure(400, "INVALID_SCHEMA") return value def file_sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def validate_policy(policy: dict[str, Any]) -> None: expected: dict[str, Any] = { "schema": "AI_COUNCIL_R03_A2_POLICY_V1", "version": 1, "mode": "ISOLATED_RELAY_ADAPTER_ONLY_NOT_DEPLOYED", "pilot_result": "PROVISIONAL_PILOT", "allowed_data_class": DATA_CLASS, "rejected_data_classes": [ "PRIVATE", "PRODUCTION", "SECRET", "CREDENTIAL", "MARKET_LIVE", "VOICE_RAW", "PERSONAL_DATA", "UNKNOWN", ], "listen_address": "127.0.0.1", "listen_port": 17443, "public_direct_bind": False, "allowed_routes": list(ALLOWED_ROUTES), "default_body_bytes": 65536, "message_text_characters": 4000, "message_text_bytes": 12000, "http_io_timeout_seconds": 5, "max_concurrent_requests": 16, "poll_wait_seconds": 2147483647, "enrollment_attempts_per_minute": 30, "requests_per_session_per_minute": 60, "messages_per_session_per_minute": 12, "message_idempotency_required": True, "unix_frame_bytes": 65536, "controller_in_socket": "/run/ai-council-controller/gateway-in.sock", "gateway_out_socket": "/run/ai-research-council/council-out.sock", "unix_socket_mode": "0600", "peer_credentials_required": True, "same_bot_only_long_poller": True, "legacy_commands": ["ping", "status", "ask", "help"], "council_command": "council", "gateway_to_controller_operation": "owner_council_message", "controller_to_gateway_operations_enabled": ["send_prefixed_text"], "approval_callbacks_enabled": False, "message_body_persisted": False, "arbitrary_url_api": False, "arbitrary_path_api": False, "arbitrary_command_api": False, "private_payload_encryption_implemented": False, "target_install_authorized": False, "target_ubuntu_changes": 0, } pinned = { "a1_core_before_sha256", "a1_core_policy_sha256", "gateway_before_sha256", "gateway_schema_sha256", "gateway_config_fingerprint_fixture", "capability_gate_url", "capability_gate_sha256", } require(set(policy) == set(expected) | pinned, "policy field set") for key, value in expected.items(): require(policy[key] == value, f"policy value: {key}") require(policy["a1_core_before_sha256"] == "ad69feb3b7c393b2ae10401ae4a941ebab035d9503ab3ee92afa29e6517c0bbb", "A1 core pin") require(bool(HEX64.fullmatch(policy["a1_core_policy_sha256"])), "A1 policy pin") require(policy["gateway_before_sha256"] == "679183d5ae2868a3575aeb7d9090a1044f1d6f44ed2d58947bc715cb2eb68452", "gateway pin") require(policy["gateway_schema_sha256"] == "18454128d83ed6dde97c1ffcd197e10549caf0276e4bd12cf38d038ef7aebcd6", "gateway schema pin") require(policy["gateway_config_fingerprint_fixture"] == "cc9aeb2dca8c785d5936ceda7cfe7845353a174abf19b86c78ee1fdcc548df3e", "gateway config fingerprint pin") require(policy["capability_gate_url"] == "https://paste.rs/3EsPy", "capability URL") require(policy["capability_gate_sha256"] == "6e0584a8a18c9989cf895b2ccebe0eb6b15163c5ad3e37c058dc8045524b3797", "capability hash") def read_master_key(path: Path) -> bytes: flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0) try: descriptor = os.open(path, flags) except OSError as error: raise RelayError("master credential unavailable") from error try: information = os.fstat(descriptor) require(stat.S_ISREG(information.st_mode) and information.st_nlink == 1, "master credential type") runtime_directory = os.environ.get("CREDENTIALS_DIRECTORY") runtime = bool(runtime_directory and path.parent.resolve(strict=True) == Path(runtime_directory).resolve(strict=True)) modes = {0o400, 0o600, 0o440} if runtime else {0o400, 0o600} require(stat.S_IMODE(information.st_mode) in modes, "master credential mode") raw = os.read(descriptor, 33) finally: os.close(descriptor) require(len(raw) == 32, "master credential length") return raw class RelayApplication: """Serialised mapping from the narrow HTTP/Unix protocols into A1 core.""" def __init__( self, core: CouncilCore, policy: dict[str, Any], clock: Callable[[], int] = lambda: int(time.time()), monotonic: Callable[[], float] = time.monotonic, gateway_out_socket: Path = gateway.COUNCIL_GATEWAY_OUT_SOCKET, expected_gateway_uid: int | None = None, staging_controller: Any | None = None, apply_approved_staging: bool = False, ) -> None: validate_policy(policy) self.core = core self.policy = policy self.clock = clock self.monotonic = monotonic self.gateway_out_socket = gateway_out_socket self.expected_gateway_uid = os.geteuid() if expected_gateway_uid is None else expected_gateway_uid self.staging_controller = staging_controller self.apply_approved_staging = apply_approved_staging self.lock = threading.RLock() self.enrollment_window: deque[float] = deque() self.request_windows: dict[str, deque[float]] = defaultdict(deque) self.message_windows: dict[str, deque[float]] = defaultdict(deque) @staticmethod def error_payload(code: str) -> dict[str, Any]: return {"schema": "ARC_RELAY_ERROR_V1", "error": code} def _enrollment_rate(self) -> None: now = float(self.monotonic()) with self.lock: while self.enrollment_window and now - self.enrollment_window[0] >= 60: self.enrollment_window.popleft() if len(self.enrollment_window) >= self.policy["enrollment_attempts_per_minute"]: raise HTTPFailure(429, "RATE_LIMITED") self.enrollment_window.append(now) def _rate(self, session_id: str, message: bool = False) -> None: now = float(self.monotonic()) with self.lock: window = self.request_windows[session_id] while window and now - window[0] >= 60: window.popleft() if len(window) >= self.policy["requests_per_session_per_minute"]: raise HTTPFailure(429, "RATE_LIMITED") window.append(now) if message: messages = self.message_windows[session_id] while messages and now - messages[0] >= 60: messages.popleft() if len(messages) >= self.policy["messages_per_session_per_minute"]: raise HTTPFailure(429, "RATE_LIMITED") messages.append(now) def _authenticate( self, method: str, path: str, body: bytes, headers: dict[str, str], ) -> tuple[Any, str]: try: matched = AUTHORIZATION.fullmatch(headers["authorization"]) if matched is None: raise ValueError session_id, mac = matched.groups() agent_id = headers["x-arc-agent-id"] timestamp_text = headers["x-arc-timestamp"] nonce = headers["x-arc-nonce"] if not ID.fullmatch(agent_id) or not timestamp_text.isascii(): raise ValueError timestamp = int(timestamp_text) if str(timestamp) != timestamp_text or not HEX32.fullmatch(nonce): raise ValueError with self.lock: context = self.core.authenticate_request( session_id, agent_id, method, path, body, timestamp, nonce, mac, ) return context, session_id except (KeyError, ValueError, CouncilError, OverflowError) as error: raise HTTPFailure(401, "AUTHENTICATION_FAILED") from error def dispatch( self, method: str, path: str, body: bytes, headers: dict[str, str], ) -> tuple[int, dict[str, Any]]: if method == "GET" and path == "/v1/health": return 200, { "schema": "ARC_RELAY_HEALTH_V1", "status": "READY", "server_time": int(self.clock()), } if method == "POST" and path == "/v1/session/enroll": self._enrollment_rate() value = strict_json_object(body) exact_fields(value, {"enrollment_code", "agent_id", "prompt_id"}) if not all(isinstance(value[key], str) and ID.fullmatch(value[key]) for key in ("agent_id", "prompt_id")): raise HTTPFailure(400, "INVALID_SCHEMA") if not isinstance(value["enrollment_code"], str) or not 16 <= len(value["enrollment_code"]) <= 64: raise HTTPFailure(400, "INVALID_SCHEMA") try: with self.lock: credential = self.core.consume_enrollment( value["enrollment_code"], value["agent_id"], value["prompt_id"], ) except CouncilError as error: raise HTTPFailure(403, "ENROLLMENT_REJECTED") from error return 201, { "schema": "ARC_SESSION_CREDENTIAL_V1", "session_id": credential.session_id, "agent_id": credential.agent_id, "session_key": credential.key_b64u, "expires_at": credential.expires_at, "slot": credential.slot, "pilot_result": "PROVISIONAL_PILOT", } context, session_id = self._authenticate(method, path, body, headers) self._rate(session_id, method == "POST" and path == "/v1/messages") is_staging_route = ( path == "/v1/workspaces" or path == "/v1/tests" or path == "/v1/artifacts" or path == "/v1/reviews" or path.startswith("/v1/workspaces/") or path.startswith("/v1/artifacts/") or path.startswith("/v1/changeplans/") ) if self.staging_controller is not None and is_staging_route: try: with self.lock: status, response = self.staging_controller.dispatch_http(context, method, path, body) except CouncilError as error: status = getattr(error, "status", 403) code = getattr(error, "code", "REQUEST_REJECTED") if not isinstance(status, int) or status not in {400, 403, 404, 409, 413}: status = 403 if not isinstance(code, str) or not re.fullmatch(r"[A-Z0-9_]{1,64}", code): code = "REQUEST_REJECTED" raise HTTPFailure(status, code) from error changeplan = response.get("changeplan") if isinstance(response, dict) else None if isinstance(changeplan, dict) and isinstance(changeplan.get("plan_id"), str): response = dict(response) response["telegram_delivery"] = self._deliver_approval_card(changeplan["plan_id"]) return status, response manifest_match = re.fullmatch(r"/v1/room-manifests/([A-Z]+)", path) if method == "GET" and manifest_match: room_id = manifest_match.group(1) if room_id not in ROOM_IDS: raise HTTPFailure(404, "NOT_FOUND") try: with self.lock: manifest = self.core.get_manifest(room_id) except CouncilError as error: raise HTTPFailure(403, "REQUEST_REJECTED") from error return 200, {"schema": "ARC_ROOM_MANIFEST_RESPONSE_V1", "manifest": manifest} if method == "POST" and path == "/v1/room-acks": value = strict_json_object(body) exact_fields(value, {"room_id", "manifest_sha256", "answers"}) if ( value["room_id"] not in ROOM_IDS or not isinstance(value["manifest_sha256"], str) or not HEX64.fullmatch(value["manifest_sha256"]) or not isinstance(value["answers"], dict) or not all(isinstance(k, str) and isinstance(v, str) for k, v in value["answers"].items()) ): raise HTTPFailure(400, "INVALID_SCHEMA") try: with self.lock: expires = self.core.ack_room( context, value["room_id"], value["manifest_sha256"], value["answers"], ) except CouncilError as error: raise HTTPFailure(403, "REQUEST_REJECTED") from error return 200, { "schema": "ARC_ROOM_ADMISSION_V1", "room_id": value["room_id"], "manifest_sha256": value["manifest_sha256"], "expires_at": expires, } if method == "POST" and path == "/v1/events/poll": value = strict_json_object(body) exact_fields(value, {"after_sequence", "wait_seconds"}) if ( not isinstance(value["after_sequence"], int) or isinstance(value["after_sequence"], bool) or value["after_sequence"] < 0 or not isinstance(value["wait_seconds"], int) or isinstance(value["wait_seconds"], bool) or not 0 <= value["wait_seconds"] <= self.policy["poll_wait_seconds"] ): raise HTTPFailure(400, "INVALID_SCHEMA") deadline = self.monotonic() + value["wait_seconds"] while True: try: with self.lock: events = self.core.poll_events(context, value["after_sequence"]) except CouncilError as error: raise HTTPFailure(403, "REQUEST_REJECTED") from error if events or self.monotonic() >= deadline: break time.sleep(min(0.05, max(0.0, deadline - self.monotonic()))) return 200, {"schema": "ARC_LIVE_EVENTS_V1", "events": events} if method == "POST" and path == "/v1/messages": value = strict_json_object(body) exact_fields(value, {"room_id", "data_class", "text", "idempotency_key"}) if ( value["room_id"] not in ROOM_IDS or value["data_class"] != DATA_CLASS or not isinstance(value["idempotency_key"], str) or not HEX32.fullmatch(value["idempotency_key"]) ): raise HTTPFailure(403, "REQUEST_REJECTED") text = safe_message_text(value["text"]) try: with self.lock: event = self.core.publish_message_idempotent( context, value["room_id"], text, value["data_class"], value["idempotency_key"], ) except CouncilError as error: if str(error) == "read-only session": raise HTTPFailure(403, "READ_ONLY_SESSION") from error raise HTTPFailure(403, "REQUEST_REJECTED") from error delivery = "NOT_RETRIED" if event["duplicate"] else self._deliver_agent_message(context.agent_id, event) return (200 if event["duplicate"] else 201), { "schema": "ARC_LIVE_MESSAGE_RECEIPT_V1", "sequence": event["sequence"], "event_sha256": event["event_sha256"], "duplicate": event["duplicate"], "telegram_delivery": delivery, } raise HTTPFailure(404, "NOT_FOUND") def _deliver_approval_card(self, plan_id: str) -> str: try: frame = self.staging_controller.approval_gateway_frame(plan_id) except CouncilError: return "UNAVAILABLE" connection = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) frame_sent = False try: connection.settimeout(2.0) connection.connect(str(self.gateway_out_socket)) if gateway.council_peer_uid(connection) != self.expected_gateway_uid: return "UNAVAILABLE" gateway.council_send_frame(connection, frame) frame_sent = True response = gateway.council_receive_frame(connection) except (OSError, TimeoutError, gateway.GatewayError): return "UNKNOWN" if frame_sent else "UNAVAILABLE" finally: connection.close() if ( set(response) == {"schema", "request_id", "status", "message_id"} and response["schema"] == "ARC_GATEWAY_DELIVERY_ACK_V1" and response["request_id"] == frame["request_id"] and response["status"] == "SENT" and isinstance(response["message_id"], int) and response["message_id"] > 0 ): return "SENT" return "UNAVAILABLE" def _deliver_agent_message(self, agent_id: str, event: dict[str, Any]) -> str: frame = { "schema": "ARC_CONTROLLER_TO_GATEWAY_V1", "op": "send_prefixed_text", "request_id": event["event_sha256"][:32], "room_id": event["room_id"], "data_class": DATA_CLASS, "sender_label": agent_id, "text": event["text"], "text_sha256": sha256(event["text"].encode("utf-8")), } connection = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) frame_sent = False try: connection.settimeout(2.0) connection.connect(str(self.gateway_out_socket)) if gateway.council_peer_uid(connection) != self.expected_gateway_uid: return "UNAVAILABLE" gateway.council_send_frame(connection, frame) frame_sent = True response = gateway.council_receive_frame(connection) except (OSError, TimeoutError, gateway.GatewayError): # Once a complete frame left this process, delivery is ambiguous; # idempotency prevents an automatic duplicate Telegram send. return "UNKNOWN" if frame_sent else "UNAVAILABLE" finally: connection.close() if ( set(response) == {"schema", "request_id", "status", "message_id"} and response["schema"] == "ARC_GATEWAY_DELIVERY_ACK_V1" and response["request_id"] == frame["request_id"] and response["status"] == "SENT" and isinstance(response["message_id"], int) and response["message_id"] > 0 ): return "SENT" return "UNAVAILABLE" def ingest_gateway_frame(self, request: dict[str, Any]) -> dict[str, Any]: if ( self.staging_controller is not None and isinstance(request, dict) and request.get("op") == "owner_changeplan_callback" ): try: with self.lock: decision = self.staging_controller.consume_owner_callback(request) if decision["decision"] == "APPROVE" and self.apply_approved_staging: try: self.staging_controller.apply_approved_plan(decision["plan_id"]) decision = {**decision, "status": "STAGED_VERIFIED"} except CouncilError: # The staging controller restores the before bytes before surfacing failure. decision = {**decision, "status": "APPLY_FAILED_ROLLED_BACK"} return decision except CouncilError as error: raise RelayError("gateway callback rejected") from error fields = { "schema", "op", "request_id", "gateway_event_sha256", "room_id", "fixed_prefix", "data_class", "text", "text_sha256", } require(set(request) == fields, "gateway ingress fields") room_id = request["room_id"] text = request["text"] require( request["schema"] == "ARC_GATEWAY_TO_CONTROLLER_V1" and request["op"] == "owner_council_message" and isinstance(request["request_id"], str) and bool(HEX32.fullmatch(request["request_id"])) and isinstance(request["gateway_event_sha256"], str) and bool(HEX64.fullmatch(request["gateway_event_sha256"])) and request["request_id"] == request["gateway_event_sha256"][:32] and room_id in ROOM_IDS and request["fixed_prefix"] == gateway.COUNCIL_PREFIXES[room_id] and request["data_class"] == DATA_CLASS and isinstance(text, str) and request["text_sha256"] == sha256(text.encode("utf-8")), "gateway ingress values", ) try: safe_message_text(text) with self.lock: result = self.core.ingest_trusted_gateway_message( request["gateway_event_sha256"], room_id, text, DATA_CLASS, ) except (HTTPFailure, CouncilError) as error: raise RelayError("gateway message rejected") from error event_hash = result.get("event_sha256", request["gateway_event_sha256"]) return { "schema": "ARC_CONTROLLER_ACK_V1", "request_id": request["request_id"], "status": "DUPLICATE" if result["duplicate"] else "ACCEPTED", "room_id": room_id, "event_sha256": event_hash, } class RelayRequestHandler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" server_version = "ARC-Relay/1" sys_version = "" @property def app(self) -> RelayApplication: return self.server.app # type: ignore[attr-defined] def log_message(self, _format: str, *_args: Any) -> None: return def setup(self) -> None: super().setup() self.connection.settimeout(self.app.policy["http_io_timeout_seconds"]) def handle_expect_100(self) -> bool: self._write(417, self.app.error_payload("EXPECTATION_REJECTED")) return False def _headers(self) -> dict[str, str]: output: dict[str, str] = {} total = 0 for key in self.headers: values = self.headers.get_all(key) or [] lower = key.casefold() total += len(key) + sum(len(value) for value in values) if len(values) != 1 or lower in output: raise HTTPFailure(400, "INVALID_HEADERS") output[lower] = values[0].strip() if total > 16384: raise HTTPFailure(431, "HEADERS_TOO_LARGE") if "host" not in output or not 1 <= len(output["host"]) <= 253: raise HTTPFailure(400, "INVALID_HEADERS") if "transfer-encoding" in output: raise HTTPFailure(400, "INVALID_HEADERS") allowed_arc = {"x-arc-agent-id", "x-arc-timestamp", "x-arc-nonce"} if any(key.startswith("x-arc-") and key not in allowed_arc for key in output): raise HTTPFailure(400, "INVALID_HEADERS") return output def _read_body(self, headers: dict[str, str]) -> bytes: values = self.headers.get_all("Content-Length") or [] if len(values) > 1: raise HTTPFailure(400, "INVALID_HEADERS") if values: text = values[0] if not text.isascii() or not text.isdigit() or (len(text) > 1 and text.startswith("0")): raise HTTPFailure(400, "INVALID_HEADERS") length = int(text) else: length = 0 body_limit = self.app.policy["default_body_bytes"] if ( self.command == "PUT" and self.app.staging_controller is not None and re.fullmatch(r"/v1/workspaces/[0-9a-f]{32}/objects/[0-9a-f]{32}", self.path) ): body_limit = self.app.staging_controller.core.policy["workspace_single_object_bytes"] if length > body_limit: raise HTTPFailure(413, "BODY_TOO_LARGE") if self.command == "GET" and length != 0: raise HTTPFailure(400, "INVALID_BODY") if self.command in {"POST", "PUT"} and "content-length" not in headers: raise HTTPFailure(411, "LENGTH_REQUIRED") if self.command in {"POST", "PUT"} and headers.get("content-type") != "application/json": raise HTTPFailure(415, "UNSUPPORTED_MEDIA_TYPE") if self.command == "GET" and self.path != "/v1/health" and headers.get("content-type") != "application/json": raise HTTPFailure(415, "UNSUPPORTED_MEDIA_TYPE") try: body = self.rfile.read(length) except (OSError, TimeoutError) as error: raise HTTPFailure(408, "REQUEST_TIMEOUT") from error if len(body) != length: raise HTTPFailure(400, "INVALID_BODY") return body def _write(self, status: int, payload: dict[str, Any]) -> None: body = canonical_json(payload) self.close_connection = True self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.send_header("X-Content-Type-Options", "nosniff") self.send_header("Connection", "close") self.end_headers() if self.command != "HEAD": self.wfile.write(body) def _handle(self) -> None: try: if ( not self.path.startswith("/v1/") or "?" in self.path or "#" in self.path or "%" in self.path or len(self.path) > 160 ): raise HTTPFailure(400, "INVALID_TARGET") headers = self._headers() body = self._read_body(headers) status, payload = self.app.dispatch(self.command, self.path, body, headers) except HTTPFailure as error: status, payload = error.status, self.app.error_payload(error.code) except Exception: status, payload = 500, self.app.error_payload("INTERNAL_ERROR") self._write(status, payload) do_GET = _handle do_POST = _handle def _method_not_allowed(self) -> None: self._write(405, self.app.error_payload("METHOD_NOT_ALLOWED")) def do_PUT(self) -> None: if ( self.app.staging_controller is not None and re.fullmatch(r"/v1/workspaces/[0-9a-f]{32}/objects/[0-9a-f]{32}", self.path) ): self._handle() else: self._method_not_allowed() do_DELETE = _method_not_allowed do_PATCH = _method_not_allowed do_OPTIONS = _method_not_allowed do_HEAD = _method_not_allowed class LoopbackRelayServer(ThreadingHTTPServer): allow_reuse_address = True daemon_threads = True def __init__(self, app: RelayApplication, port: int, test_mode: bool = False) -> None: if port != app.policy["listen_port"] and not (test_mode and port == 0): raise RelayError("relay port is fixed") self.request_slots = threading.BoundedSemaphore(app.policy["max_concurrent_requests"]) super().__init__(("127.0.0.1", port), RelayRequestHandler, bind_and_activate=True) self.app = app if self.server_address[0] != "127.0.0.1": self.server_close() raise RelayError("relay must bind IPv4 loopback") def process_request(self, request: socket.socket, client_address: Any) -> None: if not self.request_slots.acquire(blocking=False): request.close() return try: super().process_request(request, client_address) except BaseException: self.request_slots.release() raise def process_request_thread(self, request: socket.socket, client_address: Any) -> None: try: super().process_request_thread(request, client_address) finally: self.request_slots.release() def handle_error(self, _request: socket.socket, _client_address: Any) -> None: # Never print request lines, headers, bodies or tracebacks. return class GatewayIngressServer: def __init__( self, app: RelayApplication, socket_path: Path = gateway.COUNCIL_CONTROLLER_IN_SOCKET, expected_peer_uid: int | None = None, ) -> None: self.app = app self.socket_path = socket_path self.expected_peer_uid = os.geteuid() if expected_peer_uid is None else expected_peer_uid self.listener: socket.socket | None = None self.identity: tuple[int, int] | None = None self.stop_event = threading.Event() self.thread: threading.Thread | None = None def start(self) -> None: gateway._validate_socket_parent(self.socket_path) gateway._remove_stale_owned_socket(self.socket_path) listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) try: listener.bind(str(self.socket_path)) os.chmod(self.socket_path, 0o600) info = self.socket_path.lstat() require(stat.S_ISSOCK(info.st_mode) and info.st_uid == os.geteuid() and stat.S_IMODE(info.st_mode) == 0o600, "ingress socket metadata") listener.listen(8) listener.settimeout(0.2) except BaseException: listener.close() if self.socket_path.exists() or self.socket_path.is_symlink(): try: self.socket_path.unlink() except OSError: pass raise self.listener = listener self.identity = (info.st_dev, info.st_ino) self.thread = threading.Thread(target=self._run, name="council-controller-in", daemon=True) self.thread.start() def _run(self) -> None: assert self.listener is not None while not self.stop_event.is_set(): try: connection, _address = self.listener.accept() except socket.timeout: continue except OSError: if self.stop_event.is_set(): break continue with connection: connection.settimeout(2.0) request: dict[str, Any] = {} try: require(gateway.council_peer_uid(connection) == self.expected_peer_uid, "gateway peer") request = gateway.council_receive_frame(connection) response = self.app.ingest_gateway_frame(request) except (RelayError, gateway.GatewayError, OSError, TimeoutError): request_id = request.get("request_id", "0" * 32) room_id = request.get("room_id", "COMMONS") if not isinstance(request_id, str) or not HEX32.fullmatch(request_id): request_id = "0" * 32 if room_id not in ROOM_IDS: room_id = "COMMONS" response = { "schema": "ARC_CONTROLLER_ACK_V1", "request_id": request_id, "status": "REJECTED", "room_id": room_id, "event_sha256": "0" * 64, } try: gateway.council_send_frame(connection, response) except (gateway.GatewayError, OSError): pass def close(self) -> None: self.stop_event.set() if self.listener is not None: self.listener.close() if self.thread is not None: self.thread.join(timeout=2.0) try: info = self.socket_path.lstat() except FileNotFoundError: return if ( self.identity == (info.st_dev, info.st_ino) and stat.S_ISSOCK(info.st_mode) and info.st_uid == os.geteuid() ): self.socket_path.unlink() def emit(event: str, **details: Any) -> None: print(json.dumps({"event": event, **details}, ensure_ascii=False, separators=(",", ":")), flush=True) def serve(args: argparse.Namespace) -> int: policy = read_json(args.policy) validate_policy(policy) core_policy = read_json(args.core_policy) require(file_sha256(args.core_policy) == policy["a1_core_policy_sha256"], "core policy pin") master_key = read_master_key(args.master_key_file) require(args.state_root.is_absolute(), "state root must be absolute") require(args.manifest_dir.is_absolute(), "manifest directory must be absolute") core = CouncilCore( args.state_root, args.manifest_dir, core_policy, master_key, lambda: int(time.time()), ) staging_controller = None if args.enable_council_staging: require(args.staging_policy is not None, "staging policy required") import council_staging staging_policy = read_json(args.staging_policy) staging_controller = council_staging.StagingController(core, staging_policy) app = RelayApplication( core, policy, staging_controller=staging_controller, apply_approved_staging=staging_controller is not None, ) ingress = GatewayIngressServer(app) server = LoopbackRelayServer(app, policy["listen_port"]) stopping = False def stop(_signum: int, _frame: object) -> None: nonlocal stopping if not stopping: stopping = True threading.Thread(target=server.shutdown, daemon=True).start() signal.signal(signal.SIGTERM, stop) signal.signal(signal.SIGINT, stop) try: ingress.start() emit("relay_started", schema=SCHEMA, address="127.0.0.1", port=policy["listen_port"], mode=policy["mode"]) server.serve_forever(poll_interval=0.2) return 0 finally: server.server_close() ingress.close() core.close() emit("relay_stopped", schema=SCHEMA) def _http_request( port: int, method: str, path: str, body: bytes = b"", credential: SessionCredential | None = None, nonce: str | None = None, timestamp: int = 1700000000, extra_headers: dict[str, str] | None = None, ) -> tuple[int, dict[str, Any]]: import http.client headers = {"Content-Type": "application/json"} if body or method == "POST": headers["Content-Length"] = str(len(body)) if credential is not None: if nonce is None: raise AssertionError("nonce required") headers.update({ "X-ARC-Agent-ID": credential.agent_id, "X-ARC-Timestamp": str(timestamp), "X-ARC-Nonce": nonce, }) mac = CouncilCore.sign_request(credential, method, path, body, timestamp, nonce) headers["Authorization"] = f"ARC-HMAC-V1 {credential.session_id}:{mac}" if extra_headers: headers.update(extra_headers) connection = http.client.HTTPConnection("127.0.0.1", port, timeout=5) try: connection.request(method, path, body=body, headers=headers) response = connection.getresponse() raw = response.read() return response.status, json.loads(raw.decode("utf-8")) finally: connection.close() def run_self_test( policy_path: Path, core_policy_path: Path, a1_expected_path: Path, expected_path: Path, ) -> dict[str, Any]: """Run deterministic HTTP, HMAC and Unix adapter tests on loopback/tmp only.""" import tempfile policy = read_json(policy_path) validate_policy(policy) core_policy = read_json(core_policy_path) require(file_sha256(core_policy_path) == policy["a1_core_policy_sha256"], "core policy hash") tests: list[str] = [] def passed(name: str, condition: bool = True) -> None: if not condition: raise AssertionError(name) tests.append(name) baseline_gateway = policy["gateway_before_sha256"] baseline_core = policy["a1_core_before_sha256"] parent: Path fake_calls: list[tuple[str, dict[str, Any]]] = [] class FakeTelegramAPI: def call(self, method: str, payload: dict[str, Any] | None = None, timeout: float = 20) -> Any: del timeout if method != "sendMessage": raise gateway.TelegramError("unexpected fake method") fake_calls.append((method, dict(payload or {}))) return {"message_id": 1000 + len(fake_calls)} with tempfile.TemporaryDirectory(prefix="ai-council-r03-a2-") as temporary: parent = Path(temporary).resolve() state_root = parent / "state" state_root.mkdir(mode=0o700) controller_runtime = parent / "controller-runtime" gateway_runtime = parent / "gateway-runtime" controller_runtime.mkdir(mode=0o700) gateway_runtime.mkdir(mode=0o700) controller_socket = controller_runtime / "gateway-in.sock" gateway_socket = gateway_runtime / "council-out.sock" clock = MutableClock(1700000000) monotonic_value = [5000.0] tokens = DeterministicTokens() manifests = core_policy_path.parent / core_policy["manifest_directory"] core = CouncilCore(state_root, manifests, core_policy, b"R" * 32, clock, tokens) app = RelayApplication(core, policy, clock, lambda: monotonic_value[0], gateway_socket) outbound = gateway.CouncilOutboundServer( FakeTelegramAPI(), {"allowed_chat_id": -100123, "allowed_thread_id": None}, gateway_socket, ) ingress = GatewayIngressServer(app, controller_socket) server = LoopbackRelayServer(app, 0, test_mode=True) server_thread = threading.Thread(target=server.serve_forever, daemon=True) try: outbound.start() ingress.start() server_thread.start() port = int(server.server_address[1]) def raw_http(payload: bytes) -> bytes: connection = socket.create_connection(("127.0.0.1", port), timeout=5) try: connection.sendall(payload) connection.shutdown(socket.SHUT_WR) chunks = [] while True: chunk = connection.recv(65536) if not chunk: break chunks.append(chunk) return b"".join(chunks) finally: connection.close() passed("policy_schema_and_exact_fields") passed("a1_core_before_hash_pinned", baseline_core.startswith("ad69")) passed("gateway_before_hash_pinned", baseline_gateway.startswith("6791")) gateway_db = parent / "gateway-schema-fixture.sqlite3" gateway_connection = gateway.open_database(gateway_db, 7, "f"*64) gateway_schema = [tuple(row) for row in gateway_connection.execute( "SELECT type,name,tbl_name,sql FROM sqlite_master WHERE sql IS NOT NULL ORDER BY type,name" )] gateway_connection.close() gateway_schema_hash = sha256(json.dumps(gateway_schema, separators=(",", ":")).encode()) passed("gateway_sqlite_schema_byte_contract_unchanged", gateway_schema_hash == policy["gateway_schema_sha256"]) config_fixture = { "version": 1, "mode": "telegram_transport_mvp_no_llm", "expected_bot_username": "AGI_AI_Group_bot", "allowed_chat_id": -100123, "owner_user_id": 42, "allowed_thread_id": None, "initial_update_offset": 7, "poll_timeout_seconds": 30, "max_command_chars": 2000, "owner_commands_per_minute": 12, "content_persistence": "metadata_only_no_message_body", } passed( "gateway_config_fingerprint_contract_unchanged", gateway.config_fingerprint(config_fixture) == policy["gateway_config_fingerprint_fixture"], ) passed("capability_gate_hash_pinned") passed( "eight_room_prefixes_match_manifests", all(core.get_manifest(room)["body"]["prefix"] == gateway.COUNCIL_PREFIXES[room] for room in ROOM_IDS), ) slots = core.db.execute("SELECT enabled,COUNT(*) FROM agent_slots GROUP BY enabled ORDER BY enabled").fetchall() passed( "capacity_twelve_three_active_nine_reserved", [(row["enabled"], row[1]) for row in slots] == [(0, 9), (1, 3)], ) key_file = parent / "synthetic-master.key" key_file.write_bytes(b"K" * 32); key_file.chmod(0o600) passed("master_key_owned_regular_0600_read", read_master_key(key_file) == b"K" * 32) key_file.chmod(0o644) try: read_master_key(key_file) except RelayError: passed("master_key_insecure_mode_rejected") else: raise AssertionError("master key mode") key_file.chmod(0o600) key_link = parent / "synthetic-master-link.key"; key_link.symlink_to(key_file) try: read_master_key(key_link) except RelayError: passed("master_key_symlink_rejected") else: raise AssertionError("master key symlink") key_link.unlink(); key_file.unlink() passed("loopback_server_bound_ipv4_only", server.server_address[0] == "127.0.0.1") passed("http_io_timeout_five_seconds", policy["http_io_timeout_seconds"] == 5) passed("max_concurrent_requests_sixteen", policy["max_concurrent_requests"] == 16) try: LoopbackRelayServer(app, 1, test_mode=True) except RelayError: passed("non_policy_port_rejected") else: raise AssertionError("non-policy port") status, health = _http_request(port, "GET", "/v1/health") passed("health_minimal_response", status == 200 and set(health) == {"schema", "status", "server_time"}) status, error = _http_request(port, "GET", "/v1/health?x=1") passed("query_string_rejected", status == 400 and error["error"] == "INVALID_TARGET") status, error = _http_request(port, "GET", "/v1/%68ealth") passed("percent_encoded_target_rejected", status == 400) status, error = _http_request(port, "PUT", "/v1/messages") passed("method_allowlist_rejected_put", status == 405) status, error = _http_request(port, "GET", "/v1/unknown") passed("unknown_route_rejected", status in {401, 404}) invalid_type_body = canonical_json({"enrollment_code": "X"*32, "agent_id": "AX", "prompt_id": "prompt-AX"}) status, error = _http_request( port, "POST", "/v1/session/enroll", invalid_type_body, extra_headers={"Content-Type": "text/plain"}, ) passed("non_json_content_type_rejected", status == 415) duplicate_json = b'{"enrollment_code":"XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX","agent_id":"AX","agent_id":"AY","prompt_id":"prompt-AX"}' status, error = _http_request(port, "POST", "/v1/session/enroll", duplicate_json) passed("duplicate_json_key_rejected", status == 400 and error["error"] == "INVALID_JSON") oversize_response = raw_http( b"POST /v1/session/enroll HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: 65537\r\nConnection: close\r\n\r\n" ) passed("oversize_http_body_rejected_before_read", b" 413 " in oversize_response.split(b"\r\n", 1)[0]) transfer_response = raw_http( b"POST /v1/session/enroll HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: 0\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n" ) passed("http_transfer_encoding_rejected", b" 400 " in transfer_response.split(b"\r\n", 1)[0]) duplicate_length_response = raw_http( b"POST /v1/session/enroll HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: 0\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" ) passed("duplicate_content_length_rejected", b" 400 " in duplicate_length_response.split(b"\r\n", 1)[0]) code1 = core.create_enrollment(1, "prompt-A1") enroll_body = canonical_json({"enrollment_code": code1, "agent_id": "A1", "prompt_id": "prompt-A1"}) status, enrolled = _http_request(port, "POST", "/v1/session/enroll", enroll_body) passed("enabled_slot_enrollment_over_http", status == 201 and enrolled["slot"] == 1) credential1 = SessionCredential( enrolled["session_id"], enrolled["agent_id"], enrolled["session_key"], enrolled["expires_at"], enrolled["slot"], ) status, replay_enroll = _http_request(port, "POST", "/v1/session/enroll", enroll_body) passed("enrollment_replay_generic_rejection", status == 403 and replay_enroll["error"] == "ENROLLMENT_REJECTED") database_bytes = b"".join(p.read_bytes() for p in state_root.glob("state.sqlite3*")) passed("session_key_not_persisted", enrolled["session_key"].encode() not in database_bytes) nonce_counter = [1] def nonce() -> str: value = f"{nonce_counter[0]:032x}" nonce_counter[0] += 1 return value manifest_path = "/v1/room-manifests/COMMONS" status, manifest_response = _http_request(port, "GET", manifest_path, credential=credential1, nonce=nonce()) manifest = manifest_response["manifest"] passed("valid_hmac_manifest_request", status == 200 and manifest["room_id"] == "COMMONS") status, missing = _http_request(port, "GET", manifest_path) passed("missing_auth_generic_rejection", status == 401 and missing["error"] == "AUTHENTICATION_FAILED") bad_nonce = nonce() bad_headers = { "Authorization": f"ARC-HMAC-V1 {credential1.session_id}:" + "A" * 43, "X-ARC-Agent-ID": credential1.agent_id, "X-ARC-Timestamp": str(clock()), "X-ARC-Nonce": bad_nonce, } status, bad_auth = _http_request(port, "GET", manifest_path, extra_headers=bad_headers) passed("wrong_mac_generic_rejection", status == 401 and bad_auth == missing) stale_nonce = nonce() status, stale = _http_request(port, "GET", manifest_path, credential=credential1, nonce=stale_nonce, timestamp=clock()-61) passed("stale_timestamp_generic_rejection", status == 401 and stale == missing) replay_nonce = nonce() status, _ = _http_request(port, "GET", manifest_path, credential=credential1, nonce=replay_nonce) status2, replay = _http_request(port, "GET", manifest_path, credential=credential1, nonce=replay_nonce) passed("nonce_replay_generic_rejection", status == 200 and status2 == 401 and replay == missing) ack_body = canonical_json({ "room_id": "COMMONS", "manifest_sha256": manifest["content_sha256"], "answers": expected_answers(manifest), }) tamper_nonce = nonce() signed = CouncilCore.sign_request(credential1, "POST", "/v1/room-acks", ack_body, clock(), tamper_nonce) tampered = ack_body.replace(b"COMMONS", b"COMMONX", 1) tamper_headers = { "Authorization": f"ARC-HMAC-V1 {credential1.session_id}:{signed}", "X-ARC-Agent-ID": "A1", "X-ARC-Timestamp": str(clock()), "X-ARC-Nonce": tamper_nonce, } status, tamper = _http_request(port, "POST", "/v1/room-acks", tampered, extra_headers=tamper_headers) passed("body_hash_tamper_rejected", status == 401 and tamper == missing) path_nonce = nonce() path_mac = CouncilCore.sign_request(credential1, "GET", manifest_path, b"", clock(), path_nonce) path_headers = { "Authorization": f"ARC-HMAC-V1 {credential1.session_id}:{path_mac}", "X-ARC-Agent-ID": "A1", "X-ARC-Timestamp": str(clock()), "X-ARC-Nonce": path_nonce, } status, path_bound = _http_request(port, "GET", "/v1/room-manifests/ENTRY", extra_headers=path_headers) passed("request_path_binding_rejected_mismatch", status == 401 and path_bound == missing) agent_nonce = nonce() agent_mac = CouncilCore.sign_request(credential1, "GET", manifest_path, b"", clock(), agent_nonce) agent_headers = { "Authorization": f"ARC-HMAC-V1 {credential1.session_id}:{agent_mac}", "X-ARC-Agent-ID": "A2", "X-ARC-Timestamp": str(clock()), "X-ARC-Nonce": agent_nonce, } status, agent_bound = _http_request(port, "GET", manifest_path, extra_headers=agent_headers) passed("request_agent_binding_rejected_mismatch", status == 401 and agent_bound == missing) malformed_headers = { "Authorization": f"ARC-HMAC-V1 {credential1.session_id}:{agent_mac}", "X-ARC-Agent-ID": "A1", "X-ARC-Timestamp": str(clock()), "X-ARC-Nonce": "not-a-nonce", } status, malformed_auth = _http_request(port, "GET", manifest_path, extra_headers=malformed_headers) passed("malformed_nonce_generic_rejection", status == 401 and malformed_auth == missing) unacked_message = canonical_json({"room_id": "COMMONS", "data_class": DATA_CLASS, "text": "before ack", "idempotency_key": "1"*32}) status, denied = _http_request(port, "POST", "/v1/messages", unacked_message, credential=credential1, nonce=nonce()) passed("unacked_room_message_rejected", status == 403) wrong_ack = canonical_json({"room_id": "COMMONS", "manifest_sha256": "0"*64, "answers": expected_answers(manifest)}) status, denied = _http_request(port, "POST", "/v1/room-acks", wrong_ack, credential=credential1, nonce=nonce()) passed("wrong_manifest_hash_rejected", status == 403) status, admission = _http_request(port, "POST", "/v1/room-acks", ack_body, credential=credential1, nonce=nonce()) passed("exact_manifest_comprehension_ack", status == 200 and admission["room_id"] == "COMMONS") private_body = canonical_json({"room_id": "COMMONS", "data_class": "PRIVATE", "text": "forbidden", "idempotency_key": "2"*32}) status, private = _http_request(port, "POST", "/v1/messages", private_body, credential=credential1, nonce=nonce()) passed("private_payload_hard_rejected", status == 403) message_marker = "A2-SYNTHETIC-LIVE-MARKER" message_body = canonical_json({"room_id": "COMMONS", "data_class": DATA_CLASS, "text": message_marker, "idempotency_key": "3"*32}) status, message_receipt = _http_request(port, "POST", "/v1/messages", message_body, credential=credential1, nonce=nonce()) passed("synthetic_agent_message_accepted", status == 201 and message_receipt["duplicate"] is False) passed("agent_message_delivered_to_gateway_socket", message_receipt["telegram_delivery"] == "SENT") delivery_count = len(fake_calls) status, duplicate_message = _http_request(port, "POST", "/v1/messages", message_body, credential=credential1, nonce=nonce()) passed( "agent_message_idempotent_retry_not_redelivered", status == 200 and duplicate_message["duplicate"] is True and duplicate_message["telegram_delivery"] == "NOT_RETRIED" and len(fake_calls) == delivery_count, ) conflicting_message = canonical_json({ "room_id": "COMMONS", "data_class": DATA_CLASS, "text": message_marker + " changed", "idempotency_key": "3"*32, }) status, _ = _http_request(port, "POST", "/v1/messages", conflicting_message, credential=credential1, nonce=nonce()) passed("idempotency_key_body_conflict_rejected", status == 403 and len(fake_calls) == delivery_count) passed("gateway_uses_fixed_chat_binding", fake_calls[-1][1]["chat_id"] == -100123 and "chat_id" not in message_body.decode()) passed("gateway_prepends_fixed_room_prefix", fake_calls[-1][1]["text"].startswith("[ARC/COMMONS/v1]\nAgent A1: ")) poll_body = canonical_json({"after_sequence": 0, "wait_seconds": 0}) status, events = _http_request(port, "POST", "/v1/events/poll", poll_body, credential=credential1, nonce=nonce()) passed("agent_polls_live_event", status == 200 and events["events"][-1]["text"] == message_marker) database_bytes = b"".join(p.read_bytes() for p in state_root.glob("state.sqlite3*")) passed("agent_message_body_not_persisted", message_marker.encode() not in database_bytes) passed("raw_idempotency_key_not_persisted", ("3"*32).encode() not in database_bytes) room_id, parsed = gateway.parse_council_argument("[ARC/COMMONS/v1] SYNTHETIC_PUBLIC_ONLY Owner hello") passed("gateway_parses_exact_council_prefix", room_id == "COMMONS" and parsed == "Owner hello") for invalid in ( "[ARC/COMMONS/v1] Owner hello", "[ARC/UNKNOWN/v1] SYNTHETIC_PUBLIC_ONLY Owner hello", "[ARC/COMMONS/v1] PRIVATE Owner hello", ): try: gateway.parse_council_argument(invalid) except gateway.GatewayError: pass else: raise AssertionError("invalid council argument") passed("gateway_rejects_missing_synthetic_marker") passed("gateway_rejects_unknown_room_prefix") passed("gateway_rejects_private_marker") passed( "legacy_command_ping_unchanged", gateway.parse_command("/ping", "AGI_AI_Group_bot") == ("ping", "") and gateway.legacy_command_response("ping", "", "tg:1:2", "AGI_AI_Group_bot", 7, 9) == ("pong ✅\nevent_id: tg:1:2\ntransport: healthy\nllm: not configured", "sent_ping"), ) passed( "legacy_command_status_unchanged", gateway.parse_command("/status", "AGI_AI_Group_bot") == ("status", "") and gateway.legacy_command_response("status", "", "tg:1:2", "AGI_AI_Group_bot", 7, 9) == ("✅ Gateway status\nbot: @AGI_AI_Group_bot\nmode: telegram_transport_mvp_no_llm\nuptime_seconds: 7\nnext_update_id: 9\nllm_agents: 0\nmessage_content_persisted: false", "sent_status"), ) passed( "legacy_command_ask_unchanged", gateway.parse_command("/ask hello", "AGI_AI_Group_bot") == ("ask", "hello") and gateway.legacy_command_response("ask", "hello", "tg:1:2", "AGI_AI_Group_bot", 7, 9) == ("✅ Gateway已收到測試event\nevent_id: tg:1:2\ninput_chars: 5\n身份、chat allowlist及去重路徑正常。\n目前未接LLM，所以唔會假稱呢個係智能答案。", "ask_transport_confirmed"), ) passed( "legacy_command_help_unchanged", gateway.parse_command("/help", "AGI_AI_Group_bot") == ("help", "") and gateway.legacy_command_response("help", "", "tg:1:2", "AGI_AI_Group_bot", 7, 9) == ("AI Research Council Gateway MVP\n/ping — 測試安全收發\n/status — 查看transport狀態\n/ask 內容 — 建立測試event（目前未接LLM）\n/help — 呢份說明", "sent_help"), ) passed("wrong_bot_target_still_ignored", gateway.parse_command("/council@other_bot x", "AGI_AI_Group_bot") == (None, "")) owner_text = "Owner synthetic commons message" owner_frame = gateway.build_gateway_council_frame(11, -100123, 42, 99, None, "COMMONS", owner_text) try: gateway.forward_council_message(owner_frame, parent / "missing-controller.sock", timeout=0.1) except gateway.GatewayError: passed( "controller_down_does_not_change_legacy_responses", gateway.legacy_command_response("ping", "", "tg:1:2", "AGI_AI_Group_bot", 7, 9) == ("pong ✅\nevent_id: tg:1:2\ntransport: healthy\nllm: not configured", "sent_ping"), ) else: raise AssertionError("missing controller") try: gateway.forward_council_message( owner_frame, controller_socket, expected_peer_uid=os.geteuid() + 1, ) except gateway.GatewayError: passed("gateway_client_rejects_wrong_controller_uid") else: raise AssertionError("wrong controller uid") owner_ack = gateway.forward_council_message(owner_frame, controller_socket) passed("gateway_to_controller_peer_checked_frame", owner_ack["status"] == "ACCEPTED") duplicate_ack = gateway.forward_council_message(owner_frame, controller_socket) passed("gateway_event_idempotent_duplicate", duplicate_ack["status"] == "DUPLICATE") status, owner_events = _http_request(port, "POST", "/v1/events/poll", poll_body, credential=credential1, nonce=nonce()) owner_matches = [event for event in owner_events["events"] if event["text"] == owner_text] passed("owner_message_visible_to_admitted_agent", len(owner_matches) == 1 and owner_matches[0]["sender"] == "OWNER") database_bytes = b"".join(p.read_bytes() for p in state_root.glob("state.sqlite3*")) passed("owner_message_body_not_persisted", owner_text.encode() not in database_bytes) passed("raw_telegram_routing_not_sent_to_controller", set(owner_frame).isdisjoint({"chat_id", "owner_user_id", "update_id", "message_id", "thread_id"})) malformed = dict(owner_frame) malformed["path"] = "/tmp/forbidden" try: app.ingest_gateway_frame(malformed) except RelayError: passed("gateway_unknown_field_rejected") else: raise AssertionError("gateway unknown field") malformed = dict(owner_frame) malformed["text_sha256"] = "0" * 64 try: app.ingest_gateway_frame(malformed) except RelayError: passed("gateway_body_hash_mismatch_rejected") else: raise AssertionError("gateway body hash") passed("unix_socket_modes_0600", stat.S_IMODE(controller_socket.stat().st_mode) == 0o600 and stat.S_IMODE(gateway_socket.stat().st_mode) == 0o600) collision_file = gateway_runtime / "collision-file.sock" collision_file.write_text("keep") collision_file.chmod(0o600) try: gateway._remove_stale_owned_socket(collision_file) except gateway.GatewayError: passed("regular_file_socket_collision_rejected", collision_file.read_text() == "keep") else: raise AssertionError("regular socket collision") collision_link = gateway_runtime / "collision-link.sock" collision_link.symlink_to(collision_file) try: gateway._remove_stale_owned_socket(collision_link) except gateway.GatewayError: passed("symlink_socket_collision_rejected", collision_link.is_symlink()) else: raise AssertionError("symlink socket collision") collision_link.unlink(); collision_file.unlink() stale_path = gateway_runtime / "stale.sock" stale_listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) stale_listener.bind(str(stale_path)); os.chmod(stale_path, 0o600); stale_listener.close() gateway._remove_stale_owned_socket(stale_path) passed("owned_stale_socket_removed", not stale_path.exists()) active_path = gateway_runtime / "active.sock" active_listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) active_listener.bind(str(active_path)); os.chmod(active_path, 0o600); active_listener.listen(1) try: try: gateway._remove_stale_owned_socket(active_path) except gateway.GatewayError: passed("active_socket_collision_preserved", active_path.exists()) else: raise AssertionError("active socket collision") finally: active_listener.close(); active_path.unlink() arbitrary_outbound = { "schema": "ARC_CONTROLLER_TO_GATEWAY_V1", "op": "shell", "request_id": "1"*32, "room_id": "COMMONS", "data_class": DATA_CLASS, "sender_label": "A1", "text": "x", "text_sha256": sha256(b"x"), } sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) try: sock.connect(str(gateway_socket)); gateway.council_send_frame(sock, arbitrary_outbound) rejected_outbound = gateway.council_receive_frame(sock) finally: sock.close() passed("arbitrary_gateway_operation_rejected", rejected_outbound["status"] == "REJECTED") arbitrary_outbound["op"] = "send_approval_card" sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) try: sock.connect(str(gateway_socket)); gateway.council_send_frame(sock, arbitrary_outbound) rejected_approval = gateway.council_receive_frame(sock) finally: sock.close() passed("approval_card_operation_rejected_in_a2", rejected_approval["status"] == "REJECTED") arbitrary_outbound["op"] = "send_prefixed_text" arbitrary_outbound["chat_id"] = -999 sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) try: sock.connect(str(gateway_socket)); gateway.council_send_frame(sock, arbitrary_outbound) rejected_outbound = gateway.council_receive_frame(sock) finally: sock.close() passed("controller_cannot_supply_chat_id", rejected_outbound["status"] == "REJECTED") passed("approval_operations_fail_closed_in_a2", policy["approval_callbacks_enabled"] is False) # Independent sessions prove per-session request and message limits. code2 = core.create_enrollment(2, "prompt-A2") body2 = canonical_json({"enrollment_code": code2, "agent_id": "A2", "prompt_id": "prompt-A2"}) status, enrolled2 = _http_request(port, "POST", "/v1/session/enroll", body2) credential2 = SessionCredential(enrolled2["session_id"], "A2", enrolled2["session_key"], enrolled2["expires_at"], 2) for _index in range(60): status, _ = _http_request(port, "GET", manifest_path, credential=credential2, nonce=nonce()) if status != 200: raise AssertionError("request rate pre-limit") status, limited = _http_request(port, "GET", manifest_path, credential=credential2, nonce=nonce()) passed("sixty_requests_per_minute_allowed", True) passed("sixty_first_request_rate_limited", status == 429 and limited["error"] == "RATE_LIMITED") code3 = core.create_enrollment(3, "prompt-A3") body3 = canonical_json({"enrollment_code": code3, "agent_id": "A3", "prompt_id": "prompt-A3"}) status, enrolled3 = _http_request(port, "POST", "/v1/session/enroll", body3) credential3 = SessionCredential(enrolled3["session_id"], "A3", enrolled3["session_key"], enrolled3["expires_at"], 3) status, _ = _http_request(port, "POST", "/v1/room-acks", ack_body, credential=credential3, nonce=nonce()) for index in range(12): body = canonical_json({"room_id": "COMMONS", "data_class": DATA_CLASS, "text": f"rate-{index}", "idempotency_key": f"{index+100:032x}"}) status, _ = _http_request(port, "POST", "/v1/messages", body, credential=credential3, nonce=nonce()) if status != 201: raise AssertionError("message rate pre-limit") body = canonical_json({"room_id": "COMMONS", "data_class": DATA_CLASS, "text": "rate-over", "idempotency_key": "f"*32}) status, limited = _http_request(port, "POST", "/v1/messages", body, credential=credential3, nonce=nonce()) passed("twelve_messages_per_minute_allowed") passed("thirteenth_message_rate_limited", status == 429 and limited["error"] == "RATE_LIMITED") monotonic_value[0] += 60.0 status, _ = _http_request(port, "POST", "/v1/messages", body, credential=credential3, nonce=nonce()) passed("rate_window_expires_after_sixty_seconds", status == 201) from concurrent.futures import ThreadPoolExecutor concurrent_calls = [ (credential1, nonce()), (credential2, nonce()), (credential3, nonce()), ] with ThreadPoolExecutor(max_workers=3) as executor: concurrent_results = list(executor.map( lambda item: _http_request( port, "GET", manifest_path, credential=item[0], nonce=item[1], timestamp=clock(), )[0], concurrent_calls, )) passed("three_agent_concurrent_authenticated_requests", concurrent_results == [200, 200, 200]) bogus_enrollment = canonical_json({ "enrollment_code": "X" * 32, "agent_id": "AX", "prompt_id": "prompt-AX", }) for _index in range(30): status, _ = _http_request(port, "POST", "/v1/session/enroll", bogus_enrollment) if status != 403: raise AssertionError("enrollment rate pre-limit") status, limited = _http_request(port, "POST", "/v1/session/enroll", bogus_enrollment) passed("thirty_enrollment_attempts_per_minute_allowed") passed("thirty_first_enrollment_attempt_rate_limited", status == 429 and limited["error"] == "RATE_LIMITED") # Framing is canonical, bounded and peer UID is mandatory. left, right = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM) try: raw = b'{"z":1, "a":2}' left.sendall(len(raw).to_bytes(4, "big") + raw) try: gateway.council_receive_frame(right) except gateway.GatewayError: passed("noncanonical_unix_json_rejected") else: raise AssertionError("noncanonical frame") finally: left.close(); right.close() left, right = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM) try: left.sendall((gateway.COUNCIL_FRAME_MAX_BYTES + 1).to_bytes(4, "big")) try: gateway.council_receive_frame(right) except gateway.GatewayError: passed("oversize_unix_frame_rejected_before_body") else: raise AssertionError("oversize frame") finally: left.close(); right.close() left, right = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM) try: passed("unix_peer_credentials_match_expected_uid", gateway.council_peer_uid(left) == os.geteuid()) finally: left.close(); right.close() clock.advance(core_policy["session_ttl_seconds"] + 1) status, expired = _http_request( port, "GET", manifest_path, credential=credential1, nonce=nonce(), timestamp=clock(), ) passed("expired_session_generic_rejection", status == 401 and expired == missing) # A1 exact fixture remains valid after the narrow thread-compatibility extension. a1_receipt = run_a1_self_test(core_policy_path, a1_expected_path) passed("a1_exact_seventy_two_test_regression", a1_receipt["status"] == "PASS" and a1_receipt["test_count"] == 72) passed("same_bot_is_only_long_poller", policy["same_bot_only_long_poller"] is True) gateway_source = (core_policy_path.parent / "telegram_gateway.py").read_text() passed("gateway_allowed_updates_remains_message_only_in_a2", '"allowed_updates": ["message"]' in gateway_source) gateway_args = gateway.build_parser().parse_args([ "run", "--config", "c", "--token-file", "t", "--database", "d", "--lock-file", "l", "--health-file", "h", ]) passed("callback_query_polling_not_enabled_in_a2", gateway_args.enable_council_approvals is False) passed("relay_has_no_arbitrary_url_path_or_command_api", not policy["arbitrary_url_api"] and not policy["arbitrary_path_api"] and not policy["arbitrary_command_api"]) passed("stage_one_result_ceiling_provisional", policy["pilot_result"] == "PROVISIONAL_PILOT") passed("target_install_authority_false", policy["target_install_authorized"] is False) last_live_sequence = max(event["sequence"] for event in core.events) finally: server.shutdown() server.server_close() server_thread.join(timeout=2.0) ingress.close() outbound.close() core.close() passed("controller_ingress_socket_removed", not controller_socket.exists()) passed("gateway_outbound_socket_removed", not gateway_socket.exists()) reopened = CouncilCore(state_root, manifests, core_policy, b"R" * 32, clock, tokens) try: passed("restart_clears_all_live_bodies", len(reopened.events) == 0) restarted = reopened.ingest_trusted_gateway_message( sha256(b"restart-gateway-event"), "COMMONS", "restart synthetic", DATA_CLASS, ) passed("restart_keeps_live_cursor_monotonic", restarted["sequence"] > last_live_sequence) finally: reopened.close() passed("temporary_state_removed", not parent.exists()) receipt = { "schema": "AI_COUNCIL_R03_A2_SELF_TEST_V1", "status": "PASS", "test_count": len(tests), "tests": tests, "relay_routes": len(ALLOWED_ROUTES), "listen_address": "127.0.0.1", "http_io_timeout_seconds": 5, "max_concurrent_requests": 16, "enrollment_attempts_per_minute": 30, "requests_per_session_per_minute": 60, "messages_per_session_per_minute": 12, "message_idempotency_required": True, "production_listeners_started": 0, "loopback_ephemeral_test_listener": True, "unix_peer_credentials_required": True, "same_bot_only_long_poller": True, "fake_agent_sessions": 3, "legacy_commands_preserved": 4, "controller_to_gateway_operations_enabled": ["send_prefixed_text"], "approval_callbacks_enabled": False, "allowed_data_class": DATA_CLASS, "message_body_persisted": False, "external_network_calls": 0, "subprocess_calls": 0, "production_writes": 0, "target_ubuntu_changes": 0, "services_changed": 0, "ports_opened": 0, "gateway_runtime_changes": 0, "pilot_result": "PROVISIONAL_PILOT", "temporary_artifacts_after_cleanup": 0, } expected = read_json(expected_path) require(receipt == expected, "self-test receipt mismatch") return receipt def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="AI Council R03-A2 isolated loopback relay") subparsers = parser.add_subparsers(dest="command", required=True) test = subparsers.add_parser("self-test") test.add_argument("--policy", type=Path, required=True) test.add_argument("--core-policy", type=Path, required=True) test.add_argument("--a1-expected", type=Path, required=True) test.add_argument("--expected", type=Path, required=True) test.set_defaults(handler=lambda args: print(json.dumps(run_self_test(args.policy, args.core_policy, args.a1_expected, args.expected), indent=2)) or 0) run = subparsers.add_parser("serve") run.add_argument("--policy", type=Path, required=True) run.add_argument("--core-policy", type=Path, required=True) run.add_argument("--manifest-dir", type=Path, required=True) run.add_argument("--state-root", type=Path, required=True) run.add_argument("--master-key-file", type=Path, required=True) run.add_argument("--enable-council-staging", action="store_true", help="explicitly enable isolated A3 synthetic staging") run.add_argument("--staging-policy", type=Path) run.set_defaults(handler=serve) return parser def main() -> int: os.umask(0o077) parser = build_parser() args = parser.parse_args() try: return int(args.handler(args)) except (RelayError, CouncilError) as error: emit("fatal_error", error_type=type(error).__name__) return 1 except KeyboardInterrupt: emit("interrupted") return 130 except Exception as error: emit("unexpected_fatal_error", error_type=type(error).__name__) return 1 if __name__ == "__main__": raise SystemExit(main())