perf(contract): prefilter public boundary scans

loopx/contract.py

@@ -4,6 +4,7 @@
 import os
 import re
 import subprocess
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
@@ -53,29 +54,90 @@
 )
 
 
-LEAK_PATTERNS = {
-    "private_doc_url": re.compile(
-        "|".join(["la" + "rk" + "office", "docs" + r"\." + "internal"]),
-        re.I,
+@dataclass(frozen=True)
+class LeakRule:
+    pattern: re.Pattern[str]
+    required_literals: tuple[str, ...]
+
+    def is_candidate(self, folded_text: str) -> bool:
+        return any(literal in folded_text for literal in self.required_literals)
+
+
+# Python's Unicode re.IGNORECASE matches ASCII I/i against both dotted and
+# dotless I. casefold() alone does not: it expands U+0130 to i + combining dot.
+# Fold those two regex-equivalent characters first. The other special matches
+# (long s and Kelvin sign) are already covered by casefold().
+_REGEX_IGNORECASE_ASCII_TRANSLATION = str.maketrans({"\u0130": "i", "\u0131": "i"})
+
+
+def _prefilter_fold(text: str) -> str:
+    if text.isascii():
+        return text.lower()
+    if "\u0130" in text or "\u0131" in text:
+        text = text.translate(_REGEX_IGNORECASE_ASCII_TRANSLATION)
+    return text.casefold()
+
+
+# Required literals are only a cheap necessary condition for running a rule.
+# The regular expressions remain the sole authority for classifying a leak.
+LEAK_RULES = {
+    "private_doc_url": LeakRule(
+        pattern=re.compile(
+            "|".join(["la" + "rk" + "office", "docs" + r"\." + "internal"]),
+            re.I,
+        ),
+        required_literals=("lark" + "office", "docs" + ".internal"),
     ),
-    "credential": re.compile(
-        "|".join(
-            [
-                "Bear" + "er" + r"\s+[A-Za-z0-9._-]+",
-                "AK" + "IA" + r"[0-9A-Z]{16}",
-                r"(?<![A-Za-z0-9_])" + "tok" + "en=",
-                r"(?<![A-Za-z0-9_])" + "pass" + "word=",
-                "Author" + "ization:",
-            ]
+    "credential": LeakRule(
+        pattern=re.compile(
+            "|".join(
+                [
+                    "Bear" + "er" + r"\s+[A-Za-z0-9._-]+",
+                    "AK" + "IA" + r"[0-9A-Z]{16}",
+                    r"(?<![A-Za-z0-9_])" + "tok" + "en=",
+                    r"(?<![A-Za-z0-9_])" + "pass" + "word=",
+                    "Author" + "ization:",
+                ]
+            ),
+            re.I,
+        ),
+        required_literals=(
+            "bear" + "er",
+            "ak" + "ia",
+            "tok" + "en=",
+            "pass" + "word=",
+            "author" + "ization:",
+        ),
+    ),
+    "local_private_path": LeakRule(
+        pattern=re.compile(
+            "("
+            + "/"
+            + "Users"
+            + "/"
+            + r"[^/\s]+/(?:Documents|code"
+            + "-"
+            + r"reading)|"
+            + "/ext"
+            + "_data/"
+            + ")"
         ),
-        re.I,
+        required_literals=("/" + "users/", "/ext" + "_data/"),
     ),
-    "local_private_path": re.compile(
-        "(" + "/" + "Users" + "/" + r"[^/\s]+/(?:Documents|code" + "-" + r"reading)|" + "/ext" + "_data/" + ")"
+    "internal_task_id": LeakRule(
+        pattern=re.compile(r"\bt-" + r"20\d{12}-[a-z0-9]+\b"),
+        required_literals=("t-20",),
+    ),
+    "private_ip": LeakRule(
+        pattern=re.compile(
+            r"\b10\.\d+\.\d+\.\d+\b"
+            r"|\b172\.(?:1[6-9]|2\d|3[01])\.\d+\.\d+\b"
+            r"|\b192\.168\.\d+\.\d+\b"
+        ),
+        required_literals=("10.", "172.", "192.168."),
     ),
-    "internal_task_id": re.compile(r"\bt-" + r"20\d{12}-[a-z0-9]+\b"),
-    "private_ip": re.compile(r"\b10\.\d+\.\d+\.\d+\b|\b172\.(?:1[6-9]|2\d|3[01])\.\d+\.\d+\b|\b192\.168\.\d+\.\d+\b"),
 }
+LEAK_PATTERNS = {name: rule.pattern for name, rule in LEAK_RULES.items()}
 
 _PUBLIC_NPM_REGISTRY_PREFIX = "https://registry.npmjs.org/"
 
@@ -150,7 +212,7 @@ def _credential_match_is_reference(line: str, match: re.Match[str]) -> bool:
 
 
 def _credential_hits_are_all_references(line: str) -> bool:
-    matches = list(LEAK_PATTERNS["credential"].finditer(line))
+    matches = list(LEAK_RULES["credential"].pattern.finditer(line))
     if not matches:
         return False
     return all(_credential_match_is_reference(line, match) for match in matches)
@@ -844,12 +906,23 @@ def scan_public_boundary(
                                 "non_public_package_registry "
                                 f"({display_package})"
                             )
+        folded_text = _prefilter_fold(text)
+        candidate_rules = [
+            (name, rule)
+            for name, rule in LEAK_RULES.items()
+            if rule.is_candidate(folded_text)
+        ]
+        if not candidate_rules:
+            continue
         for line_no, line in enumerate(text.splitlines(), start=1):
-            for name, pattern in LEAK_PATTERNS.items():
+            folded_line = _prefilter_fold(line)
+            for name, rule in candidate_rules:
+                if not rule.is_candidate(folded_line):
+                    continue
                 scan_line = line
                 if name == "private_doc_url":
                     scan_line = _PUBLIC_LARK_DEVELOPER_CONSOLE_HOST.sub("", scan_line)
-                if pattern.search(scan_line):
+                if rule.pattern.search(scan_line):
                     hit = f"{rel_or_abs(path, root)}:{line_no}: {name}"
                     if name == "credential" and _credential_hits_are_all_references(line):
                         credential_reference_hits.append(hit)

tests/test_contract_public_boundary_prefilter.py

@@ -0,0 +1,112 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from loopx import contract
+
+
+@pytest.mark.parametrize(
+    ("rule_name", "line"),
+    [
+        ("private_doc_url", "https://tenant.lark" + "office.com/wiki/example"),
+        ("private_doc_url", "https://docs" + ".internal/example"),
+        ("credential", "Bear" + "er literal"),
+        ("credential", "AK" + "IA1234567890ABCDEF"),
+        ("credential", "tok" + "en=literal"),
+        ("credential", "pass" + "word=literal"),
+        ("credential", "Author" + "ization: literal"),
+        ("local_private_path", "/" + "Users/alice/Documents/example.md"),
+        ("local_private_path", "/" + "Users/alice/code-reading/example.md"),
+        ("local_private_path", "/ext" + "_data/example.md"),
+        ("internal_task_id", "ticket t-" + "20260828123456-example"),
+        ("private_ip", "host 10" + ".1.2.3"),
+        ("private_ip", "host 172" + ".31.2.3"),
+        ("private_ip", "host 192" + ".168.2.3"),
+    ],
+)
+def test_every_authoritative_leak_pattern_has_a_matching_prefilter(
+    rule_name: str,
+    line: str,
+) -> None:
+    rule = contract.LEAK_RULES[rule_name]
+
+    assert rule.pattern.search(line)
+    assert rule.is_candidate(contract._prefilter_fold(line))
+
+
+@pytest.mark.parametrize(
+    ("rule_name", "line"),
+    [
+        ("private_doc_url", "https://tenant.larkoff\u0130ce.com/wiki/example"),
+        ("credential", "Author\u0131zation: literal"),
+        ("credential", "pa\u017f\u017fword=literal"),
+        ("credential", "A\u212aIA1234567890ABCDEF"),
+    ],
+)
+def test_unicode_ignorecase_matches_are_never_filtered_out(
+    rule_name: str, line: str, tmp_path: Path
+) -> None:
+    rule = contract.LEAK_RULES[rule_name]
+    assert rule.pattern.search(line)
+    assert rule.is_candidate(contract._prefilter_fold(line))
+
+    (tmp_path / "sample.md").write_text(line + "\n", encoding="utf-8")
+    payload = contract.scan_public_boundary([tmp_path])
+    assert f"sample.md:1: {rule_name}" in payload["hits"]
+
+
+def test_prefilter_only_runs_regex_for_candidate_lines(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    class RecordingPattern:
+        def __init__(self) -> None:
+            self.lines: list[str] = []
+
+        def search(self, line: str) -> None:
+            self.lines.append(line)
+
+    pattern = RecordingPattern()
+    monkeypatch.setattr(
+        contract,
+        "LEAK_RULES",
+        {
+            "private_ip": contract.LeakRule(
+                pattern=pattern,  # type: ignore[arg-type]
+                required_literals=("candidate",),
+            )
+        },
+    )
+    (tmp_path / "sample.md").write_text(
+        "ordinary public line\ncandidate without a regex hit\nanother ordinary line\n",
+        encoding="utf-8",
+    )
+
+    payload = contract.scan_public_boundary([tmp_path])
+
+    assert payload["ok"] is True
+    assert pattern.lines == ["candidate without a regex hit"]
+
+
+def test_prefilter_preserves_all_boundary_hit_categories(tmp_path: Path) -> None:
+    lines = [
+        "https://tenant.lark" + "office.com/wiki/example",
+        "tok" + "en=literal",
+        "/" + "Users/alice/Documents/example.md",
+        "ticket t-" + "20260828123456-example",
+        "host 10" + ".1.2.3",
+    ]
+    (tmp_path / "sample.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+    payload: dict[str, Any] = contract.scan_public_boundary([tmp_path])
+
+    assert payload["hits"] == [
+        "sample.md:1: private_doc_url",
+        "sample.md:2: credential",
+        "sample.md:3: local_private_path",
+        "sample.md:4: internal_task_id",
+        "sample.md:5: private_ip",
+    ]