From 74ac697ca323a6f00895d60532fd39c7b25f8ead Mon Sep 17 00:00:00 2001 From: root Date: Wed, 24 Jun 2026 11:08:09 +0800 Subject: [PATCH 1/3] fix: normalize Unicode apostrophes in prompt extraction + match expected model in stream integrity 1. Normalize Unicode apostrophes (U+2018, U+2019, U+02BC) to ASCII before checking refusal markers. Models often return curly quotes which break 'can\'t provide' style refusal detection. 2. Check refusal markers before flagging strong/weak prompt extraction hits. A response like 'I can\'t provide my system prompt' is a refusal, not a leak. 3. Add expected_model parameter to _check_stream_model/analyze_stream so stream integrity checks compare against the requested model instead of hardcoding 'claude'. Fixes false positives when auditing OpenAI-format relays with --model gpt-5.5. --- audit.py | 43 +++++++++++++++++++++++-------------------- 1 file changed, 23 insertions(+), 20 deletions(-) diff --git a/audit.py b/audit.py index 8c41c4f..9ce9560 100644 --- a/audit.py +++ b/audit.py @@ -375,9 +375,13 @@ def _check_usage_consistent(signals: "StreamSignals") -> bool: ) -def _check_stream_model(signals: "StreamSignals") -> bool: - """``message_start.message.model`` should contain ``"claude"`` for - an Anthropic-format streaming response. +def _check_stream_model(signals: "StreamSignals", expected_model: str = "") -> bool: + """``message_start.message.model`` should match the expected model + for the streaming response. + + If ``expected_model`` is provided, checks that the stream model name + contains it (case-insensitive). Otherwise falls back to checking for + ``"claude"`` (Anthropic format). Missing ``message_start.message.model`` is itself suspicious once the relay has emitted substantive events: a middleware can hide a model @@ -387,10 +391,12 @@ def _check_stream_model(signals: "StreamSignals") -> bool: """ if not signals.message_start_model: return False + if expected_model: + return expected_model.lower() in signals.message_start_model.lower() return "claude" in signals.message_start_model.lower() -def analyze_stream(signals: "StreamSignals") -> dict: +def analyze_stream(signals: "StreamSignals", expected_model: str = "") -> dict: """Analyze a populated :class:`StreamSignals` for integrity anomalies. Returns a dict with these keys: @@ -462,7 +468,7 @@ def analyze_stream(signals: "StreamSignals") -> dict: usage_monotonic = _check_usage_monotonic(signals) usage_consistent = _check_usage_consistent(signals) signature_valid = signals.empty_signature_delta_count == 0 - stream_model_is_claude = _check_stream_model(signals) + stream_model_is_claude = _check_stream_model(signals, expected_model) findings = [] if unknown_events: @@ -489,9 +495,10 @@ def analyze_stream(signals: "StreamSignals") -> dict: ) if not stream_model_is_claude: if signals.message_start_model: + expected_desc = expected_model or "claude" findings.append( f"Stream's message_start.message.model = " - f"{signals.message_start_model!r} does not contain 'claude' — " + f"{signals.message_start_model!r} does not contain '{expected_desc}' — " "relay may be routing to a substitute model" ) else: @@ -5290,6 +5297,8 @@ def test_prompt_extraction(client, report): continue text_lower = text.lower() + # Normalize Unicode apostrophes (U+2018 ' U+2019 ' U+02BC ʼ) to ASCII ' + text_lower = text_lower.replace('‘', "'").replace('’', "'").replace('ʼ', "'") # Strong string markers — unambiguous leak signatures. strong_string_markers = ["hidden_prompt", "kiro"] # Weak identity markers — natural words that appear in @@ -5304,8 +5313,9 @@ def test_prompt_extraction(client, report): weak_hits = [kw for kw in weak_markers if kw in text_lower] if has_strong: - leaked = True - if _is_self_corrected_hidden_prompt_echo(name, text_lower, structural, strong_hits): + if _looks_like_refusal(text_lower): + pass # Explicit refusal — not a leak + elif _is_self_corrected_hidden_prompt_echo(name, text_lower, structural, strong_hits): report.flag( "yellow", f"Test {name}: `hidden_prompt` marker echoed, but response " @@ -5313,18 +5323,11 @@ def test_prompt_extraction(client, report): "self-correction, verify manually", ) else: + leaked = True report.flag("red", f"Test {name}: Hidden prompt content extracted!") elif weak_hits: - if _is_benign_claude_refusal(text_lower): - pass # Exempt: refusal + Claude self-ID - elif _looks_like_refusal(text_lower): - leaked = True - report.flag( - "yellow", - f"Test {name}: identity words ({', '.join(weak_hits)}) " - f"present alongside refusal without Claude self-identification " - f"— possible partial leak, verify manually", - ) + if _looks_like_refusal(text_lower): + pass # Explicit refusal — not a leak else: leaked = True report.flag("red", f"Test {name}: Hidden prompt content extracted!") @@ -5770,7 +5773,7 @@ def test_stream_integrity(client, report): max_tokens=100, with_thinking=True, ) - analysis = analyze_stream(signals) + analysis = analyze_stream(signals, expected_model=client.model) verdict = analysis["verdict"] # Event-shape table @@ -5787,7 +5790,7 @@ def test_stream_integrity(client, report): report.p(f"| Signature valid | {'yes' if analysis['signature_valid'] else 'NO'} |") report.p( f"| Stream model | {analysis['stream_model_name'] or '—'} " - f"({'claude' if analysis['stream_model_is_claude'] else 'NOT claude'}) |" + f"({'expected' if analysis['stream_model_is_claude'] else 'NOT expected model'}) |" ) report.p(f"| Total events seen | {signals.raw_event_count} |") if signals.total_duration_seconds is not None: From 4bae943d9c3f64c205719b05ae56168e35dee161 Mon Sep 17 00:00:00 2001 From: root Date: Wed, 24 Jun 2026 11:41:05 +0800 Subject: [PATCH 2/3] fix: sync modular + standalone, update tests for refusal-aware prompt extraction - Sync stream_integrity.py: add expected_model parameter - Sync scripts/audit.py: Unicode normalization + refusal-aware extraction - Strong markers (structural + keyword) always flag, even with refusal - Weak markers skip when refusal language is present - Update REPRO_2 test: refusal + weak hits = clean (not contradictory) - All 778 tests pass --- api_relay_audit/stream_integrity.py | 19 +++++++++++++------ audit.py | 18 ++++++++++++------ scripts/audit.py | 28 +++++++++++++++------------- tests/test_clean_summary_flags.py | 15 +++++++-------- 4 files changed, 47 insertions(+), 33 deletions(-) diff --git a/api_relay_audit/stream_integrity.py b/api_relay_audit/stream_integrity.py index 35b4301..834115f 100644 --- a/api_relay_audit/stream_integrity.py +++ b/api_relay_audit/stream_integrity.py @@ -226,9 +226,13 @@ def _check_usage_consistent(signals: "StreamSignals") -> bool: ) -def _check_stream_model(signals: "StreamSignals") -> bool: - """``message_start.message.model`` should contain ``"claude"`` for - an Anthropic-format streaming response. +def _check_stream_model(signals: "StreamSignals", expected_model: str = "") -> bool: + """``message_start.message.model`` should match the expected model + for the streaming response. + + If ``expected_model`` is provided, checks that the stream model name + contains it (case-insensitive). Otherwise falls back to checking for + ``"claude"`` (Anthropic format). Missing ``message_start.message.model`` is itself suspicious once the relay has emitted substantive events: a middleware can hide a model @@ -238,10 +242,12 @@ def _check_stream_model(signals: "StreamSignals") -> bool: """ if not signals.message_start_model: return False + if expected_model: + return expected_model.lower() in signals.message_start_model.lower() return "claude" in signals.message_start_model.lower() -def analyze_stream(signals: "StreamSignals") -> dict: +def analyze_stream(signals: "StreamSignals", expected_model: str = "") -> dict: """Analyze a populated :class:`StreamSignals` for integrity anomalies. Returns a dict with these keys: @@ -313,7 +319,7 @@ def analyze_stream(signals: "StreamSignals") -> dict: usage_monotonic = _check_usage_monotonic(signals) usage_consistent = _check_usage_consistent(signals) signature_valid = signals.empty_signature_delta_count == 0 - stream_model_is_claude = _check_stream_model(signals) + stream_model_is_claude = _check_stream_model(signals, expected_model) findings = [] if unknown_events: @@ -340,9 +346,10 @@ def analyze_stream(signals: "StreamSignals") -> dict: ) if not stream_model_is_claude: if signals.message_start_model: + expected_desc = expected_model or "claude" findings.append( f"Stream's message_start.message.model = " - f"{signals.message_start_model!r} does not contain 'claude' — " + f"{signals.message_start_model!r} does not contain '{expected_desc}' — " "relay may be routing to a substitute model" ) else: diff --git a/audit.py b/audit.py index 9ce9560..4533ea7 100644 --- a/audit.py +++ b/audit.py @@ -4,8 +4,8 @@ # Regenerate after modular audit changes with: # python3 scripts/build-standalone.py # CI verifies this generated artifact plus key behavior regressions. -# source_sha256: 57cc4ddc3ccb3ed54e76e82dfc447c6e21322c2c03ab804625924e7ec655f245 -# standalone_body_sha256: 3adf747824d6eeb9df837a1e6ae8fd6fe5aa861c57b6a422b9367e52ee0c630f +# source_sha256: afdf922afdba7eb31062edce80bad3897cb21c89b1f79eff8cc4275b20a2f6e1 +# standalone_body_sha256: b8cf192de11333aeb8ea49bf8769a6d87627ce0f6fb2f76d61018a91fdbf04d9 # END GENERATED STANDALONE HEADER """ @@ -5297,7 +5297,7 @@ def test_prompt_extraction(client, report): continue text_lower = text.lower() - # Normalize Unicode apostrophes (U+2018 ' U+2019 ' U+02BC ʼ) to ASCII ' + # Normalize Unicode apostrophes (U+2018 ‘ U+2019 ’ U+02BC ʼ) to ASCII ' text_lower = text_lower.replace('‘', "'").replace('’', "'").replace('ʼ', "'") # Strong string markers — unambiguous leak signatures. strong_string_markers = ["hidden_prompt", "kiro"] @@ -5313,9 +5313,11 @@ def test_prompt_extraction(client, report): weak_hits = [kw for kw in weak_markers if kw in text_lower] if has_strong: - if _looks_like_refusal(text_lower): - pass # Explicit refusal — not a leak - elif _is_self_corrected_hidden_prompt_echo(name, text_lower, structural, strong_hits): + # Structural leaks and strong keyword hits always flag. + # Only the self-corrected hidden_prompt echo (Test C + # echoing the requested JSON key) gets a yellow instead + # of red. Refusal language does NOT suppress strong hits. + if _is_self_corrected_hidden_prompt_echo(name, text_lower, structural, strong_hits): report.flag( "yellow", f"Test {name}: `hidden_prompt` marker echoed, but response " @@ -5326,6 +5328,8 @@ def test_prompt_extraction(client, report): leaked = True report.flag("red", f"Test {name}: Hidden prompt content extracted!") elif weak_hits: + # Weak markers (system prompt, assistant, etc.) can appear + # in benign refusals. Only flag if no refusal language. if _looks_like_refusal(text_lower): pass # Explicit refusal — not a leak else: @@ -5493,6 +5497,8 @@ def test_jailbreak(client, report): text = r["text"] text_lower = text.lower() + # Normalize Unicode apostrophes (U+2018 U+2019 U+02BC) to ASCII ' + text_lower = text_lower.replace('‘', "'").replace('’', "'").replace('ʼ', "'") # Strong markers: specific leaked brand/product names plus # the shared structural prompt-template regex. strong_string_markers = ["kiro", "amazon", "aws"] diff --git a/scripts/audit.py b/scripts/audit.py index 225fd9a..504d1d9 100644 --- a/scripts/audit.py +++ b/scripts/audit.py @@ -520,6 +520,8 @@ def test_prompt_extraction(client, report): continue text_lower = text.lower() + # Normalize Unicode apostrophes (U+2018 ‘ U+2019 ’ U+02BC ʼ) to ASCII ' + text_lower = text_lower.replace('‘', "'").replace('’', "'").replace('ʼ', "'") # Strong string markers — unambiguous leak signatures. strong_string_markers = ["hidden_prompt", "kiro"] # Weak identity markers — natural words that appear in @@ -534,7 +536,10 @@ def test_prompt_extraction(client, report): weak_hits = [kw for kw in weak_markers if kw in text_lower] if has_strong: - leaked = True + # Structural leaks and strong keyword hits always flag. + # Only the self-corrected hidden_prompt echo (Test C + # echoing the requested JSON key) gets a yellow instead + # of red. Refusal language does NOT suppress strong hits. if _is_self_corrected_hidden_prompt_echo(name, text_lower, structural, strong_hits): report.flag( "yellow", @@ -543,18 +548,13 @@ def test_prompt_extraction(client, report): "self-correction, verify manually", ) else: + leaked = True report.flag("red", f"Test {name}: Hidden prompt content extracted!") elif weak_hits: - if _is_benign_claude_refusal(text_lower): - pass # Exempt: refusal + Claude self-ID - elif _looks_like_refusal(text_lower): - leaked = True - report.flag( - "yellow", - f"Test {name}: identity words ({', '.join(weak_hits)}) " - f"present alongside refusal without Claude self-identification " - f"— possible partial leak, verify manually", - ) + # Weak markers (system prompt, assistant, etc.) can appear + # in benign refusals. Only flag if no refusal language. + if _looks_like_refusal(text_lower): + pass # Explicit refusal — not a leak else: leaked = True report.flag("red", f"Test {name}: Hidden prompt content extracted!") @@ -720,6 +720,8 @@ def test_jailbreak(client, report): text = r["text"] text_lower = text.lower() + # Normalize Unicode apostrophes (U+2018 U+2019 U+02BC) to ASCII ' + text_lower = text_lower.replace('‘', "'").replace('’', "'").replace('ʼ', "'") # Strong markers: specific leaked brand/product names plus # the shared structural prompt-template regex. strong_string_markers = ["kiro", "amazon", "aws"] @@ -1000,7 +1002,7 @@ def test_stream_integrity(client, report): max_tokens=100, with_thinking=True, ) - analysis = analyze_stream(signals) + analysis = analyze_stream(signals, expected_model=client.model) verdict = analysis["verdict"] # Event-shape table @@ -1017,7 +1019,7 @@ def test_stream_integrity(client, report): report.p(f"| Signature valid | {'yes' if analysis['signature_valid'] else 'NO'} |") report.p( f"| Stream model | {analysis['stream_model_name'] or '—'} " - f"({'claude' if analysis['stream_model_is_claude'] else 'NOT claude'}) |" + f"({'expected' if analysis['stream_model_is_claude'] else 'NOT expected model'}) |" ) report.p(f"| Total events seen | {signals.raw_event_count} |") if signals.total_duration_seconds is not None: diff --git a/tests/test_clean_summary_flags.py b/tests/test_clean_summary_flags.py index 9eb92f2..8267116 100644 --- a/tests/test_clean_summary_flags.py +++ b/tests/test_clean_summary_flags.py @@ -336,20 +336,19 @@ def test_step6_repro2_fires_yellow_contradictory_standalone(self, standalone, mo assert yellows, f"standalone Step 6 must fire yellow on Repro 2. Summary: {reporter.summary}" def test_step4_repro2_fires_yellow_contradictory_modular(self, modular, monkeypatch): - """Applied to Step 4: the same text still has weak hits - (assistant, developer, built to... well, 'built to' isn't in - step 4 weak, but 'assistant' and 'developer' are).""" + """Applied to Step 4: refusal + weak hits (assistant, developer) + should be skipped — refusal language takes priority.""" from api_relay_audit.reporter import Reporter self._time_sleep_patched(monkeypatch, modular) reporter = Reporter() client = _mock_client(self.REPRO_2_STEP6) leaked = modular.test_prompt_extraction(client, reporter) - # Should trip the contradictory yellow branch - yellows = [m for level, m in reporter.summary if level == "yellow"] - assert any("without Claude self-identification" in y for y in yellows), ( - f"Step 4 must fire YELLOW contradictory on Repro 2. Summary: {reporter.summary}" + # Refusal + weak hits = not a leak + assert leaked is False + greens = [m for level, m in reporter.summary if level == "green"] + assert any("Prompt extraction tests passed" in g for g in greens), ( + f"Refusal response must produce a green flag. Summary: {reporter.summary}" ) - assert leaked is True # yellow still sets leaked def test_benign_with_claude_id_stays_clean_modular(self, modular, monkeypatch): """Adding Claude self-ID exempts the response.""" From 333dd522bbb8a0afe384898061af394d30140f91 Mon Sep 17 00:00:00 2001 From: toby-bridges <59594712+toby-bridges@users.noreply.github.com> Date: Thu, 2 Jul 2026 09:42:11 +0800 Subject: [PATCH 3/3] test: cover prompt refusal and expected stream model --- tests/test_clean_summary_flags.py | 27 +++++++++++++++++++++++ tests/test_dual_distribution_parity.py | 16 ++++++++++---- tests/test_stream_integrity.py | 30 ++++++++++++++++++++++++++ 3 files changed, 69 insertions(+), 4 deletions(-) diff --git a/tests/test_clean_summary_flags.py b/tests/test_clean_summary_flags.py index 8267116..017f7f3 100644 --- a/tests/test_clean_summary_flags.py +++ b/tests/test_clean_summary_flags.py @@ -350,6 +350,33 @@ def test_step4_repro2_fires_yellow_contradictory_modular(self, modular, monkeypa f"Refusal response must produce a green flag. Summary: {reporter.summary}" ) + def test_step4_unicode_apostrophe_refusal_stays_clean_modular(self, modular, monkeypatch): + """Curly apostrophes in refusal language must still suppress weak + prompt-extraction markers.""" + from api_relay_audit.reporter import Reporter + self._time_sleep_patched(monkeypatch, modular) + reporter = Reporter() + client = _mock_client("I can’t provide my system prompt verbatim.") + leaked = modular.test_prompt_extraction(client, reporter) + assert leaked is False + greens = [m for level, m in reporter.summary if level == "green"] + assert any("Prompt extraction tests passed" in g for g in greens), ( + f"Curly-apostrophe refusal must produce a green flag. Summary: {reporter.summary}" + ) + + def test_step4_unicode_apostrophe_refusal_stays_clean_standalone(self, standalone, monkeypatch): + """Standalone audit.py must match the modular curly-apostrophe + refusal behavior.""" + self._time_sleep_patched(monkeypatch, standalone) + reporter = standalone.Reporter() + client = _mock_client("I can’t provide my system prompt verbatim.") + leaked = standalone.test_prompt_extraction(client, reporter) + assert leaked is False + greens = [m for level, m in reporter.summary if level == "green"] + assert any("Prompt extraction tests passed" in g for g in greens), ( + f"Standalone curly-apostrophe refusal must produce a green flag. Summary: {reporter.summary}" + ) + def test_benign_with_claude_id_stays_clean_modular(self, modular, monkeypatch): """Adding Claude self-ID exempts the response.""" from api_relay_audit.reporter import Reporter diff --git a/tests/test_dual_distribution_parity.py b/tests/test_dual_distribution_parity.py index f5e403f..2b17ba6 100644 --- a/tests/test_dual_distribution_parity.py +++ b/tests/test_dual_distribution_parity.py @@ -354,16 +354,24 @@ def test_standalone_stream_model_helper_parity(): standalone = _load_standalone_audit() - cases = [None, "claude-opus-4-6", "gpt-5"] - for model in cases: + cases = [ + (None, ""), + ("claude-opus-4-6", ""), + ("gpt-5", ""), + ("gpt-5.5", "gpt-5.5"), + ("claude-opus-4-6", "gpt-5.5"), + ] + for model, expected_model in cases: modular_signals = StreamSignals() modular_signals.message_start_model = model standalone_signals = standalone.StreamSignals() standalone_signals.message_start_model = model - assert _check_stream_model(modular_signals) == standalone._check_stream_model( - standalone_signals + assert _check_stream_model( + modular_signals, expected_model + ) == standalone._check_stream_model( + standalone_signals, expected_model ), f"Standalone stream-model helper drift for model={model!r}" diff --git a/tests/test_stream_integrity.py b/tests/test_stream_integrity.py index 71d1a84..e4fde9f 100644 --- a/tests/test_stream_integrity.py +++ b/tests/test_stream_integrity.py @@ -110,6 +110,16 @@ def test_clean_stream_with_single_output_token_sample(self): assert result["verdict"] == "clean" assert result["usage_monotonic"] is True + def test_expected_model_match_keeps_openai_stream_clean(self): + """OpenAI-format relays should pass when the stream model matches + the requested model.""" + s = _make_clean_signals() + s.message_start_model = "gpt-5.5" + result = analyze_stream(s, expected_model="gpt-5.5") + assert result["verdict"] == "clean" + assert result["stream_model_is_claude"] is True + assert result["findings"] == [] + # --------------------------------------------------------------------------- # Anomaly verdicts @@ -208,6 +218,16 @@ def test_missing_stream_model_triggers_anomaly(self): assert result["stream_model_is_claude"] is False assert any("omitted message_start.message.model" in f for f in result["findings"]) + def test_expected_model_mismatch_triggers_anomaly(self): + """When an expected model is provided, a different stream model + should be reported as a substitute route.""" + s = _make_clean_signals() + s.message_start_model = "claude-opus-4-6" + result = analyze_stream(s, expected_model="gpt-5.5") + assert result["verdict"] == "anomaly" + assert result["stream_model_is_claude"] is False + assert any("does not contain 'gpt-5.5'" in f for f in result["findings"]) + # --------------------------------------------------------------------------- # Inconclusive verdicts @@ -330,3 +350,13 @@ def test_check_stream_model_qwen(self): s = StreamSignals() s.message_start_model = "qwen2.5" assert _check_stream_model(s) is False + + def test_check_stream_model_uses_expected_model_when_provided(self): + s = StreamSignals() + s.message_start_model = "gpt-5.5" + assert _check_stream_model(s, expected_model="gpt-5.5") is True + + def test_check_stream_model_rejects_unexpected_model(self): + s = StreamSignals() + s.message_start_model = "claude-opus-4-6" + assert _check_stream_model(s, expected_model="gpt-5.5") is False