#!/usr/bin/env python3 """Pure-Python cassette guard — replacement for ``tests/check_cassettes_clean.sh``. Walks ``tests/cassettes/*.yaml`` (or any explicit paths passed on the command line) and reports any cassette that contains sensitive data. Uses the canonical pattern registry in ``tests/cassette_patterns.py``. Key differences vs. the legacy bash script: * Cross-platform — pure stdlib, runs on Linux / macOS / Windows. * Explicit placeholder allowlist (``SCRUB_PLACEHOLDERS``) instead of the bash "starts with S" heuristic — closes the cookie-value leak gap (a real token whose first byte is ``S`` no longer slips through). * ``--strict`` flag disables the repair allowlist for CI gating once cleanup is done. * Reports ``file:line`` for every leak so a developer can jump straight to the offending interaction. Usage:: uv run python tests/scripts/check_cassettes_clean.py uv run python tests/scripts/check_cassettes_clean.py path/to/cassette.yaml uv run python tests/scripts/check_cassettes_clean.py --strict Exit codes: 0 — every scanned cassette is clean 1 — one or more leaks detected (printed to stdout) Implementation note: the tool reads each cassette line-by-line (no PyYAML parse) so that: * It runs in O(stream) memory even on multi-megabyte cassettes. * Reported line numbers map directly to the file on disk. * It can also scan partial / malformed YAML that a real recorder might emit. """ from __future__ import annotations import argparse import sys from pathlib import Path # Allow `python tests/scripts/check_cassettes_clean.py` from anywhere by # putting the repo root on sys.path so ``tests.cassette_patterns`` resolves. _REPO_ROOT = Path(__file__).resolve().parents[2] if str(_REPO_ROOT) not in sys.path: sys.path.insert(0, str(_REPO_ROOT)) from tests.cassette_patterns import ( # noqa: E402 find_cookie_leaks, find_credential_leaks, is_clean, ) DEFAULT_CASSETTE_DIR = _REPO_ROOT / "tests" / "cassettes" # Extensions scanned in ``--secrets-only`` mode. Golden RPC fixtures are # ``.json`` and the captured-page fixture is ``.html``; the default cassette # scan only globs ``.yaml``, so a Google API key smuggled into a golden HTML # fixture (the GET_INTERACTIVE_HTML.json shape) would slip past the cassette # guard entirely. ``--secrets-only`` widens both the file types AND the # directories scanned. _SECRETS_ONLY_EXTENSIONS: tuple[str, ...] = (".yaml", ".yml", ".json", ".html", ".txt") DEFAULT_ALLOWLIST = _REPO_ROOT / "tests" / "scripts" / "cassette_repair_allowlist.txt" def _load_allowlist(path: Path) -> set[str]: """Read the repair allowlist as a set of cassette basenames. Blank lines and ``#`` comments are ignored. Entries are interpreted as cassette basenames (e.g. ``chat_ask.yaml``) — paths or globs are not supported, by design, to keep the file boring and auditable. """ if not path.exists(): return set() return { line.strip() for line in path.read_text(encoding="utf-8").splitlines() if line.strip() and not line.strip().startswith("#") } # Subdirectories of ``tests/cassettes/`` that hold ILLUSTRATIVE fixtures # (not real recordings) and are filtered out of the recursive scan. These # files use placeholder cookie / token values with intentional formatting # quirks (truncated YAML strings, hand-edited content) that would trip the # leak detector even though they contain no actual secrets — see the # README in ``tests/cassettes/examples/`` for the design intent. # # The ``tests/integration/conftest.py`` cassette-availability check already # excludes ``example_*`` cassettes from the "real recordings present" # count via a similar filter; this constant carries the same exclusion # semantic onto the guard tool. _EXAMPLE_SUBDIRS: frozenset[str] = frozenset({"examples"}) def _iter_cassettes( paths: list[str], recursive: bool = False, extensions: tuple[str, ...] = (".yaml",), skip_examples: bool = True, ) -> list[Path]: """Resolve CLI arguments into a concrete list of cassette files. * If no paths are given, scan ``tests/cassettes/*.yaml`` (non-recursive) OR ``tests/cassettes/**/*.yaml`` when ``recursive=True``. * If a directory is given, scan ``*.yaml`` inside it (recursively when ``recursive=True``). * If a file is given, scan it directly. * Non-existent paths are silently skipped — matches the bash original's "scan what exists" behaviour and keeps the tool friendly to pre-commit hooks that may pass deleted-but-still-staged paths. * When ``skip_examples`` is set (the default), files under an ``examples/`` directory (any depth) and ``example_*`` files are excluded from directory scans — they are illustrative fixtures with placeholder cookies and intentional YAML quirks, not real recordings (see :data:`_EXAMPLE_SUBDIRS`). Their placeholder content trips the full :func:`is_clean` heuristics, so the default cassette scan filters them. ``skip_examples`` is set to ``False`` for ``--secrets-only`` scans: that mode only matches high-severity credential shapes (which never occur in placeholder fixtures), so there is no false-positive reason to skip ``examples/`` — and skipping them WOULD be a blind spot for credential hunting over a directory like ``tests/fixtures/`` (coderabbit review on #1266). Explicitly-named file paths are always scanned regardless. The ``recursive`` flag is what P1-5 adds: CI now scans subdirectories of ``tests/cassettes/`` (e.g. ``gzip_coverage/``) so a recorder cannot smuggle a leak into a nested folder. The default stays non-recursive so existing developer workflows (running the guard on a single file or the top-level directory) are unchanged. """ glob_patterns = [(f"**/*{ext}" if recursive else f"*{ext}") for ext in extensions] def _globdir(d: Path) -> list[Path]: found: list[Path] = [] for pat in glob_patterns: found.extend(d.glob(pat)) return sorted(set(found)) def _is_example_path(p: Path) -> bool: # An ``example_`` file at the top level OR any file under an # ``examples/`` directory anywhere is treated as illustrative and # excluded from directory scans (only when ``skip_examples``). if p.name.startswith("example_"): return True return any(part in _EXAMPLE_SUBDIRS for part in p.parts) def _keep(p: Path) -> bool: return not (skip_examples and _is_example_path(p)) if not paths: if not DEFAULT_CASSETTE_DIR.exists(): return [] return [p for p in _globdir(DEFAULT_CASSETTE_DIR) if _keep(p)] resolved: list[Path] = [] for raw in paths: candidate = Path(raw) if candidate.is_dir(): resolved.extend(p for p in _globdir(candidate) if _keep(p)) elif candidate.is_file(): # Explicit file paths are always scanned, even if under # ``examples/`` — the operator asked for them by name. resolved.append(candidate) return resolved def _scan_cookie_headers_yaml(path: Path) -> list[tuple[int, str]]: """YAML-aware, name-agnostic cookie-value leak scan over a cassette. The line-by-line :func:`is_clean` pass below cannot see a cookie pair that a folded YAML scalar split onto a continuation line (the session-cookie anchor that scopes the name-agnostic check lives on a different physical line). This pass parses the cassette with PyYAML, recovers each request ``Cookie:`` and response ``Set-Cookie:`` header as its LOGICAL (joined) value, and runs the name-agnostic :func:`tests.cassette_patterns.find_cookie_leaks` on it — so an off-allowlist cookie (``_ga`` / ``_gcl_au`` / ``AEC`` …) whose real value survives in a folded scalar is caught regardless of line wrapping. Line number ``0`` is reported (the leak is logical, not tied to one physical line of a folded scalar). A YAML parse failure is non-fatal — the line-by-line pass still runs — so a malformed cassette degrades to the streaming scan rather than crashing the guard. """ leaks: list[tuple[int, str]] = [] try: import yaml try: from yaml import CSafeLoader as _Loader except ImportError: # pragma: no cover - libyaml optional from yaml import SafeLoader as _Loader # type: ignore[assignment] raw = path.read_text(encoding="utf-8", errors="replace") data = yaml.load(raw, Loader=_Loader) except Exception: # noqa: BLE001 - degrade to line-scan on any parse/read error return leaks if not isinstance(data, dict): return leaks def _values(header_block: object, key: str) -> list[str]: if not isinstance(header_block, dict): return [] for hk, hv in header_block.items(): if isinstance(hk, str) and hk.lower() == key.lower(): if isinstance(hv, list): return [v for v in hv if isinstance(v, str)] if isinstance(hv, str): return [hv] return [] for interaction in data.get("interactions") or []: if not isinstance(interaction, dict): continue request = interaction.get("request") or {} response = interaction.get("response") or {} for value in _values(request.get("headers"), "Cookie"): for desc in find_cookie_leaks(value): leaks.append((0, desc)) for value in _values(response.get("headers"), "Set-Cookie"): for desc in find_cookie_leaks(value, set_cookie=True): leaks.append((0, desc)) return leaks def _scan_file(path: Path, secrets_only: bool = False) -> list[tuple[int, str]]: """Return ``(line_number, leak_description)`` for each leak. Reads the cassette line-by-line (no PyYAML parse) so the tool runs in streaming memory even on multi-megabyte cassettes, ``file:line`` numbers map directly to the on-disk file, and the guard can scan partial/malformed YAML a recorder might emit before VCR finalises the file. Each leak description is the human-readable string emitted by :func:`tests.cassette_patterns.is_clean` (e.g. ``"Leak (cookie header): cookie 'SID' value 'abc' is not a known scrub placeholder"``). When ``secrets_only`` is set the per-line check uses :func:`tests.cassette_patterns.find_credential_leaks` instead — only the high-severity credential shapes (auth tokens + Google API keys), which never match legitimate placeholder fixture content. That is what makes scanning fixture directories full of ``"Scrubbed ..."`` placeholders viable. For full (non-``secrets_only``) scans a SECOND, YAML-aware cookie pass (:func:`_scan_cookie_headers_yaml`) runs name-agnostic cookie-value detection on each cassette's joined ``Cookie:`` / ``Set-Cookie:`` header values — catching off-allowlist cookies (``_ga`` …) that a folded YAML scalar split across lines the streaming pass cannot stitch back together. """ leaks: list[tuple[int, str]] = [] try: # ``errors="replace"`` guarantees a corrupted cassette never crashes # the guard mid-scan; we still produce useful output. with path.open("r", encoding="utf-8", errors="replace") as fh: for line_no, line in enumerate(fh, start=1): if secrets_only: line_leaks = find_credential_leaks(line) else: ok, line_leaks = is_clean(line) if ok: continue for description in line_leaks: leaks.append((line_no, description)) except OSError as exc: print(f"{path}: could not read ({exc})", file=sys.stderr) if not secrets_only: # De-duplicate against the line-pass: the YAML pass catches folded-scalar # cookie leaks the streaming pass cannot stitch together; an on-allowlist # leak found by both is reported once. seen = {desc for _, desc in leaks} for line_no, desc in _scan_cookie_headers_yaml(path): if desc not in seen: seen.add(desc) leaks.append((line_no, desc)) return leaks def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( description=( "Scan VCR cassettes for sensitive-data leaks. " "Returns exit 1 on any leak; otherwise exit 0." ) ) parser.add_argument( "paths", nargs="*", help=( "Cassette file(s) or directory. If omitted, scans " "tests/cassettes/*.yaml from the repo root." ), ) parser.add_argument( "--strict", action="store_true", help=( "Ignore the repair allowlist AND fail with exit code 1 if any " "allowlist entry is present in the working tree. Use this in " "CI: the allowlist is a one-way ratchet for cleanup, and " "``--strict`` flips it from 'best-effort suppressor' to " "'must be empty'." ), ) parser.add_argument( "--recursive", action="store_true", help=( "Scan ``tests/cassettes/**/*.yaml`` (recurse into subdirectories) " "instead of the default top-level-only ``tests/cassettes/*.yaml``. " "Required in CI so a recorder cannot smuggle a leak into a nested " "folder like ``tests/cassettes/gzip_coverage/``." ), ) parser.add_argument( "--secrets-only", action="store_true", help=( "Scan only for high-severity credential shapes (Google auth tokens " "and Google API keys) instead of the full cookie/PII heuristics. " "Also widens the scanned file types to .json/.html/.txt (not just " ".yaml). Use this to scan fixture directories such as " "tests/fixtures/ that contain intentional placeholder content " "(e.g. 'Scrubbed Note Title') which would false-positive under the " "full is_clean() heuristics." ), ) parser.add_argument( "--allowlist", type=Path, default=DEFAULT_ALLOWLIST, help=( "Path to the repair allowlist file " "(default: tests/scripts/cassette_repair_allowlist.txt)." ), ) args = parser.parse_args(argv) extensions = _SECRETS_ONLY_EXTENSIONS if args.secrets_only else (".yaml",) # ``--secrets-only`` matches only credential shapes (no false positives on # placeholder content), so it must NOT skip ``examples/`` — doing so would # be a blind spot for credential hunting over fixture dirs (#1266). Default # (full-heuristic) scans keep skipping examples to avoid placeholder noise. cassettes = _iter_cassettes( args.paths, recursive=args.recursive, extensions=extensions, skip_examples=not args.secrets_only, ) if not cassettes: # Fresh checkout with no recorded cassettes — that's a valid clean # state, matching the bash original's behaviour. print("OK: no cassettes to scan") return 0 if args.strict: # Strict mode: ignore the allowlist for scan-skip purposes AND # require the allowlist to be empty (or non-existent). Any entry # present in the file is treated as a CI failure so the allowlist # cannot quietly grow past the cleanup phase. allowlist: set[str] = set() present_in_allowlist = _load_allowlist(args.allowlist) if present_in_allowlist: print( f"ERROR (--strict): {args.allowlist} contains " f"{len(present_in_allowlist)} entries; --strict requires the " "allowlist to be empty. Either drop the entries or run " "without --strict for the cleanup-in-progress workflow." ) for entry in sorted(present_in_allowlist): print(f" - {entry}") return 1 else: allowlist = _load_allowlist(args.allowlist) scanned = 0 skipped = 0 leaked_files: list[Path] = [] total_leaks = 0 for cassette in cassettes: if not args.strict and cassette.name in allowlist: skipped += 1 continue scanned += 1 leaks = _scan_file(cassette, secrets_only=args.secrets_only) if leaks: leaked_files.append(cassette) total_leaks += len(leaks) try: rel = cassette.relative_to(_REPO_ROOT) except ValueError: rel = cassette for line_no, description in leaks: print(f"{rel}:{line_no}: {description}") # Summary always prints, even on the happy path, so the operator gets a # one-line confirmation that the guard actually ran. print( f"\nSummary: {scanned} cassettes scanned" + (f", {skipped} allow-listed" if skipped else "") + f", {total_leaks} leaks found in {len(leaked_files)} files." ) return 1 if leaked_files else 0 if __name__ == "__main__": raise SystemExit(main())