190 lines
8.5 KiB
Python
190 lines
8.5 KiB
Python
#!/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())
|