diff --git a/loopx/contract.py b/loopx/contract.py index 5436f6ad9..b6278e88a 100644 --- a/loopx/contract.py +++ b/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"(? 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) diff --git a/tests/test_contract_public_boundary_prefilter.py b/tests/test_contract_public_boundary_prefilter.py new file mode 100644 index 000000000..cf6380f8d --- /dev/null +++ b/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", + ]