Files
firmware-lowlevel-development/skills/firmware-lowlevel-psp-development/scripts/psp/validate_bsp_handoff.py
T

512 lines
21 KiB
Python
Executable File

#!/usr/bin/env python3
"""Validate a BSP-to-PSP handoff, output permissions, and artifact hashes."""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import stat
import xml.etree.ElementTree as ET
from pathlib import Path
REQUIRED_IMAGE_MARGIN_BYTES = 512 * 1024
REQUIRED_PARTITION_IMAGES = {"kernel.bin", "rootfs.bin"}
DIRECTORY_ACCESS_BITS = 0o555
FILE_READ_BITS = 0o444
FILE_EXECUTE_BITS = 0o111
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 within(path: Path, root: Path) -> bool:
try:
path.relative_to(root)
return True
except ValueError:
return False
def validate_output_permissions(
payload: object,
build: dict[str, object],
producer: dict[str, object],
imgs_path: Path | None,
errors: list[str],
repair: bool,
) -> dict[str, object] | None:
"""Check BSP evidence and live modes; optionally repair only its exact output."""
if not isinstance(payload, dict) or payload.get("passed") is not True:
errors.append("BSP handoff 缺少通过的输出路径权限证据")
return None
if payload.get("schema_version") != 1 or payload.get("policy") != "a+rX":
errors.append("输出路径权限证据的 schema_version/policy 不匹配")
return None
root_value = payload.get("output_root")
version_value = payload.get("imgs_version_path")
project_name = producer.get("project")
build_version = build.get("version")
if (
not isinstance(root_value, str)
or not Path(root_value).is_absolute()
or not isinstance(version_value, str)
or not Path(version_value).is_absolute()
or not isinstance(project_name, str)
or project_name in ("", ".", "..")
or Path(project_name).name != project_name
or not isinstance(build_version, str)
or build_version in ("", ".", "..")
or Path(build_version).name != build_version
or imgs_path is None
):
errors.append("输出路径权限证据缺少精确的绝对输出目录、项目或构建版本")
return None
output_root = Path(root_value)
if (
len(output_root.parts) < 3
or output_root == Path.home()
or output_root.is_symlink()
):
errors.append("输出路径权限证据的根路径过宽或是软链接")
return None
expected_imgs_path = output_root / project_name / "imgs" / build_version
if Path(version_value) != expected_imgs_path or imgs_path != expected_imgs_path:
errors.append("输出路径权限证据与 BSP 项目、构建版本或 imgs_version_path 不一致")
return None
evidence_value = build.get("output_permissions_path")
project_value = producer.get("project_path")
if (
not isinstance(project_value, str)
or not Path(project_value).is_absolute()
or not isinstance(evidence_value, str)
or not Path(evidence_value).is_absolute()
):
errors.append("build.output_permissions_path 必须是 BSP 项目内的绝对路径")
return None
evidence_path = Path(evidence_value)
expected_evidence = (
Path(project_value)
/ "doc" / "bsp_adaptation" / "validation"
/ f"output_permissions_{build_version}.json"
)
if evidence_path != expected_evidence or evidence_path.is_symlink():
errors.append("输出路径权限证据文件路径不是预期 BSP 项目内的普通文件")
return None
try:
external_payload = json.loads(evidence_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
errors.append(f"输出路径权限证据文件不可读或解析失败: {exc}")
return None
if external_payload != payload:
errors.append("handoff 内嵌权限证据与外部证据文件不一致")
return None
paths: list[tuple[Path, int, str]] = []
for path in (
output_root,
output_root / project_name,
output_root / project_name / "imgs",
expected_imgs_path,
):
if path.is_symlink() or not path.is_dir():
errors.append(f"构建输出路径节点不是普通目录: {path}")
return None
paths.append((path, DIRECTORY_ACCESS_BITS, "目录"))
def walk_error(exc: OSError) -> None:
raise exc
skipped_symlinks = 0
try:
for parent, directory_names, file_names in os.walk(
expected_imgs_path, followlinks=False, onerror=walk_error
):
parent_path = Path(parent)
for name in list(directory_names):
path = parent_path / name
if path.is_symlink():
directory_names.remove(name)
skipped_symlinks += 1
elif path.is_dir():
paths.append((path, DIRECTORY_ACCESS_BITS, "目录"))
else:
errors.append(f"构建输出包含非普通目录: {path}")
return None
for name in file_names:
path = parent_path / name
if path.is_symlink():
skipped_symlinks += 1
elif path.is_file():
bits = FILE_READ_BITS
if stat.S_IMODE(path.stat().st_mode) & FILE_EXECUTE_BITS:
bits |= FILE_EXECUTE_BITS
paths.append((path, bits, "文件"))
else:
errors.append(f"构建输出包含非普通文件: {path}")
return None
except OSError as exc:
errors.append(f"无法遍历构建输出目录: {exc}")
return None
changed = 0
for path, bits, kind in paths:
try:
if path.is_symlink():
errors.append(f"权限处理拒绝软链接: {path}")
return None
before = stat.S_IMODE(path.stat().st_mode)
if before & bits != bits:
if not repair:
errors.append(f"构建输出{kind}缺少共享读取/进入权限: {path}")
continue
os.chmod(path, before | bits, follow_symlinks=False)
changed += 1
after = stat.S_IMODE(path.stat().st_mode)
if after & bits != bits:
errors.append(f"构建输出{kind}权限修复后仍不满足要求: {path}")
except OSError as exc:
errors.append(f"无法检查或修复构建输出权限 {path}: {exc}")
return {
**payload,
"current_check": {
"passed": not errors,
"repaired": changed > 0,
"changed_count": changed,
"checked_paths": len(paths),
"symlinks_skipped": skipped_symlinks,
},
}
def validate_image_partition_margin(
payload: object,
build: dict[str, object],
producer: dict[str, object],
verified_artifacts: list[dict[str, object]],
errors: list[str],
) -> dict[str, object] | None:
if not isinstance(payload, dict) or payload.get("passed") is not True:
errors.append("BSP handoff 缺少通过的 kernel/rootfs 分区余量证据")
return None
if payload.get("schema_version") != 1:
errors.append("kernel/rootfs 分区余量证据 schema_version 必须为 1")
if payload.get("required_margin_bytes") != REQUIRED_IMAGE_MARGIN_BYTES:
errors.append("kernel/rootfs 分区余量阈值必须精确为 524288 字节")
project_value = producer.get("project_path")
table_value = payload.get("partition_table")
table_hash = payload.get("partition_table_sha256")
table_path: Path | None = None
if (
not isinstance(project_value, str)
or not Path(project_value).is_absolute()
or not isinstance(table_value, str)
or not Path(table_value).is_absolute()
):
errors.append("分区余量证据缺少绝对 BSP project_path/partition_table")
else:
project_path = Path(project_value).resolve(strict=False)
table_path = Path(table_value).resolve(strict=False)
if not within(table_path, project_path / "configs"):
errors.append("分区余量证据的 partition_table 不在 BSP 项目 configs/ 内")
if not table_path.is_file() or table_path.is_symlink():
errors.append(f"分区余量证据的 partition_table 不可读: {table_path}")
table_path = None
elif not isinstance(table_hash, str) or sha256_file(table_path) != table_hash:
errors.append("分区余量证据的 partition_table SHA-256 不匹配")
evidence_value = build.get("image_partition_margin_path")
if not isinstance(evidence_value, str) or not Path(evidence_value).is_absolute():
errors.append("build.image_partition_margin_path 必须是绝对路径")
else:
evidence_path = Path(evidence_value).resolve(strict=False)
if not evidence_path.is_file() or evidence_path.is_symlink():
errors.append(f"镜像分区余量证据文件不可读: {evidence_path}")
else:
try:
external_payload = json.loads(evidence_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as exc:
errors.append(f"镜像分区余量证据文件解析失败: {exc}")
else:
if external_payload != payload:
errors.append("handoff 内嵌分区余量证据与外部证据文件不一致")
partitions_by_image: dict[str, list[int]] = {}
if table_path is not None:
try:
table_root = ET.parse(table_path).getroot()
except ET.ParseError as exc:
errors.append(f"分区表 XML 解析失败: {exc}")
else:
for element in table_root.iter("partition"):
image_name = Path(element.attrib.get("file", "")).name
if image_name not in REQUIRED_PARTITION_IMAGES:
continue
raw_size = element.attrib.get("size_in_kb")
try:
size_bytes = int(raw_size or "", 10) * 1024
except ValueError:
errors.append(f"分区表中 {image_name} 的 size_in_kb 非法")
continue
partitions_by_image.setdefault(image_name, []).append(size_bytes)
artifacts_by_name: dict[str, list[dict[str, object]]] = {}
for artifact in verified_artifacts:
name = Path(str(artifact.get("relative_path", ""))).name
artifacts_by_name.setdefault(name, []).append(artifact)
checks = payload.get("checks")
if not isinstance(checks, list):
errors.append("kernel/rootfs 分区余量证据 checks 必须是数组")
return dict(payload)
checks_by_image = {
item.get("image"): item for item in checks if isinstance(item, dict)
}
if set(checks_by_image) != REQUIRED_PARTITION_IMAGES or len(checks) != 2:
errors.append("分区余量证据必须且只能包含 kernel.bin、rootfs.bin 两项")
for image_name in sorted(REQUIRED_PARTITION_IMAGES):
check = checks_by_image.get(image_name)
artifacts_for_image = artifacts_by_name.get(image_name, [])
if not isinstance(check, dict):
continue
if len(artifacts_for_image) != 1:
errors.append(f"已验证 artifacts 中 {image_name} 必须唯一")
continue
artifact = artifacts_for_image[0]
sizes = partitions_by_image.get(image_name, [])
if not sizes:
errors.append(f"当前分区表未找到 {image_name} 的有效分区")
continue
image_size = artifact.get("size")
effective_size = min(sizes)
actual_margin = effective_size - int(image_size) if isinstance(image_size, int) else None
if check.get("passed") is not True:
errors.append(f"{image_name} 分区余量检查未通过")
if check.get("image_path") != artifact.get("absolute_path"):
errors.append(f"{image_name} 分区余量证据的镜像路径不一致")
if check.get("image_size_bytes") != image_size:
errors.append(f"{image_name} 分区余量证据的镜像大小不一致")
if check.get("effective_partition_size_bytes") != effective_size:
errors.append(f"{image_name} 分区余量证据的最小分区大小不一致")
if check.get("required_margin_bytes") != REQUIRED_IMAGE_MARGIN_BYTES:
errors.append(f"{image_name} 要求余量不是 524288 字节")
if check.get("actual_margin_bytes") != actual_margin:
errors.append(f"{image_name} 分区实际余量计算不一致")
if actual_margin is None or actual_margin < REQUIRED_IMAGE_MARGIN_BYTES:
errors.append(f"{image_name} 相对最小对应分区的余量不足 512 KiB")
return dict(payload)
def validate_handoff(
manifest_path: Path,
expected_project: str | None = None,
expected_chip: str | None = None,
expected_sdk: str | None = None,
expected_build_type: str | None = None,
expected_system_form: str | None = None,
repair_output_permissions: bool = False,
) -> dict[str, object]:
errors: list[str] = []
warnings: list[str] = []
manifest_path = manifest_path.expanduser().resolve(strict=True)
payload = json.loads(manifest_path.read_text(encoding="utf-8"))
if payload.get("schema_version") != 1:
errors.append("schema_version 必须为 1")
if payload.get("handoff_type") != "bsp-to-psp":
errors.append("handoff_type 必须为 bsp-to-psp")
producer = payload.get("producer")
baseline = payload.get("baseline")
build = payload.get("build")
validation = payload.get("validation")
artifacts = payload.get("artifacts")
if not isinstance(producer, dict):
errors.append("缺少 producer 对象")
producer = {}
if not isinstance(baseline, dict):
errors.append("缺少 baseline 对象")
baseline = {}
if not isinstance(build, dict):
errors.append("缺少 build 对象")
build = {}
if not isinstance(validation, dict):
errors.append("缺少 validation 对象")
validation = {}
if not isinstance(artifacts, list) or not artifacts:
errors.append("artifacts 必须是非空数组")
artifacts = []
expected_values = {
"project": (producer.get("project"), expected_project),
"chip": (baseline.get("chip"), expected_chip),
"sdk": (baseline.get("sdk"), expected_sdk),
"build_type": (baseline.get("board_build_type"), expected_build_type),
"system_form": (baseline.get("system_form"), expected_system_form),
}
for field, (actual, expected) in expected_values.items():
if expected is not None and actual != expected:
errors.append(f"{field} 不匹配:期望 {expected!r},实际 {actual!r}")
if repair_output_permissions and expected_project is None:
errors.append("自动修复输出权限时必须提供预期的 BSP 项目名 --project")
if build.get("status") != "build_passed" or build.get("exit_code") != 0:
errors.append("BSP 构建状态不是 build_passed/exit_code=0")
if build.get("completion_marker_seen") is not True:
errors.append("BSP 构建未记录正常完成标记")
imgs_value = build.get("imgs_version_path")
imgs_path: Path | None = None
if not isinstance(imgs_value, str) or not Path(imgs_value).is_absolute():
errors.append("imgs_version_path 必须是精确的绝对构建版本目录")
else:
imgs_path = Path(imgs_value).resolve(strict=False)
if not imgs_path.is_dir() or imgs_path.is_symlink():
errors.append(f"imgs_version_path 不存在或不是普通目录: {imgs_path}")
# Never modify BSP output if the handoff identity is already inconsistent.
output_permissions = validate_output_permissions(
validation.get("output_permissions"),
build,
producer,
imgs_path,
errors,
repair=repair_output_permissions and not errors and expected_project == producer.get("project"),
)
verified = 0
verified_artifacts: list[dict[str, object]] = []
seen_relative_paths: set[str] = set()
for index, artifact in enumerate(artifacts):
prefix = f"artifacts[{index}]"
if not isinstance(artifact, dict):
errors.append(f"{prefix} 必须是对象")
continue
relative = artifact.get("relative_path")
absolute = artifact.get("absolute_path")
size = artifact.get("size")
digest = artifact.get("sha256")
if not isinstance(relative, str) or not relative or Path(relative).is_absolute():
errors.append(f"{prefix}.relative_path 非法")
continue
if relative in seen_relative_paths:
errors.append(f"{prefix}.relative_path 重复: {relative}")
continue
seen_relative_paths.add(relative)
if not isinstance(absolute, str) or not Path(absolute).is_absolute():
errors.append(f"{prefix}.absolute_path 非法")
continue
path = Path(absolute).resolve(strict=False)
if imgs_path is None or not within(path, imgs_path):
errors.append(f"{prefix} 不在 imgs_version_path 内")
continue
if path != imgs_path / relative:
errors.append(f"{prefix} 的绝对/相对路径不一致")
continue
if not path.is_file() or path.is_symlink():
errors.append(f"{prefix} 的产物不存在或不是普通文件: {path}")
continue
if not isinstance(size, int) or size <= 0 or path.stat().st_size != size:
errors.append(f"{prefix} 的大小不匹配")
continue
if not isinstance(digest, str) or sha256_file(path) != digest:
errors.append(f"{prefix} 的 SHA-256 不匹配")
continue
verified += 1
verified_artifacts.append(
{
"relative_path": relative,
"absolute_path": str(path),
"size": size,
"sha256": digest,
}
)
image_partition_margin = validate_image_partition_margin(
validation.get("image_partition_margin"),
build,
producer,
verified_artifacts,
errors,
)
if validation.get("hardware") != "passed":
warnings.append(f"继承 BSP 硬件验证状态: {validation.get('hardware')!r}")
if validation.get("release") != "authorized":
warnings.append(f"继承 BSP 发布状态: {validation.get('release')!r}")
producer_git = producer.get("git")
if not isinstance(producer_git, dict) or not isinstance(producer_git.get("commit"), str):
warnings.append("BSP handoff 未记录可追踪的源码 commit")
return {
"passed": not errors,
"errors": errors,
"warnings": warnings,
"dependency": {
"module": "BSP",
"project": producer.get("project"),
"reference_project": producer.get("reference_project"),
"bsp_commit": producer_git.get("commit") if isinstance(producer_git, dict) else None,
"bsp_source_dirty": producer_git.get("dirty") if isinstance(producer_git, dict) else None,
"manifest_path": str(manifest_path),
"manifest_sha256": sha256_file(manifest_path),
"chip": baseline.get("chip"),
"sdk": baseline.get("sdk"),
"architecture": baseline.get("architecture"),
"flash": baseline.get("flash"),
"build_type": baseline.get("board_build_type"),
"system_form": baseline.get("system_form"),
"build_id": build.get("version"),
"imgs_version_path": str(imgs_path) if imgs_path else imgs_value,
"artifact_count": len(artifacts),
"verified_artifact_count": verified,
"artifacts": verified_artifacts,
"image_partition_margin": image_partition_margin,
"output_permissions_path": build.get("output_permissions_path"),
"output_permissions": output_permissions,
"hardware_validation": validation.get("hardware"),
"owner_acceptance": validation.get("owner_acceptance"),
"release": validation.get("release"),
},
}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("manifest", type=Path)
parser.add_argument("--project")
parser.add_argument("--chip")
parser.add_argument("--sdk")
parser.add_argument("--build-type")
parser.add_argument("--system-form", choices=("single-os", "dual-os"))
parser.add_argument(
"--repair-output-permissions",
action="store_true",
help="仅对已核对项目身份及证据的本次 imgs 路径补齐 a+rX 权限",
)
args = parser.parse_args()
try:
result = validate_handoff(
args.manifest,
args.project,
args.chip,
args.sdk,
args.build_type,
args.system_form,
repair_output_permissions=args.repair_output_permissions,
)
except (OSError, json.JSONDecodeError) as exc:
result = {"passed": False, "errors": [str(exc)], "warnings": []}
print(json.dumps(result, ensure_ascii=False, indent=2))
return 0 if result.get("passed") is True else 2
if __name__ == "__main__":
raise SystemExit(main())