Files
firmware-lowlevel-development/skills/firmware-lowlevel-bsp-development/scripts/bsp/project_guard.py
T

602 lines
23 KiB
Python
Executable File

#!/usr/bin/env python3
"""Safely stage, verify, finalize, and audit an EmxRes BSP project clone."""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import re
import shutil
import stat
import sys
import uuid
from datetime import datetime, timezone
from pathlib import Path, PurePosixPath
ROOT_SUFFIX = Path("EapilBsp") / "EMX2.0" / "EmxRes"
PROJECT_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
EXCLUDED_DIR_NAMES = {
".git",
".cache",
".mypy_cache",
".pytest_cache",
".ruff_cache",
"__pycache__",
"build",
"logs",
"out",
"output",
}
EXCLUDED_FILE_SUFFIXES = {".log", ".pyc", ".tmp"}
CREDENTIAL_FILE_NAMES = {
".env",
"credentials.json",
"credentials.yaml",
"credentials.yml",
"id_ed25519",
"id_rsa",
}
CREDENTIAL_FILE_SUFFIXES = {".key", ".p12", ".pem", ".pfx"}
DOC_PREFIX = "doc/bsp_adaptation/"
DELTA_REPORT_PATH = "doc/bsp_adaptation/validation/project_delta.json"
FINAL_AUDIT_REPORT_PATH = "doc/bsp_adaptation/validation/final_audit.json"
class GuardError(RuntimeError):
"""A safety or validation gate failed."""
def now_utc() -> str:
return datetime.now(timezone.utc).isoformat(timespec="seconds")
def production_root() -> Path:
return Path.home() / ROOT_SUFFIX
def normalize_root(root: Path) -> Path:
root = root.expanduser()
if not root.is_dir():
raise GuardError(f"EmxRes 根目录不存在: {root}")
if root.is_symlink():
raise GuardError(f"EmxRes 根目录不能是软链接: {root}")
return root.resolve()
def validate_project_name(name: str, field: str) -> str:
if not PROJECT_NAME_RE.fullmatch(name or ""):
raise GuardError(
f"{field} 必须是 EmxRes 直接子目录名,只能包含字母、数字、点、下划线和连字符: {name!r}"
)
return name
def direct_child(
root: Path, name: str, field: str, must_exist: bool, allow_hidden: bool = False
) -> Path:
if allow_hidden and name.startswith("."):
validate_project_name(name[1:], field)
else:
name = validate_project_name(name, field)
path = root / name
if path.parent != root:
raise GuardError(f"{field} 发生路径逃逸: {name}")
if must_exist:
if not path.is_dir() or path.is_symlink():
raise GuardError(f"{field} 不是普通项目目录: {path}")
if path.resolve().parent != root:
raise GuardError(f"{field} 解析后不在 EmxRes 内: {path}")
return path
def is_excluded_name(path: Path, is_dir: bool) -> bool:
if is_dir and path.name in EXCLUDED_DIR_NAMES:
return True
return not is_dir and path.suffix.lower() in EXCLUDED_FILE_SUFFIXES
def copy_ignore(_directory: str, names: list[str]) -> set[str]:
ignored = set()
base = Path(_directory)
for name in names:
path = base / name
is_dir = path.is_dir()
if is_excluded_name(path, is_dir):
ignored.add(name)
return ignored
def scan_credentials(reference: Path) -> list[str]:
hits = []
for current, dirs, files in os.walk(reference, followlinks=False):
base = Path(current)
dirs[:] = [
name
for name in dirs
if not is_excluded_name(base / name, True) and not (base / name).is_symlink()
]
for name in files:
path = base / name
if path.is_symlink() or is_excluded_name(path, False):
continue
if name in CREDENTIAL_FILE_NAMES or path.suffix.lower() in CREDENTIAL_FILE_SUFFIXES:
hits.append(path.relative_to(reference).as_posix())
return sorted(hits)
def excluded_paths(reference: Path) -> list[str]:
hits = []
for current, dirs, files in os.walk(reference, followlinks=False):
base = Path(current)
kept_dirs = []
for name in dirs:
path = base / name
if is_excluded_name(path, True):
hits.append(path.relative_to(reference).as_posix() + "/")
elif path.is_symlink():
continue
else:
kept_dirs.append(name)
dirs[:] = kept_dirs
for name in files:
path = base / name
if not path.is_symlink() and is_excluded_name(path, False):
hits.append(path.relative_to(reference).as_posix())
return sorted(hits)
def within(path: Path, root: Path) -> bool:
try:
path.relative_to(root)
return True
except ValueError:
return False
def expected_imgs_target(project_name: str) -> str:
return f"/epbuild/EMX2.0/EmxRes/{project_name}/imgs"
def scan_symlinks(reference: Path, root: Path, reference_name: str) -> list[dict[str, str]]:
links = []
link_paths = []
for current, dirs, files in os.walk(reference, followlinks=False):
base = Path(current)
kept_dirs = []
for name in dirs:
path = base / name
if is_excluded_name(path, True):
continue
if path.is_symlink():
link_paths.append(path)
else:
kept_dirs.append(name)
dirs[:] = kept_dirs
for name in files:
path = base / name
if path.is_symlink() and not is_excluded_name(path, False):
link_paths.append(path)
for path in sorted(link_paths):
relative = path.relative_to(reference).as_posix()
raw_target = os.readlink(path)
if relative == "imgs" and raw_target == expected_imgs_target(reference_name):
kind = "imgs"
elif os.path.isabs(raw_target):
resolved = Path(raw_target).resolve(strict=False)
if not within(resolved, root):
raise GuardError(f"存在未经确认的外部绝对软链接: {relative} -> {raw_target}")
kind = "absolute-in-root"
else:
resolved = (path.parent / raw_target).resolve(strict=False)
if not within(resolved, root):
raise GuardError(f"存在未经确认的外部相对软链接: {relative} -> {raw_target}")
kind = "relative-in-root"
links.append({"path": relative, "target": raw_target, "kind": kind})
return links
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for block in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
def inventory(
project: Path, omit_paths: set[str] | None = None
) -> dict[str, dict[str, object]]:
omit_paths = omit_paths or set()
entries: dict[str, dict[str, object]] = {}
for current, dirs, files in os.walk(project, followlinks=False):
base = Path(current)
kept_dirs = []
for name in sorted(dirs):
path = base / name
if is_excluded_name(path, True):
continue
if path.is_symlink():
rel = path.relative_to(project).as_posix()
entries[rel] = {
"kind": "symlink",
"mode": stat.S_IMODE(path.lstat().st_mode),
"target": os.readlink(path),
}
else:
kept_dirs.append(name)
dirs[:] = kept_dirs
for name in sorted(files):
path = base / name
if is_excluded_name(path, False):
continue
rel = path.relative_to(project).as_posix()
if rel in omit_paths:
continue
if path.is_symlink():
entries[rel] = {
"kind": "symlink",
"mode": stat.S_IMODE(path.lstat().st_mode),
"target": os.readlink(path),
}
elif path.is_file():
file_stat = path.stat()
entries[rel] = {
"kind": "file",
"mode": stat.S_IMODE(file_stat.st_mode),
"size": file_stat.st_size,
"sha256": sha256_file(path),
}
return dict(sorted(entries.items()))
def write_json(path: Path, payload: dict[str, object]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
def prepare_project(reference_name: str, target_name: str, root: Path | None = None) -> dict[str, object]:
root = normalize_root(root or production_root())
reference = direct_child(root, reference_name, "reference-project", True)
target = direct_child(root, target_name, "target-project", False)
if target.exists() or target.is_symlink():
raise GuardError(f"目标项目已存在,禁止覆盖: {target}")
credentials = scan_credentials(reference)
if credentials:
raise GuardError("参考项目包含疑似凭据文件,未复制: " + ", ".join(credentials))
links = scan_symlinks(reference, root, reference_name)
source_before = inventory(reference)
staging_name = f".{target_name}.bsp-staging-{uuid.uuid4().hex[:12]}"
staging = root / staging_name
shutil.copytree(
reference,
staging,
symlinks=True,
copy_function=shutil.copy2,
ignore=copy_ignore,
)
imgs = staging / "imgs"
imgs_retargeted = False
if imgs.is_symlink() and os.readlink(imgs) == expected_imgs_target(reference_name):
imgs.unlink()
imgs.symlink_to(expected_imgs_target(target_name))
imgs_retargeted = True
doc_root = staging / "doc" / "bsp_adaptation"
(doc_root / "inputs").mkdir(parents=True, exist_ok=True)
validation = doc_root / "validation"
validation.mkdir(parents=True, exist_ok=True)
source_after = inventory(reference)
if source_before != source_after:
raise GuardError(f"复制期间参考项目发生变化,临时目录保留供检查: {staging}")
manifest: dict[str, object] = {
"schema_version": 1,
"created_at": now_utc(),
"emxres_root": str(root),
"reference_project": reference_name,
"target_project": target_name,
"staging_directory": staging_name,
"excluded_paths": excluded_paths(reference),
"symlinks": links,
"imgs_retargeted": imgs_retargeted,
"imgs_target": os.readlink(imgs) if imgs.is_symlink() else None,
"source_inventory": source_before,
"staging_inventory_before_edits": inventory(staging),
}
write_json(validation / "clone_manifest.json", manifest)
return {
"staging_directory": staging_name,
"staging_path": str(staging),
"manifest": str(validation / "clone_manifest.json"),
"imgs_retargeted": imgs_retargeted,
}
def normalize_allow_paths(values: list[str]) -> set[str]:
allowed = set()
for value in values:
pure = PurePosixPath(value)
if pure.is_absolute() or ".." in pure.parts or value in {"", "."}:
raise GuardError(f"allow-path 必须是项目内精确相对路径: {value!r}")
if any(char in value for char in "*?[]"):
raise GuardError(f"allow-path 不允许通配符,必须逐文件列出: {value!r}")
allowed.add(pure.as_posix())
return allowed
def compare_inventories(
source: dict[str, dict[str, object]], target: dict[str, dict[str, object]]
) -> dict[str, list[str]]:
source_paths = set(source)
target_paths = set(target)
return {
"added": sorted(target_paths - source_paths),
"removed": sorted(source_paths - target_paths),
"modified": sorted(path for path in source_paths & target_paths if source[path] != target[path]),
}
def load_clone_manifest(
staging: Path, root: Path, reference_name: str, target_name: str, staging_name: str
) -> dict[str, object]:
path = staging / "doc" / "bsp_adaptation" / "validation" / "clone_manifest.json"
if not path.is_file():
raise GuardError("缺少 clone_manifest.json;临时目录不是受控 prepare 结果")
manifest = json.loads(path.read_text(encoding="utf-8"))
expected = {
"emxres_root": str(root),
"reference_project": reference_name,
"target_project": target_name,
"staging_directory": staging_name,
}
mismatched = [key for key, value in expected.items() if manifest.get(key) != value]
if mismatched:
raise GuardError("clone_manifest 与当前操作不匹配: " + ", ".join(mismatched))
return manifest
def verify_project(
reference_name: str,
staging_name: str,
target_name: str,
allow_paths: list[str],
root: Path | None = None,
) -> dict[str, object]:
root = normalize_root(root or production_root())
reference = direct_child(root, reference_name, "reference-project", True)
validate_project_name(target_name, "target-project")
staging = direct_child(root, staging_name, "staging-dir", True, allow_hidden=True)
expected_prefix = f".{target_name}.bsp-staging-"
if not staging_name.startswith(expected_prefix):
raise GuardError(f"临时目录与目标项目不匹配: {staging_name}")
clone_manifest = load_clone_manifest(staging, root, reference_name, target_name, staging_name)
scan_symlinks(staging, root, target_name)
credentials = scan_credentials(staging)
if credentials:
raise GuardError("临时目录包含疑似凭据文件,禁止最终化: " + ", ".join(credentials))
unexpected_excluded = excluded_paths(staging)
if unexpected_excluded:
raise GuardError(
"临时目录新增了被排除的缓存、构建输出或日志,必须先处理: "
+ ", ".join(unexpected_excluded)
)
allowed = normalize_allow_paths(allow_paths)
source_inventory = clone_manifest.get("source_inventory")
if not isinstance(source_inventory, dict):
raise GuardError("clone_manifest 缺少 prepare 时的参考项目清单")
if inventory(reference) != source_inventory:
raise GuardError("prepare 后参考项目发生变化;必须重新确认并重新 prepare")
staging_inventory = inventory(staging, {DELTA_REPORT_PATH})
changes = compare_inventories(source_inventory, staging_inventory)
def permitted(path: str) -> bool:
return path == "imgs" or path.startswith(DOC_PREFIX) or path in allowed
violations = sorted(
{path for group in changes.values() for path in group if not permitted(path)}
)
report: dict[str, object] = {
"schema_version": 1,
"checked_at": now_utc(),
"reference_project": reference_name,
"target_project": target_name,
"staging_directory": staging_name,
"allowed_paths": sorted(allowed),
"changes": changes,
"violations": violations,
"passed": not violations,
"verified_staging_inventory": staging_inventory,
}
write_json(staging / "doc" / "bsp_adaptation" / "validation" / "project_delta.json", report)
return report
def finalize_project(
reference_name: str,
staging_name: str,
target_name: str,
allow_paths: list[str],
root: Path | None = None,
) -> dict[str, object]:
root = normalize_root(root or production_root())
target = direct_child(root, target_name, "target-project", False)
if target.exists() or target.is_symlink():
raise GuardError(f"目标项目已存在,禁止覆盖: {target}")
staging = direct_child(root, staging_name, "staging-dir", True, allow_hidden=True)
if not staging_name.startswith(f".{target_name}.bsp-staging-"):
raise GuardError(f"临时目录与目标项目不匹配: {staging_name}")
report_path = staging / "doc" / "bsp_adaptation" / "adaptation_report.md"
delta = verify_project(reference_name, staging_name, target_name, allow_paths, root=root)
if delta.get("passed") is not True:
raise GuardError("差异白名单未通过,禁止最终化")
verified_inventory = delta.get("verified_staging_inventory")
if not isinstance(verified_inventory, dict):
raise GuardError("差异报告缺少已验证目录清单;必须重新运行 verify")
current_inventory = inventory(staging, {DELTA_REPORT_PATH})
if current_inventory != verified_inventory:
raise GuardError("verify 后临时目录又发生变化;必须重新运行 verify")
if not report_path.is_file() or not report_path.read_text(encoding="utf-8").strip():
raise GuardError("缺少非空的 doc/bsp_adaptation/adaptation_report.md")
os.rename(staging, target)
finalized_at = now_utc()
clone_path = target / "doc" / "bsp_adaptation" / "validation" / "clone_manifest.json"
if clone_path.is_file():
clone = json.loads(clone_path.read_text(encoding="utf-8"))
clone["finalized_at"] = finalized_at
clone["final_path"] = str(target)
write_json(clone_path, clone)
delta_path = target / "doc" / "bsp_adaptation" / "validation" / "project_delta.json"
delta["finalized_at"] = finalized_at
delta["final_path"] = str(target)
write_json(delta_path, delta)
return {"target_project": target_name, "target_path": str(target), "finalized_at": finalized_at}
def audit_final_project(
target_name: str,
allow_paths: list[str] | None = None,
root: Path | None = None,
) -> dict[str, object]:
"""Audit a finalized project against its immutable prepare-time baseline."""
root = normalize_root(root or production_root())
target = direct_child(root, target_name, "target-project", True)
clone_path = target / "doc" / "bsp_adaptation" / "validation" / "clone_manifest.json"
if not clone_path.is_file():
raise GuardError("正式项目缺少 clone_manifest.json,无法恢复 prepare 时基线")
clone = json.loads(clone_path.read_text(encoding="utf-8"))
expected = {"emxres_root": str(root), "target_project": target_name}
mismatched = [key for key, value in expected.items() if clone.get(key) != value]
if mismatched:
raise GuardError("clone_manifest 与正式项目不匹配: " + ", ".join(mismatched))
source_inventory = clone.get("source_inventory")
if not isinstance(source_inventory, dict):
raise GuardError("clone_manifest 缺少 prepare 时的参考项目清单")
delta_path = target / DELTA_REPORT_PATH
delta: dict[str, object] = {}
if delta_path.is_file():
delta = json.loads(delta_path.read_text(encoding="utf-8"))
effective_allow_paths = allow_paths
if effective_allow_paths is None or not effective_allow_paths:
stored = delta.get("allowed_paths")
if not isinstance(stored, list) or not all(isinstance(item, str) for item in stored):
raise GuardError("未提供 allow-path,且 project_delta.json 中没有可复用白名单")
effective_allow_paths = stored
allowed = normalize_allow_paths(effective_allow_paths)
scan_symlinks(target, root, target_name)
credentials = scan_credentials(target)
if credentials:
raise GuardError("正式项目包含疑似凭据文件: " + ", ".join(credentials))
unexpected_excluded = excluded_paths(target)
if unexpected_excluded:
raise GuardError(
"正式项目包含被排除的缓存、构建输出或日志: " + ", ".join(unexpected_excluded)
)
target_inventory = inventory(target, {DELTA_REPORT_PATH, FINAL_AUDIT_REPORT_PATH})
changes = compare_inventories(source_inventory, target_inventory)
def permitted(path: str) -> bool:
return path == "imgs" or path.startswith(DOC_PREFIX) or path in allowed
violations = sorted(
{path for group in changes.values() for path in group if not permitted(path)}
)
reference_name = clone.get("reference_project")
reference_drifted: bool | None = None
if isinstance(reference_name, str):
reference = root / reference_name
if reference.is_dir() and not reference.is_symlink():
reference_drifted = inventory(reference) != source_inventory
report: dict[str, object] = {
"schema_version": 1,
"checked_at": now_utc(),
"reference_project": reference_name,
"reference_drifted_since_prepare": reference_drifted,
"target_project": target_name,
"target_path": str(target),
"baseline": "clone_manifest.source_inventory",
"allowed_paths": sorted(allowed),
"changes": changes,
"violations": violations,
"passed": not violations,
"audited_target_inventory": target_inventory,
}
write_json(target / FINAL_AUDIT_REPORT_PATH, report)
return report
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
subparsers = parser.add_subparsers(dest="command", required=True)
prepare = subparsers.add_parser("prepare", help="复制参考项目到唯一临时目录")
prepare.add_argument("--reference-project", required=True)
prepare.add_argument("--target-project", required=True)
verify = subparsers.add_parser("verify", help="按精确文件白名单检查差异")
verify.add_argument("--reference-project", required=True)
verify.add_argument("--staging-dir", required=True)
verify.add_argument("--target-project", required=True)
verify.add_argument("--allow-path", action="append", default=[])
finalize = subparsers.add_parser("finalize", help="将已验证临时目录原子落为正式项目")
finalize.add_argument("--reference-project", required=True)
finalize.add_argument("--staging-dir", required=True)
finalize.add_argument("--target-project", required=True)
finalize.add_argument("--allow-path", action="append", default=[])
audit = subparsers.add_parser("audit-final", help="复核已落盘正式项目的差异白名单")
audit.add_argument("--target-project", required=True)
audit.add_argument("--allow-path", action="append", default=[])
return parser
def main() -> int:
args = build_parser().parse_args()
try:
if args.command == "prepare":
result = prepare_project(args.reference_project, args.target_project)
elif args.command == "verify":
result = verify_project(
args.reference_project,
args.staging_dir,
args.target_project,
args.allow_path,
)
if not result["passed"]:
print(json.dumps(result, ensure_ascii=False, indent=2))
return 2
elif args.command == "finalize":
result = finalize_project(
args.reference_project,
args.staging_dir,
args.target_project,
args.allow_path,
)
else:
result = audit_final_project(args.target_project, args.allow_path)
if not result["passed"]:
print(json.dumps(result, ensure_ascii=False, indent=2))
return 2
print(json.dumps(result, ensure_ascii=False, indent=2))
return 0
except (GuardError, OSError, json.JSONDecodeError) as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return 2
if __name__ == "__main__":
raise SystemExit(main())