499 lines
19 KiB
Python
Executable File
499 lines
19 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""Render a PSP evidence handoff for user review or Etunel Hub; never send it."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import tempfile
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
|
||
STATUSES = ("candidate", "packaged", "device_verified", "app_integrated", "accepted")
|
||
|
||
|
||
class HandoffError(RuntimeError):
|
||
"""Handoff evidence is incomplete or inconsistent."""
|
||
|
||
|
||
def now_utc() -> str:
|
||
return datetime.now(timezone.utc).isoformat(timespec="seconds")
|
||
|
||
|
||
def load_object(path: Path, name: str) -> dict[str, Any]:
|
||
payload = json.loads(path.expanduser().read_text(encoding="utf-8"))
|
||
if not isinstance(payload, dict):
|
||
raise HandoffError(f"{name} JSON 根节点必须是对象: {path}")
|
||
return payload
|
||
|
||
|
||
def write_json_new(path: Path, payload: dict[str, object]) -> None:
|
||
path = path.expanduser().resolve(strict=False)
|
||
if path.exists() or path.is_symlink():
|
||
raise HandoffError(f"交接文件已存在，禁止覆盖: {path}")
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
with tempfile.NamedTemporaryFile(
|
||
"w", encoding="utf-8", dir=path.parent, prefix=f".{path.name}.", delete=False
|
||
) as handle:
|
||
json.dump(payload, handle, ensure_ascii=False, indent=2)
|
||
handle.write("\n")
|
||
temporary = Path(handle.name)
|
||
os.replace(temporary, path)
|
||
|
||
|
||
def run_git(path: Path, *args: str) -> subprocess.CompletedProcess[str]:
|
||
return subprocess.run(
|
||
["git", "-C", str(path), *args],
|
||
text=True,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
check=False,
|
||
)
|
||
|
||
|
||
def psp_git_state(path: Path) -> dict[str, object]:
|
||
path = path.expanduser().resolve(strict=True)
|
||
if not path.is_dir() or path.is_symlink():
|
||
raise HandoffError(f"PSP 仓库必须是普通目录: {path}")
|
||
top = run_git(path, "rev-parse", "--show-toplevel")
|
||
commit = run_git(path, "rev-parse", "HEAD")
|
||
status = run_git(path, "status", "--porcelain", "--untracked-files=all", "--", ".")
|
||
if top.returncode or commit.returncode or status.returncode:
|
||
raise HandoffError("PSP 必须是可读取 HEAD 和工作区状态的 Git worktree")
|
||
lines = [line for line in status.stdout.splitlines() if line.strip()]
|
||
commit_value = commit.stdout.strip()
|
||
tags = run_git(path, "tag", "--points-at", commit_value)
|
||
if tags.returncode != 0:
|
||
raise HandoffError(f"无法读取 PSP 状态 tag: {tags.stderr.strip()}")
|
||
return {
|
||
"path": str(path),
|
||
"repository_root": str(Path(top.stdout.strip()).resolve()),
|
||
"commit": commit_value,
|
||
"dirty": bool(lines),
|
||
"status": lines,
|
||
"tags": sorted(line for line in tags.stdout.splitlines() if line),
|
||
}
|
||
|
||
|
||
def annotated_tag(path: Path, name: str) -> bool:
|
||
result = run_git(path, "cat-file", "-t", f"refs/tags/{name}")
|
||
return result.returncode == 0 and result.stdout.strip() == "tag"
|
||
|
||
|
||
def shared_repository_url(value: str) -> bool:
|
||
stripped = value.strip()
|
||
return bool(stripped) and not stripped.startswith(("/", "./", "../", "~", "file:"))
|
||
|
||
|
||
def verify_remote(repository_url: str, commit: str) -> dict[str, object]:
|
||
checked_at = now_utc()
|
||
environment = os.environ.copy()
|
||
environment["GIT_TERMINAL_PROMPT"] = "0"
|
||
try:
|
||
result = subprocess.run(
|
||
["git", "ls-remote", "--heads", "--tags", repository_url],
|
||
text=True,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
check=False,
|
||
timeout=30,
|
||
env=environment,
|
||
)
|
||
except subprocess.TimeoutExpired:
|
||
return {
|
||
"schema_version": 1,
|
||
"passed": False,
|
||
"checked_at": checked_at,
|
||
"repository_url": repository_url,
|
||
"commit": commit,
|
||
"refs": [],
|
||
"errors": ["git ls-remote 超时"],
|
||
}
|
||
except OSError as exc:
|
||
return {
|
||
"schema_version": 1,
|
||
"passed": False,
|
||
"checked_at": checked_at,
|
||
"repository_url": repository_url,
|
||
"commit": commit,
|
||
"refs": [],
|
||
"errors": [f"无法执行 git ls-remote: {exc}"],
|
||
}
|
||
refs: list[str] = []
|
||
for line in result.stdout.splitlines():
|
||
fields = line.split("\t", 1)
|
||
if len(fields) == 2 and fields[0] == commit:
|
||
refs.append(fields[1])
|
||
errors: list[str] = []
|
||
if result.returncode != 0:
|
||
errors.append(f"git ls-remote 失败: {result.stderr.strip() or result.returncode}")
|
||
elif not refs:
|
||
errors.append("共享远端 heads/tags 中不存在当前完整 PSP commit")
|
||
return {
|
||
"schema_version": 1,
|
||
"passed": not errors,
|
||
"checked_at": checked_at,
|
||
"repository_url": repository_url,
|
||
"commit": commit,
|
||
"refs": sorted(refs),
|
||
"errors": errors,
|
||
}
|
||
|
||
|
||
def require_passed(payload: dict[str, Any] | None, name: str, errors: list[str]) -> None:
|
||
if payload is None:
|
||
errors.append(f"缺少 {name} 证据")
|
||
elif payload.get("passed") is not True and payload.get("result") != "PASS":
|
||
errors.append(f"{name} 结果不是 PASS")
|
||
|
||
|
||
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 validate_packaging_record(payload: dict[str, Any] | None, errors: list[str]) -> None:
|
||
require_passed(payload, "PSP 打包", errors)
|
||
if payload is None:
|
||
return
|
||
if payload.get("schema_version") != 2:
|
||
errors.append("PSP 打包证据必须由契约版工具生成（schema_version=2）")
|
||
contract = payload.get("contract")
|
||
if not isinstance(contract, dict):
|
||
errors.append("PSP 打包证据缺少固定契约 path/hash/id")
|
||
else:
|
||
path_value = contract.get("path")
|
||
expected_hash = contract.get("sha256")
|
||
if not isinstance(path_value, str) or not isinstance(expected_hash, str):
|
||
errors.append("PSP 打包契约 path/hash 非法")
|
||
else:
|
||
contract_path = Path(path_value)
|
||
if not contract_path.is_file() or contract_path.is_symlink():
|
||
errors.append("PSP 打包契约文件当前不可读")
|
||
elif sha256_file(contract_path) != expected_hash:
|
||
errors.append("PSP 打包契约文件 hash 已变化")
|
||
image = payload.get("image")
|
||
if not isinstance(image, dict):
|
||
errors.append("PSP 打包证据缺少 image")
|
||
else:
|
||
path_value = image.get("path")
|
||
expected_hash = image.get("sha256")
|
||
if not isinstance(path_value, str) or not isinstance(expected_hash, str):
|
||
errors.append("PSP 烧录包 path/hash 非法")
|
||
else:
|
||
image_path = Path(path_value)
|
||
if not image_path.is_file() or image_path.is_symlink():
|
||
errors.append("PSP 烧录包当前不可读")
|
||
elif sha256_file(image_path) != expected_hash:
|
||
errors.append("PSP 烧录包 hash 与打包证据不一致")
|
||
|
||
|
||
def evidence_commit(payload: dict[str, Any] | None) -> str | None:
|
||
if payload is None:
|
||
return None
|
||
direct = payload.get("psp_commit")
|
||
if isinstance(direct, str):
|
||
return direct
|
||
expected = payload.get("expected_psp_commit")
|
||
if isinstance(expected, str):
|
||
return expected
|
||
psp = payload.get("psp")
|
||
if isinstance(psp, dict) and isinstance(psp.get("commit"), str):
|
||
return psp["commit"]
|
||
return None
|
||
|
||
|
||
def validate_device(
|
||
payload: dict[str, Any] | None,
|
||
packaging: dict[str, Any] | None,
|
||
commit: str,
|
||
errors: list[str],
|
||
) -> None:
|
||
if payload is None:
|
||
errors.append("缺少设备验证证据")
|
||
return
|
||
if payload.get("result") != "PASS":
|
||
errors.append("设备验证结果不是 PASS")
|
||
if payload.get("evidence_source") != "USER_REPORTED":
|
||
errors.append("当前人工烧录阶段的 evidence_source 必须为 USER_REPORTED")
|
||
if payload.get("psp_commit") != commit:
|
||
errors.append("设备验证的 PSP commit 与当前仓库不一致")
|
||
unexecuted = payload.get("unexecuted")
|
||
if not isinstance(unexecuted, list) or unexecuted:
|
||
errors.append("设备验证仍有未执行的适用检查")
|
||
checks = payload.get("checks")
|
||
if not isinstance(checks, list) or not checks:
|
||
errors.append("设备验证缺少非空 checks")
|
||
elif any(
|
||
not isinstance(item, dict) or item.get("result") not in {"PASS", "N/A"}
|
||
for item in checks
|
||
):
|
||
errors.append("设备验证 checks 含失败或非法结果")
|
||
if packaging:
|
||
image = packaging.get("image")
|
||
if not isinstance(image, dict):
|
||
errors.append("打包证据缺少 image")
|
||
else:
|
||
if payload.get("image_path") != image.get("path"):
|
||
errors.append("设备验证镜像路径与打包结果不一致")
|
||
if payload.get("image_sha256") != image.get("sha256"):
|
||
errors.append("设备验证镜像 hash 与打包结果不一致")
|
||
|
||
|
||
def validate_application(
|
||
payload: dict[str, Any] | None, commit: str, errors: list[str]
|
||
) -> None:
|
||
if payload is None:
|
||
errors.append("缺少应用集成证据")
|
||
return
|
||
if payload.get("result") != "PASS":
|
||
errors.append("应用集成结果不是 PASS")
|
||
if payload.get("psp_commit") != commit:
|
||
errors.append("应用集成的 PSP commit 与当前仓库不一致")
|
||
if not isinstance(payload.get("application_commit"), str) or not payload.get(
|
||
"application_commit"
|
||
):
|
||
errors.append("应用集成缺少 application_commit")
|
||
unexecuted = payload.get("unexecuted")
|
||
if not isinstance(unexecuted, list) or unexecuted:
|
||
errors.append("应用集成仍有未执行的适用检查")
|
||
approval = payload.get("internal_approved")
|
||
if approval is not True:
|
||
errors.append("应用集成缺少应用负责人确认")
|
||
|
||
|
||
def validate_approval(payload: dict[str, Any] | None, errors: list[str]) -> None:
|
||
if payload is None:
|
||
errors.append("缺少嵌入式底层负责人确认")
|
||
return
|
||
if payload.get("internal_approved") is not True:
|
||
errors.append("internal_approved 不是 true")
|
||
for field in ("confirmed_by", "confirmed_at", "confirmed_scope"):
|
||
if not isinstance(payload.get(field), str) or not payload.get(field):
|
||
errors.append(f"负责人确认缺少 {field}")
|
||
|
||
|
||
def render(
|
||
project: str,
|
||
chip: str,
|
||
status: str,
|
||
psp_repo: Path,
|
||
repository_url: str | None,
|
||
reference_psp: str | None,
|
||
reference_commit: str | None,
|
||
static_path: Path,
|
||
bsp_lock_path: Path,
|
||
packaging_path: Path | None,
|
||
device_path: Path | None,
|
||
application_path: Path | None,
|
||
etunel_context_path: Path | None,
|
||
approval_path: Path | None,
|
||
media_version: str | None,
|
||
ai_plugin_version: str | None,
|
||
risks: list[str],
|
||
warnings: list[str],
|
||
unexecuted: list[str],
|
||
remote_verification: dict[str, object] | None,
|
||
for_etunel: bool,
|
||
) -> dict[str, object]:
|
||
errors: list[str] = []
|
||
git = psp_git_state(psp_repo)
|
||
commit = str(git["commit"])
|
||
if git["repository_root"] != git["path"]:
|
||
errors.append("PSP 必须是独立 Git 仓库根目录")
|
||
if git["dirty"] is True:
|
||
errors.append("PSP 工作区不干净，不能形成不可变交接")
|
||
|
||
static = load_object(static_path, "static validation")
|
||
bsp_lock = load_object(bsp_lock_path, "BSP lock")
|
||
expected_bsp_lock = Path(str(git["path"])) / "metadata" / "bsp.lock.json"
|
||
if bsp_lock_path.expanduser().resolve(strict=True) != expected_bsp_lock:
|
||
errors.append("BSP lock 必须是当前 PSP 仓库的 metadata/bsp.lock.json")
|
||
if isinstance(static.get("psp_root"), str) and Path(static["psp_root"]).resolve(
|
||
strict=False
|
||
) != Path(str(git["path"])):
|
||
errors.append("静态验证结果不属于当前 PSP 仓库")
|
||
if isinstance(static.get("project"), str) and static["project"] != project:
|
||
errors.append("静态验证结果的项目名与交接项目不一致")
|
||
packaging = load_object(packaging_path, "packaging") if packaging_path else None
|
||
device = load_object(device_path, "device validation") if device_path else None
|
||
application = (
|
||
load_object(application_path, "application integration") if application_path else None
|
||
)
|
||
etunel_context = (
|
||
load_object(etunel_context_path, "Etunel context") if etunel_context_path else None
|
||
)
|
||
approval = load_object(approval_path, "owner approval") if approval_path else None
|
||
|
||
require_passed(static, "PSP 静态验证", errors)
|
||
if bsp_lock.get("schema_version") != 1:
|
||
errors.append("BSP lock schema_version 必须为 1")
|
||
if status in {"packaged", "device_verified", "app_integrated", "accepted"}:
|
||
validate_packaging_record(packaging, errors)
|
||
if packaging and evidence_commit(packaging) != commit:
|
||
errors.append("打包证据的 PSP commit 与当前仓库不一致")
|
||
if status in {"device_verified", "accepted"}:
|
||
validate_device(device, packaging, commit, errors)
|
||
if status in {"app_integrated", "accepted"}:
|
||
validate_application(application, commit, errors)
|
||
if for_etunel or status == "accepted":
|
||
validate_approval(approval, errors)
|
||
if for_etunel:
|
||
if status not in {"packaged", "accepted"}:
|
||
errors.append("Etunel 应用交接只允许 packaged 或 accepted 状态")
|
||
if not repository_url or not shared_repository_url(repository_url):
|
||
errors.append("Etunel 交接必须提供团队共享 PSP 仓库地址")
|
||
if remote_verification is None or remote_verification.get("passed") is not True:
|
||
errors.append("Etunel 交接缺少通过的共享远端 commit 可达性验证")
|
||
elif (
|
||
remote_verification.get("repository_url") != repository_url
|
||
or remote_verification.get("commit") != commit
|
||
):
|
||
errors.append("远端可达性验证与当前仓库地址或 PSP commit 不一致")
|
||
if etunel_context is None:
|
||
errors.append("Etunel 交接缺少有效上下文")
|
||
else:
|
||
for field in ("work_id", "subtask_id", "phase", "wave", "role"):
|
||
if not isinstance(etunel_context.get(field), str) or not etunel_context.get(field):
|
||
errors.append(f"Etunel 上下文缺少 {field}")
|
||
if status == "accepted":
|
||
accepted_tag = f"accepted/{commit[:12]}"
|
||
if accepted_tag not in git["tags"] or not annotated_tag(Path(str(git["path"])), accepted_tag):
|
||
errors.append(f"当前 commit 缺少 annotated accepted 状态 tag: {accepted_tag}")
|
||
revoked_tag = f"revoked/{commit[:12]}"
|
||
if revoked_tag in git["tags"]:
|
||
errors.append(f"当前 PSP commit 已被 revoked: {revoked_tag}")
|
||
|
||
evidence = {
|
||
"static_validation": static,
|
||
"packaging": packaging,
|
||
"device_validation": device,
|
||
"application_integration": application,
|
||
}
|
||
result: dict[str, object] = {
|
||
"schema_version": 1,
|
||
"handoff_type": "psp-to-application",
|
||
"generated_at": now_utc(),
|
||
"passed": not errors,
|
||
"project": project,
|
||
"chip": chip,
|
||
"status": status,
|
||
"psp": {
|
||
"repository_url": repository_url,
|
||
"remote_verification": remote_verification,
|
||
**git,
|
||
},
|
||
"reference_psp": {
|
||
"source": reference_psp,
|
||
"commit": reference_commit,
|
||
},
|
||
"bsp_lock": bsp_lock,
|
||
"component_baseline": {
|
||
"media_version": media_version,
|
||
"ai_plugin_version": ai_plugin_version,
|
||
},
|
||
"evidence": evidence,
|
||
"etunel": {
|
||
"context": etunel_context,
|
||
"remote_commit_reachable": bool(
|
||
remote_verification and remote_verification.get("passed") is True
|
||
),
|
||
"prepared_for_send": for_etunel and not errors,
|
||
"sent": False,
|
||
},
|
||
"owner_approval": approval,
|
||
"self_check_result": "PASS" if not errors else "FAIL",
|
||
"unexecuted": unexecuted,
|
||
"warnings": warnings,
|
||
"risks": risks,
|
||
"errors": errors,
|
||
}
|
||
return result
|
||
|
||
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("--project", required=True)
|
||
parser.add_argument("--chip", required=True)
|
||
parser.add_argument("--status", required=True, choices=STATUSES)
|
||
parser.add_argument("--psp-repo", type=Path, required=True)
|
||
parser.add_argument("--repository-url")
|
||
reference_group = parser.add_mutually_exclusive_group(required=True)
|
||
reference_group.add_argument("--reference-psp")
|
||
reference_group.add_argument("--no-reference", action="store_true")
|
||
parser.add_argument("--reference-commit")
|
||
parser.add_argument("--static-validation", type=Path, required=True)
|
||
parser.add_argument("--bsp-lock", type=Path, required=True)
|
||
parser.add_argument("--packaging-result", type=Path)
|
||
parser.add_argument("--device-validation", type=Path)
|
||
parser.add_argument("--application-integration", type=Path)
|
||
parser.add_argument("--etunel-context", type=Path)
|
||
parser.add_argument("--owner-approval", type=Path)
|
||
parser.add_argument("--media-version")
|
||
parser.add_argument("--ai-plugin-version")
|
||
parser.add_argument("--risk", action="append", default=[])
|
||
parser.add_argument("--warning", action="append", default=[])
|
||
parser.add_argument("--unexecuted", action="append", default=[])
|
||
parser.add_argument("--for-etunel", action="store_true")
|
||
parser.add_argument(
|
||
"--verify-remote",
|
||
action="store_true",
|
||
help="用 git ls-remote 证明当前 commit 可从 repository-url 获取",
|
||
)
|
||
parser.add_argument("--output", type=Path, required=True)
|
||
return parser
|
||
|
||
|
||
def main() -> int:
|
||
args = build_parser().parse_args()
|
||
try:
|
||
if args.no_reference and args.reference_commit:
|
||
raise HandoffError("无参考 PSP 时不能填写 --reference-commit")
|
||
remote_verification = None
|
||
if args.verify_remote:
|
||
if not args.repository_url:
|
||
raise HandoffError("--verify-remote 必须同时提供 --repository-url")
|
||
current_commit = str(psp_git_state(args.psp_repo)["commit"])
|
||
remote_verification = verify_remote(args.repository_url, current_commit)
|
||
result = render(
|
||
args.project,
|
||
args.chip,
|
||
args.status,
|
||
args.psp_repo,
|
||
args.repository_url,
|
||
args.reference_psp,
|
||
args.reference_commit,
|
||
args.static_validation,
|
||
args.bsp_lock,
|
||
args.packaging_result,
|
||
args.device_validation,
|
||
args.application_integration,
|
||
args.etunel_context,
|
||
args.owner_approval,
|
||
args.media_version,
|
||
args.ai_plugin_version,
|
||
args.risk,
|
||
args.warning,
|
||
args.unexecuted,
|
||
remote_verification,
|
||
args.for_etunel,
|
||
)
|
||
write_json_new(args.output, result)
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||
return 0 if result.get("passed") is True else 2
|
||
except (HandoffError, OSError, json.JSONDecodeError) as exc:
|
||
print(json.dumps({"passed": False, "error": str(exc)}, ensure_ascii=False, indent=2))
|
||
return 2
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|