pii_guard

#!/usr/bin/env python3
"""pii_guard — structural PII detector for the pre-commit dispatcher (#269).

The control that was missing when child initials + DOBs entered tracked code at
d2b57be (7/6) and sat undetected for six days until the 7/12 purge. It scans the
STAGED DIFF (added lines only) for personal-data pattern CLASSES and WARNs — it
is ADVISORY (never blocks); promotion to blocking is a human ruling (one-line
hooks.conf edit: LEVEL_PII_GUARD=block).

⚠️ DESIGN INVARIANT — this file NEVER contains a real personal token. It detects
   pattern CLASSES only. An optional exact-value denylist lives OUTSIDE the repo
   at /Users/dev/.pii_denylist (mode 600, never tracked) and is strictly OPTIONAL
   — the structural detector below is authoritative and works with it absent.
   Findings NEVER print the matched value — only file · line · category.

Detection classes:
  1. dob-in-kid-context    — a DOB-like date (ISO YYYY-M-D, M/D/YY, M-D-YY) within
                             a birth/child context (same line or same JSON object).
  2. unmarked-household-rec — a structured kids/household record (name/kid/child
                             field with an initials-like value) not marked SYNTHETIC.
  3. personal-media-url    — a video-call / recording / file-share URL.
  4. test-fixture-pii      — any of the above inside tests/ (synthetic-fixtures-only).
A record is EXEMPT when a SYNTHETIC marker (the memo-173 convention) appears within
SYNTH_WINDOW lines.

COVERAGE IS REPORTED, NOT ASSUMED. Files this check does not read are NAMED as
UNSCANNED and counted in the summary: "N findings, M files unscanned". A bare
"0 findings" would conceal the gap — "no findings" about a file nobody read is
not evidence.

THE NAME NEVER DECIDES COVERAGE ALONE (ruled 2026-07-25). Every candidate's first
8KB is read and checked against its suffix: a binary name WITH NUL bytes is
skipped by policy and reported UNSCANNED (expected: fonts, images — this does not
block, since the site self-hosts two fonts and a blocking rule would refuse every
publish forever); a text name WITH NUL bytes BLOCKS (exit 2) — that is the UTF-16
class this check used to accept, decoding it to replacement characters and
"scanning" a file whose patterns could never match; a binary name WITHOUT NUL
bytes BLOCKS (exit 2) — text wearing a binary name, where the skip would hide
readable content. The blocking rules target the anomaly, not the volume, and both
fired zero times against the tracked tree and dist/ when adopted. A coverage
failure outranks a finding: anomalies exit 2 even when findings exist.

Enumeration is FAIL-CLOSED. File lists come from git NUL-delimited (-z): a plain
whitespace split loses any path containing a space or a tab and mangles non-ASCII
ones (git quote-escapes those by default), and a lost path is silently never
scanned — a fail-open in a gate whose whole purpose is refusal. A git command
that fails, a missing git binary, staged content git will not hand over, and a
tracked file that cannot be opened are all exit 2, never a quiet "0 findings".

Usage:
  pii_guard --staged        scan the staged diff; exit 1 if findings (advisory signal)
  pii_guard --scan-tree      one-shot scan of all tracked files; report by CATEGORY+FILE
  pii_guard --scan-paths F [F...]  scan the given files from disk (artifact choke:
                             covers UNTRACKED build output like dist/**, which
                             --scan-tree by design cannot see)
Exit codes: 0 = no findings · 1 = findings · 2 = usage/internal error.
stdlib only; runs under /opt/homebrew/bin/python3 (3.14) and 3.9.
"""
import os
import re
import subprocess
import sys

REPO = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
DENYLIST_PATH = "/Users/dev/.pii_denylist"
SYNTH_WINDOW = 10          # lines up/down to look for a SYNTHETIC marker
CTX_WINDOW = 2             # lines up/down that count as "same record" for context

# ── pattern classes (NO real tokens — shapes only) ───────────────────────────
# DOB-like dates. Negative lookarounds keep them from matching inside longer runs.
_ISO = r"(?<!\d)\d{4}-\d{1,2}-\d{1,2}(?!\d)"
_SLASH = r"(?<!\d)\d{1,2}/\d{1,2}/\d{2,4}(?!\d)"
_DASH = r"(?<!\d)\d{1,2}-\d{1,2}-\d{2}(?!\d)"        # 2-digit year only (ISO owns 4-2-2)
_DATE = "(?:" + "|".join((_ISO, _SLASH, _DASH)) + ")"

# class 1 signal: a birth/DOB keyword IMMEDIATELY followed (<=10 sep chars, same
# line) by a DOB-like date -- e.g. "born <slash-date>", or a "dob" key then an
# ISO value. The adjacency is what separates a birthdate from an event date: a
# changelog line where the date comes BEFORE the word "born" does not match.
# Deliberately NOT "kid" alone ("kid-safety" is not a birth record).
# (Placeholders only in this comment -- the detector's own source carries no
# date-shaped or record-shaped example tokens, so it never self-flags.)
DOB_RE = re.compile(r"(?:born|birth(?:day|date)?|d\.?o\.?b\.?|date of birth)\b\W{0,10}" + _DATE, re.I)

# A structured household/kid record field (name/kid/child/initials key) carrying
# a short UPPERCASE initials value, e.g. a "name" key set to two capital letters.
# Lookbehind rejects "className"/"filename" (key must be a standalone word); the
# value group is strictly UPPERCASE (case-sensitive) so JS like className='on'
# cannot match.
NAME_FIELD_RE = re.compile(
    r"""(?<![A-Za-z])["']?(?i:name|initials?|kid|child)["']?\s*[:=]\s*["']?[A-Z]{2,3}\b"""
)
# Household framing that promotes a NAME_FIELD hit to a personal record.
HOUSEHOLD_RE = re.compile(r"\b(kids?|child(?:ren)?|household|famil(?:y|ies)|guardian|dob|born)\b", re.I)

# Personal media / scheduling / file-share URLs (scheme-anchored so prose that
# merely names a platform, or a doc *describing* a purge, does not match).
URL_RE = re.compile(
    r"""https?://\S*?(zoom\.us|meet\.google|teams\.microsoft|webex\.com|whereby\.com|
        calendly\.com|dropbox\.com|drive\.google|\.mp4|\.mov|\.m4a)""",
    re.I | re.X,
)

SYNTH_RE = re.compile(r"\b(synthetic|fictional|illustrative|invented|not real|placeholder example)\b", re.I)


def _git(args):
    """Run git inside REPO. git absent or failing is a config error, never 'clean'."""
    try:
        return subprocess.run(["git"] + args, cwd=REPO, capture_output=True)
    except OSError:
        print("pii_guard: BLOCKED — git is not runnable (fail-closed; cannot "
              "determine what to scan)", file=sys.stderr)
        sys.exit(2)


def _git_text(args, what):
    """git stdout as text, or refuse. Never returns partial output on failure."""
    proc = _git(args)
    if proc.returncode != 0:
        print("pii_guard: BLOCKED — `git {}` failed (fail-closed; not a git repo, "
              "or git error). Nothing was scanned.".format(what), file=sys.stderr)
        sys.exit(2)
    return proc.stdout.decode("utf-8", errors="replace")


def _git_paths(args, what):
    """NUL-delimited path enumeration. Plain .split() loses any path containing a
    space or tab and mangles non-ASCII ones (git quote-escapes those by default),
    and a lost path is silently never scanned — a fail-OPEN in a fail-closed gate."""
    return [p for p in _git_text(args + ["-z"], what).split("\0") if p]


def _load_denylist():
    """Optional exact-value denylist from OUTSIDE the repo. Absent by design in CI."""
    try:
        if not os.path.exists(DENYLIST_PATH):
            return []
        with open(DENYLIST_PATH, encoding="utf-8", errors="replace") as fh:
            return [ln.strip() for ln in fh if ln.strip() and not ln.startswith("#")]
    except OSError:
        return []


def _window_has(lines, idx, rx, radius):
    lo, hi = max(0, idx - radius), min(len(lines), idx + radius + 1)
    return any(rx.search(lines[j]) for j in range(lo, hi))


def scan(path, lines, check_idxs, denylist):
    """Return list of (category, line_no) for the checked line indices. Values are
    never captured or returned — only category + 1-based line number."""
    out = []
    in_tests = path.startswith("tests/")
    for i in check_idxs:
        if i >= len(lines):
            continue
        line = lines[i]
        synthetic = _window_has(lines, i, SYNTH_RE, SYNTH_WINDOW)
        cats = []

        # class 3 — personal media URL (marker does not excuse a real link)
        if URL_RE.search(line):
            cats.append("personal-media-url")

        # class 1 — DOB-like date adjacent to a birth keyword (same line), unmarked
        if DOB_RE.search(line) and not synthetic:
            cats.append("dob-in-kid-context")

        # class 2 — structured household record w/ initials value, unmarked
        if (NAME_FIELD_RE.search(line) and _window_has(lines, i, HOUSEHOLD_RE, CTX_WINDOW)
                and not synthetic):
            cats.append("unmarked-household-record")

        # optional external exact-value denylist
        if denylist and any(tok in line for tok in denylist):
            cats.append("denylist-exact")

        for c in cats:
            # class 4 supersedes for tests/ fixtures (synthetic-fixtures-only rule)
            out.append(("test-fixture-pii" if in_tests else c, i + 1))
    return out


def _added_line_numbers(path):
    """New-side line numbers added in the staged diff for `path` (via -U0 hunks)."""
    diff = _git_text(["diff", "--cached", "-U0", "--", path], "diff --cached -U0")
    added, new_ln = [], 0
    for ln in diff.splitlines():
        if ln.startswith("@@"):
            m = re.search(r"\+(\d+)(?:,(\d+))?", ln)
            new_ln = int(m.group(1)) if m else 0
            continue
        if ln.startswith("+") and not ln.startswith("+++"):
            added.append(new_ln - 1)   # 0-based index into staged blob
            new_ln += 1
        elif ln.startswith("-") and not ln.startswith("---"):
            continue
        else:
            new_ln += 1
    return added


BINARY_SUFFIXES = (".png", ".jpg", ".jpeg", ".gif", ".pdf", ".mp4", ".mov",
                   ".zip", ".gz", ".bundle", ".pyc", ".ico", ".woff", ".woff2")


def _is_texty(path):
    return not path.endswith(BINARY_SUFFIXES)


def _classify(path, blob=None):
    """('scan'|'skip'|'block', reason). The name alone never decides coverage.

    Concealment has two shapes and the head is read for both: text wearing a
    binary name (skipped, so its content is never read) and binary wearing a
    text name — the UTF-16 case this check used to accept, decoding it
    utf-8/replace and "scanning" a file whose patterns could never match, which
    counted as coverage while providing none. When name and content disagree the
    check REFUSES; when they agree a binary is skipped by policy and named."""
    if blob is None:
        try:
            with open(path, "rb") as fh:
                blob = fh.read(8192)
        except OSError:
            return "scan", None      # unreadable: the caller refuses
    head = blob[:8192]
    if not head:
        return "scan", None          # empty file: nothing to scan, nothing odd
    has_nul = b"\x00" in head if isinstance(head, bytes) else "\x00" in head
    named_binary = path.endswith(BINARY_SUFFIXES)
    if named_binary:
        if has_nul:
            return "skip", "non-text suffix"
        return "block", ("non-text suffix but no NUL byte in first 8KB — text "
                         "wearing a binary name, so the skip would hide content")
    if has_nul:
        return "block", ("NUL byte in first 8KB despite a text name — binary "
                         "wearing a text name (e.g. UTF-16); decoding it would "
                         "produce replacement characters no pattern can match")
    return "scan", None


def staged_mode(denylist):
    files = _git_paths(["diff", "--cached", "--name-only", "--diff-filter=ACM"],
                       "diff --cached --name-only")
    findings, unscanned, anomalies = [], [], []
    for f in files:
        show = _git(["show", ":" + f])
        if show.returncode != 0:
            # git listed the path but will not hand over its staged content;
            # treating that as clean is the fail-open this gate exists to avoid
            print("pii_guard: BLOCKED — cannot read staged content of {} "
                  "(fail-closed)".format(f), file=sys.stderr)
            sys.exit(2)
        if not show.stdout:
            continue                      # genuinely empty blob: nothing to scan
        verdict, reason = _classify(f, show.stdout)
        if verdict == "block":
            anomalies.append((f, reason))
            continue
        if verdict == "skip":
            unscanned.append((f, reason))
            continue
        lines = show.stdout.decode("utf-8", errors="replace").split("\n")
        idxs = _added_line_numbers(f)
        findings += [(f, cat, ln) for cat, ln in scan(f, lines, idxs, denylist)]
    return findings, unscanned, anomalies


def tree_mode(denylist):
    files = _git_paths(["ls-files"], "ls-files")
    findings, unscanned, anomalies = [], [], []
    for f in files:
        p = os.path.join(REPO, f)
        verdict, reason = _classify(p)
        if verdict == "block":
            anomalies.append((f, reason))
            continue
        if verdict == "skip":
            unscanned.append((f, reason))
            continue
        try:
            with open(p, encoding="utf-8", errors="replace") as fh:
                lines = fh.read().split("\n")
        except OSError:
            # a tracked file the scanner cannot open is a coverage hole, not a pass
            print("pii_guard: BLOCKED — cannot read tracked file {} "
                  "(fail-closed)".format(f), file=sys.stderr)
            sys.exit(2)
        findings += [(f, cat, ln) for cat, ln in scan(f, lines, range(len(lines)), denylist)]
    return findings, unscanned, anomalies


def paths_mode(denylist, paths):
    """Scan explicit files from disk (all lines). Fail-closed: an artifact the
    scanner cannot read must block the publish, so unreadable = exit 2."""
    findings, unscanned, anomalies = [], [], []
    for f in paths:
        verdict, reason = _classify(f)
        if verdict == "block":
            anomalies.append((f, reason))
            continue
        if verdict == "skip":
            unscanned.append((f, reason))
            continue
        try:
            with open(f, encoding="utf-8", errors="replace") as fh:
                lines = fh.read().split("\n")
        except OSError:
            print("pii_guard[scan-paths]: cannot read {} (fail-closed)".format(f),
                  file=sys.stderr)
            sys.exit(2)
        findings += [(f, cat, ln) for cat, ln in scan(f, lines, range(len(lines)), denylist)]
    return findings, unscanned, anomalies


def _coverage(unscanned):
    return "{} file{} unscanned".format(len(unscanned),
                                        "" if len(unscanned) == 1 else "s")


def _print_unscanned(unscanned, stream):
    """Name every file the scan did not cover. A skipped file is a COVERAGE GAP,
    not a pass: 'no findings' about a file nobody read is not evidence. After
    the name/content agreement check, everything listed here is a confirmed
    binary skipped by policy — never a file whose nature is in doubt."""
    for f, reason in sorted(unscanned):
        print("  UNSCANNED  {}  ({})".format(f, reason), file=stream)


def _print_anomalies(anomalies, mode):
    print("pii_guard[{}]: BLOCKED — {} file(s) whose name and content disagree; "
          "coverage cannot be established (fail-closed)".format(mode, len(anomalies)),
          file=sys.stderr)
    for f, reason in sorted(anomalies):
        print("  UNSCANNABLE  {}\n      {}".format(f, reason), file=sys.stderr)
    print("pii_guard[{}]: rename the file to match its content, or remove it."
          .format(mode), file=sys.stderr)


def _report(findings, unscanned, anomalies, mode):
    if not findings and not anomalies:
        print("pii_guard[{}]: 0 findings, {}".format(mode, _coverage(unscanned)))
        _print_unscanned(unscanned, sys.stdout)
        return 0
    if findings:
        # group by category → files (never print values or line content)
        by_cat = {}
        for f, cat, ln in findings:
            by_cat.setdefault(cat, []).append((f, ln))
        print("pii_guard[{}]: {} finding(s), {} — category · file:line (values "
              "never shown)".format(mode, len(findings), _coverage(unscanned)),
              file=sys.stderr)
        for cat in sorted(by_cat):
            print("  [{}]".format(cat), file=sys.stderr)
            for f, ln in sorted(by_cat[cat]):
                print("    {}:{}".format(f, ln), file=sys.stderr)
    _print_unscanned(unscanned, sys.stderr)
    if anomalies:
        # a coverage failure outranks a finding: the check could not do its job
        _print_anomalies(anomalies, mode)
        return 2
    return 1


def main(argv):
    if argv and argv[0] == "--scan-paths" and len(argv) > 1:
        findings, unscanned, anomalies = paths_mode(_load_denylist(), argv[1:])
        return _report(findings, unscanned, anomalies, "scan-paths")
    if len(argv) != 1 or argv[0] not in ("--staged", "--scan-tree"):
        print(__doc__.splitlines()[0], file=sys.stderr)
        print("usage: pii_guard --staged | --scan-tree | --scan-paths FILE [FILE...]",
              file=sys.stderr)
        return 2
    denylist = _load_denylist()
    if argv[0] == "--staged":
        findings, unscanned, anomalies = staged_mode(denylist)
        return _report(findings, unscanned, anomalies, "staged")
    findings, unscanned, anomalies = tree_mode(denylist)
    return _report(findings, unscanned, anomalies, "scan-tree")


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))