"""Detection + mask + demask. Run from backend/: python tests/test_privacy_detect.py""" from __future__ import annotations import os import sys import tempfile from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) os.environ["KANSHO_DB_PATH"] = str(Path(tempfile.gettempdir()) / "kansho-privacy-detect-test.sqlite") os.environ["KANSHO_FAKE_PROVIDER"] = "1" os.environ["KANSHO_FAKE_DETECT"] = "1" Path(os.environ["KANSHO_DB_PATH"]).unlink(missing_ok=True) from fastapi.testclient import TestClient from identity_store import list_mappings from main import app from privacy_gateway import reset_debug def expect(ok: bool, message: str) -> None: if not ok: raise SystemExit(f"FAIL: {message}") print(f"OK {message}") def main() -> None: reset_debug() with TestClient(app) as client: setup = client.post( "/api/auth/setup", json={"email": "lars@example.test", "name": "Lars", "password": "test-pass"}, ) headers = {"X-Auth-Token": setup.json()["token"]} space = client.post("/api/journal/spaces", headers=headers, json={"title": "Alltag"}) day = client.post( f"/api/journal/spaces/{space.json()['id']}/days", headers=headers, json={"calendar_date": "2026-08-22"}, ) conv = client.post( f"/api/journal/days/{day.json()['day']['id']}/conversations", headers=headers, json={"title": "Gespräch"}, ) turn = client.post( f"/api/journal/conversations/{conv.json()['id']}/turn", headers=headers, json={"body": "Ich war mit Anna in Hamburg und danach allein."}, ) expect(turn.status_code == 200, f"turn {turn.text}") from privacy_gateway import debug_last expect(debug_last is not None, "gateway recorded request") expect(debug_last["detect_provider"] == "fake-detect", "detect provider is separate") expect("Anna" not in debug_last["masked"], "Anna masked before generate") expect("Hamburg" not in debug_last["masked"], "Hamburg masked before generate") expect("[[PERSON:" in debug_last["masked"] or "[[PLACE:" in debug_last["masked"], "placeholders in egress") expect("Anna" in turn.json()["user"]["body"], "source keeps cleartext") mappings = list_mappings(setup.json()["profile_id"]) labels = {item["local_label"] for item in mappings} expect("Anna" in labels, "Anna stored in local dictionary") expect("Hamburg" in labels, "Hamburg stored in local dictionary") expect(all(not (item.get("token") or "").startswith("[[") or True for item in mappings), "tokens stored") reset_debug() again = client.post( f"/api/journal/conversations/{conv.json()['id']}/turn", headers=headers, json={"body": "Anna war ruhig."}, ) expect(again.status_code == 200, "second turn") from privacy_gateway import debug_last as last2 expect("nicht" in last2["masked"].lower() or "Anna" not in last2["masked"], "dictionary remasks Anna without new invention") expect(last2["mapping_count"] >= 2, "dictionary reused") from entity_detect import _pattern_entities from identity_store import is_maskable_label from privacy_gateway import _mask patterned = {item["text"] for item in _pattern_entities("Ich war mit Anna in Hamburg.", set())} expect("Anna" in patterned and "Hamburg" in patterned, "pattern finds proper names") expect("Himmel" not in {item["text"] for item in _pattern_entities("Der Himmel war klar.", set())}, "pattern skips weather") expect(not is_maskable_label("ich"), "pronoun ich is not a mapping") expect(not is_maskable_label("user"), "role user is not a mapping") expect(not is_maskable_label("Himmel"), "weather is not identity") expect(not is_maskable_label("Der Himmel"), "determiner plus weather is not identity") expect(not is_maskable_label("6:00 Uhr"), "clock time is not identity") expect(not is_maskable_label("meinen Tee"), "tea is not identity") expect(not is_maskable_label("Relativ"), "sentence adverb is not identity") expect(not is_maskable_label("Allerdings"), "conjunction is not identity") expect(not is_maskable_label("Tages"), "common noun is not identity") expect(not is_maskable_label("Brote"), "food is not identity") expect(not is_maskable_label("Überlegen"), "common noun is not identity") expect(not is_maskable_label("Ortes"), "place word is not identity") expect(not is_maskable_label("Zwiebeln"), "food is not identity") expect(not is_maskable_label("letzten"), "function word letzten is not identity") expect(not is_maskable_label("Getöse"), "collective noun is not identity") expect(not is_maskable_label("Schiff"), "vessel is not identity") expect(not is_maskable_label("Faden"), "prompt word is not identity") expect("Getöse" not in {item["text"] for item in _pattern_entities("Zweimaster mit Getöse im Hafen.", set())}, "mit plus common noun is not a name") expect("Anna" in {item["text"] for item in _pattern_entities("Ich war mit Anna im Hafen.", set())}, "mit plus given name stays") expect(is_maskable_label("Sushi"), "given name stays maskable") expect(is_maskable_label("meine Frau Sushi"), "kinship stripped, name remains") from identity_store import coalesce_mappings, normalize_token, remember_mapping from db import get_db import uuid expect(normalize_token("PERSON:KURZ") is None, "template token KURZ rejected") first = remember_mapping(setup.json()["profile_id"], "meine Frau Sushi", "PERSON:08") second = remember_mapping(setup.json()["profile_id"], "Sushi", "PERSON:KURZ") expect(first["token"] == second["token"], "alias reuses token") expect(second["local_label"] == "Sushi", "canonical label is the name") extra_id = str(uuid.uuid4()) with get_db() as conn: conn.execute( """ INSERT INTO identity_mappings (id, profile_id, token, local_label) VALUES (?, ?, ?, ?) """, (extra_id, setup.json()["profile_id"], "PERSON:KURZ", "meine Frau Sushi"), ) coalesce_mappings(setup.json()["profile_id"]) sushi_rows = [ item for item in list_mappings(setup.json()["profile_id"]) if (item.get("local_label") or "").lower() == "sushi" ] expect(len(sushi_rows) == 1, "coalesce keeps one Sushi row") expect(sushi_rows[0]["token"].upper() != "PERSON:KURZ", "stable token wins over KURZ") joined = _mask( "meine Frau Sushi kam. Sushi stellte fest.", sushi_rows, ) expect(joined.count(f"[[{sushi_rows[0]['token']}]]") == 2, "both mentions share one placeholder") expect("Sushi" not in joined, "name fully masked") expect("meine Frau " in joined, "kinship stays visible") homonym = _mask( "Ich esse gerne Sushi mit meiner Frau Sushi.", sushi_rows, ) expect("esse gerne Sushi" in homonym, "food homonym stays") expect("Frau [[" in homonym or f"Frau [[{sushi_rows[0]['token']}]]" in homonym, "kinship name is masked") expect(homonym.count("[[") == 1, "only the person mention is masked") eating = _mask("Heute nur von Sushi essen erzählt.", sushi_rows) expect("Sushi essen" in eating, "noun before eat stays food") expect("[[" not in eating, "food-only sentence is not the person") from pronoun_bind import bind_user_lines, bind_pronouns from privacy_gateway import mask_for_egress anna = [{"local_label": "Anna", "token": "PERSON:01"}] bound = bind_user_lines( _mask( "user: Ich war mit Anna.\nuser: Sie setzte sich zu mir, und ich gab ihr Tee.", anna, ) ) expect(bound.count("[[PERSON:01]]") == 3, "name and both pronouns share one token") expect("Sie" not in bound and "ihr" not in bound, "pronouns rebound to person") journal_masked = mask_for_egress( "user: Ich war mit Anna.\nuser: Sie setzte sich zu mir, und ich gab ihr Tee.", anna, "journal_generate", ) expect(journal_masked.count("[[PERSON:01]]") == 1, "journal masks the name only") expect("Sie setzte" in journal_masked and "ihr Tee" in journal_masked, "journal keeps anaphora") dialogue_masked = mask_for_egress( "user: Ich war mit Anna.\nuser: Sie setzte sich zu mir, und ich gab ihr Tee.", anna, "dialogue_turn", ) expect(dialogue_masked.count("[[PERSON:01]]") == 3, "dialogue still binds pronouns") instruction = bind_user_lines( "user: Anna kam.\n" "Der Impuls darf eine Frage sein. Er darf keine neue Station setzen.\n" ) expect("Er darf keine neue Station" in instruction, "instruction pronoun stays") instruction_name = _mask( "Nenne Berg nicht in der Anweisung.\nuser: Berg kam vorbei.", [{"local_label": "Berg", "token": "PERSON:02"}], personal_lines_only=True, ) expect("Nenne Berg nicht" in instruction_name, "instruction line is not remasked") expect("[[PERSON:02]]" in instruction_name, "user line still remasked") herr = [{"local_label": "Berg", "token": "PERSON:02"}] male, _ = bind_pronouns(_mask("Herr Berg kam. Er setzte sich.", herr)) expect(male.count("[[PERSON:02]]") == 2, "masculine pronoun follows last man") address, _ = bind_pronouns(_mask("Anna kam. Seid ihr schon da?", anna)) expect("ihr" in address, "address ihr is not a person") plural, _ = bind_pronouns(_mask("Anna kam. Sie waren zu dritt.", anna)) expect("Sie waren" in plural or "sie waren" in plural.lower(), "plural sie stays") croatia = ( "Heute bin ich ca. gegen 6:00 Uhr aufgestanden. Relativ früh gesellte sich " "auch meine Frau Sushi zu mir. Allerdings sollte die Bootstour erst um 11:00 Uhr starten." ) patterned_live = {item["text"] for item in _pattern_entities(croatia, set())} expect("Sushi" in patterned_live, "pattern keeps given name") expect( patterned_live <= {"Sushi"}, f"pattern does not treat German function words as names: {patterned_live}", ) from entity_detect import uses_llm_detect from providers import ProviderConfig remote = ProviderConfig( role="detect", name="openrouter", mode="http", url="https://openrouter.ai/api/v1/chat/completions", model="x", key="sk-test", local=False, zdr=True, no_train=True, ) expect(uses_llm_detect(remote), "development allows remote detect") os.environ["KANSHO_ENV"] = "production" expect(not uses_llm_detect(remote), "production blocks remote detect") local = ProviderConfig( role="detect", name="ollama", mode="http", url="http://127.0.0.1:11434/v1/chat/completions", model="x", key="", local=True, zdr=True, no_train=True, ) expect(uses_llm_detect(local), "production still allows local detect") os.environ.pop("KANSHO_ENV", None) poisoned = _mask( "Nicht erwähnt ist nicht geschehen. Ich kochte Tee.", [{"local_label": "ich", "token": "PERSON:05"}], ) expect("N[[PERSON:05]]t" not in poisoned, "ich does not split nicht") expect("Nicht erwähnt" in poisoned, "instruction word bleibt") from privacy_gateway import PrivacyGatewayError, _demask, _validate_response mapped = [{"local_label": "Anna", "token": "PERSON:01"}] kept = _validate_response("[[PERSON:01]] war ruhig.", mapped) expect(kept.startswith("[[PERSON:01]]"), "placeholders pass validation") expect(_demask(kept, mapped) == "Anna war ruhig.", "demask after validation") blocked = False try: _validate_response("Anna war ruhig.", mapped) except PrivacyGatewayError as exc: blocked = exc.code == "response_validation_failed" expect(blocked, "plaintext identity is blocked before demask") print("All privacy detect tests passed.") if __name__ == "__main__": main()