"""Bridge-Protokoll v1: echter Prozessaufruf wie durch das Plugin (python3 -I + Launcher).""" import hashlib import hmac import json import subprocess import sys from pathlib import Path from aria_router.authorization import KEY_LABEL, Clearance from aria_router.gateway import BRIDGE_REQUEST_FIELDS from tests.helpers import OWNER_ID, GatewayTestCase, config_dict from tests.schema_check import load_schema, validate LAUNCHER = str(Path(__file__).resolve().parents[1] / "bin" / "aria_bridge.py") class BridgeCliTests(GatewayTestCase): def setUp(self): super().setUp() self.config_path = self.root / "gateway.json" self.config_path.write_text(json.dumps(config_dict(self.root)), encoding="utf-8") def run_cli(self, payload, *args, raw: bytes | None = None, config: str | None = None): data = raw if raw is not None else json.dumps(payload).encode("utf-8") proc = subprocess.run( [sys.executable, "-I", LAUNCHER, "--config", config or str(self.config_path), *args], input=data, capture_output=True, timeout=60, env={"PATH": "/usr/bin:/bin"}, cwd="/", ) self.assertEqual(proc.returncode, 0, proc.stderr.decode()) lines = proc.stdout.decode("utf-8").splitlines() self.assertEqual(len(lines), 1) return json.loads(lines[0]) def inbound(self, message_id, text, mode="execute", **extra): payload = {"protocol": 1, "mode": mode, "transport": "telegram", "sender_id": OWNER_ID, "sender_username": "kais_test", "chat_id": OWNER_ID, "chat_type": "private", "message_id": message_id, "ts": "2026-10-24T18:00:00.000Z", "text": text} payload.update(extra) self.assertEqual(validate(payload, "inbound_request"), []) return payload def assertResponse(self, response): self.assertEqual(validate(response, "response"), []) self.assertFalse(response["delivered"]) def test_roundtrip_execute_codex(self): response = self.run_cli(self.inbound(1, "#public hallo")) self.assertResponse(response) self.assertEqual((response["handling"], response["backend"]), ("reply", "codex")) # Default ohne real_execution: MockRunner, kein Prozessstart. self.assertEqual(response["reply_text"], "MOCK_CODEX_RESPONSE") def test_roundtrip_forward_claude(self): response = self.run_cli(self.inbound(2, "hallo")) self.assertResponse(response) self.assertEqual(response["handling"], "forward_to_claude_channel") forward = response["forward"] self.assertEqual(forward["method"], "notifications/claude/channel") self.assertEqual(forward["params"]["content"], "hallo") self.assertEqual(forward["params"]["meta"]["user_id"], OWNER_ID) self.assertEqual(forward["clearance"]["backend"], "claude") def test_roundtrip_shadow(self): response = self.run_cli(self.inbound(3, "#public hallo", mode="shadow")) self.assertResponse(response) self.assertEqual((response["handling"], response["planned_handling"]), ("shadow_decision", "reply")) def test_unknown_field_rejected_and_logged(self): payload = self.inbound(4, "hallo") payload["user_id"] = OWNER_ID response = self.run_cli(payload) self.assertEqual((response["handling"], response["code"]), ("error", "UNKNOWN_FIELD")) log = [json.loads(line) for line in (self.root / "state" / "decisions.jsonl").read_text().splitlines()] self.assertEqual(log[-1]["rejected_at"], "bridge") def test_invalid_json(self): self.assertEqual(self.run_cli(None, raw=b"{kaputt")["code"], "INVALID_JSON") def test_input_limit(self): self.assertEqual(self.run_cli(None, raw=b" " * (1024 * 1024 + 1))["code"], "INPUT_LIMIT") def test_unsupported_protocol(self): payload = self.inbound(5, "hallo") payload["protocol"] = 2 self.assertEqual(self.run_cli(payload)["code"], "UNSUPPORTED_PROTOCOL") def test_config_unavailable(self): response = self.run_cli(self.inbound(6, "hallo"), config=str(self.root / "missing.json")) self.assertEqual((response["handling"], response["code"]), ("error", "CONFIG_UNAVAILABLE")) def test_status_command_without_secrets(self): status = self.run_cli(None, "--status", raw=b"") self.assertTrue(status["key_available"]) self.assertEqual(set(status["backends"]), {"codex", "claude"}) key_hex = (self.root / "keys" / "clearance.key").read_bytes().hex() self.assertNotIn(key_hex[:16], json.dumps(status)) def test_delivery_report(self): forward = self.run_cli(self.inbound(7, "hallo"))["forward"] report = {"protocol": 1, "type": "delivery_report", "job_id": forward["job_id"], "nonce": forward["clearance"]["nonce"], "ok": False} self.assertEqual(validate(report, "delivery_report"), []) ack = self.run_cli(report) self.assertEqual(validate(ack, "report_ack"), []) self.assertEqual(ack["code"], "REPORTED") unknown = dict(report, nonce="0" * 32) self.assertEqual(self.run_cli(unknown)["code"], "UNKNOWN_CLEARANCE") def test_text_with_many_quotes(self): response = self.run_cli(self.inbound(8, '"' * 65536)) self.assertResponse(response) self.assertNotEqual(response["handling"], "error") def test_schema_matches_implementation(self): schema = load_schema()["$defs"]["inbound_request"] self.assertEqual(set(schema["properties"]) - {"protocol", "mode", "type"}, set(BRIDGE_REQUEST_FIELDS)) self.assertEqual(set(schema["required"]) - {"protocol", "mode"}, {"transport", "sender_id", "chat_id", "chat_type", "message_id", "text"}) def test_plugin_side_verification_with_derived_key(self): forward = self.run_cli(self.inbound(9, "hallo"))["forward"] clearance = Clearance.from_dict(forward["clearance"]) master = (self.root / "keys" / "clearance.key").read_bytes() claude_key = hmac.new(master, KEY_LABEL + b"claude", hashlib.sha256).digest() codex_key = hmac.new(master, KEY_LABEL + b"codex", hashlib.sha256).digest() payload = clearance.signing_payload() self.assertEqual(hmac.new(claude_key, payload, hashlib.sha256).hexdigest(), clearance.mac) self.assertNotEqual(hmac.new(codex_key, payload, hashlib.sha256).hexdigest(), clearance.mac)