#!/usr/bin/env python3 """预览并导出 Git 变更文件，同时生成安全分类清单。""" from __future__ import annotations import argparse import fnmatch import json import re import subprocess import sys from pathlib import Path, PurePosixPath PROTECTED = ( ".env", ".env.*", "application*.yml", "application*.yaml", "application*.properties", "bootstrap*.yml", "bootstrap*.yaml", "config/local/*", "config/dev/*", "*.local.*", "*.secret.*", ".craftkit/local/*", ".craftkit/cache/*", ) BLOCKED_SUFFIXES = {".key", ".pem", ".p12", ".pfx"} SECRET_PATTERN = re.compile( rb"(?i)(api[_-]?key|password|access[_-]?token)\s*[:=]\s*['\"]?[A-Za-z0-9_./+@-]{8,}" ) def git(repo: Path, *args: str, check: bool = True) -> bytes: """执行只读 Git 子命令，保留原始字节以支持任意合法文件名。""" result = subprocess.run(["git", "-C", str(repo), *args], capture_output=True, check=False) if check and result.returncode: raise ValueError(result.stderr.decode("utf-8", errors="replace").strip()) return result.stdout def paths_from_nul(data: bytes) -> list[str]: """将 Git 的 NUL 分隔路径转换为去重后的正斜杠相对路径。""" values = {item.decode("utf-8", errors="surrogateescape").replace("\\", "/") for item in data.split(b"\0") if item} return sorted(values) def selected_paths(args: argparse.Namespace, repo: Path) -> list[str]: """根据模式选择路径；时间只负责选路径，不决定内容快照。""" if args.mode == "worktree": values = paths_from_nul(git(repo, "diff", "--name-only", "-z", "HEAD", "--")) if args.include_untracked: values = sorted(set(values) | set(paths_from_nul(git(repo, "ls-files", "--others", "--exclude-standard", "-z")))) return values if args.mode == "range": if not args.base or not args.head: raise ValueError("range 模式必须提供 --base 和 --head") return paths_from_nul(git(repo, "diff", "--name-only", "-z", args.base, args.head, "--")) if not args.since: raise ValueError("time 模式必须提供 --since") command = ["log", f"--since={args.since}"] if args.until: command.append(f"--until={args.until}") command.extend(["--name-only", "--format=", "-z", "--"]) return paths_from_nul(git(repo, *command)) def safe_relative(value: str) -> PurePosixPath: """拒绝绝对路径、空路径和目录逃逸。""" path = PurePosixPath(value) if path.is_absolute() or not path.parts or any(part in {"", ".", ".."} for part in path.parts): raise ValueError(f"不安全的 Git 路径：{value}") return path def is_protected(value: str) -> bool: """识别默认不应进入交付快照的环境和本地配置。""" lowered = value.casefold() name = PurePosixPath(lowered).name if name in {".env.example", ".env.sample", ".env.template"}: return False return any(fnmatch.fnmatch(lowered, pattern) or fnmatch.fnmatch(name, pattern) for pattern in PROTECTED) def evidence_allows(value: str, evidence: str) -> bool: """规范必须明确写出路径、文件名或可匹配该路径的 glob。""" lowered = value.casefold() name = PurePosixPath(lowered).name content = evidence.casefold() if lowered in content or name in content: return True tokens = re.findall(r"[a-z0-9_.*/?-]+", content) return any("*" in token and (fnmatch.fnmatch(lowered, token) or fnmatch.fnmatch(name, token)) for token in tokens) def read_snapshot(repo: Path, value: str, snapshot: str) -> bytes | None: """从工作区或指定 Git 树读取文件；不存在时返回空值。""" if snapshot == "worktree": path = (repo / Path(*PurePosixPath(value).parts)).resolve() try: path.relative_to(repo) except ValueError as error: raise ValueError(f"来源路径逃逸仓库：{value}") from error return path.read_bytes() if path.is_file() else None result = subprocess.run(["git", "-C", str(repo), "show", f"{snapshot}:{value}"], capture_output=True, check=False) return result.stdout if result.returncode == 0 else None def policy_text(args: argparse.Namespace, repo: Path) -> tuple[str, str | None]: """验证规范证据位于允许的项目文档范围内。""" if not args.allow_protected: return "", None if not args.policy_evidence: raise ValueError("允许环境配置时必须提供 --policy-evidence") path = (repo / args.policy_evidence).resolve() try: relative = path.relative_to(repo).as_posix() except ValueError as error: raise ValueError("规范证据必须位于项目仓库内") from error allowed = relative == "AGENTS.md" or relative.endswith("/AGENTS.md") or relative.startswith((".craftkit/agents/", ".craftkit/standards/")) if not allowed or not path.is_file(): raise ValueError("规范证据必须是适用的 AGENTS.md 或 .craftkit Agent/规范文档") return path.read_text(encoding="utf-8-sig"), relative def build_manifest(args: argparse.Namespace, repo: Path, target: Path) -> tuple[dict, dict[str, bytes]]: """读取候选内容并分类，blocked 项始终不能被覆盖授权解除。""" evidence, evidence_path = policy_text(args, repo) manifest = {"mode": args.mode, "snapshot": args.snapshot, "policyEvidence": evidence_path, "exported": [], "protected": [], "blocked": [], "deleted": [], "missing": []} payloads: dict[str, bytes] = {} target_prefix = None try: target_prefix = target.relative_to(repo).as_posix().rstrip("/") + "/" except ValueError: pass for raw in selected_paths(args, repo): value = safe_relative(raw).as_posix() if target_prefix and value.startswith(target_prefix): continue data = read_snapshot(repo, value, args.snapshot) if data is None: manifest["deleted"].append(value); continue if PurePosixPath(value).suffix.casefold() in BLOCKED_SUFFIXES or b"PRIVATE KEY" in data or SECRET_PATTERN.search(data): manifest["blocked"].append(value); continue if is_protected(value) and not (args.allow_protected and evidence_allows(value, evidence)): manifest["protected"].append(value); continue manifest["exported"].append(value) payloads[value] = data return manifest, payloads def main(argv: list[str] | None = None) -> int: """默认输出预演 JSON；显式 --apply 后才创建交付目录。""" parser = argparse.ArgumentParser(description="预览并导出 Git 变更文件") parser.add_argument("--repo", type=Path, required=True); parser.add_argument("--target", type=Path, required=True) parser.add_argument("--mode", choices=("range", "time", "worktree"), required=True) parser.add_argument("--base"); parser.add_argument("--head"); parser.add_argument("--since"); parser.add_argument("--until") parser.add_argument("--snapshot", default="HEAD"); parser.add_argument("--include-untracked", action="store_true") parser.add_argument("--allow-protected", action="store_true"); parser.add_argument("--policy-evidence") parser.add_argument("--apply", action="store_true"); parser.add_argument("--force", action="store_true") args = parser.parse_args(argv) repo, target = args.repo.resolve(), args.target.resolve() try: git(repo, "rev-parse", "--is-inside-work-tree") if target == repo: raise ValueError("导出目标不得等于仓库根目录") if args.snapshot != "worktree": git(repo, "rev-parse", "--verify", f"{args.snapshot}^{{commit}}") if args.apply and target.exists() and not args.force: raise ValueError(f"目标目录已存在，覆盖需明确使用 --force：{target}") manifest, payloads = build_manifest(args, repo, target) manifest["operation"] = "apply" if args.apply else "dry-run" if args.apply: target.mkdir(parents=True, exist_ok=True) for value, data in payloads.items(): output = target / Path(*PurePosixPath(value).parts) output.parent.mkdir(parents=True, exist_ok=True); output.write_bytes(data) (target / "export-manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(json.dumps(manifest, ensure_ascii=False, indent=2)) return 2 if manifest["blocked"] else 0 except (OSError, ValueError) as error: print(f"错误：{error}", file=sys.stderr); return 1 if __name__ == "__main__": raise SystemExit(main())