"""HMAC-Clearance: Replay, Ablauf, Backend-Bindung, Manipulation, Schlüsseldatei.""" import hashlib import hmac import os import tempfile import unittest from dataclasses import replace from pathlib import Path from aria_router.adapter_contract import ResultStatus from aria_router.adapters import MockAdapter from aria_router.authorization import ( KEY_LABEL, AuthorizationError, ClearanceIssuer, ClearanceVerifier, KeyUnavailableError, ) from aria_router.core import PrivacyMatrix, PrivacyTier from aria_router.gateway import Handling from aria_router.state import NonceStore from tests.helpers import AdapterHarness, GatewayTestCase, write_key class ClearanceTests(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.root = Path(self._tmp.name) self.h = AdapterHarness(self.root) self.codex = MockAdapter(self.h.verifier, "codex") self.claude = MockAdapter(self.h.verifier, "claude") self.job = self.h.job("#public hallo") def tearDown(self): self._tmp.cleanup() def assertBlocked(self, result, code, adapter=None): self.assertEqual(result.status, ResultStatus.BLOCKED) self.assertEqual(result.error_code, code) self.assertEqual((adapter or self.codex).calls, 0) def test_valid_clearance_executes_once(self): result = self.codex.execute(self.h.clear(self.codex, self.job), self.job) self.assertEqual(result.status, ResultStatus.SUCCESS) self.assertEqual(self.codex.calls, 1) def test_replay_rejected(self): clearance = self.h.clear(self.codex, self.job) self.assertEqual(self.codex.execute(clearance, self.job).status, ResultStatus.SUCCESS) second = self.codex.execute(clearance, self.job) self.assertEqual(second.error_code, "REPLAY") self.assertEqual(self.codex.calls, 1) def test_replay_rejected_after_restart(self): clearance = self.h.clear(self.codex, self.job) self.codex.execute(clearance, self.job) # Neuer Prozess: neuer Verifier, neuer Nonce-Speicher auf derselben Datei. restarted = ClearanceVerifier(self.h.keyring, self.h.matrix, NonceStore(self.root / "state" / "nonces-clearance.key.json"), self.h.classifier, clock=self.h.clock) adapter = MockAdapter(restarted, "codex") self.assertBlocked(adapter.execute(clearance, self.job), "REPLAY", adapter) def test_expired_clearance(self): clearance = self.h.clear(self.codex, self.job) self.h.clock.advance(121) self.assertBlocked(self.codex.execute(clearance, self.job), "EXPIRED") def test_expiry_boundary_still_valid(self): clearance = self.h.clear(self.codex, self.job) self.h.clock.advance(119) self.assertEqual(self.codex.execute(clearance, self.job).status, ResultStatus.SUCCESS) def test_not_yet_valid(self): future = ClearanceIssuer(self.h.keyring, self.h.matrix, clock=lambda: self.h.clock.now + 60) clearance = future.issue(job_id="job-1", backend="codex", privacy=PrivacyTier.PUBLIC, context_digest=self.codex.outbound_digest(self.job), sender_id=self.job.sender_id, chat_id=self.job.chat_id) self.assertBlocked(self.codex.execute(clearance, self.job), "NOT_YET_VALID") def test_lifetime_too_long(self): long_issuer = ClearanceIssuer(self.h.keyring, self.h.matrix, ttl_seconds=600, clock=self.h.clock) strict = ClearanceVerifier(self.h.keyring, self.h.matrix, self.h.nonces, self.h.classifier, max_ttl_seconds=120, clock=self.h.clock) adapter = MockAdapter(strict, "codex") clearance = long_issuer.issue(job_id="job-1", backend="codex", privacy=PrivacyTier.PUBLIC, context_digest=adapter.outbound_digest(self.job), sender_id=self.job.sender_id, chat_id=self.job.chat_id) self.assertBlocked(adapter.execute(clearance, self.job), "INVALID_LIFETIME", adapter) def test_clearance_for_other_backend(self): clearance = self.h.clear(self.claude, self.job) self.assertBlocked(self.codex.execute(clearance, self.job), "BACKEND_MISMATCH") def test_tampered_backend_field(self): clearance = replace(self.h.clear(self.codex, self.job), backend="claude") self.assertBlocked(self.claude.execute(clearance, self.job), "INVALID_SIGNATURE", self.claude) def test_tampered_privacy_field(self): job = self.h.job("hallo") # UNKNOWN clearance = self.h.clear(self.claude, job) forged = replace(clearance, privacy="public") self.assertBlocked(self.claude.execute(forged, job), "INVALID_SIGNATURE", self.claude) def test_tampered_expiry(self): clearance = replace(self.h.clear(self.codex, self.job), expires_at=10 ** 12) self.assertBlocked(self.codex.execute(clearance, self.job), "INVALID_SIGNATURE") def test_subject_mismatch(self): clearance = self.h.clear(self.codex, self.job, chat_id="555") self.assertBlocked(self.codex.execute(clearance, self.job), "SUBJECT_MISMATCH") def test_job_mismatch(self): clearance = self.h.clear(self.codex, self.job, job_id="job-2") self.assertBlocked(self.codex.execute(clearance, self.job), "JOB_MISMATCH") def test_privacy_mismatch(self): clearance = self.h.clear(self.claude, self.job, privacy=PrivacyTier.UNKNOWN) self.assertBlocked(self.claude.execute(clearance, self.job), "PRIVACY_MISMATCH", self.claude) def test_adapter_rechecks_matrix(self): # Gültig signiert, aber der Verifier des Adapters lässt PUBLIC für Codex nicht zu. strict_matrix = PrivacyMatrix.from_mapping({"codex": set(), "claude": {PrivacyTier.PUBLIC}}) strict = ClearanceVerifier(self.h.keyring, strict_matrix, self.h.nonces, self.h.classifier, clock=self.h.clock) adapter = MockAdapter(strict, "codex") clearance = self.h.clear(adapter, self.job) self.assertBlocked(adapter.execute(clearance, self.job), "PRIVACY_NOT_PERMITTED", adapter) def test_issuer_refuses_unpermitted_privacy(self): with self.assertRaises(AuthorizationError) as ctx: self.h.issuer.issue(job_id="j", backend="codex", privacy=PrivacyTier.UNKNOWN, context_digest="c" * 64, sender_id="1", chat_id="1") self.assertEqual(ctx.exception.code, "PRIVACY_NOT_PERMITTED") def test_missing_key_fail_closed(self): clearance = self.h.clear(self.codex, self.job) os.unlink(self.h.key_file) self.assertBlocked(self.codex.execute(clearance, self.job), "KEY_UNAVAILABLE") with self.assertRaises(KeyUnavailableError): self.h.clear(self.codex, self.job) def test_key_permissions_too_open(self): clearance = self.h.clear(self.codex, self.job) os.chmod(self.h.key_file, 0o644) self.assertBlocked(self.codex.execute(clearance, self.job), "KEY_PERMISSIONS_TOO_OPEN") def test_key_too_short(self): write_key(self.h.key_file, size=16) with self.assertRaises(KeyUnavailableError) as ctx: self.h.keyring.key_for("codex") self.assertEqual(ctx.exception.code, "KEY_TOO_SHORT") def test_key_symlink_rejected(self): link = self.root / "keys" / "link.key" os.symlink(self.h.key_file, link) harness = AdapterHarness(self.root / "other") harness.keyring.key_file = link with self.assertRaises(KeyUnavailableError) as ctx: harness.keyring.key_for("codex") self.assertEqual(ctx.exception.code, "KEY_UNAVAILABLE") def test_key_derivation_per_backend(self): self.assertNotEqual(self.h.keyring.key_for("codex"), self.h.keyring.key_for("claude")) master = self.h.key_file.read_bytes() expected = hmac.new(master, KEY_LABEL + b"claude", hashlib.sha256).digest() self.assertEqual(self.h.keyring.key_for("claude"), expected) def test_dict_clearance_roundtrip(self): clearance = self.h.clear(self.codex, self.job).to_dict() self.assertEqual(self.codex.execute(clearance, self.job).status, ResultStatus.SUCCESS) def test_malformed_clearance(self): good = self.h.clear(self.codex, self.job).to_dict() missing = {k: v for k, v in good.items() if k != "nonce"} extra = {**good, "approved_by": "trusted-policy-engine"} wrong_type = {**good, "issued_at": str(good["issued_at"])} for bad in (missing, extra, wrong_type): self.assertBlocked(self.codex.execute(bad, self.job), "MALFORMED_CLEARANCE") def test_garbage_clearance(self): self.assertBlocked(self.codex.execute("approved", self.job), "INVALID_CLEARANCE") class GatewayKeyTests(GatewayTestCase): def test_gateway_missing_key_fail_closed(self): gateway = self.gateway() os.unlink(self.root / "keys" / "clearance.key") result = gateway.handle(self.request(1, "#public hallo")) self.assertEqual(result.handling, Handling.ERROR) self.assertEqual(result.code, "KEY_UNAVAILABLE") self.assertEqual(gateway.adapters["codex"].runner.calls, []) self.assertEqual(gateway.status()["backends"]["codex"]["budget"]["jobs"], 0) self.assertFalse(gateway.status()["key_available"]) if __name__ == "__main__": unittest.main()