jesse-uxandClaude Code
8497587e65
Document P0 collision fix and precise privacy declaration permit. Co-Authored-By: Claude Code <noreply@anthropic.com>
703 lines
32 KiB
Python
703 lines
32 KiB
Python
"""Tracked-file privacy gate. Diagnostics never include matching source text.
|
||
|
||
The upstream snapshot exception expires when its owner supplies a cleaned import.
|
||
All rectification version files remain in scope, including hash-bound packages.
|
||
Local rules are optional, one literal per nonblank line; # starts a comment.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import ast
|
||
import codecs
|
||
import json
|
||
from collections import Counter
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path, PurePosixPath
|
||
import re
|
||
import stat
|
||
import subprocess
|
||
import sys
|
||
|
||
import pytest
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
SELF_PATH = "tests/test_repo_privacy_markers.py"
|
||
GOLDEN_PATH = "tests/test_consultation_contract_golden.py"
|
||
LOCAL_PATH = "tests/privacy_markers.local.txt"
|
||
UPSTREAM_EXCEPTION = "references/upstream/yinduzhanxing/SKILL.md"
|
||
PROTECTED_PREFIX = "skills/jyotish-birth-time-rectification/versions/"
|
||
USER_HOME_PREFIX = "/Users/"
|
||
|
||
# Reviewed synthetic security fixtures, not a username or file-wide allowlist.
|
||
# Pin the location AND full source line; moving or extending one fails closed.
|
||
# Only the matching prefix span is allowed, never the rest of the line.
|
||
SAFE_LINES = {
|
||
"frontend/tests/agent-evals.test.ts": {
|
||
603: 'path: "/Users/example/private.txt",',
|
||
},
|
||
"frontend/tests/personal-report-contract.test.ts": {
|
||
131: '"file:///Users/example/private/chart.json",',
|
||
},
|
||
"tests/test_personal_report_contract.py": {
|
||
190: '"file:///Users/example/private/chart.json",',
|
||
},
|
||
"frontend/tests/agent-observability.test.ts": {
|
||
122: 'modelVersion: "/Users/example/private/model.json",',
|
||
219: 'const rawFailure = "provider rejected user@example.com at /Users/alice/private.json";',
|
||
},
|
||
"frontend/tests/personal-report-api.test.ts": {
|
||
832: 'workflow.internal_path = "/Users/private/project/engine.py";',
|
||
860: 'summary: "Bearer opaque-conflict at /Users/private/trace.py",',
|
||
890: '"/Users/private",',
|
||
},
|
||
"frontend/tests/personal-report-generation.test.ts": {
|
||
1097: 'summary: "Authorization: Bearer secret-token at /Users/private/trace.py",',
|
||
1271: 'wealthWorkflow.internal_path = "/Users/private/project/engine.py";',
|
||
1288: 'summary: "Authorization: Bearer secret-token at /Users/private/trace.py",',
|
||
1291: 'wealthSections["Authorization: Bearer section-secret at /Users/private/tool.py"] = {',
|
||
1373: '"/Users/private",',
|
||
},
|
||
"tests/test_calculation_profile_contract.py": {
|
||
243: 'first = build_with_path("/Users/alice/repo/references/open_source_sources/vedic-astro-skills/ephe")',
|
||
},
|
||
"tests/test_commercial_skill_truth_contract.py": {
|
||
24: 'assert "/Users/" not in str(overlay)',
|
||
},
|
||
"tests/test_professional_report_reference_api.py": {
|
||
113: '"source=scripts/private_engine.py::compute path=/Users/example/report.json "',
|
||
124: 'assert "/Users/" not in sanitized',
|
||
134: '"fixture_path": "/Users/example/references/oracle.json",',
|
||
153: 'assert "/Users/" not in rendered',
|
||
},
|
||
"docs/tasks/UPSTREAM-INSTRUCTION-owner-case-purge-20260919.md": {
|
||
68: '验收：`git grep -c "/Users/" -- . \':!references/open_source_sources\' | wc -l` 为 0。',
|
||
72: '新增 `tests/test_repo_privacy_markers.py`：读 `tests/privacy_markers.local.txt`（gitignore 的）里的模式，扫 `git ls-files`（排除 pdf / wasm / `open_source_sources`），命中即失败；文件不存在时 skip 并打印提示。`/Users/` 前缀作为内置模式始终检查。',
|
||
99: 'git grep -l "/Users/" -- . \':!references/open_source_sources\' | wc -l # 0',
|
||
},
|
||
}
|
||
|
||
|
||
class ScanError(Exception):
|
||
"""Only stable codes, never exception strings containing source or paths."""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Rule:
|
||
code: str
|
||
value: str = field(repr=False)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Finding:
|
||
rule: str
|
||
path: str # already redacted
|
||
line: int # zero means filename / file-level failure
|
||
|
||
|
||
@dataclass
|
||
class ScanResult:
|
||
findings: list[Finding] = field(default_factory=list)
|
||
scanned: int = 0
|
||
excluded: int = 0
|
||
upstream_exceptions: int = 0
|
||
deleted: int = 0
|
||
protected: int = 0
|
||
permitted: int = 0
|
||
|
||
|
||
def _assignment(tree: ast.Module, name: str) -> ast.Assign | None:
|
||
nodes = [n for n in tree.body if isinstance(n, ast.Assign)
|
||
and len(n.targets) == 1 and isinstance(n.targets[0], ast.Name)
|
||
and n.targets[0].id == name]
|
||
return nodes[0] if len(nodes) == 1 else None
|
||
|
||
|
||
def extract_markers(source: str) -> tuple[str, ...]:
|
||
"""Parse, never import or execute the API golden test module."""
|
||
try:
|
||
node = _assignment(ast.parse(source), "FORBIDDEN_PRIVACY_MARKERS")
|
||
if node is None or not isinstance(node.value, (ast.Tuple, ast.List)):
|
||
raise ValueError
|
||
if not all(isinstance(n, ast.Constant) and isinstance(n.value, str)
|
||
and n.value for n in node.value.elts):
|
||
raise ValueError
|
||
values = tuple(n.value for n in node.value.elts)
|
||
if not values or len(set(values)) != len(values):
|
||
raise ValueError
|
||
return values
|
||
except (SyntaxError, ValueError, TypeError):
|
||
raise ScanError("CONFIG_GOLDEN") from None
|
||
|
||
|
||
def build_rules(golden_source: str, local_source: str = "") -> tuple[Rule, ...]:
|
||
rules = [Rule(f"R{i:03}", value)
|
||
for i, value in enumerate(extract_markers(golden_source), 1)]
|
||
rules.append(Rule("HOME_PREFIX", USER_HOME_PREFIX))
|
||
for i, line in enumerate(local_source.removeprefix(codecs.BOM_UTF8.decode("utf8")).splitlines(), 1):
|
||
value = line.strip()
|
||
if value and not value.startswith("#"):
|
||
rules.append(Rule(f"LOCAL_{i:03}", value))
|
||
return tuple(rules)
|
||
|
||
|
||
def _node_span(source: str, node: ast.AST) -> tuple[int, int]:
|
||
# AST columns count UTF-8 bytes, not Unicode code points.
|
||
lines = source.splitlines(keepends=True)
|
||
start = sum(map(len, lines[:node.lineno - 1]))
|
||
end = sum(map(len, lines[:node.end_lineno - 1]))
|
||
start += len(lines[node.lineno - 1].encode("utf8")[:node.col_offset].decode("utf8"))
|
||
end += len(lines[node.end_lineno - 1].encode("utf8")[:node.end_col_offset].decode("utf8"))
|
||
return start, end
|
||
|
||
|
||
def _declaration_spans(path: str, source: str, rules: tuple[Rule, ...]) -> dict[str, list[tuple[int, int]]]:
|
||
spans: dict[str, list[tuple[int, int]]] = {}
|
||
if path not in {GOLDEN_PATH, SELF_PATH}:
|
||
return spans
|
||
try:
|
||
tree = ast.parse(source)
|
||
except SyntaxError:
|
||
return spans # malformed files still get scanned, with no exemptions
|
||
if path == GOLDEN_PATH:
|
||
node = _assignment(tree, "FORBIDDEN_PRIVACY_MARKERS")
|
||
if node is not None and isinstance(node.value, (ast.Tuple, ast.List)):
|
||
for child in node.value.elts:
|
||
for rule in rules:
|
||
if rule.code.startswith("R") and isinstance(child, ast.Constant) and child.value == rule.value:
|
||
spans.setdefault(rule.code, []).append(_node_span(source, child))
|
||
else:
|
||
prefix = _assignment(tree, "USER_HOME_PREFIX")
|
||
if prefix is not None and isinstance(prefix.value, ast.Constant) and prefix.value.value == USER_HOME_PREFIX:
|
||
spans.setdefault("HOME_PREFIX", []).append(_node_span(source, prefix.value))
|
||
table = _assignment(tree, "SAFE_LINES")
|
||
if table is not None:
|
||
try:
|
||
matches = ast.literal_eval(table.value) == SAFE_LINES
|
||
except (ValueError, TypeError):
|
||
matches = False
|
||
if matches:
|
||
for child in ast.walk(table.value):
|
||
if isinstance(child, ast.Constant) and isinstance(child.value, str):
|
||
spans.setdefault("HOME_PREFIX", []).append(_node_span(source, child))
|
||
return spans
|
||
|
||
|
||
def _collision_spans(path: str, source: str, rules: tuple[Rule, ...]) -> list[tuple[int, int]]:
|
||
"""Reviewed non-personal R003 collisions; preserve existing numerical evidence.
|
||
|
||
The replay generator declares this case public/synthetic, not an owner case.
|
||
Permit only its computed endpoint, not any birth input or another JSON node.
|
||
"""
|
||
marker = next((r.value for r in rules if r.code == "R003"), None)
|
||
if marker is None:
|
||
return []
|
||
service_path = "tests/test_rectification_v5_services.py"
|
||
replay_path = "references/oracle/pyjhora_multi_case_panchanga_gulika_replay_2026_07_23.json"
|
||
if path not in {service_path, replay_path}:
|
||
return []
|
||
try:
|
||
tree = ast.parse(source)
|
||
except SyntaxError:
|
||
return []
|
||
if path == service_path:
|
||
name = "test_hour_window_keeps_all_seventeen_signature_clusters_including_late_tail"
|
||
functions = [n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and n.name == name]
|
||
if len(functions) != 1:
|
||
return []
|
||
spans = []
|
||
for node in ast.walk(functions[0]):
|
||
if not isinstance(node, ast.Set) or node.lineno not in {1182, 1190}:
|
||
continue
|
||
try:
|
||
values = ast.literal_eval(node)
|
||
except (ValueError, TypeError):
|
||
continue
|
||
if values == {f"14:{minute:02}" for minute in range(46, 52)} and len(node.elts) == 6:
|
||
for child in node.elts:
|
||
if isinstance(child, ast.Constant) and child.value == marker:
|
||
spans.append(_node_span(source, child))
|
||
return spans
|
||
try:
|
||
data = json.loads(source)
|
||
case = data["cases"][1]
|
||
expected_input = {
|
||
"date": "1996-12-07", "time": "10:34:00", "place": "Chennai",
|
||
"latitude": 13.0878, "longitude": 80.2785, "timezone": "+5.5", "ayanamsa": "Lahiri",
|
||
}
|
||
if case["case_id"] != "public_smoke_chennai_1996" or case["input"] != expected_input:
|
||
return []
|
||
if case["pyjhora_raw"]["yamaganda_kaalam"] != ["13:24:50", marker + ":18"]:
|
||
return []
|
||
# Follow the JSON pointer in its syntax tree, retaining exact value span.
|
||
node = tree.body[0].value
|
||
for key in ("cases", 1, "pyjhora_raw", "yamaganda_kaalam", 1):
|
||
if isinstance(key, int):
|
||
node = node.elts[key]
|
||
else:
|
||
matches = [v for k, v in zip(node.keys, node.values)
|
||
if isinstance(k, ast.Constant) and k.value == key]
|
||
if len(matches) != 1:
|
||
return []
|
||
node = matches[0]
|
||
if isinstance(node, ast.Constant) and node.lineno == 149 and node.value == marker + ":18":
|
||
return [_node_span(source, node)]
|
||
except (ValueError, KeyError, IndexError, AttributeError, TypeError):
|
||
pass
|
||
return []
|
||
|
||
|
||
def safe_path(path: str, rules: tuple[Rule, ...]) -> str:
|
||
# Mask whole user-path tails, not just the prefix. Literal markers in
|
||
# filenames must also never reach pytest assertion output or log summaries.
|
||
result = path
|
||
if USER_HOME_PREFIX in result:
|
||
result = result[:result.index(USER_HOME_PREFIX)] + "/[redacted-path]"
|
||
for rule in sorted(rules, key=lambda r: len(r.value), reverse=True):
|
||
result = result.replace(rule.value, "[redacted]")
|
||
return "".join(c if c.isprintable() else "?" for c in result)
|
||
|
||
|
||
def _excluded(path: str) -> bool:
|
||
# Root-relative exclusions only: no broad 'references' or version exemption.
|
||
return (PurePosixPath(path).suffix.lower() in {".pdf", ".wasm"}
|
||
or path.startswith(("vendor/", "references/open_source_sources/")))
|
||
|
||
|
||
def scan_text(path: str, source: str, rules: tuple[Rule, ...]) -> tuple[list[Finding], int]:
|
||
allowed = _declaration_spans(path, source, rules)
|
||
allowed.setdefault("R003", []).extend(_collision_spans(path, source, rules))
|
||
offset = 0
|
||
for line_number, line in enumerate(source.splitlines(keepends=True), 1):
|
||
approved = SAFE_LINES.get(path, {}).get(line_number)
|
||
if approved is not None and line.strip() == approved:
|
||
for match in re.finditer(re.escape(USER_HOME_PREFIX), line):
|
||
allowed.setdefault("HOME_PREFIX", []).append((offset + match.start(), offset + match.end()))
|
||
offset += len(line)
|
||
findings: list[Finding] = []
|
||
permitted = 0
|
||
for rule in rules:
|
||
for match in re.finditer(re.escape(rule.value), source):
|
||
if any(start <= match.start() and match.end() <= end
|
||
for start, end in allowed.get(rule.code, ())):
|
||
permitted += 1
|
||
continue
|
||
findings.append(Finding(rule.code, safe_path(path, rules), source.count("\n", 0, match.start()) + 1))
|
||
return findings, permitted
|
||
|
||
|
||
def _git(root: Path, *args: str) -> bytes:
|
||
try:
|
||
# Required even for read-only git operations in this shared worktree.
|
||
subprocess.run(["git", "status", "-sb"], cwd=root, check=True,
|
||
stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=30)
|
||
return subprocess.run(["git", *args], cwd=root, check=True,
|
||
stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=30).stdout
|
||
except (OSError, subprocess.SubprocessError):
|
||
raise ScanError("GIT_READ") from None
|
||
|
||
|
||
def tracked_paths(root: Path) -> tuple[str, ...]:
|
||
try:
|
||
return tuple(sorted(set(p for p in _git(root, "ls-files", "-z").decode("utf8").split("\0") if p)))
|
||
except UnicodeError:
|
||
raise ScanError("PATH_ENCODING") from None
|
||
|
||
|
||
def _read(root: Path, path: str) -> str:
|
||
relative = PurePosixPath(path)
|
||
if relative.is_absolute() or ".." in relative.parts or "\\" in path or ":" in path:
|
||
raise OSError("READ_ERROR")
|
||
target = root / path
|
||
# Check from the trusted repository root down before touching a child.
|
||
# On Windows junctions are reparse points even when S_ISLNK is false.
|
||
for ancestor in (root, *(root.joinpath(*relative.parts[:i])
|
||
for i in range(1, len(relative.parts)))):
|
||
info = ancestor.lstat()
|
||
if (stat.S_ISLNK(info.st_mode)
|
||
or getattr(info, "st_file_attributes", 0) & stat.FILE_ATTRIBUTE_REPARSE_POINT
|
||
or not stat.S_ISDIR(info.st_mode)):
|
||
raise OSError("READ_ERROR")
|
||
info = target.lstat()
|
||
if stat.S_ISLNK(info.st_mode):
|
||
try:
|
||
return str(target.readlink())
|
||
except OSError:
|
||
# A failed readlink is not an ordinary deleted tracked file.
|
||
raise OSError("READ_ERROR") from None
|
||
if (getattr(info, "st_file_attributes", 0) & stat.FILE_ATTRIBUTE_REPARSE_POINT
|
||
or not stat.S_ISREG(info.st_mode)):
|
||
raise OSError("READ_ERROR")
|
||
data = target.read_bytes()
|
||
# Binary extensions other than the explicit exclusions are not silently
|
||
# skipped. Replacement decoding preserves embedded ASCII privacy markers.
|
||
if data.startswith((b"\xff\xfe", b"\xfe\xff")):
|
||
return data.decode("utf16", errors="replace")
|
||
return data.decode("utf8", errors="replace")
|
||
|
||
|
||
def scan_repository(root: Path) -> ScanResult:
|
||
try:
|
||
golden = _read(root, GOLDEN_PATH)
|
||
try:
|
||
local = _read(root, LOCAL_PATH)
|
||
except FileNotFoundError:
|
||
local = ""
|
||
except OSError:
|
||
raise ScanError("CONFIG_READ") from None
|
||
rules = build_rules(golden, local)
|
||
paths = tracked_paths(root)
|
||
result = ScanResult()
|
||
if LOCAL_PATH in paths:
|
||
result.findings.append(Finding("LOCAL_TRACKED", LOCAL_PATH, 0))
|
||
if _git(root, "check-ignore", "--no-index", LOCAL_PATH).decode("utf8").strip() != LOCAL_PATH:
|
||
raise ScanError("LOCAL_NOT_IGNORED")
|
||
for path in paths:
|
||
if _excluded(path):
|
||
result.excluded += 1
|
||
continue
|
||
if path == UPSTREAM_EXCEPTION:
|
||
result.upstream_exceptions += 1
|
||
continue
|
||
for rule in rules:
|
||
if rule.value in path:
|
||
result.findings.append(Finding(rule.code, safe_path(path, rules), 0))
|
||
try:
|
||
source = _read(root, path)
|
||
except FileNotFoundError:
|
||
# Unstaged tracked deletions remain in ls-files. Report their count;
|
||
# committed CI checkouts naturally have none. Never delete anything.
|
||
result.deleted += 1
|
||
continue
|
||
except OSError:
|
||
result.findings.append(Finding("READ_ERROR", safe_path(path, rules), 0))
|
||
continue
|
||
result.scanned += 1
|
||
if path.startswith(PROTECTED_PREFIX):
|
||
result.protected += 1
|
||
findings, permitted = scan_text(path, source, rules)
|
||
result.findings.extend(findings)
|
||
result.permitted += permitted
|
||
return result
|
||
|
||
|
||
def format_report(result: ScanResult) -> str:
|
||
counts = Counter((f.path, f.rule) for f in result.findings)
|
||
rows = [f"scanned={result.scanned} excluded={result.excluded} deleted={result.deleted} "
|
||
f"upstream_exceptions={result.upstream_exceptions} protected={result.protected} "
|
||
f"permitted={result.permitted} files={len({f.path for f in result.findings})} hits={len(result.findings)}"]
|
||
for (path, rule), count in sorted(counts.items()):
|
||
lines = sorted({f.line for f in result.findings if f.path == path and f.rule == rule})
|
||
rows.append(f"{rule} {path} count={count} lines={','.join(map(str, lines))}")
|
||
return "\n".join(rows)
|
||
|
||
|
||
def test_tracked_repository_has_no_privacy_markers() -> None:
|
||
__tracebackhide__ = True
|
||
try:
|
||
result = scan_repository(ROOT)
|
||
except ScanError as error:
|
||
pytest.fail(str(error), pytrace=False)
|
||
if result.findings:
|
||
pytest.fail(format_report(result), pytrace=False)
|
||
|
||
|
||
# All self-test inputs below are synthetic. Private marker values are never
|
||
# copied out of the existing golden declaration, hashed, or encoded here.
|
||
def _synthetic_rules() -> tuple[Rule, ...]:
|
||
return (Rule("R001", "fictional-forbidden-token"), Rule("HOME_PREFIX", USER_HOME_PREFIX))
|
||
|
||
|
||
def test_extract_uses_ast_without_executing_module() -> None:
|
||
source = 'raise RuntimeError("must not execute")\nFORBIDDEN_PRIVACY_MARKERS = ("synthetic",)'
|
||
assert extract_markers(source) == ("synthetic",)
|
||
|
||
|
||
@pytest.mark.parametrize("source", ["", "FORBIDDEN_PRIVACY_MARKERS = load()",
|
||
'FORBIDDEN_PRIVACY_MARKERS = ("",)', 'FORBIDDEN_PRIVACY_MARKERS = (1,)',
|
||
'FORBIDDEN_PRIVACY_MARKERS = ("x", "x")',
|
||
'FORBIDDEN_PRIVACY_MARKERS = ("x",)\nFORBIDDEN_PRIVACY_MARKERS = ("y",)',
|
||
'FORBIDDEN_PRIVACY_MARKERS = ("unterminated',
|
||
])
|
||
def test_invalid_marker_declaration_fails_closed(source: str) -> None:
|
||
with pytest.raises(ScanError, match="^CONFIG_GOLDEN$"):
|
||
extract_markers(source)
|
||
|
||
|
||
def test_golden_node_not_comments_other_nodes_or_new_files() -> None:
|
||
rules = _synthetic_rules()
|
||
source = f'FORBIDDEN_PRIVACY_MARKERS = ({rules[0].value!r},)'
|
||
assert scan_text(GOLDEN_PATH, source, rules)[0] == []
|
||
dirty = source + f' # {rules[0].value}\nother = {rules[0].value!r}'
|
||
findings, _ = scan_text(GOLDEN_PATH, dirty, rules)
|
||
assert [f.line for f in findings] == [1, 2]
|
||
assert len(scan_text("tests/new_test.py", source, rules)[0]) == 1
|
||
|
||
|
||
def test_prefix_declaration_is_not_a_private_path_permission() -> None:
|
||
rules = _synthetic_rules()
|
||
source = f'USER_HOME_PREFIX = {USER_HOME_PREFIX!r}'
|
||
assert scan_text(SELF_PATH, source, rules)[0] == []
|
||
assert scan_text(SELF_PATH, f'USER_HOME_PREFIX = {USER_HOME_PREFIX + "new-person/private.txt"!r}', rules)[0]
|
||
dirty = source + f' # {USER_HOME_PREFIX}new-person/private.txt\nother = {USER_HOME_PREFIX!r}'
|
||
assert len(scan_text(SELF_PATH, dirty, rules)[0]) == 2
|
||
assert scan_text("tests/new_test.py", source, rules)[0]
|
||
|
||
|
||
@pytest.mark.parametrize("path,line_number,source_line", [
|
||
(path, number, line) for path, lines in SAFE_LINES.items() for number, line in lines.items()
|
||
], ids=[f"safe-{i:02}" for i in range(sum(map(len, SAFE_LINES.values())))])
|
||
def test_exact_synthetic_line_permission_rejects_mutations(path: str, line_number: int, source_line: str) -> None:
|
||
rules = _synthetic_rules()
|
||
source = "\n" * (line_number - 1) + source_line
|
||
assert scan_text(path, source, rules)[0] == []
|
||
assert scan_text("tests/new_test.py", source, rules)[0]
|
||
assert scan_text(path, "\n" + source, rules)[0]
|
||
assert scan_text(path, source + "\n" + source_line, rules)[0]
|
||
assert scan_text(path, source.replace(USER_HOME_PREFIX, USER_HOME_PREFIX + "new-person/"), rules)[0]
|
||
assert scan_text(path, source + " # " + rules[0].value, rules)[0]
|
||
|
||
|
||
def test_self_file_and_allowlist_nodes_are_not_whole_file_exemptions() -> None:
|
||
source = (ROOT / SELF_PATH).read_text(encoding="utf8")
|
||
rules = (Rule("HOME_PREFIX", USER_HOME_PREFIX),)
|
||
assert scan_text(SELF_PATH, source, rules)[0] == []
|
||
dirty = source + f'\nextra = {USER_HOME_PREFIX + "new-person/private.txt"!r}'
|
||
assert len(scan_text(SELF_PATH, dirty, rules)[0]) == 1
|
||
mutated = source.replace(USER_HOME_PREFIX + "example/private.txt", USER_HOME_PREFIX + "new-person/private.txt")
|
||
assert scan_text(SELF_PATH, mutated, rules)[0]
|
||
|
||
|
||
@pytest.mark.parametrize("path,excluded", [
|
||
("assets/file.PDF", True), ("assets/file.wasm", True), ("vendor/lib.txt", True),
|
||
("references/open_source_sources/source.txt", True),
|
||
("references/oracle/evidence.json", False),
|
||
("nested/vendor/local.txt", False),
|
||
(PROTECTED_PREFIX + "sample/references/evidence.json", False),
|
||
])
|
||
def test_explicit_exclusions_only(path: str, excluded: bool) -> None:
|
||
assert _excluded(path) is excluded
|
||
|
||
|
||
def test_optional_local_markers_are_additive_and_not_values_in_logs() -> None:
|
||
rules = build_rules('FORBIDDEN_PRIVACY_MARKERS = ("synthetic",)', "# note\n\nfictional-local-secret\n")
|
||
assert [r.code for r in rules] == ["R001", "HOME_PREFIX", "LOCAL_003"]
|
||
path = USER_HOME_PREFIX + "new-person/fictional-local-secret.txt"
|
||
findings, _ = scan_text(path, "fictional-local-secret\nsynthetic", rules)
|
||
report = format_report(ScanResult(findings=findings))
|
||
assert "fictional-local-secret" not in report
|
||
assert "new-person" not in report
|
||
assert "synthetic" not in report
|
||
assert "fictional-local-secret" not in repr(rules)
|
||
assert "lines=1" in report and "lines=2" in report
|
||
|
||
|
||
def test_git_enumeration_is_nul_delimited_and_status_first(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
calls = []
|
||
def run(argv, **kwargs):
|
||
calls.append(argv)
|
||
return subprocess.CompletedProcess(argv, 0, b"tests/a b.py\0tests/\xe4\xb8\xad.py\0" if argv[1] == "ls-files" else b"", b"")
|
||
monkeypatch.setattr(subprocess, "run", run)
|
||
assert tracked_paths(ROOT) == ("tests/a b.py", "tests/中.py")
|
||
assert calls == [["git", "status", "-sb"], ["git", "ls-files", "-z"]]
|
||
|
||
|
||
def test_repository_scan_includes_new_tracked_and_all_protected_files(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
rules = _synthetic_rules()
|
||
paths = {f"{PROTECTED_PREFIX}sample/{i}.md": rules[0].value for i in range(186)}
|
||
paths.update({"tests/new_test.py": rules[0].value,
|
||
UPSTREAM_EXCEPTION: rules[0].value,
|
||
"nested/" + UPSTREAM_EXCEPTION: rules[0].value,
|
||
"vendor/source.py": rules[0].value})
|
||
paths[GOLDEN_PATH] = f'FORBIDDEN_PRIVACY_MARKERS = ({rules[0].value!r},)'
|
||
monkeypatch.setattr(Path, "exists", lambda _: False)
|
||
monkeypatch.setattr(sys.modules[__name__], "tracked_paths", lambda _: tuple(paths))
|
||
monkeypatch.setattr(sys.modules[__name__], "_read", lambda root, path: paths.get(path, ""))
|
||
monkeypatch.setattr(sys.modules[__name__], "_git", lambda *args: LOCAL_PATH.encode())
|
||
result = scan_repository(ROOT)
|
||
assert result.protected == 186
|
||
assert len(result.findings) == 188
|
||
assert result.upstream_exceptions == 1
|
||
assert result.excluded == 1
|
||
|
||
|
||
@pytest.mark.parametrize("path,count", [
|
||
("tests/test_rectification_v5_services.py", 2),
|
||
("references/oracle/pyjhora_multi_case_panchanga_gulika_replay_2026_07_23.json", 1),
|
||
])
|
||
def test_reviewed_numerical_collisions_are_exact_nodes_only(path: str, count: int) -> None:
|
||
__tracebackhide__ = True
|
||
rules = build_rules((ROOT / GOLDEN_PATH).read_text(encoding="utf8"))
|
||
source = (ROOT / path).read_text(encoding="utf8")
|
||
marker = next(r.value for r in rules if r.code == "R003")
|
||
findings, permitted = scan_text(path, source, rules)
|
||
assert len(findings) == 0 and permitted == count
|
||
assert len(scan_text("new/" + path, source, rules)[0]) == count
|
||
# A same-file appended field cannot borrow a reviewed value's permission.
|
||
if path.endswith(".py"):
|
||
dirty = source + f'\nextra = {marker!r}\n'
|
||
assert len(scan_text(path, dirty, rules)[0]) == 1
|
||
dirty = source.replace("test_hour_window_keeps_all_seventeen_signature_clusters_including_late_tail", "test_unreviewed")
|
||
assert len(scan_text(path, dirty, rules)[0]) == count
|
||
dirty = source.replace('"14:46", "14:47"', '"14:45", "14:47"')
|
||
assert len(scan_text(path, dirty, rules)[0]) == count
|
||
else:
|
||
dirty = source.replace('"scope":', f'"extra": {json.dumps(marker)}, "scope":', 1)
|
||
assert len(scan_text(path, dirty, rules)[0]) == 1
|
||
for old, new in [("public_smoke_chennai_1996", "unreviewed_case"),
|
||
("yamaganda_kaalam", "birth_time"),
|
||
("1996-12-07", "2000-01-01")]:
|
||
assert len(scan_text(path, source.replace(old, new), rules)[0]) == count
|
||
assert len(scan_text(path, source.replace(marker + ":18", marker + ":19"), rules)[0]) == count
|
||
|
||
|
||
def test_local_rule_is_not_exempted_by_golden_declaration() -> None:
|
||
source = 'FORBIDDEN_PRIVACY_MARKERS = ("synthetic",)'
|
||
rules = build_rules(source, "synthetic")
|
||
findings, permitted = scan_text(GOLDEN_PATH, source, rules)
|
||
assert permitted == 1
|
||
assert [f.rule for f in findings] == ["LOCAL_001"]
|
||
|
||
|
||
def test_unicode_ast_columns_and_same_line_extra_marker() -> None:
|
||
rules = _synthetic_rules()
|
||
source = f'FORBIDDEN_PRIVACY_MARKERS = ("虚构", {rules[0].value!r}); extra = {rules[0].value!r}'
|
||
findings, permitted = scan_text(GOLDEN_PATH, source, rules)
|
||
assert permitted == 1 and len(findings) == 1
|
||
assert findings[0].line == 1
|
||
|
||
|
||
def test_repository_scan_reports_local_tracked_deleted_and_unreadable(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
source = 'FORBIDDEN_PRIVACY_MARKERS = ("synthetic",)'
|
||
paths = (GOLDEN_PATH, LOCAL_PATH, "deleted.md", "unreadable.md", "binary.dat")
|
||
def read(root, path):
|
||
if path == "deleted.md":
|
||
raise FileNotFoundError
|
||
if path == "unreadable.md":
|
||
raise PermissionError("must never be rendered")
|
||
return {GOLDEN_PATH: source, LOCAL_PATH: "fictional-local", "binary.dat": "\0synthetic\0"}[path]
|
||
monkeypatch.setattr(Path, "exists", lambda _: True)
|
||
monkeypatch.setattr(sys.modules[__name__], "tracked_paths", lambda _: paths)
|
||
monkeypatch.setattr(sys.modules[__name__], "_read", read)
|
||
monkeypatch.setattr(sys.modules[__name__], "_git", lambda *args: LOCAL_PATH.encode())
|
||
result = scan_repository(ROOT)
|
||
assert result.deleted == 1
|
||
assert {f.rule for f in result.findings} == {"LOCAL_TRACKED", "LOCAL_001", "READ_ERROR", "R001"}
|
||
assert "must never be rendered" not in format_report(result)
|
||
|
||
|
||
def test_git_failures_do_not_leak_stderr_or_command_values(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
def run(*args, **kwargs):
|
||
raise subprocess.CalledProcessError(1, ["git", "fictional-sensitive-argv"], stderr="fictional-sensitive-stderr")
|
||
monkeypatch.setattr(subprocess, "run", run)
|
||
with pytest.raises(ScanError, match="^GIT_READ$"):
|
||
tracked_paths(ROOT)
|
||
|
||
|
||
def test_optional_file_missing_does_not_disable_repository_scan(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
source = 'FORBIDDEN_PRIVACY_MARKERS = ("synthetic",)'
|
||
monkeypatch.setattr(Path, "exists", lambda _: False)
|
||
monkeypatch.setattr(sys.modules[__name__], "tracked_paths", lambda _: ("new.md",))
|
||
def read(root, path):
|
||
if path == LOCAL_PATH:
|
||
raise FileNotFoundError
|
||
return source if path == GOLDEN_PATH else "synthetic"
|
||
monkeypatch.setattr(sys.modules[__name__], "_read", read)
|
||
monkeypatch.setattr(sys.modules[__name__], "_git", lambda *args: LOCAL_PATH.encode())
|
||
result = scan_repository(ROOT)
|
||
assert len(result.findings) == 1 and result.findings[0].rule == "R001"
|
||
|
||
|
||
@pytest.mark.parametrize("target_text", [
|
||
"../fictional-outside/fictional-forbidden-token.txt",
|
||
"../fictional-missing/fictional-forbidden-token.txt",
|
||
], ids=["external-target", "dangling-target"])
|
||
def test_symlink_scans_target_text_without_opening_target(monkeypatch: pytest.MonkeyPatch, target_text: str) -> None:
|
||
from types import SimpleNamespace
|
||
target = ROOT / "tracked-link"
|
||
calls = []
|
||
def lstat(path):
|
||
calls.append(path)
|
||
return SimpleNamespace(st_mode=stat.S_IFLNK if path == target else stat.S_IFDIR)
|
||
def forbidden_read(path):
|
||
pytest.fail("symlink content must not be opened")
|
||
monkeypatch.setattr(Path, "lstat", lstat)
|
||
monkeypatch.setattr(Path, "readlink", lambda path: Path(target_text))
|
||
monkeypatch.setattr(Path, "read_bytes", forbidden_read)
|
||
text = _read(ROOT, "tracked-link")
|
||
findings, _ = scan_text("tracked-link", text, _synthetic_rules())
|
||
assert len(findings) == 1 and findings[0].rule == "R001"
|
||
assert calls == [ROOT, target]
|
||
|
||
|
||
@pytest.mark.parametrize("junction", [False, True], ids=["symlink", "junction"])
|
||
def test_linked_ancestor_fails_without_accessing_descendants(monkeypatch: pytest.MonkeyPatch, junction: bool) -> None:
|
||
from types import SimpleNamespace
|
||
ancestor = ROOT / "linked-directory"
|
||
calls = []
|
||
def lstat(path):
|
||
calls.append(path)
|
||
if path == ROOT:
|
||
return SimpleNamespace(st_mode=stat.S_IFDIR)
|
||
assert path == ancestor
|
||
return SimpleNamespace(st_mode=stat.S_IFDIR if junction else stat.S_IFLNK,
|
||
st_file_attributes=stat.FILE_ATTRIBUTE_REPARSE_POINT if junction else 0)
|
||
def forbidden_access(path):
|
||
pytest.fail("ancestor link must not be followed")
|
||
monkeypatch.setattr(Path, "lstat", lstat)
|
||
monkeypatch.setattr(Path, "readlink", forbidden_access)
|
||
monkeypatch.setattr(Path, "read_bytes", forbidden_access)
|
||
with pytest.raises(OSError, match="^READ_ERROR$"):
|
||
_read(ROOT, "linked-directory/child.txt")
|
||
assert calls == [ROOT, ancestor]
|
||
|
||
|
||
def test_readlink_failure_is_not_reported_as_deleted(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
from types import SimpleNamespace
|
||
target = ROOT / "tracked-link"
|
||
monkeypatch.setattr(Path, "lstat", lambda path: SimpleNamespace(st_mode=stat.S_IFLNK if path == target else stat.S_IFDIR))
|
||
def readlink(path):
|
||
raise FileNotFoundError("fictional-sensitive-error")
|
||
monkeypatch.setattr(Path, "readlink", readlink)
|
||
with pytest.raises(OSError, match="^READ_ERROR$") as error:
|
||
_read(ROOT, "tracked-link")
|
||
assert not isinstance(error.value, FileNotFoundError)
|
||
|
||
|
||
def test_missing_regular_tracked_file_remains_deleted(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
from types import SimpleNamespace
|
||
def lstat(path):
|
||
if path == ROOT:
|
||
return SimpleNamespace(st_mode=stat.S_IFDIR)
|
||
raise FileNotFoundError
|
||
monkeypatch.setattr(Path, "lstat", lstat)
|
||
with pytest.raises(FileNotFoundError):
|
||
_read(ROOT, "deleted.txt")
|
||
|
||
|
||
def test_regular_file_still_scans_bytes(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
from types import SimpleNamespace
|
||
target = ROOT / "regular.txt"
|
||
calls = []
|
||
monkeypatch.setattr(Path, "lstat", lambda path: SimpleNamespace(st_mode=stat.S_IFREG if path == target else stat.S_IFDIR))
|
||
def read_bytes(path):
|
||
calls.append(path)
|
||
return b"fictional-forbidden-token"
|
||
monkeypatch.setattr(Path, "read_bytes", read_bytes)
|
||
assert _read(ROOT, "regular.txt") == "fictional-forbidden-token"
|
||
assert calls == [target]
|
||
|
||
|
||
def test_optional_local_utf8_bom_does_not_disable_first_rule() -> None:
|
||
source = 'FORBIDDEN_PRIVACY_MARKERS = ("synthetic",)'
|
||
local = "fictional-local-secret"
|
||
plain = build_rules(source, local)
|
||
with_bom = build_rules(source, codecs.BOM_UTF8.decode("utf8") + local)
|
||
assert plain == with_bom
|
||
findings, _ = scan_text("tests/synthetic.txt", local, with_bom)
|
||
assert len(findings) == 1 and findings[0].rule == "LOCAL_001"
|
||
|
||
|
||
def test_quick_profile_explicitly_includes_repository_guard() -> None:
|
||
source = (ROOT / "scripts/run_quality_gate.py").read_text(encoding="utf8")
|
||
node = _assignment(ast.parse(source), "CORE_PYTEST_TARGETS")
|
||
assert node is not None and SELF_PATH in ast.literal_eval(node.value)
|