jesse-uxandGrok (agent loop)
8e0503f9c0
Cache work that depends only on the candidate minute, drop unread response fields, and reject windows above 241 candidates before any chart is built. Eleven canonical scenario hashes match 77057a819. Sealed and reported-offset replays match the 2026-10-08 records (changed_trial_count 8 and 581) and are stored as scoring_perf_2026_10_13.
Co-Authored-By: Grok (agent loop)
789 lines
37 KiB
Python
789 lines
37 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"
|
||
report_path = "frontend/tests/fixtures/report-density-fictional-reader.json"
|
||
if path not in {service_path, replay_path, report_path}:
|
||
return []
|
||
try:
|
||
tree = ast.parse(source)
|
||
except SyntaxError:
|
||
return []
|
||
if path == report_path:
|
||
# BUG-1010: engine timestamp from fictional input, independently replayed
|
||
# during acceptance (TASK-report-density-fix-20260923, P1-B; BUG-1008).
|
||
# Pin provenance, JSON node, source line and complete computed row; allow
|
||
# only the colliding timestamp span, not the large markdown string.
|
||
try:
|
||
data = json.loads(source)
|
||
if data["fixtureProvenance"] != {
|
||
"fictional": True,
|
||
"generator": "cmd_full_reading -> build_professional_report_reference_packet",
|
||
"input": {
|
||
"year": 2000, "month": 1, "day": 1, "hour": 12,
|
||
"minute": 0, "second": 0, "lat": 0.0, "lon": 0.0,
|
||
"tz": 0, "ayanamsa": "lahiri", "node_mode": "mean",
|
||
"house_system": "whole_sign", "today": "2026-09-22",
|
||
"target_year": 2026,
|
||
},
|
||
} or data["fixtureProvenance"]["fictional"] is not True:
|
||
return []
|
||
nodes = [v for k, v in zip(tree.body[0].value.keys, tree.body[0].value.values)
|
||
if isinstance(k, ast.Constant) and k.value == "markdown"]
|
||
if len(nodes) != 1 or not isinstance(nodes[0], ast.Constant) or nodes[0].lineno != 22:
|
||
return []
|
||
row = (f"| 5 | 4 | 5 | 2028-01-24 {marker}:39 (unresolved_external_tuple_boundary) "
|
||
"| 0.019276268255984996 | pyjhora_behavior_only / not_multiengine_parity |")
|
||
if data["markdown"].splitlines().count(row) != 1:
|
||
return []
|
||
start, end = _node_span(source, nodes[0])
|
||
literal = source[start:end]
|
||
encoded_row = r"\n" + json.dumps(row, ensure_ascii=False)[1:-1] + r"\n"
|
||
if literal.count(encoded_row) != 1:
|
||
return []
|
||
offset = start + literal.index(encoded_row) + encoded_row.index(marker)
|
||
return [(offset, offset + len(marker))]
|
||
except (ValueError, KeyError, IndexError, AttributeError, TypeError):
|
||
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]):
|
||
# 原值: node.lineno not in {1182, 1190}
|
||
# 新值: node.lineno not in {1195, 1203}
|
||
# 原因: BUG-1325 在本文件前面加了三栏注释，这两处时钟集合整段下移，数值本身没变
|
||
if not isinstance(node, ast.Set) or node.lineno not in {1195, 1203}:
|
||
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_local_python_venv_is_not_tracked() -> None:
|
||
paths = tracked_paths(ROOT)
|
||
assert ".venv" not in paths
|
||
assert not any(path.startswith(".venv/") for path in paths)
|
||
|
||
|
||
def test_gitignore_covers_venv_symlink_name() -> None:
|
||
assert _git(ROOT, "check-ignore", "--no-index", ".venv").decode("utf8").strip() == ".venv"
|
||
|
||
|
||
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_report_fixture_collision_is_one_reviewed_timestamp_only() -> None:
|
||
__tracebackhide__ = True
|
||
path = "frontend/tests/fixtures/report-density-fictional-reader.json"
|
||
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 == 1
|
||
for dirty_path, dirty in [
|
||
("new/" + path, source),
|
||
(path, "\n" + source),
|
||
(path, source.replace('"fictional": true', '"fictional": false', 1)),
|
||
(path, source.replace('"year": 2000', '"year": 2001', 1)),
|
||
(path, source.replace('"generator": "cmd_full_reading', '"generator": "unreviewed', 1)),
|
||
(path, source.replace('"markdown":', '"other":', 1)),
|
||
(path, source.replace(marker + ":39", marker + ":40", 1)),
|
||
(path, source.replace("| 5 | 4 | 5 | 2028-01-24", "| 6 | 4 | 5 | 2028-01-24", 1)),
|
||
]:
|
||
findings, permitted = scan_text(dirty_path, dirty, rules)
|
||
assert len(findings) == 1 and permitted == 0
|
||
# Extra markers in the same markdown node or another field remain findings.
|
||
for dirty in [
|
||
source.replace('"markdown": "', '"markdown": "' + marker + r"\n", 1),
|
||
source.replace('"markdown":', f'"extra": {json.dumps(marker)}, "markdown":', 1),
|
||
]:
|
||
findings, permitted = scan_text(path, dirty, rules)
|
||
assert len(findings) == 1 and permitted == 1
|
||
row = next(line for line in json.loads(source)["markdown"].splitlines() if marker in line)
|
||
encoded_row = json.dumps(row, ensure_ascii=False)[1:-1]
|
||
dirty = source.replace(encoded_row, encoded_row + r"\n" + encoded_row, 1)
|
||
findings, permitted = scan_text(path, dirty, rules)
|
||
assert len(findings) == 2 and permitted == 0
|
||
findings, permitted = scan_text(path, source, rules + (Rule("LOCAL_001", marker),))
|
||
assert [finding.rule for finding in findings] == ["LOCAL_001"] and permitted == 1
|
||
|
||
|
||
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)
|