Embedding privacy: measure the orphan-run width on the base64 tail only (#394)

ingest_protection.py

@@ -1802,25 +1802,31 @@ def _has_orphan_full_width_base64_run(text: str) -> bool:
     surviving BEGIN/END marker and no placeholder in range — invisible to every
     marker- or placeholder-keyed check. Two or more contiguous strict base64
     lines of full PEM wrap width (>=40 chars) are the private-key body
-    signature; ordinary prose/config never produces them. Lines already inside
+    signature; PREFIXED_B64 width measures only the tail token (#394).
+    >=2 contiguous prefixed lines with 40+ character base64-charset tokens
+    (such as numbered lists of long IDs) still block. Lines already inside
     a redaction placeholder do not count. Fail-closed: a genuine multi-line
     base64 attachment on the cloud path blocks rather than ships (the durable
     store is untouched; a base64 blob has no embedding value anyway).
     """
     run = 0
     for kind, redact_start, content_end, _line_start, _marker_end in _pem_line_model(text):
-        if kind in (_PEM_LINE_KIND_STRICT_B64, _PEM_LINE_KIND_PREFIXED_B64) and (
-            content_end - redact_start
-        ) >= 40:
-            segment = text[redact_start:content_end]
-            if "[LCM embedding privacy:" in segment or "[LCM sensitive redaction:" in segment:
-                run = 0
+        if kind in (_PEM_LINE_KIND_STRICT_B64, _PEM_LINE_KIND_PREFIXED_B64):
+            width = content_end - redact_start
+            if kind == _PEM_LINE_KIND_PREFIXED_B64:
+                # #394: measure the base64 tail with the classification's rsplit.
+                tail = text[redact_start:content_end].rsplit(None, 1)
+                width = len(tail[-1]) if tail else 0
+            if width >= 40:
+                segment = text[redact_start:content_end]
+                if "[LCM embedding privacy:" in segment or "[LCM sensitive redaction:" in segment:
+                    run = 0
+                    continue
+                run += 1
+                if run >= 2:
+                    return True
                 continue
-            run += 1
-            if run >= 2:
-                return True
-        else:
-            run = 0
+        run = 0
     return False
 
 

tests/test_issue_394_orphan_run_tail_width.py

@@ -0,0 +1,117 @@
+"""#394 tail-width precision and #389 pure-hex exclusion mutation pins."""
+
+import pytest
+
+import hermes_lcm.command as command
+import hermes_lcm.ingest_protection as ip
+from hermes_lcm.config import LCMConfig
+
+
+_PROSE = (
+    "We met at the station after the long meeting.\n"
+    "Please file the expense report under REF2026QX7781ABCD\n"
+    "Then forward the same report to ACCT4410ZZ8812WXYZ\n"
+    "Thanks again."
+)
+_PEM_B0 = "MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQC7VJTUt9Us8cKj"
+_PEM_B1 = "MHcCAQEEIQD1eJ7yhkG0987xyzABCDEFghijkLMNOPqrstuvwxyz0987654321pq"
+
+
+def _config(tmp_path):
+    return LCMConfig(
+        database_path=str(tmp_path / "privacy.db"),
+        embeddings_enabled=True,
+        embedding_provider="voyage",
+        embedding_model="voyage-4-large",
+        sensitive_patterns_enabled=False,
+        embedding_privacy_enabled=None,
+        sensitive_patterns=[
+            "api_key", "bearer_token", "password_assignment", "private_key"
+        ],
+    )
+
+
+def test_prose_lines_ending_in_long_token_are_not_refused(tmp_path):
+    cfg = _config(tmp_path)
+    assert ip._has_orphan_full_width_base64_run(_PROSE) is False
+    protected, revision, changed = ip.protect_embedding_text(_PROSE, cfg)
+    assert protected == _PROSE
+    assert changed is False
+    ip.validate_embedding_privacy_dispatch(
+        [protected], cfg, expected_revision=revision
+    )
+
+
+def test_numbered_list_with_sub_40_tail_is_not_refused(tmp_path):
+    text = "Items:\n1. " + "Ab3" * 13 + "\n2. " + "Qz9x" * 19 + "Q9\nend"
+    cfg = _config(tmp_path)
+    assert ip._has_orphan_full_width_base64_run(text) is False
+    protected, revision, changed = ip.protect_embedding_text(text, cfg)
+    assert protected == text
+    assert changed is False
+    ip.validate_embedding_privacy_dispatch(
+        [protected], cfg, expected_revision=revision
+    )
+
+
+def test_backfill_prepare_does_not_refuse_prose(tmp_path):
+    cfg = _config(tmp_path)
+    documents, revision, transformed, blocked, error = (
+        command._prepare_embedding_provider_documents(
+            [("n1", _PROSE, 10)],
+            config=cfg,
+            provider_name="voyage",
+            expected_revision=ip.embedding_privacy_revision(cfg),
+        )
+    )
+    assert blocked == 0
+    assert error is None
+    assert transformed == 0
+    assert revision == ip.embedding_privacy_revision(cfg)
+    assert len(documents) == 1
+    assert documents[0][:2] == ("n1", _PROSE)
+
+
+def test_prefixed_full_width_body_lines_still_block(tmp_path):
+    text = f"INFO {_PEM_B0}\nINFO {_PEM_B1}"
+    cfg = _config(tmp_path)
+    assert ip._has_orphan_full_width_base64_run(text) is True
+    try:
+        protected, revision, _changed = ip.protect_embedding_text(text, cfg)
+    except ip.EmbeddingPrivacyPolicyError:
+        return
+    with pytest.raises(ip.EmbeddingPrivacyPolicyError):
+        ip.validate_embedding_privacy_dispatch(
+            [protected], cfg, expected_revision=revision
+        )
+
+
+def test_unprefixed_full_width_body_lines_still_block():
+    assert ip._has_orphan_full_width_base64_run(f"{_PEM_B0}\n{_PEM_B1}") is True
+
+
+def test_long_token_numbered_list_stays_blocked_boundary():
+    # Long prefixed IDs remain indistinguishable from a key body: fail closed.
+    text = "\n".join(f"{i}. " + "Xy7" * 20 for i in range(1, 18))
+    assert ip._has_orphan_full_width_base64_run(text) is True
+
+
+def test_389_pure_hex_31_run_near_private_key_placeholder_blocks():
+    ph = ip._embedding_privacy_placeholder("private_key")
+    assert ip._pem_fragment_near_private_key_placeholder(
+        ph + " 0123456789abcdef0123456789abcde"
+    ) is True
+
+
+def test_389_mixed_run_with_embedded_40_hex_blocks():
+    ph = ip._embedding_privacy_placeholder("private_key")
+    assert ip._pem_fragment_near_private_key_placeholder(
+        ph + " Zq" + "0123456789abcdef" * 2 + "01234567Xy"
+    ) is True
+
+
+def test_389_pure_hex_40_run_dispatches():
+    ph = ip._embedding_privacy_placeholder("private_key")
+    assert ip._pem_fragment_near_private_key_placeholder(
+        ph + " " + "0123456789abcdef0123456789abcdef01234567"
+    ) is False