109 lines
4.1 KiB
Python
109 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""签发或校验 staging RC 的 Python 运行树字节清单。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import stat
|
|
from pathlib import Path
|
|
|
|
|
|
IGNORED_PARTS = frozenset({"__pycache__", ".pytest_cache"})
|
|
|
|
|
|
def sha256_file(path: Path) -> str:
|
|
"""流式计算文件摘要，避免大依赖文件一次性进入内存。"""
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def should_ignore(path: Path) -> bool:
|
|
"""忽略 Python 运行时可再生缓存，其余新增文件一律计入漂移。"""
|
|
return path.suffix == ".pyc" or any(part in IGNORED_PARTS for part in path.parts)
|
|
|
|
|
|
def capture(roots: list[str]) -> dict:
|
|
"""采集普通文件与符号链接；符号链接同时绑定文本目标和解析后文件摘要。"""
|
|
entries: list[dict] = []
|
|
cwd = Path.cwd().resolve()
|
|
normalized_roots: list[str] = []
|
|
for raw_root in roots:
|
|
root = Path(raw_root)
|
|
if root.is_absolute() or ".." in root.parts or not root.is_dir() or root.is_symlink():
|
|
raise ValueError(f"运行树根必须是仓内非符号链接目录: {raw_root}")
|
|
normalized_roots.append(root.as_posix())
|
|
for path in root.rglob("*"):
|
|
relative = path.relative_to(cwd) if path.is_absolute() else path
|
|
if should_ignore(relative):
|
|
continue
|
|
metadata = path.lstat()
|
|
mode = stat.S_IMODE(metadata.st_mode)
|
|
if path.is_symlink():
|
|
target = os.readlink(path)
|
|
resolved = path.resolve(strict=False)
|
|
entries.append({
|
|
"path": relative.as_posix(),
|
|
"type": "symlink",
|
|
"mode": mode,
|
|
"target": target,
|
|
"resolvedSha256": sha256_file(resolved) if resolved.is_file() else None,
|
|
})
|
|
elif path.is_file():
|
|
entries.append({
|
|
"path": relative.as_posix(),
|
|
"type": "file",
|
|
"mode": mode,
|
|
"sha256": sha256_file(path),
|
|
"size": metadata.st_size,
|
|
})
|
|
entries.sort(key=lambda item: item["path"])
|
|
return {"schema": "staging-runtime-tree/1", "roots": normalized_roots, "entries": entries}
|
|
|
|
|
|
def write_manifest(output: Path, roots: list[str]) -> None:
|
|
"""以 0600 原子替换清单，避免半写文件被签入 RC。"""
|
|
payload = capture(roots)
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = output.with_name(output.name + ".tmp")
|
|
temporary.write_text(json.dumps(payload, ensure_ascii=False, sort_keys=True, indent=2) + "\n",
|
|
encoding="utf-8")
|
|
temporary.chmod(0o600)
|
|
temporary.replace(output)
|
|
|
|
|
|
def verify_manifest(manifest: Path, roots: list[str]) -> None:
|
|
"""逐结构比较当前树，新增、删除、权限、字节或链接目标漂移均拒绝。"""
|
|
expected = json.loads(manifest.read_text(encoding="utf-8"))
|
|
actual = capture(roots)
|
|
if expected != actual:
|
|
raise RuntimeError("Python 运行树与 RC 清单不一致")
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="staging RC Python 运行树完整性清单")
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
capture_parser = subparsers.add_parser("capture")
|
|
capture_parser.add_argument("--output", required=True, type=Path)
|
|
capture_parser.add_argument("roots", nargs="+")
|
|
verify_parser = subparsers.add_parser("verify")
|
|
verify_parser.add_argument("--manifest", required=True, type=Path)
|
|
verify_parser.add_argument("roots", nargs="+")
|
|
args = parser.parse_args()
|
|
if args.command == "capture":
|
|
write_manifest(args.output, args.roots)
|
|
print(f"RUNTIME_TREE_CAPTURE_PASS manifest={args.output}")
|
|
else:
|
|
verify_manifest(args.manifest, args.roots)
|
|
print(f"RUNTIME_TREE_VERIFY_PASS manifest={args.manifest}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|