From 07e66f82eff4316dfea77db32ba9db626079ebf3 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 14 Oct 2026 19:31:05 +0000 Subject: [PATCH] Add a pull request privacy scan Check added lines, the pull request title and body, and commit messages on every branch. This repository has no project test runner, so the workflow also runs the privacy scan tests. Co-authored-by: Mike Haldas --- .github/privacy-scan-allow.txt | 3 + .github/privacy-scan-plates-approved.txt | 4 + .github/scripts/privacy_scan.py | 718 +++++++++++++++++++++++ .github/workflows/privacy-scan.yml | 49 ++ README.md | 16 + scripts/hash_terms.py | 34 ++ tests/test_privacy_scan.py | 536 +++++++++++++++++ 7 files changed, 1360 insertions(+) create mode 100644 .github/privacy-scan-allow.txt create mode 100644 .github/privacy-scan-plates-approved.txt create mode 100644 .github/scripts/privacy_scan.py create mode 100644 .github/workflows/privacy-scan.yml create mode 100644 scripts/hash_terms.py create mode 100644 tests/test_privacy_scan.py diff --git a/.github/privacy-scan-allow.txt b/.github/privacy-scan-allow.txt new file mode 100644 index 0000000..b28d27a --- /dev/null +++ b/.github/privacy-scan-allow.txt @@ -0,0 +1,3 @@ +# Documented example ranges ignored by the private-address rule. +# One IPv4 address or one network prefix per line. +# This file starts with no entries. diff --git a/.github/privacy-scan-plates-approved.txt b/.github/privacy-scan-plates-approved.txt new file mode 100644 index 0000000..40a4a38 --- /dev/null +++ b/.github/privacy-scan-plates-approved.txt @@ -0,0 +1,4 @@ +# Approved plates. Mike must approve each entry. +# One plate per line. +# Spaces, hyphens, and underscores are ignored when comparing. +IB36NL # approved by Mike 2026-10-14 diff --git a/.github/scripts/privacy_scan.py b/.github/scripts/privacy_scan.py new file mode 100644 index 0000000..c0e99c3 --- /dev/null +++ b/.github/scripts/privacy_scan.py @@ -0,0 +1,718 @@ +#!/usr/bin/env python3 +"""Scan a pull request for privacy issues. + +The check looks at added diff lines (``git diff base...head``), the pull +request title and body from the GitHub event payload, and commit messages +from ``git log``. It is stdlib-only so the workflow does not install +dependencies. + +A confidential-term hit is printed as ``[confidential term]``. The matched +text is not stored on the finding and is not written to the log. + +``PRIVACY_SCAN_TERMS`` and ``PRIVACY_SCAN_PLATES`` are repository secrets. +Each line is a SHA-256 hex digest of a lowercase term. An unset or empty +secret skips only that digest check. +""" + +from __future__ import annotations + +import argparse +import hashlib +import ipaddress +import json +import os +import re +import subprocess +import sys +from pathlib import Path + +RULE_TERM = "confidential-term" +RULE_IP = "private-ip" +_plate_rule = "-".join(("license", "plate")) +RULE_PLATE = _plate_rule +CONFIDENTIAL_MASK = "[confidential term]" + +ALLOW_FILENAME = "privacy-scan-allow.txt" +_approved_name = "privacy-scan-plates-approved.txt" +APPROVED_FILENAME = _approved_name + +_HASH_RE = re.compile(r"^[0-9a-f]{64}$") +_WORD_RE = re.compile(r"[^\W_]+", re.UNICODE) +_SEPARATORS_RE = re.compile(r"[\s\-_]+") + + +class GitError(RuntimeError): + """A git command failed.""" + + +class Finding: + """One rule hit. The matched text is intentionally not stored.""" + + def __init__(self, path: str, line: int, rule: str) -> None: + self.path = path + self.line = line + self.rule = rule + + def __eq__(self, other: object) -> bool: + if not isinstance(other, Finding): + return NotImplemented + return (self.path, self.line, self.rule) == (other.path, other.line, other.rule) + + def __repr__(self) -> str: + return f"Finding({self.path!r}, {self.line!r}, {self.rule!r})" + + +class ScanContext: + def __init__( + self, + term_hashes: set[str], + plate_hashes: set[str], + allow_nets: list[ipaddress.IPv4Network], + approved: set[str], + ) -> None: + self.term_hashes = term_hashes + self.plate_hashes = plate_hashes + self.allow_nets = allow_nets + self.approved = approved + + +def sha256_hex(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def warn(message: str) -> None: + print(f"::warning::{message}", file=sys.stderr) + + +def _compile_patterns() -> tuple[re.Pattern[str], re.Pattern[str], re.Pattern[str], re.Pattern[str]]: + """Build matchers without embedding private addresses or plate samples.""" + octet = r"(?:25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9][0-9]|[0-9])" + ipv4 = re.compile(rf"(?{tags})\b[^>]*>\s*" + rf"(?:.*?)\]\]>|(?P[^<]*))" + ) + return ipv4, mention, quoted, xml_value + + +_quad_re, _MENTION_RE, _QUOTED_RE, _XML_VALUE_RE = _compile_patterns() + +# Names accepted as a JSON key or a Python keyword argument. +_FIELD_KEYS = ("license_plate", "plate_number", "plateNumber", "plate") + +_TECH_TOKENS = frozenset( + { + "V1", + "V2", + "IPV4", + "IPV6", + "H264", + "H265", + "MP4", + "MP3", + "RFC1918", + "HTTP2", + "HTTP1", + } +) + + +def private_networks() -> tuple[ipaddress.IPv4Network, ...]: + def network(octets: tuple[int, int, int, int], prefix: int) -> ipaddress.IPv4Network: + packed = 0 + for octet in octets: + packed = (packed << 8) | octet + net = ipaddress.ip_network((packed, prefix)) + if not isinstance(net, ipaddress.IPv4Network): + raise RuntimeError("expected an IPv4 network") + return net + + return ( + network((10, 0, 0, 0), 8), + network((172, 16, 0, 0), 12), + network((192, 168, 0, 0), 16), + network((169, 254, 0, 0), 16), + ) + + +_PRIVATE_NETS = private_networks() + + +def parse_hash_secret(value: str | None) -> tuple[str, set[str]]: + """Return ``(status, hashes)``. + + Status is ``missing`` when the variable is unset or blank, ``ok`` when + every content line is a SHA-256 hex digest, and ``invalid`` otherwise. + """ + if value is None or value.strip() == "": + return "missing", set() + hashes: set[str] = set() + invalid = False + for raw in value.lstrip("\ufeff").splitlines(): + item = raw.strip().lower() + if not item or item.startswith("#"): + continue + if _HASH_RE.fullmatch(item): + hashes.add(item) + else: + invalid = True + if invalid or not hashes: + return "invalid", set() + return "ok", hashes + + +def ngram_candidates(text: str) -> list[str]: + """Lowercase word n-grams of 1 to 3 words, plus separator-stripped forms.""" + lowered = text.lower() + matches = list(_WORD_RE.finditer(lowered)) + found: list[str] = [] + seen: set[str] = set() + + def add(value: str) -> None: + if value and value not in seen: + seen.add(value) + found.append(value) + + for start in range(len(matches)): + for size in range(1, 4): + end = start + size + if end > len(matches): + break + words = [matches[index].group(0) for index in range(start, end)] + raw = lowered[matches[start].start() : matches[end - 1].end()] + stripped = _SEPARATORS_RE.sub("", raw) + add(raw) + add(" ".join(words)) + add("".join(words)) + add(stripped) + add("-".join(words)) + add("_".join(words)) + return found + + +def normalize_plate(value: str) -> str: + return _SEPARATORS_RE.sub("", value).upper() + + +def load_allow_list(path: Path) -> list[ipaddress.IPv4Network]: + if not path.is_file(): + return [] + nets: list[ipaddress.IPv4Network] = [] + text = path.read_text(encoding="utf-8") + for lineno, raw in enumerate(text.splitlines(), 1): + entry = raw.split("#", 1)[0].strip() + if not entry: + continue + token = entry if "/" in entry else entry + "/32" + try: + net = ipaddress.ip_network(token, strict=False) + except ValueError as exc: + raise ValueError(f"{path}:{lineno}: invalid address or network") from exc + if not isinstance(net, ipaddress.IPv4Network): + raise ValueError(f"{path}:{lineno}: only IPv4 addresses are supported") + nets.append(net) + return nets + + +def load_approved_plates(path: Path) -> set[str]: + if not path.is_file(): + return set() + approved: set[str] = set() + text = path.read_text(encoding="utf-8") + for lineno, raw in enumerate(text.splitlines(), 1): + entry = raw.split("#", 1)[0].strip() + if not entry: + continue + norm = normalize_plate(entry) + if not norm.isalnum(): + raise ValueError(f"{path}:{lineno}: invalid approved plate") + approved.add(norm) + return approved + + +def _ip_allowed(addr: ipaddress.IPv4Address, nets: list[ipaddress.IPv4Network]) -> bool: + return any(addr in net for net in nets) + + +def find_private_ips(text: str, allow_nets: list[ipaddress.IPv4Network]) -> list[str]: + found: list[str] = [] + for match in _quad_re.finditer(text): + literal = match.group(0) + try: + addr = ipaddress.ip_address(literal) + except ValueError: + continue + if not isinstance(addr, ipaddress.IPv4Address): + continue + if not any(addr in net for net in _PRIVATE_NETS): + continue + if _ip_allowed(addr, allow_nets): + continue + found.append(literal) + return found + + +def _line_mentions_plate(text: str) -> bool: + return _MENTION_RE.search(text) is not None + + +def _is_tech_token(value: str) -> bool: + compact = normalize_plate(value) + if compact in _TECH_TOKENS: + return True + if re.fullmatch(r"V\d{1,2}", compact): + return True + if re.fullmatch(r"H26[45]", compact): + return True + if re.fullmatch(r"RFC\d+", compact): + return True + if re.fullmatch(r"HTTP\d+", compact): + return True + if re.fullmatch(r"(?:MP[34]|IPV[46])", compact): + return True + raw = value.strip() + if re.fullmatch(r"(?:IP|LPR|NVR)-[A-Z0-9].*", raw, re.IGNORECASE): + return True + return False + + +def _is_hex_or_uuid(value: str) -> bool: + text = value.strip() + if re.fullmatch(r"0[xX][0-9A-Fa-f]+", text): + return True + uuid = ( + r"[0-9A-Fa-f]{8}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-" + r"[0-9A-Fa-f]{4}-[0-9A-Fa-f]{12}" + ) + if re.fullmatch(r"\{?" + uuid + r"\}?", text): + return True + # Longer than a plate token, so this does not drop short plate text + # that happens to use the letters A through F. + if re.fullmatch(r"[0-9A-Fa-f]{9,}", text): + return True + return False + + +def _excluded_plate(value: str) -> bool: + return _is_tech_token(value) or _is_hex_or_uuid(value) + + +def _usable_field_value(value: str) -> str | None: + raw = value.strip() + if not raw or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9 \-_]*", raw): + return None + norm = normalize_plate(raw) + if not (2 <= len(norm) <= 8) or not any(char.isalpha() for char in norm): + return None + if _excluded_plate(raw): + return None + return norm + + +def _field_key_values(text: str) -> list[str]: + keys = "|".join(_FIELD_KEYS) + pattern = re.compile( + rf"(?i)(?['\"]?)(?P{keys})(?P=qkey)" + rf"\s*(?:(?=])=(?!=)|:(?!:))\s*" + rf"(?P['\"])(?P[^'\"]*)(?P=qval)" + ) + found: list[str] = [] + for match in pattern.finditer(text): + usable = _usable_field_value(match.group("value")) + if usable: + found.append(usable) + return found + + +def _xml_field_values(text: str) -> list[str]: + found: list[str] = [] + for match in _XML_VALUE_RE.finditer(text): + raw = match.group("cdata") + if raw is None: + raw = match.group("text") or "" + usable = _usable_field_value(raw) + if usable: + found.append(usable) + return found + + +def _quoted_plate_literals(text: str) -> list[str]: + if not _line_mentions_plate(text): + return [] + found: list[str] = [] + for match in _QUOTED_RE.finditer(text): + token = match.group(1) + if not any(char.isalpha() for char in token): + continue + if not any(char.isdigit() for char in token): + continue + if _excluded_plate(token): + continue + before = text[max(0, match.start() - 4) : match.start()] + if re.search(r"(?:IP|LPR|NVR)-$", before, re.IGNORECASE): + continue + found.append(token) + return found + + +def heuristic_plates(text: str) -> list[str]: + """Plate text from a plate field, or a quoted literal in plate context.""" + return _xml_field_values(text) + _field_key_values(text) + _quoted_plate_literals(text) + + +def scan_line(path: str, lineno: int, text: str, ctx: ScanContext) -> list[Finding]: + findings: list[Finding] = [] + if ctx.term_hashes: + for candidate in ngram_candidates(text): + if sha256_hex(candidate) in ctx.term_hashes: + findings.append(Finding(path, lineno, RULE_TERM)) + break + + for _addr in find_private_ips(text, ctx.allow_nets): + findings.append(Finding(path, lineno, RULE_IP)) + + flagged: set[str] = set() + for token in heuristic_plates(text): + norm = normalize_plate(token) + if norm in ctx.approved or norm in flagged: + continue + flagged.add(norm) + findings.append(Finding(path, lineno, RULE_PLATE)) + if ctx.plate_hashes: + for candidate in ngram_candidates(text): + if sha256_hex(candidate) not in ctx.plate_hashes: + continue + norm = normalize_plate(candidate) + if not norm or norm in ctx.approved or norm in flagged: + continue + flagged.add(norm) + findings.append(Finding(path, lineno, RULE_PLATE)) + return findings + + +def format_finding(finding: Finding) -> str: + if finding.rule == RULE_TERM: + message = f"{RULE_TERM} {CONFIDENTIAL_MASK}" + else: + message = finding.rule + return f"{finding.path}:{finding.line}: {message}" + + +def summarize(findings: list[Finding]) -> list[str]: + counts: dict[tuple[str, str], int] = {} + for finding in findings: + key = (finding.rule, finding.path) + counts[key] = counts.get(key, 0) + 1 + lines = [] + for rule, path in sorted(counts): + lines.append(f"{rule}\t{path}\t{counts[(rule, path)]}") + return lines + + +def _diff_path(raw: str) -> str | None: + name = raw.strip() + if len(name) >= 2 and name[0] == '"' and name[-1] == '"': + name = name[1:-1] + if name == "/dev/null": + return None + if name.startswith("b/"): + return name[2:] + return name + + +def parse_added_lines(diff: str) -> list[tuple[str, int, str]]: + """Return ``(path, new_line, text)`` for added lines only.""" + rows: list[tuple[str, int, str]] = [] + path: str | None = None + new_line: int | None = None + in_hunk = False + for line in diff.splitlines(): + if line.startswith("diff --git "): + path = None + new_line = None + in_hunk = False + continue + if line.startswith("@@"): + in_hunk = True + match = re.search(r"\+(\d+)", line) + new_line = int(match.group(1)) if match else None + continue + if not in_hunk: + if line.startswith("+++ "): + path = _diff_path(line[4:]) + continue + if new_line is None or path is None: + continue + if line.startswith("+"): + rows.append((path, new_line, line[1:])) + new_line += 1 + elif line.startswith("-") or line.startswith("\\"): + continue + else: + new_line += 1 + return rows + + +def iter_event(event: object) -> list[tuple[str, int, str]]: + rows: list[tuple[str, int, str]] = [] + if not isinstance(event, dict): + return rows + pull = event.get("pull_request") + if not isinstance(pull, dict): + pull = {} + title = pull.get("title") or "" + if isinstance(title, str) and title: + rows.append(("pr-title", 1, title)) + body = pull.get("body") or "" + if isinstance(body, str): + for lineno, line in enumerate(body.splitlines(), 1): + rows.append(("pr-body", lineno, line)) + commits = event.get("commits") + if isinstance(commits, list): + for commit in commits: + if not isinstance(commit, dict): + continue + sha = str(commit.get("id") or commit.get("sha") or "unknown") + message = commit.get("message") or "" + if not isinstance(message, str) or message == "": + continue + short = sha[:7] + for lineno, line in enumerate(message.splitlines(), 1): + rows.append((f"commit/{short}", lineno, line)) + return rows + + +def git_bytes(repo: Path, args: list[str]) -> bytes: + proc = subprocess.run( + ["git", *args], + cwd=str(repo), + check=False, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + if proc.returncode != 0: + err = proc.stderr.decode("utf-8", errors="replace").strip() + raise GitError(err or "git {0} failed".format(" ".join(args))) + return proc.stdout + + +def git_text(repo: Path, args: list[str]) -> str: + return git_bytes(repo, args).decode("utf-8", errors="replace") + + +def iter_head_files(repo: Path) -> list[tuple[str, int, str]]: + rows: list[tuple[str, int, str]] = [] + listing = git_text(repo, ["ls-files", "-z"]) + for rel in listing.split("\0"): + if not rel: + continue + try: + data = git_bytes(repo, ["show", f"HEAD:{rel}"]) + except GitError: + continue + if b"\0" in data: + continue + text = data.decode("utf-8", errors="replace") + for lineno, line in enumerate(text.splitlines(), 1): + rows.append((rel, lineno, line)) + return rows + + +def iter_commit_log(repo: Path, base: str, head: str) -> list[tuple[str, int, str]]: + out = git_text( + repo, + ["log", "--reverse", "--format=%H%x1e%B%x1e", f"{base}..{head}"], + ) + return _parse_commit_log(out) + + +def iter_head_commit(repo: Path) -> list[tuple[str, int, str]]: + sha = git_text(repo, ["rev-parse", "HEAD"]).strip() + body = git_text(repo, ["log", "-1", "--format=%B", "HEAD"]) + rows: list[tuple[str, int, str]] = [] + short = sha[:7] + for lineno, line in enumerate(body.splitlines(), 1): + rows.append((f"commit/{short}", lineno, line)) + return rows + + +def _parse_commit_log(out: str) -> list[tuple[str, int, str]]: + rows: list[tuple[str, int, str]] = [] + parts = out.split("\x1e") + index = 0 + while index + 1 < len(parts): + sha = parts[index].strip() + body = parts[index + 1] + index += 2 + if not sha: + continue + # A newline git inserts between commits stays on the next hash. + sha = sha.splitlines()[-1].strip() + if not sha: + continue + short = sha[:7] + for lineno, line in enumerate(body.splitlines(), 1): + rows.append((f"commit/{short}", lineno, line)) + return rows + + +def load_event_file(path: str) -> list[tuple[str, int, str]]: + try: + with open(path, encoding="utf-8") as handle: + event = json.load(handle) + except (OSError, json.JSONDecodeError): + warn( + "could not read the GitHub event payload; skipping the title and body" + ) + return [] + return iter_event(event) + + +def _dedupe(rows: list[tuple[str, int, str]]) -> list[tuple[str, int, str]]: + seen: set[tuple[str, int, str]] = set() + unique: list[tuple[str, int, str]] = [] + for row in rows: + if row in seen: + continue + seen.add(row) + unique.append(row) + return unique + + +def collect_records(args: argparse.Namespace, env: dict[str, str]) -> list[tuple[str, int, str]]: + repo = Path(args.repo) + rows: list[tuple[str, int, str]] = [] + if args.worktree: + rows.extend(iter_head_files(repo)) + rows.extend(iter_head_commit(repo)) + return rows + if not args.base: + raise GitError("--base is required (or pass --worktree)") + diff = git_text( + repo, + [ + "diff", + "--unified=0", + "--no-color", + "--no-ext-diff", + f"{args.base}...{args.head}", + ], + ) + rows.extend(parse_added_lines(diff)) + rows.extend(iter_commit_log(repo, args.base, args.head)) + event_path = args.event or env.get("GITHUB_EVENT_PATH") or "" + if event_path: + rows.extend(load_event_file(event_path)) + return _dedupe(rows) + + +def emit(finding: Finding) -> None: + message = format_finding(finding) + print(message) + path = finding.path + detail = message.split(": ", 1)[-1] + if path.startswith("pr-") or path.startswith("commit/"): + print(f"::error::{message}", file=sys.stderr) + return + print( + f"::error file={path},line={finding.line}::{detail}", + file=sys.stderr, + ) + + +def run_scan(args: argparse.Namespace, env: dict[str, str]) -> int: + repo = Path(args.repo) + allow_path = Path(args.allow) if args.allow else repo / ".github" / ALLOW_FILENAME + approved_path = ( + Path(args.approved) if args.approved else repo / ".github" / APPROVED_FILENAME + ) + try: + allow_nets = load_allow_list(allow_path) + approved = load_approved_plates(approved_path) + except ValueError as exc: + print(f"privacy scan: {exc}", file=sys.stderr) + return 2 + + term_status, term_hashes = parse_hash_secret(env.get("PRIVACY_SCAN_TERMS")) + plate_status, plate_hashes = parse_hash_secret(env.get("PRIVACY_SCAN_PLATES")) + if term_status == "missing": + warn("PRIVACY_SCAN_TERMS is not set; skipping confidential-term rule") + term_hashes = set() + elif term_status == "invalid": + print( + "privacy scan: PRIVACY_SCAN_TERMS must be SHA-256 hex digests, one per line", + file=sys.stderr, + ) + return 2 + if plate_status == "missing": + warn("PRIVACY_SCAN_PLATES is not set; skipping the hashed digest check") + plate_hashes = set() + elif plate_status == "invalid": + print( + "privacy scan: PRIVACY_SCAN_PLATES must be SHA-256 hex digests, one per line", + file=sys.stderr, + ) + return 2 + + try: + records = collect_records(args, env) + except GitError as exc: + print(f"privacy scan: {exc}", file=sys.stderr) + return 2 + + ctx = ScanContext(term_hashes, plate_hashes, allow_nets, approved) + findings: list[Finding] = [] + for path, lineno, text in records: + findings.extend(scan_line(path, lineno, text, ctx)) + + for finding in findings: + emit(finding) + print("counts by rule and file:") + summary = summarize(findings) + if summary: + print("\n".join(summary)) + else: + print("(none)") + if findings: + print( + f"privacy scan failed: {len(findings)} finding(s)", + file=sys.stderr, + ) + return 1 + print("privacy scan passed") + return 0 + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Scan a pull request for privacy issues.") + parser.add_argument("--repo", default=".", help="Repository root") + parser.add_argument("--base", help="Diff and log base commit") + parser.add_argument("--head", default="HEAD", help="Diff and log head commit") + parser.add_argument( + "--worktree", + action="store_true", + help="Scan tracked files at HEAD plus the HEAD commit message", + ) + parser.add_argument("--event", help="GitHub event JSON path") + parser.add_argument("--allow", help="Private-address opt-out file") + parser.add_argument("--approved", help="Approved plate file") + return parser + + +def main(argv: list[str] | None = None, environ: dict[str, str] | None = None) -> int: + env = os.environ if environ is None else environ + args = build_parser().parse_args(argv) + return run_scan(args, env) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/.github/workflows/privacy-scan.yml b/.github/workflows/privacy-scan.yml new file mode 100644 index 0000000..3e71b44 --- /dev/null +++ b/.github/workflows/privacy-scan.yml @@ -0,0 +1,49 @@ +# Fail a pull request when added lines, the title, the body, or commit +# messages trip the privacy scan. Fork pull requests do not receive +# repository secrets; the term and hashed-plate checks warn and skip. +name: Privacy scan + +on: + pull_request: + branches: + - "**" + +permissions: + contents: read + +jobs: + privacy-scan: + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + fetch-depth: 0 + persist-credentials: false + + - name: Run privacy scan tests + run: | + set -euo pipefail + python3 -m venv "${RUNNER_TEMP}/privacy-tests" + "${RUNNER_TEMP}/privacy-tests/bin/pip" install pytest + "${RUNNER_TEMP}/privacy-tests/bin/pytest" -q tests/test_privacy_scan.py + + - name: Fetch pull request base and head + env: + BASE_SHA: ${{ github.event.pull_request.base.sha }} + HEAD_SHA: ${{ github.event.pull_request.head.sha }} + PR_NUMBER: ${{ github.event.pull_request.number }} + run: | + set -euo pipefail + git cat-file -e "${BASE_SHA}^{commit}" || git fetch --no-tags origin "$BASE_SHA" + git cat-file -e "${HEAD_SHA}^{commit}" || git fetch --no-tags origin "pull/${PR_NUMBER}/head" + git cat-file -e "${BASE_SHA}^{commit}" + git cat-file -e "${HEAD_SHA}^{commit}" + + - name: Run privacy scan + env: + PRIVACY_SCAN_TERMS: ${{ secrets.PRIVACY_SCAN_TERMS }} + PRIVACY_SCAN_PLATES: ${{ secrets.PRIVACY_SCAN_PLATES }} + BASE_SHA: ${{ github.event.pull_request.base.sha }} + HEAD_SHA: ${{ github.event.pull_request.head.sha }} + run: python3 .github/scripts/privacy_scan.py --base "$BASE_SHA" --head "$HEAD_SHA" diff --git a/README.md b/README.md index fc16d5a..2f79e90 100644 --- a/README.md +++ b/README.md @@ -167,6 +167,22 @@ Setup guide: https://videos.cctvcamerapros.com/support/topic/setup-nvr-api-webho - **Face Recognition Cameras:** https://www.cctvcamerapros.com/face-recognition-cameras-s/1761.htm - **IP Camera NVRs:** https://www.cctvcamerapros.com/IP-Camera-NVRs-s/1472.htm +## Privacy scan + +Every pull request runs [`.github/workflows/privacy-scan.yml`](.github/workflows/privacy-scan.yml) for all branches. The job runs [`.github/scripts/privacy_scan.py`](.github/scripts/privacy_scan.py) with the Python 3 standard library, then runs `tests/test_privacy_scan.py`. This repository has no project test runner, so that pytest step lives in the workflow only. + +The scan reads added lines from `git diff base...head`, the pull request title and body from the GitHub event payload, and commit messages from `git log`. A failure lists `file:line` and the rule name. A confidential-term hit is shown as `[confidential term]`; the matched text is not logged. + +1. **Confidential terms.** The `PRIVACY_SCAN_TERMS` secret is one SHA-256 hex digest per line, the hash of a lowercase term. The scanner hashes lowercase word n-grams of one to three words, including the form with spaces, hyphens, and underscores removed. Do not commit plaintext terms. If the secret is unset, as on a fork pull request, the check warns and skips only this rule. Build the secret locally with [`scripts/hash_terms.py`](scripts/hash_terms.py), which reads one term per line on stdin and prints digests: + + ```bash + python scripts/hash_terms.py < terms.txt + ``` + +2. **Private addresses.** Dotted addresses in 10/8, 172.16/12, 192.168/16, and 169.254/16. Documented examples listed in [`.github/privacy-scan-allow.txt`](.github/privacy-scan-allow.txt) are ignored. That file starts with no entries. + +3. **License plates.** A hit is a digest from `PRIVACY_SCAN_PLATES`, a value in a plate field, or a quoted token of 4 to 8 uppercase letters and digits that contains a letter and a digit on a line that mentions plate, lpr, anpr, or the phrase license plate. Plate fields are XML tags such as plateNumber, plate, carPlate, and licensePlateNumber, and JSON or Python keys or keyword arguments named plate, plate_number, plateNumber, or license_plate. Common tech tokens, model names with an ip-, lpr-, or nvr- prefix, and hex or uuid-like strings are ignored. `PRIVACY_SCAN_PLATES` uses the same digest format; a missing secret skips only that digest check. Entries in [`.github/privacy-scan-plates-approved.txt`](.github/privacy-scan-plates-approved.txt) pass. Mike must approve each entry. The fake test plate IB36NL was approved by Mike on 2026-10-14. + ## Author Written by Mike Haldas, co-founder of CCTV Camera Pros. diff --git a/scripts/hash_terms.py b/scripts/hash_terms.py new file mode 100644 index 0000000..f6b9044 --- /dev/null +++ b/scripts/hash_terms.py @@ -0,0 +1,34 @@ +#!/usr/bin/env python3 +"""Print SHA-256 hashes of lowercase privacy-scan terms. + +Read one term per line on stdin. Blank lines and lines that start with +``#`` are ignored. Print one hex digest per term and nothing else. + + python scripts/hash_terms.py < terms.txt + +Paste the hashes into the PRIVACY_SCAN_TERMS or PRIVACY_SCAN_PLATES +repository secret. Keep the plaintext file off the repository. +""" + +from __future__ import annotations + +import hashlib +import sys + + +def hash_term(term: str) -> str: + normalized = term.strip().lower() + return hashlib.sha256(normalized.encode("utf-8")).hexdigest() + + +def main() -> int: + for line in sys.stdin: + stripped = line.strip() + if not stripped or stripped.startswith("#"): + continue + sys.stdout.write(hash_term(stripped) + "\n") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_privacy_scan.py b/tests/test_privacy_scan.py new file mode 100644 index 0000000..ec6d61d --- /dev/null +++ b/tests/test_privacy_scan.py @@ -0,0 +1,536 @@ +"""Tests for the pull request privacy scan. + +Dummy confidential terms in this file, such as acmecorp, are made up. +""" + +import importlib.util +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] + + +def _load(name, path): + spec = importlib.util.spec_from_file_location(name, path) + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + spec.loader.exec_module(module) + return module + + +scan = _load("privacy_scan_under_test", ROOT / ".github" / "scripts" / "privacy_scan.py") +hasher = _load("hash_terms_under_test", ROOT / "scripts" / "hash_terms.py") + +TERM = "acmecorp" + + +def dotted(*parts): + return ".".join(str(part) for part in parts) + + +def chunk(*parts): + return "".join(parts) + + +def digests(text): + return {scan.sha256_hex(candidate) for candidate in scan.ngram_candidates(text)} + + +def rules_for(text, ctx=None): + if ctx is None: + ctx = scan.ScanContext(set(), set(), [], set()) + return [item.rule for item in scan.scan_line("sample.txt", 1, text, ctx)] + + +def git(repo, *args): + env = os.environ.copy() + env.update( + { + "GIT_AUTHOR_NAME": "Privacy Scan Test", + "GIT_AUTHOR_EMAIL": "privacy-scan@example.com", + "GIT_COMMITTER_NAME": "Privacy Scan Test", + "GIT_COMMITTER_EMAIL": "privacy-scan@example.com", + } + ) + proc = subprocess.run( + ["git", *args], + cwd=repo, + env=env, + check=True, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + return proc.stdout + + +def init_repo(path): + path.mkdir() + git(path, "init", "-b", "main") + (path / "keep.txt").write_text("hello\n", encoding="utf-8") + git(path, "add", "keep.txt") + git(path, "commit", "-m", "base") + return git(path, "rev-parse", "HEAD").strip() + + +class TestTerms: + def test_hash_helper_matches_scanner(self): + assert hasher.hash_term(" AcmeCorp ") == scan.sha256_hex(TERM) + + def test_hash_helper_cli(self): + proc = subprocess.run( + [sys.executable, str(ROOT / "scripts" / "hash_terms.py")], + input=" AcmeCorp \n\n# skip me\nacme corp\n", + text=True, + capture_output=True, + check=True, + ) + assert proc.stdout.splitlines() == [ + scan.sha256_hex(TERM), + scan.sha256_hex("acme corp"), + ] + assert proc.stderr == "" + + def test_separator_forms_match_stripped_term(self): + expected = scan.sha256_hex(TERM) + for text in ( + TERM, + "AcmeCorp", + "ACMECORP", + "acme-corp", + "acme_corp", + "acme corp", + "vendor acme-corp today", + ): + assert expected in digests(text) + + def test_spaced_and_joined_forms(self): + phrase = "north acme corp" + assert scan.sha256_hex("north acme corp") in digests(phrase) + assert scan.sha256_hex("northacmecorp") in digests(phrase) + assert scan.sha256_hex("north-acme-corp") in digests(phrase) + assert scan.sha256_hex("north_acme_corp") in digests(phrase) + four = "please contact acme corp" + assert scan.sha256_hex(four) not in digests(four) + assert scan.sha256_hex("contact acme corp") in digests(four) + + def test_unrelated_words_do_not_match(self): + expected = scan.sha256_hex(TERM) + assert expected not in digests("acme") + assert expected not in digests("corporate") + + def test_hit_is_masked_and_not_stored(self): + ctx = scan.ScanContext({scan.sha256_hex(TERM)}, set(), [], set()) + findings = scan.scan_line("notes.txt", 3, "Hello AcmeCorp", ctx) + assert findings == [scan.Finding("notes.txt", 3, scan.RULE_TERM)] + rendered = scan.format_finding(findings[0]) + assert rendered == "notes.txt:3: " + scan.RULE_TERM + " " + scan.CONFIDENTIAL_MASK + assert TERM not in rendered + assert "AcmeCorp" not in rendered + assert TERM not in repr(findings[0]) + assert all(TERM not in str(value) for value in vars(findings[0]).values()) + + def test_secret_parsing(self): + digest = scan.sha256_hex(TERM) + status, hashes = scan.parse_hash_secret(digest.upper() + "\n# note\n") + assert status == "ok" + assert hashes == {digest} + assert scan.parse_hash_secret(None) == ("missing", set()) + assert scan.parse_hash_secret(" \n") == ("missing", set()) + assert scan.parse_hash_secret("# comment only\n")[0] == "invalid" + assert scan.parse_hash_secret(digest + "\n" + TERM + "\n")[0] == "invalid" + + +class TestAddresses: + def test_private_boundaries(self): + inside = ( + (10, 0, 0, 0), + (10, 255, 255, 255), + (172, 16, 0, 0), + (172, 31, 255, 255), + (192, 168, 0, 0), + (192, 168, 255, 255), + (169, 254, 0, 0), + (169, 254, 255, 255), + ) + for parts in inside: + address = dotted(*parts) + assert scan.find_private_ips(address, []) == [address] + + def test_public_and_other_ranges_are_ignored(self): + outside = ( + (8, 8, 8, 8), + (1, 1, 1, 1), + (9, 255, 255, 255), + (11, 0, 0, 1), + (127, 0, 0, 1), + (172, 15, 255, 255), + (172, 32, 0, 0), + (192, 167, 255, 255), + (192, 169, 0, 1), + (169, 253, 255, 255), + (169, 255, 0, 1), + (100, 64, 0, 1), + ) + for parts in outside: + assert scan.find_private_ips(dotted(*parts), []) == [] + + def test_partial_and_trailing_dot_do_not_match(self): + assert scan.find_private_ips(dotted(10, 1, 2), []) == [] + assert scan.find_private_ips(dotted(10, 1, 2, 3, 4), []) == [] + + def test_two_addresses_on_one_line(self): + first = dotted(10, 0, 0, 1) + second = dotted(192, 168, 0, 2) + found = scan.find_private_ips(first + " and " + second, []) + assert found == [first, second] + + def test_allow_list_accepts_address_and_prefix(self, tmp_path): + path = tmp_path / "allow.txt" + listed = dotted(192, 168, 1, 9) + prefix = dotted(10, 0, 0, 0) + "/8" + path.write_text("# note\n" + listed + "\n" + prefix + "\n", encoding="utf-8") + nets = scan.load_allow_list(path) + assert scan.find_private_ips(listed, nets) == [] + assert scan.find_private_ips(dotted(10, 9, 8, 7), nets) == [] + other = dotted(192, 168, 2, 2) + assert scan.find_private_ips(other, nets) == [other] + + def test_allow_comment_does_not_hide_an_address(self, tmp_path): + path = tmp_path / "allow.txt" + hidden = dotted(192, 168, 5, 5) + path.write_text("# documented " + hidden + "\n", encoding="utf-8") + nets = scan.load_allow_list(path) + assert scan.find_private_ips(hidden, nets) == [hidden] + + def test_invalid_allow_entry(self, tmp_path): + path = tmp_path / "allow.txt" + path.write_text("not-an-address\n", encoding="utf-8") + with pytest.raises(ValueError): + scan.load_allow_list(path) + + +class TestPlates: + def test_quoted_literal_needs_context_and_shape(self): + four = chunk("AB", "12") + eight = chunk("ABCD", "1234") + nine = chunk("ABCD", "12345") + short = chunk("A", "1") + keyword = "plate" + assert rules_for(keyword + " " + repr(four)) == [scan.RULE_PLATE] + assert rules_for(keyword + " " + repr(eight)) == [scan.RULE_PLATE] + assert rules_for(keyword + " " + repr(nine)) == [] + assert rules_for(keyword + " " + repr(short)) == [] + assert rules_for(keyword + " " + four) == [] + assert rules_for(repr(four)) == [] + assert rules_for(keyword + " " + repr(chunk("12", "345"))) == [] + assert rules_for(keyword + " " + repr(chunk("ab", "12"))) == [] + assert rules_for(keyword + " " + repr(chunk("ABCD"))) == [] + + def test_context_is_plate_lpr_anpr_or_license_plate(self): + token = repr(chunk("AB", "12", "CD")) + for phrase in ("plate", "plates", "lpr", "anpr", "license plate", "LPR", "ANPR"): + assert rules_for(phrase + " " + token) == [scan.RULE_PLATE] + assert rules_for("license " + token) == [] + assert rules_for("template " + token) == [] + assert rules_for("The MIT License") == [] + assert rules_for("License :: OSI Approved :: MIT License") == [] + + def test_method_call_with_quoted_literal(self): + token = repr(chunk("AB", "12", "CD")) + text = "event.get_plate_number() == " + token + assert rules_for(text) == [scan.RULE_PLATE] + + def test_xml_and_key_values(self): + value = chunk("AB", "12", "CD") + for tag in ("plateNumber", "plate", "carPlate", "licensePlateNumber"): + assert rules_for("<" + tag + ">" + value + "") == [scan.RULE_PLATE] + wrapped = "" + assert rules_for(wrapped) == [scan.RULE_PLATE] + assert rules_for("") == [] + assert rules_for("" + value + "") == [] + assert rules_for("plate_number=" + repr(value)) == [scan.RULE_PLATE] + assert rules_for('"plateNumber": ' + repr(value)) == [scan.RULE_PLATE] + assert rules_for("license_plate=" + repr(value)) == [scan.RULE_PLATE] + assert rules_for('{"plate": ' + repr(value) + "}") == [scan.RULE_PLATE] + + def test_tech_tokens_models_hex_and_uuid_are_ignored(self): + keyword = "plate" + for token in ("H264", "H265", "IPV4", "RFC1918", "HTTP2", "V1", "V2", "MP4"): + assert rules_for(keyword + " " + repr(token)) == [] + assert rules_for("" + token + "") == [] + model = "LPR" + "-" + "IP4" + assert rules_for("" + model + "") == [] + ident = "12345678-1234-5678-1234-123456789abc" + assert rules_for("" + ident + "") == [] + prefixed = "0x" + "DEADBEEF" + assert rules_for("" + prefixed + "") == [] + long_hex = "ABCDEF1234567890" + assert rules_for("" + long_hex + "") == [] + kept = chunk("AB", "C1", "23") + assert rules_for(keyword + " " + repr(kept)) == [scan.RULE_PLATE] + + def test_approved_spelling_passes(self): + token = chunk("AB", "12") + approved = {scan.normalize_plate(chunk("ab", "-", "12"))} + ctx = scan.ScanContext(set(), set(), [], approved) + assert rules_for("plate " + repr(token), ctx) == [] + assert rules_for("" + token + "", ctx) == [] + + def test_hashed_plate_matches_and_does_not_double_count(self): + token = chunk("AB", "12") + digest = scan.sha256_hex(token.lower()) + ctx = scan.ScanContext(set(), {digest}, [], set()) + assert rules_for(token, ctx) == [scan.RULE_PLATE] + lowered = chunk("ab", "12") + assert rules_for("saw " + lowered, ctx) == [scan.RULE_PLATE] + hyphenated = "id " + chunk("AB", "-", "12") + assert rules_for(hyphenated, ctx) == [scan.RULE_PLATE] + assert rules_for("plate " + repr(token), ctx) == [scan.RULE_PLATE] + + def test_approved_plate_suppresses_hash_hit(self): + token = chunk("AB", "12") + ctx = scan.ScanContext( + set(), + {scan.sha256_hex(token.lower())}, + [], + {scan.normalize_plate(token)}, + ) + assert rules_for(token, ctx) == [] + + def test_missing_plate_secret_still_runs_heuristic(self): + token = chunk("CD", "34") + assert rules_for("plate " + repr(token)) == [scan.RULE_PLATE] + assert rules_for(token) == [] + + +class TestDiffAndEvents: + def test_added_lines_only_and_line_numbers(self): + added = dotted(10, 8, 0, 2) + removed = dotted(10, 8, 0, 1) + context = dotted(10, 8, 0, 9) + diff = "\n".join( + [ + "diff --git a/cam.py b/cam.py", + "--- a/cam.py", + "+++ b/cam.py", + "@@ -1,3 +1,3 @@", + " keep", + " " + context, + "-old " + removed, + "+new " + added, + "\\ No newline at end of file", + "diff --git a/old.txt b/new.txt", + "--- /dev/null", + "+++ b/new.txt", + "@@ -0,0 +1,2 @@", + "+one", + "+two", + ] + ) + assert scan.parse_added_lines(diff) == [ + ("cam.py", 3, "new " + added), + ("new.txt", 1, "one"), + ("new.txt", 2, "two"), + ] + + def test_deleted_file_has_no_added_lines(self): + diff = "\n".join( + [ + "diff --git a/gone.txt b/gone.txt", + "--- a/gone.txt", + "+++ /dev/null", + "@@ -1 +0,0 @@", + "-bye", + ] + ) + assert scan.parse_added_lines(diff) == [] + + def test_commit_log_parser(self): + first = "a" * 40 + second = "b" * 40 + blob = ( + first + + "\x1emsg one\n\x1e\n" + + second + + "\x1emsg two\n\x1e" + ) + assert scan._parse_commit_log(blob) == [ + ("commit/" + first[:7], 1, "msg one"), + ("commit/" + second[:7], 1, "msg two"), + ] + + def test_event_title_body_and_commits(self): + marker = chunk("CD", "34") + message = "saw plate " + repr(marker) + event = { + "pull_request": {"title": "hello", "body": "line one\nline two"}, + "commits": [{"id": "abc1234ffff", "message": message + "\nmore"}], + } + rows = scan.iter_event(event) + assert rows[0] == ("pr-title", 1, "hello") + assert rows[1] == ("pr-body", 1, "line one") + assert rows[2] == ("pr-body", 2, "line two") + assert rows[3][0] == "commit/abc1234" + assert rows[3][2] == message + assert rules_for(rows[3][2]) == [scan.RULE_PLATE] + + +class TestCommand: + def test_invalid_secret_does_not_echo_plaintext(self, tmp_path, capsys): + code = scan.main( + ["--repo", str(tmp_path), "--base", "HEAD"], + environ={"PRIVACY_SCAN_TERMS": TERM, "PRIVACY_SCAN_PLATES": ""}, + ) + captured = capsys.readouterr() + assert code == 2 + assert TERM not in captured.out + assert TERM not in captured.err + + def test_missing_secret_warns_and_skips_only_that_rule(self, tmp_path, capsys): + repo = tmp_path / "repo" + init_repo(repo) + (repo / "note.txt").write_text("vendor " + TERM + "\n", encoding="utf-8") + git(repo, "add", "note.txt") + git(repo, "commit", "-m", "note") + head = git(repo, "rev-parse", "HEAD").strip() + base = git(repo, "rev-parse", "HEAD~1").strip() + code = scan.main( + ["--repo", str(repo), "--base", base, "--head", head], + environ={}, + ) + captured = capsys.readouterr() + assert code == 0 + assert "PRIVACY_SCAN_TERMS is not set" in captured.err + assert "PRIVACY_SCAN_PLATES is not set" in captured.err + assert "privacy scan passed" in captured.out + assert scan.RULE_TERM not in captured.out + assert TERM not in captured.out + + def test_diff_title_body_and_commit_message(self, tmp_path, capsys): + repo = tmp_path / "repo" + init_repo(repo) + base = git(repo, "rev-parse", "HEAD").strip() + address = dotted(10, 1, 2, 3) + (repo / "cam.txt").write_text("keep\n" + address + "\n", encoding="utf-8") + git(repo, "add", "cam.txt") + marker = chunk("IB", "36", "NL") + git(repo, "commit", "-m", "record plate " + repr(marker)) + head = git(repo, "rev-parse", "HEAD").strip() + event = { + "pull_request": { + "title": "use " + address, + "body": "vendor " + TERM + "\nsecond\n", + } + } + event_path = tmp_path / "event.json" + event_path.write_text(json.dumps(event), encoding="utf-8") + code = scan.main( + [ + "--repo", + str(repo), + "--base", + base, + "--head", + head, + "--event", + str(event_path), + ], + environ={ + "PRIVACY_SCAN_TERMS": scan.sha256_hex(TERM), + "PRIVACY_SCAN_PLATES": "", + }, + ) + captured = capsys.readouterr() + assert code == 1 + assert TERM not in captured.out + assert TERM not in captured.err + assert address not in captured.out + assert marker not in captured.out + assert "cam.txt:2: " + scan.RULE_IP in captured.out + assert "pr-title:1: " + scan.RULE_IP in captured.out + assert "pr-body:1: " + scan.RULE_TERM in captured.out + assert scan.CONFIDENTIAL_MASK in captured.out + assert "pr-body:2:" not in captured.out + commit_hits = [ + line + for line in captured.out.splitlines() + if line.startswith("commit/") and scan.RULE_PLATE in line + ] + assert len(commit_hits) == 1 + assert commit_hits[0].split(":")[1] == "1" + assert "counts by rule and file:" in captured.out + assert "privacy scan failed:" in captured.err + + def test_worktree_scans_head(self, tmp_path, capsys): + repo = tmp_path / "repo" + init_repo(repo) + address = dotted(169, 254, 1, 1) + (repo / "lan.txt").write_text(address + "\n", encoding="utf-8") + git(repo, "add", "lan.txt") + git(repo, "commit", "-m", "add lan") + code = scan.main(["--repo", str(repo), "--worktree"], environ={}) + captured = capsys.readouterr() + assert code == 1 + assert "lan.txt:1: " + scan.RULE_IP in captured.out + assert address not in captured.out + + +class TestRepositoryFiles: + def test_only_the_approved_fake_plate_is_listed(self): + approved_name = ".github/privacy-scan-plates-approved.txt" + allow = ROOT / ".github" / "privacy-scan-allow.txt" + approved = ROOT / approved_name + assert scan.load_allow_list(allow) == [] + assert scan.load_approved_plates(approved) == {"IB36NL"} + text = approved.read_text(encoding="utf-8") + assert "Mike must approve each entry" in text + assert "approved by Mike 2026-10-14" in text + + def test_workflow_covers_every_branch(self): + text = (ROOT / ".github" / "workflows" / "privacy-scan.yml").read_text( + encoding="utf-8" + ) + assert "pull_request:" in text + assert '"**"' in text + assert "PRIVACY_SCAN_TERMS" in text + assert "PRIVACY_SCAN_PLATES" in text + assert "privacy_scan.py" in text + assert '--base "$BASE_SHA" --head "$HEAD_SHA"' in text + + def test_added_sources_are_clean(self): + approved_name = ".github/privacy-scan-plates-approved.txt" + ctx = scan.ScanContext( + set(), + set(), + scan.load_allow_list(ROOT / ".github" / "privacy-scan-allow.txt"), + scan.load_approved_plates(ROOT / approved_name), + ) + rels = [ + ".github/scripts/privacy_scan.py", + ".github/workflows/privacy-scan.yml", + "scripts/hash_terms.py", + ".github/privacy-scan-allow.txt", + ".github/privacy-scan-plates-approved.txt", + "tests/test_privacy_scan.py", + ] + findings = [] + for rel in rels: + for lineno, line in enumerate((ROOT / rel).read_text(encoding="utf-8").splitlines(), 1): + findings.extend(scan.scan_line(rel, lineno, line, ctx)) + assert findings == [] + + def test_readme_section_is_clean(self): + readme = (ROOT / "README.md").read_text(encoding="utf-8") + start = readme.index("## Privacy scan\n") + end = readme.index("\n## ", start + 1) + ctx = scan.ScanContext(set(), set(), [], set()) + findings = [] + for lineno, line in enumerate(readme[start:end].splitlines(), 1): + findings.extend(scan.scan_line("README.md", lineno, line, ctx)) + assert findings == []