"""Datenschutz: deterministischer Klassifikator, Policy-Matrix, manuelle Wahl gegen Datenschutz.""" import secrets import string import tempfile import unittest from pathlib import Path from aria_router.classify import PrivacyClassifier from aria_router.config import ConfigurationError, parse_gateway_config from aria_router.context import Attachment, ContextEnvelope, MemorySnippet from aria_router.core import PrivacyTier from aria_router.gateway import Handling from tests.helpers import GatewayTestCase, config_dict, document, photo def fake_secret(prefix: str, length: int, alphabet: str = string.ascii_letters + string.digits) -> str: return prefix + "".join(secrets.choice(alphabet) for _ in range(length)) class ClassifierTests(unittest.TestCase): def setUp(self): self.c = PrivacyClassifier() def assess(self, text, **kwargs): return self.c.classify(ContextEnvelope(text, **kwargs)) def tier(self, text, **kwargs): return self.assess(text, **kwargs).tier def test_default_unknown(self): self.assertEqual(self.tier("Wie spät ist es?"), PrivacyTier.UNKNOWN) self.assertEqual(self.tier(""), PrivacyTier.UNKNOWN) def test_public_marker(self): self.assertEqual(self.tier("#public Was ist ein Monoid?"), PrivacyTier.PUBLIC) self.assertEqual(self.tier(" #öffentlich Erkläre Rekursion"), PrivacyTier.PUBLIC) self.assertEqual(self.tier("[public] Erkläre Rekursion"), PrivacyTier.PUBLIC) def test_public_marker_only_at_start(self): self.assertEqual(self.tier("Erkläre Rekursion #public"), PrivacyTier.UNKNOWN) self.assertEqual(self.tier("public Erkläre Rekursion"), PrivacyTier.UNKNOWN) self.assertEqual(self.tier("#publicity Erkläre"), PrivacyTier.UNKNOWN) def test_marker_can_be_disabled(self): classifier = PrivacyClassifier(public_marker_enabled=False) self.assertEqual(classifier.classify(ContextEnvelope("#public x")).tier, PrivacyTier.UNKNOWN) def test_confidential_beats_marker(self): assessment = self.assess("#public Angebot für BMW") self.assertEqual(assessment.tier, PrivacyTier.CONFIDENTIAL) self.assertIn("customer.bmw", assessment.rule_ids) def test_secret_patterns(self): cases = { "secret.telegram_bot_token": fake_secret("".join(secrets.choice(string.digits) for _ in range(9)) + ":", 35, string.ascii_letters + string.digits + "_-"), "secret.openai_key": fake_secret("sk-", 32), "secret.anthropic_key": fake_secret("sk-ant-", 32), "secret.github_token": fake_secret("ghp_", 36), "secret.aws_access_key": fake_secret("AKIA", 16, string.ascii_uppercase + string.digits), "secret.jwt": fake_secret("eyJ", 20) + "." + fake_secret("eyJ", 20) + "." + fake_secret("", 20), "secret.private_key_block": "-----BEGIN " + "PRIVATE KEY-----\nMIIB\n-----END " + "PRIVATE KEY-----", "secret.assignment": "password: " + fake_secret("", 12), "secret.supabase_key": fake_secret("sb_secret_", 24), } for rule, value in cases.items(): with self.subTest(rule=rule): assessment = self.assess(f"#public bitte prüfen {value}") self.assertEqual(assessment.tier, PrivacyTier.CONFIDENTIAL) self.assertIn(rule, assessment.rule_ids) def test_telegram_token_edge_forms(self): # Regression: Token mit "-" am Ende und Token in der Bot-API-URL wurden nicht erkannt. body = "A" * 34 for text in (f"123456789:{body}-", f"https://api.telegram.org/bot123456789:{body}-/getMe", f"https://api.telegram.org/file/bot1234567890:{body}_/doc.pdf"): with self.subTest(text=text): assessment = self.assess("#public " + text) self.assertEqual(assessment.tier, PrivacyTier.CONFIDENTIAL) self.assertIn("secret.telegram_bot_token", assessment.rule_ids) # Gegenprobe: zu kurz bzw. zu viele Ziffern ist kein Token. self.assertNotIn("secret.telegram_bot_token", self.assess(f"#public 123456789:{body}").rule_ids) self.assertNotIn("secret.telegram_bot_token", self.assess(f"#public 12345678901:{body}-").rule_ids) def test_customer_keywords(self): for text, rule in (("Angebot BMW", "customer.bmw"), ("Meeting mit adesso", "customer.adesso"), ("SupplierPulse Backlog", "customer.supplierpulse"), ("Kundendaten exportieren", "customer.kundendaten"), ("Mandantenliste", "customer.mandant"), ("streng vertraulich", "confidential.marker")): with self.subTest(text=text): assessment = self.assess("#public " + text) self.assertEqual(assessment.tier, PrivacyTier.CONFIDENTIAL) self.assertIn(rule, assessment.rule_ids) def test_personal_data(self): for text, rule in (("schreib an max.muster@example.org", "pii.email"), ("ruf +49 170 1234567 an", "pii.phone"), ("IBAN DE89 3704 0044 0532 0130 00", "pii.iban"), ("Geburtsdatum eintragen", "pii.keyword")): with self.subTest(text=text): assessment = self.assess("#public " + text) self.assertEqual(assessment.tier, PrivacyTier.CONFIDENTIAL) self.assertIn(rule, assessment.rule_ids) def test_internal_markers(self): self.assertEqual(self.tier("Log unter /home/aria/scripts prüfen"), PrivacyTier.INTERNAL) self.assertEqual(self.tier("#public Datei /root/x"), PrivacyTier.INTERNAL) self.assertEqual(self.tier("das ist intern"), PrivacyTier.INTERNAL) def test_attachments_never_public(self): self.assertEqual(self.tier("#public Bild", attachments=(photo(),)), PrivacyTier.UNKNOWN) clean = Attachment(kind="document", file_id="x", name="x.txt", data=b"hello world") self.assertEqual(self.tier("#public Text", attachments=(clean,)), PrivacyTier.UNKNOWN) def test_attachment_name_confidential(self): self.assertEqual(self.tier("#public Datei", attachments=(document("BMW_Angebot.pdf"),)), PrivacyTier.CONFIDENTIAL) def test_identifiers_with_underscores_and_digits(self): # Regression: \b behandelte "_" als Wortzeichen, BMW_Angebot entging der Regel. for text in ("BMW_Angebot.pdf", "Kundendaten_2026.xlsx", "Supplier_Pulse", "adesso2026", "customer_data"): with self.subTest(text=text): self.assertEqual(self.tier("#public " + text), PrivacyTier.CONFIDENTIAL) for text in ("Urkunde", "Kundgebung", "internationale Lage"): with self.subTest(text=text): self.assertEqual(self.tier("#public " + text), PrivacyTier.PUBLIC) def test_attachment_text_content_scanned(self): secret_doc = Attachment(kind="document", file_id="x", name="n.txt", data=("password=" + fake_secret("", 12)).encode()) self.assertEqual(self.tier("#public Datei", attachments=(secret_doc,)), PrivacyTier.CONFIDENTIAL) def test_memory_classified(self): self.assertEqual(self.tier("#public x", memory=(MemorySnippet("brain/a", "neutral"),)), PrivacyTier.UNKNOWN) self.assertEqual(self.tier("#public x", memory=(MemorySnippet("brain/a", "neutral", PrivacyTier.PUBLIC),)), PrivacyTier.PUBLIC) self.assertEqual(self.tier("#public x", memory=(MemorySnippet("brain/a", "BMW Termin", PrivacyTier.PUBLIC),)), PrivacyTier.CONFIDENTIAL) self.assertEqual(self.tier("#public x", memory=(MemorySnippet("brain/a", "neutral", PrivacyTier.CONFIDENTIAL),)), PrivacyTier.CONFIDENTIAL) def test_memory_source_label_classified(self): # Review #16: ein vertraulicher QUELLNAME (wird per render_prompt uebertragen) muss den # Ausschnitt vertraulich machen, auch wenn Text und declared_tier harmlos sind. self.assertEqual( self.tier("#public x", memory=(MemorySnippet("BMW-Geheimprojekt", "Guten Morgen", PrivacyTier.PUBLIC),)), PrivacyTier.CONFIDENTIAL) # harmlose Quelle bleibt bei der deklarierten Stufe self.assertEqual( self.tier("#public x", memory=(MemorySnippet("brain/tagesplan.md", "Guten Morgen", PrivacyTier.PUBLIC),)), PrivacyTier.PUBLIC) def test_extra_keywords(self): classifier = PrivacyClassifier(extra_confidential_keywords=["Projekt Falke"]) self.assertEqual(classifier.classify(ContextEnvelope("#public Projekt Falke")).tier, PrivacyTier.CONFIDENTIAL) def test_deterministic(self): text = "#public Bitte debug mein Python" self.assertEqual(self.assess(text), self.assess(text)) class MatrixPolicyTests(GatewayTestCase): def test_default_matrix_codex_only_public(self): gateway = self.gateway() self.assertEqual(gateway.handle(self.request(1, "#public Monoid?")).backend, "codex") self.assertEqual(gateway.handle(self.request(2, "Monoid?")).backend, "claude") self.assertEqual(gateway.adapters["codex"].runner.calls.__len__(), 1) def test_default_matrix_claude_status_quo(self): gateway = self.gateway() for message_id, text in ((3, "hallo"), (4, "Pfad /home/aria"), (5, "Angebot BMW")): result = gateway.handle(self.request(message_id, text)) self.assertEqual(result.handling, Handling.FORWARD_TO_CLAUDE_CHANNEL, text) self.assertEqual(len(self.sink.delivered), 3) def test_status_quo_switch_off_blocks_confidential(self): gateway = self.gateway({"privacy": {"claude_status_quo_confidential": False}}) blocked = gateway.handle(self.request(6, "Angebot BMW")) self.assertEqual((blocked.handling, blocked.code), (Handling.BLOCKED, "PRIVACY_DENIED")) self.assertEqual(gateway.handle(self.request(7, "hallo")).backend, "claude") def test_manual_codex_against_privacy_blocked_and_explained(self): gateway = self.gateway() gateway.handle(self.request(8, "/codex")) result = gateway.handle(self.request(9, "Angebot für BMW")) self.assertEqual((result.handling, result.code), (Handling.BLOCKED, "MANUAL_CHOICE_NOT_PERMITTED")) self.assertIn("/codex", result.reply_text) self.assertIn("confidential", result.reply_text) self.assertIn("customer.bmw", result.reply_text) self.assertEqual(self.sink.delivered, []) self.assertEqual(gateway.adapters["codex"].runner.calls, []) def test_one_shot_codex_against_privacy(self): gateway = self.gateway() result = gateway.handle(self.request(10, "/codex Angebot für BMW")) self.assertEqual(result.code, "MANUAL_CHOICE_NOT_PERMITTED") self.assertEqual(gateway.adapters["codex"].runner.calls, []) def test_manual_mode_never_reaches_codex_with_unknown(self): gateway = self.gateway() gateway.handle(self.request(11, "/codex")) result = gateway.handle(self.request(12, "hallo")) self.assertEqual(result.code, "MANUAL_CHOICE_NOT_PERMITTED") self.assertEqual(self.sink.delivered, []) def test_checked_fallback_to_claude_when_enabled(self): gateway = self.gateway({"routing": {"manual_fallback": "checked_fallback"}}) gateway.handle(self.request(13, "/codex")) result = gateway.handle(self.request(14, "Angebot für BMW")) self.assertEqual(result.handling, Handling.FORWARD_TO_CLAUDE_CHANNEL) self.assertIn("CHECKED_FALLBACK", result.flags) self.assertIn("vollständiger Prüfung", result.reply_text) record = [r for r in self.log_records(gateway) if r.get("job_id") == result.job_id][0] self.assertEqual(record["routing"]["code"], "CHECKED_FALLBACK") self.assertEqual(record["routing"]["fallback_from"], "codex") def test_checked_fallback_requires_claude_policy(self): gateway = self.gateway({"routing": {"manual_fallback": "checked_fallback"}, "privacy": {"claude_status_quo_confidential": False}}) result = gateway.handle(self.request(15, "/codex Angebot für BMW")) self.assertEqual((result.handling, result.code), (Handling.BLOCKED, "MANUAL_CHOICE_NOT_PERMITTED")) self.assertEqual(self.sink.delivered, []) def test_checked_fallback_respects_claude_health(self): gateway = self.gateway({"routing": {"manual_fallback": "checked_fallback"}}) for _ in range(3): gateway._health.record_failure("claude", "DELIVERY_FAILED") result = gateway.handle(self.request(16, "/codex Angebot für BMW")) self.assertEqual(result.handling, Handling.BLOCKED) self.assertEqual(self.sink.delivered, []) def test_config_rejects_confidential_in_matrix(self): with tempfile.TemporaryDirectory() as tmp: for backend in ("codex", "claude"): with self.subTest(backend=backend), self.assertRaises(ConfigurationError): parse_gateway_config(config_dict(Path(tmp), {"privacy": {"matrix": { backend: ["public", "confidential"]}}})) if __name__ == "__main__": unittest.main()