356 lines
14 KiB
Python
356 lines
14 KiB
Python
"""Response integrity, span masking, journal rejection, and diagnosis contract."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
REPO = ROOT.parent
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from tests.harness import configure_test_engine
|
|
|
|
configure_test_engine()
|
|
|
|
os.environ["KANSHO_FAKE_PROVIDER"] = "1"
|
|
os.environ["KANSHO_FAKE_DETECT"] = "1"
|
|
|
|
from fastapi.testclient import TestClient
|
|
from identity_store import list_mappings, remember_mapping
|
|
from journal_generate import JOURNAL_NOT_ACCEPTED, JOURNAL_NOT_ACCEPTED_MESSAGE
|
|
from journal_store import current_draft
|
|
from main import app
|
|
from privacy_gateway import (
|
|
ERROR_EGRESS_VALIDATION,
|
|
ActiveReplacement,
|
|
GatewayRequest,
|
|
MaskingManifest,
|
|
PrivacyGatewayError,
|
|
complete,
|
|
mask_prompt,
|
|
reset_debug,
|
|
validate_pre_egress,
|
|
)
|
|
from providers import ChatResult
|
|
|
|
|
|
def expect(ok: bool, message: str) -> None:
|
|
if not ok:
|
|
raise SystemExit(f"FAIL: {message}")
|
|
print(f"OK {message}")
|
|
|
|
|
|
def header(token: str) -> dict:
|
|
return {"X-Auth-Token": token}
|
|
|
|
|
|
def _run_gateway(profile_id: str, rendered: str, *, purpose: str = "dialogue_turn"):
|
|
return complete(
|
|
GatewayRequest(
|
|
prompt_id="response-integrity",
|
|
purpose=purpose,
|
|
data_class="B",
|
|
profile_id=profile_id,
|
|
payload={"rendered": rendered, "source_text": rendered, "prompt_slug": "mvp.journal_generate"},
|
|
)
|
|
)
|
|
|
|
|
|
def _day_with_turn(client: TestClient, headers: dict, body: str, title: str = "Tag") -> tuple[str, str]:
|
|
space = client.post("/api/journal/spaces", headers=headers, json={"title": title})
|
|
day = client.post(
|
|
f"/api/journal/spaces/{space.json()['id']}/days",
|
|
headers=headers,
|
|
json={"calendar_date": "2026-08-27"},
|
|
)
|
|
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": body},
|
|
)
|
|
if turn.status_code != 200:
|
|
raise SystemExit(f"FAIL: setup turn {turn.status_code}")
|
|
print("OK setup turn 200")
|
|
return day.json()["day"]["id"], conv.json()["id"]
|
|
|
|
|
|
def _generate(client: TestClient, headers: dict, day_id: str, conv_id: str, reply: str, usage=None):
|
|
calls = {"n": 0}
|
|
|
|
def fake(_messages, _policy):
|
|
calls["n"] += 1
|
|
return ChatResult(
|
|
content=reply,
|
|
model="fake-gpt",
|
|
usage=usage or {"prompt_tokens": 7, "completion_tokens": 5, "total_tokens": 12, "cost": 0.001},
|
|
context_compression="disabled",
|
|
)
|
|
|
|
with patch("privacy_gateway.complete_model", fake):
|
|
response = client.post(
|
|
f"/api/journal/days/{day_id}/generate",
|
|
headers=headers,
|
|
json={"conversation_ids": [conv_id]},
|
|
)
|
|
return response, calls["n"]
|
|
|
|
|
|
def test_span_masking_and_pre_egress() -> None:
|
|
text = "Ich aß Sushi. Sushi kam später."
|
|
second = text.find("Sushi", text.find("Sushi") + 1)
|
|
manifest = mask_prompt(
|
|
text,
|
|
[
|
|
{
|
|
"local_label": "Sushi",
|
|
"token": "PERSON:01",
|
|
"entity_type": "PERSON",
|
|
"start": second,
|
|
"end": second + 5,
|
|
"text": "Sushi",
|
|
"source": "request_local",
|
|
}
|
|
],
|
|
"journal_generate",
|
|
)
|
|
expect(manifest.masked_text == "Ich aß Sushi. [[PERSON:01]] kam später.", "only the detected span is masked")
|
|
validate_pre_egress(manifest.masked_text, manifest)
|
|
|
|
overlapping = mask_prompt(
|
|
"Anna-Lena kam vorbei.",
|
|
[
|
|
{
|
|
"local_label": "Anna",
|
|
"token": "PERSON:01",
|
|
"entity_type": "PERSON",
|
|
"start": 0,
|
|
"end": 4,
|
|
"text": "Anna",
|
|
"source": "request_local",
|
|
},
|
|
{
|
|
"local_label": "Anna-Lena",
|
|
"token": "PERSON:08",
|
|
"entity_type": "PERSON",
|
|
"start": 0,
|
|
"end": 9,
|
|
"text": "Anna-Lena",
|
|
"source": "request_local",
|
|
},
|
|
],
|
|
"journal_generate",
|
|
)
|
|
expect(overlapping.masked_text.startswith("[[PERSON:08]]"), "overlapping spans pick the longest")
|
|
|
|
leaky = MaskingManifest(
|
|
masked_text="Anna war am Markt.",
|
|
available_mapping_count=1,
|
|
replacements=(ActiveReplacement("PERSON:01", "PERSON", 1, "Anna"),),
|
|
)
|
|
blocked = False
|
|
try:
|
|
validate_pre_egress("Anna war am Markt.", leaky)
|
|
except PrivacyGatewayError as exc:
|
|
blocked = exc.code == ERROR_EGRESS_VALIDATION
|
|
expect(blocked, "privacy stays fail-closed before the provider")
|
|
|
|
|
|
def test_runtime_has_no_incident_wordlists() -> None:
|
|
forbidden = (
|
|
"Kinder",
|
|
"Delfine",
|
|
"delphine",
|
|
"Wohnzimmer",
|
|
"Balkon",
|
|
"Tagesreflektion",
|
|
"Lebenwesen",
|
|
"Baguette",
|
|
)
|
|
runtime = []
|
|
for path in ROOT.glob("*.py"):
|
|
if path.name.startswith("entity_detect_eval") or path.name.startswith("_"):
|
|
continue
|
|
text = path.read_text(encoding="utf-8")
|
|
for word in forbidden:
|
|
if word in text:
|
|
runtime.append(f"{path.name}:{word}")
|
|
expect(not runtime, f"incident words must not be hardcoded in runtime: {runtime}")
|
|
|
|
|
|
def test_frontend_degraded_copy() -> None:
|
|
page = (REPO / "frontend" / "src" / "pages" / "JournalDayPage.jsx").read_text(encoding="utf-8")
|
|
trace = (REPO / "frontend" / "src" / "components" / "CallTrace.jsx").read_text(encoding="utf-8")
|
|
log = (REPO / "frontend" / "src" / "components" / "RunLogPopup.jsx").read_text(encoding="utf-8")
|
|
expect("Generierung nicht übernommen" in page, "journal page names a rejected generate")
|
|
expect("Generierung nicht übernommen" in trace, "trace names a rejected generate")
|
|
expect("Generierung nicht übernommen" in log, "run log names a rejected generate")
|
|
expect("nur user-Zeilen" not in trace, "masking copy is not limited to user lines")
|
|
expect("zwei Modellaufrufe nacheinander" not in log, "running copy does not claim two model calls")
|
|
|
|
|
|
def main() -> None:
|
|
reset_debug()
|
|
test_span_masking_and_pre_egress()
|
|
test_runtime_has_no_incident_wordlists()
|
|
test_frontend_degraded_copy()
|
|
|
|
with TestClient(app) as client:
|
|
setup = client.post(
|
|
"/api/auth/setup",
|
|
json={"email": "response@example.test", "name": "Response", "password": "test-pass"},
|
|
)
|
|
headers = header(setup.json()["token"])
|
|
profile_id = setup.json()["profile_id"]
|
|
|
|
calls = {"n": 0}
|
|
|
|
def false_positive_cleartext(_messages, _policy):
|
|
calls["n"] += 1
|
|
return ChatResult(
|
|
content="Danach Sushi essen am Markt.",
|
|
model="fake-gpt",
|
|
usage={"prompt_tokens": 8, "completion_tokens": 6, "total_tokens": 14, "cost": 0.002},
|
|
context_compression="not_applicable",
|
|
)
|
|
|
|
remember_mapping(profile_id, "Sushi", "PERSON:01")
|
|
with patch("privacy_gateway.complete_model", false_positive_cleartext):
|
|
result = _run_gateway(profile_id, "Ich aß Sushi. Meine Frau Sushi kam später.")
|
|
expect(calls["n"] == 1, "false-positive cleartext still uses exactly one generate call")
|
|
expect("Sushi" in (result.content or ""), "reconstructed common noun is kept after local normalization")
|
|
expect(result.diagnostics.get("generate_calls") == 1, "gateway reports one generate call")
|
|
expect(
|
|
result.diagnostics.get("response_normalization") in {"active_cleartext_normalized", "none"},
|
|
"normalization is recorded without a retry",
|
|
)
|
|
|
|
remember_mapping(profile_id, "Anna", "PERSON:02")
|
|
day_id, conv_id = _day_with_turn(client, headers, "Heute war ich mit Anna am Markt.")
|
|
gen, n = _generate(
|
|
client,
|
|
headers,
|
|
day_id,
|
|
conv_id,
|
|
"Ein Markttag\n\nAnna stand den ganzen Nachmittag am Markt.",
|
|
)
|
|
expect(gen.status_code == 200, f"attested person cleartext {gen.status_code}")
|
|
expect(n == 1, "attested person cleartext does not retry")
|
|
expect("Anna" in (gen.json().get("body") or ""), "attested person is demasked")
|
|
expect((gen.json().get("trace") or {}).get("model_text_accepted") is True, "attested model text is accepted")
|
|
|
|
remember_mapping(profile_id, "Clarissa", "PERSON:99")
|
|
hist_day, hist_conv = _day_with_turn(client, headers, "Heute nur Markt und Kirschen.", title="Historisch")
|
|
blocked, n = _generate(
|
|
client,
|
|
headers,
|
|
hist_day,
|
|
hist_conv,
|
|
"Ein Markttag\n\nClarissa kaufte Kirschen am Markt.",
|
|
)
|
|
expect(blocked.status_code == 409, f"style-only person {blocked.status_code}")
|
|
detail = blocked.json().get("detail") or {}
|
|
expect(detail.get("code") == JOURNAL_NOT_ACCEPTED, "historical-only name is rejected")
|
|
expect(detail.get("message") == JOURNAL_NOT_ACCEPTED_MESSAGE, "API message is explicit")
|
|
expect(n == 1, "provenance reject does not retry")
|
|
expect(current_draft(profile_id, hist_day) is None, "rejected generate inserts no draft")
|
|
diag = detail.get("diagnostics") or {}
|
|
trace = diag.get("trace") or {}
|
|
expect(trace.get("abort_reason") == "unattested_identity", "provenance abort reason remains")
|
|
expect(trace.get("model_text_accepted") is False, "model text is marked not accepted")
|
|
expect(
|
|
trace.get("model") == "fake-gpt" or (trace.get("budget") or {}).get("model") == "fake-gpt",
|
|
"error trace keeps the model",
|
|
)
|
|
expect(
|
|
(trace.get("generate_calls") == 1)
|
|
or ((trace.get("budget") or {}).get("generate_calls") == 1)
|
|
or diag.get("generate_calls") == 1,
|
|
"error trace keeps generate_calls",
|
|
)
|
|
expect(
|
|
(trace.get("budget") or {}).get("completion_tokens") == 5 or diag.get("completion_tokens") == 5,
|
|
"error trace keeps completion tokens",
|
|
)
|
|
|
|
unknown_day, unknown_conv = _day_with_turn(client, headers, "Heute nur der Markt.", title="Platzhalter")
|
|
unknown, n = _generate(
|
|
client,
|
|
headers,
|
|
unknown_day,
|
|
unknown_conv,
|
|
"Ein Markttag\n\n[[PERSON:99]] stand am Markt.",
|
|
)
|
|
expect(unknown.status_code == 409, f"unknown placeholder {unknown.status_code}")
|
|
expect((unknown.json().get("detail") or {}).get("code") == JOURNAL_NOT_ACCEPTED, "unknown placeholder is rejected")
|
|
expect(n == 1, "unknown placeholder does not retry")
|
|
expect(current_draft(profile_id, unknown_day) is None, "unknown placeholder inserts no draft")
|
|
|
|
generic_day, generic_conv = _day_with_turn(client, headers, "Heute nur der Hafen.", title="Generisch")
|
|
generic, n = _generate(
|
|
client,
|
|
headers,
|
|
generic_day,
|
|
generic_conv,
|
|
"Ein Hafentag\n\n[[...]] blieb am Hafen.",
|
|
)
|
|
expect(generic.status_code == 409, f"generic placeholder {generic.status_code}")
|
|
expect((generic.json().get("detail") or {}).get("code") == JOURNAL_NOT_ACCEPTED, "generic placeholder is rejected")
|
|
expect(current_draft(profile_id, generic_day) is None, "generic placeholder inserts no draft")
|
|
|
|
ok_day, ok_conv = _day_with_turn(client, headers, "Heute war ich mit Anna am Hafen.", title="Tokens")
|
|
anna_token = next(
|
|
(
|
|
item.get("token")
|
|
for item in list_mappings(profile_id)
|
|
if (item.get("canonical_label") or item.get("local_label") or "") == "Anna"
|
|
),
|
|
"PERSON:02",
|
|
)
|
|
ok_gen, n = _generate(
|
|
client,
|
|
headers,
|
|
ok_day,
|
|
ok_conv,
|
|
f"Ein Hafentag\n\n[[{anna_token}]] stand am Hafen.",
|
|
)
|
|
expect(ok_gen.status_code == 200, f"active tokens {ok_gen.status_code}")
|
|
expect("Anna" in (ok_gen.json().get("body") or ""), "active tokens demask normally")
|
|
expect("[[" not in (ok_gen.json().get("body") or ""), "no leftover placeholder after demask")
|
|
expect(n == 1, "normal demask uses one generate call")
|
|
|
|
provider_calls = {"n": 0}
|
|
|
|
def boom(_messages, _policy):
|
|
provider_calls["n"] += 1
|
|
raise AssertionError("provider must not be called after pre-egress failure")
|
|
|
|
def leaky_mask(rendered, mappings, purpose):
|
|
return MaskingManifest(
|
|
masked_text=rendered,
|
|
available_mapping_count=len(mappings or []),
|
|
replacements=(ActiveReplacement("PERSON:02", "PERSON", 1, "Anna"),),
|
|
)
|
|
|
|
with patch("privacy_gateway.mask_prompt", leaky_mask), patch("privacy_gateway.complete_model", boom):
|
|
failed = False
|
|
try:
|
|
_run_gateway(profile_id, "Anna war am Markt.")
|
|
except PrivacyGatewayError as exc:
|
|
failed = exc.code == ERROR_EGRESS_VALIDATION
|
|
expect(failed, "pre-egress still fail-closed")
|
|
expect(provider_calls["n"] == 0, "provider is not called when confirmed identity remains")
|
|
|
|
print("All privacy response integrity tests passed.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|