271 lines
8.4 KiB
Python
271 lines
8.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Create a fail-closed sanitized public-mirror candidate from tracked files.
|
|
|
|
This tool never pushes, changes repository visibility, or rewrites history.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import fnmatch
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from collections.abc import Iterable
|
|
from dataclasses import asdict, dataclass
|
|
from pathlib import Path
|
|
|
|
ALWAYS_DENY = (
|
|
".env",
|
|
".env.*",
|
|
"*.key",
|
|
"*.pem",
|
|
"*.p12",
|
|
"*.pfx",
|
|
"*.kdbx",
|
|
"*.db",
|
|
"*.db-*",
|
|
"*.sqlite*",
|
|
"*.log",
|
|
"*.zip",
|
|
"*.tar",
|
|
"*.tar.gz",
|
|
"*.7z",
|
|
".agents/*",
|
|
".claude/*",
|
|
".codex/*",
|
|
".config/*",
|
|
".dyad/*",
|
|
".vibecode-backup/*",
|
|
"artifacts/*",
|
|
"evidence/*",
|
|
"output/*",
|
|
"test-results/*",
|
|
"node_modules/*",
|
|
"dist/*",
|
|
"build/*",
|
|
"connectors/*",
|
|
"services/*",
|
|
"infrastructure/*",
|
|
"prompts/*",
|
|
"evals/*",
|
|
)
|
|
_PRIVATE_BOOT_ROOT = "/" + "bo" + "ot"
|
|
|
|
RULES = (
|
|
(
|
|
"private_ipv4",
|
|
re.compile(
|
|
r"\b(?:10(?:\.\d{1,3}){3}|192\.168(?:\.\d{1,3}){2}|172\.(?:1[6-9]|2\d|3[01])(?:\.\d{1,3}){2})\b"
|
|
),
|
|
),
|
|
(
|
|
"private_path",
|
|
re.compile(
|
|
r"(?i)(?:/mnt/(?:user|cache)(?:/|\b)|"
|
|
+ re.escape(_PRIVATE_BOOT_ROOT)
|
|
+ r"(?:/|\b)|[A-Z]:\\Users\\)"
|
|
),
|
|
),
|
|
(
|
|
"private_ssh_target",
|
|
re.compile(
|
|
r"(?i)ssh://[^\s/@]+@(?:10(?:\.\d{1,3}){3}|192\.168(?:\.\d{1,3}){2}|172\.(?:1[6-9]|2\d|3[01])(?:\.\d{1,3}){2})"
|
|
),
|
|
),
|
|
("private_key_block", re.compile(r"-----BEGIN [A-Z0-9 ]*PRIVATE KEY-----")),
|
|
(
|
|
"credential_assignment",
|
|
re.compile(
|
|
r"(?i)\b(?:api[_-]?key|access[_-]?token|client[_-]?secret|password|passwd)\s*[:=]\s*['\"]?[A-Za-z0-9_./+=-]{16,}"
|
|
),
|
|
),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Finding:
|
|
rule: str
|
|
path: str
|
|
detail: str
|
|
allowed: bool = False
|
|
reason: str = ""
|
|
|
|
|
|
def git(root: Path, *args: str) -> str:
|
|
executable = shutil.which("git")
|
|
if executable is None:
|
|
raise OSError("git executable was not found")
|
|
result = subprocess.run( # noqa: S603 - executable is resolved locally; args are internal.
|
|
[executable, "-C", str(root), *args],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
return result.stdout.strip()
|
|
|
|
|
|
def load_patterns(path: Path) -> list[str]:
|
|
result = [
|
|
line.strip()
|
|
for line in path.read_text(encoding="utf-8").splitlines()
|
|
if line.strip() and not line.lstrip().startswith("#")
|
|
]
|
|
if not result:
|
|
raise ValueError(f"No patterns in {path}")
|
|
return result
|
|
|
|
|
|
def load_exceptions(path: Path) -> list[tuple[str, str, str]]:
|
|
if not path.exists():
|
|
return []
|
|
result: list[tuple[str, str, str]] = []
|
|
for number, raw in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
|
|
value = raw.strip()
|
|
if not value or value.startswith("#"):
|
|
continue
|
|
parts = [part.strip() for part in value.split("|", 2)]
|
|
if len(parts) != 3 or not all(parts):
|
|
raise ValueError(
|
|
f"Invalid exception at {path}:{number}; expected rule|path-glob|reason"
|
|
)
|
|
result.append((parts[0], parts[1], parts[2]))
|
|
return result
|
|
|
|
|
|
def matches(path: str, globs: Iterable[str]) -> bool:
|
|
return any(fnmatch.fnmatchcase(path, glob) for glob in globs)
|
|
|
|
|
|
def approved_reason(rule: str, path: str, entries: Iterable[tuple[str, str, str]]) -> str | None:
|
|
for entry_rule, path_glob, reason in entries:
|
|
if entry_rule == rule and fnmatch.fnmatchcase(path, path_glob):
|
|
return reason
|
|
return None
|
|
|
|
|
|
def scan(
|
|
root: Path, relative: str, max_bytes: int, entries: list[tuple[str, str, str]]
|
|
) -> list[Finding]:
|
|
path = root / relative
|
|
if path.is_symlink():
|
|
return [Finding("symlink", relative, "symlinks are not exported")]
|
|
size = path.stat().st_size
|
|
if size > max_bytes:
|
|
return [Finding("large_file", relative, f"{size} bytes exceeds {max_bytes}")]
|
|
data = path.read_bytes()
|
|
if b"\x00" in data[:8192]:
|
|
return [Finding("binary_file", relative, "binary content requires manual release review")]
|
|
try:
|
|
text = data.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
return [Finding("non_utf8", relative, "file is not valid UTF-8")]
|
|
findings: list[Finding] = []
|
|
for rule, expression in RULES:
|
|
match = expression.search(text)
|
|
if match:
|
|
reason = approved_reason(rule, relative, entries)
|
|
findings.append(
|
|
Finding(rule, relative, match.group(0)[:120], reason is not None, reason or "")
|
|
)
|
|
return findings
|
|
|
|
|
|
def sha256(path: Path) -> str:
|
|
value = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
value.update(chunk)
|
|
return value.hexdigest()
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--repository", type=Path, default=Path.cwd())
|
|
parser.add_argument("--allowlist", type=Path, default=Path("public-mirror.allowlist"))
|
|
parser.add_argument("--exceptions", type=Path, default=Path("public-mirror.exceptions"))
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
parser.add_argument("--report", type=Path)
|
|
parser.add_argument("--max-file-bytes", type=int, default=5_000_000)
|
|
args = parser.parse_args()
|
|
|
|
root = args.repository.resolve()
|
|
allowlist_file = args.allowlist if args.allowlist.is_absolute() else root / args.allowlist
|
|
exception_file = args.exceptions if args.exceptions.is_absolute() else root / args.exceptions
|
|
output = args.output.resolve()
|
|
report_path = (
|
|
args.report.resolve() if args.report else output.with_suffix(output.suffix + ".report.json")
|
|
)
|
|
try:
|
|
git(root, "rev-parse", "--is-inside-work-tree")
|
|
revision = git(root, "rev-parse", "HEAD")
|
|
tracked = [item for item in git(root, "ls-files", "-z").split("\x00") if item]
|
|
allow = load_patterns(allowlist_file)
|
|
approved = load_exceptions(exception_file)
|
|
except (OSError, ValueError, subprocess.CalledProcessError) as error:
|
|
print(f"public mirror preflight failed: {error}", file=sys.stderr)
|
|
return 2
|
|
|
|
selected = sorted(
|
|
path for path in tracked if matches(path, allow) and not matches(path, ALWAYS_DENY)
|
|
)
|
|
findings: list[Finding] = []
|
|
for relative in selected:
|
|
try:
|
|
findings.extend(scan(root, relative, args.max_file_bytes, approved))
|
|
except OSError as error:
|
|
findings.append(Finding("read_error", relative, str(error)))
|
|
blocking = [item for item in findings if not item.allowed]
|
|
report = {
|
|
"schemaVersion": 1,
|
|
"repository": str(root),
|
|
"revision": revision,
|
|
"selectedFiles": selected,
|
|
"findings": [asdict(item) for item in findings],
|
|
"blockingFindingCount": len(blocking),
|
|
}
|
|
report_path.parent.mkdir(parents=True, exist_ok=True)
|
|
report_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
|
|
if not selected:
|
|
print("allowlist selected no tracked files", file=sys.stderr)
|
|
return 2
|
|
if blocking:
|
|
print(
|
|
f"export blocked by {len(blocking)} finding(s); review {report_path}", file=sys.stderr
|
|
)
|
|
return 2
|
|
if output.exists() and any(output.iterdir()):
|
|
print(f"output directory is not empty: {output}", file=sys.stderr)
|
|
return 2
|
|
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
manifest: list[dict[str, object]] = []
|
|
for relative in selected:
|
|
source = root / relative
|
|
destination = output / relative
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(source, destination, follow_symlinks=False)
|
|
manifest.append(
|
|
{"path": relative, "size": destination.stat().st_size, "sha256": sha256(destination)}
|
|
)
|
|
(output / "PUBLIC_MIRROR_MANIFEST.json").write_text(
|
|
json.dumps(
|
|
{"schemaVersion": 1, "sourceRevision": revision, "files": manifest},
|
|
indent=2,
|
|
sort_keys=True,
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
print(f"exported {len(selected)} files to {output}")
|
|
print(f"sanitation report: {report_path}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|