repro-bundle-v0/repro/make_repro_bundle.py

#!/usr/bin/env python3
"""Stage a reproduction bundle for upload, refusing to ship secrets.

Copies this project's repro/ (code + figures), results/ (JSON), poster_embed.html
and PROMPT.md into ./repro_bundle, scans for secrets, and (optionally) syncs to
the HF bucket. Adapted from FirstPriceAuctionRevenue/repro/make_repro_bundle.py
(copied, not edited in place, per workspace rules).

Usage:
    python repro/make_repro_bundle.py --src . --out ./repro_bundle
    python repro/make_repro_bundle.py --src . --out ./repro_bundle --sync
"""

from __future__ import annotations

import argparse
import os
import pathlib
import re
import shutil
import sys

SKIP_DIRS = {"__pycache__", ".git", "node_modules", ".pytest_cache", ".venv",
             "venv", ".mypy_cache", ".ruff_cache", ".ipynb_checkpoints",
             ".trackio"}
SKIP_NAMES = {".env", ".env.local", "credentials.json", "token.json", ".netrc"}
BINARY_EXT = {".png", ".pdf", ".jpg", ".jpeg", ".gif", ".pyc", ".ico", ".gz",
              ".zip", ".tar", ".safetensors", ".pt", ".bin", ".parquet", ".woff2"}

SECRET_PATTERNS = {
    "openrouter": re.compile(r"sk-or-[A-Za-z0-9_\-]{12,}"),
    "openai": re.compile(r"\bsk-[A-Za-z0-9]{20,}"),
    "hf": re.compile(r"\bhf_[A-Za-z0-9]{20,}"),
    "anthropic": re.compile(r"\bsk-ant-[A-Za-z0-9_\-]{20,}"),
    "aws": re.compile(r"\bAKIA[0-9A-Z]{16}\b"),
    "google": re.compile(r"\bAIza[0-9A-Za-z_\-]{35}\b"),
    "bearer": re.compile(r"[Bb]earer\s+[A-Za-z0-9_\-\.]{20,}"),
    "jwt": re.compile(r"\beyJ[A-Za-z0-9_\-]{20,}\.[A-Za-z0-9_\-]{10,}"),
    "pkey": re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"),
}
PLACEHOLDER = re.compile(
    r"(your|xxx+|placeholder|example|dummy|redact|<[^>]*>|\.\.\.|foo|bar|test_?key)", re.I)


def ignore_factory(extra_dirs: set[str]):
    def _ignore(_dir: str, names: list[str]) -> set[str]:
        return {n for n in names
                if n in SKIP_DIRS or n in SKIP_NAMES or n in extra_dirs}
    return _ignore


def scan(root: pathlib.Path) -> list[str]:
    hits = []
    for dirpath, dirnames, filenames in os.walk(root, onerror=lambda e: None):
        dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS]
        for fn in filenames:
            f = pathlib.Path(dirpath) / fn
            if f.suffix.lower() in BINARY_EXT:
                continue
            try:
                txt = f.read_text(encoding="utf-8", errors="ignore")
            except OSError:
                continue
            for label, pat in SECRET_PATTERNS.items():
                for m in pat.finditer(txt):
                    if PLACEHOLDER.search(m.group(0)):
                        continue
                    hits.append(f"{f.relative_to(root)} [{label}]")
    return sorted(set(hits))


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--src", required=True, help="project directory to bundle")
    ap.add_argument("--out", default="./repro_bundle", help="staging directory")
    ap.add_argument("--include", nargs="*", default=None,
                    help="only these top-level entries (default: everything)")
    ap.add_argument("--exclude-dir", nargs="*", default=[],
                    help="extra directory names to skip anywhere in the tree")
    ap.add_argument("--allow-child", action="store_true",
                    help="allow --out to live inside --src (e.g. a committed "
                         "bundle dir that the scanner walks without following "
                         "into its own repro/ copy)")
    ap.add_argument("--sync", action="store_true",
                    help="sync the staged bundle to the HF bucket")
    ap.add_argument("--bucket", default="algorise/cssa-repro-artifacts",
                    help="HF bucket id for --sync")
    args = ap.parse_args()

    src = pathlib.Path(args.src).resolve()
    out = pathlib.Path(args.out).resolve()
    if not src.is_dir():
        sys.exit(f"no such directory: {src}")
    if out == src or src in out.parents:
        if not args.allow_child:
            sys.exit("--out must be outside --src")
    if out.exists():
        shutil.rmtree(out)
    out.mkdir(parents=True)

    ignore = ignore_factory(set(args.exclude_dir))
    default_include = ["repro", "results", "poster_embed.html", "PROMPT.md"]
    entries = ([src / e for e in args.include] if args.include
               else [src / e for e in default_include if (src / e).exists()])
    for p in entries:
        if not p.exists() or p.name in SKIP_DIRS or p.name in SKIP_NAMES:
            continue
        if p.is_dir():
            shutil.copytree(p, out / p.name, ignore=ignore, dirs_exist_ok=True)
        else:
            shutil.copy2(p, out / p.name)

    leaks = scan(out)
    if leaks:
        print("REFUSING TO BUNDLE -- key-shaped strings found:", file=sys.stderr)
        for x in leaks[:25]:
            print("   ", x, file=sys.stderr)
        shutil.rmtree(out)
        print(f"\nstaging directory removed. Fix the source, then re-run.",
              file=sys.stderr)
        return 1

    files = [f for f in out.rglob("*") if f.is_file()]
    size = sum(f.stat().st_size for f in files)
    print(f"bundle -> {out}")
    print(f"  {len(files):,} files, {size / 1e6:.1f} MB, no secrets detected")
    for child in sorted(out.iterdir()):
        if child.is_dir():
            cs = sum(f.stat().st_size for f in child.rglob("*") if f.is_file())
            print(f"    {child.name + '/':<26} {cs / 1e6:6.1f} MB")
    print("\n  no secrets detected")

    if args.sync:
        dest = f"hf://buckets/{args.bucket}/repro-bundle-v0"
        print(f"\n  syncing to {dest} ...")
        rc = os.system(f'MSYS_NO_PATHCONV=1 hf buckets sync "{out}" "{dest}"')
        if rc != 0:
            print("  sync failed (see above)", file=sys.stderr)
            return rc
        print("  sync done")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())