#!/usr/bin/env python3 """Validate hardware evidence and its explicit sensor_cfg.ini expectations.""" from __future__ import annotations import argparse import configparser import hashlib import json import os import sys import tempfile from pathlib import Path def sha256_file(path: Path) -> str: value = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(1024 * 1024), b""): value.update(block) return value.hexdigest() def write_json(path: Path, payload: dict[str, object]) -> None: path = path.expanduser().resolve(strict=False) path.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, 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 normalized(value: object) -> str: text = str(value).strip() try: return str(int(text, 0)) except ValueError: return text.casefold() def validate(psp: Path, topology_path: Path) -> dict[str, object]: errors: list[str] = [] psp = psp.expanduser().resolve(strict=True) topology_path = topology_path.expanduser().resolve(strict=True) topology = json.loads(topology_path.read_text(encoding="utf-8")) if not isinstance(topology, dict) or topology.get("schema_version") != 1: raise ValueError("hardware-topology.json 必须是 schema_version=1 的对象") for field in ("project", "chip", "system_mode", "storage"): if not topology.get(field): errors.append(f"硬件拓扑缺少字段: {field}") storage = topology.get("storage") if not isinstance(storage, dict) or not storage.get("type") or not isinstance(storage.get("capacity_bytes"), int): errors.append("storage 必须包含 type 和整数 capacity_bytes；型号未知时可省略 model") sources = topology.get("sources") source_results: list[dict[str, object]] = [] if not isinstance(sources, list) or not sources: errors.append("硬件拓扑至少需要一个可校验证据源") sources = [] for index, item in enumerate(sources): if not isinstance(item, dict): errors.append(f"sources[{index}] 必须是对象") continue raw_path, expected = item.get("path"), item.get("sha256") if not isinstance(raw_path, str) or not raw_path: errors.append(f"sources[{index}].path 不能为空") continue source_path = Path(raw_path).expanduser() if not source_path.is_absolute(): source_path = (topology_path.parent / source_path).resolve(strict=False) exists = source_path.is_file() and not source_path.is_symlink() actual = sha256_file(source_path) if exists else None if not exists: errors.append(f"硬件证据源不存在或不是普通文件: {source_path}") if not isinstance(expected, str) or len(expected) != 64 or expected != actual: errors.append(f"硬件证据源 hash 不匹配: {source_path}") source_results.append({"path": str(source_path), "expected_sha256": expected, "actual_sha256": actual}) conflicts = topology.get("conflicts") if not isinstance(conflicts, list): errors.append("conflicts 必须是数组") elif conflicts: errors.append("硬件资料仍有未解决冲突，禁止进入 PSP 适配") sensor_files = sorted( path for path in psp.rglob("sensor_cfg.ini") if path.is_file() and not path.is_symlink() and ".git" not in path.parts ) if len(sensor_files) != 1: errors.append(f"必须唯一定位 sensor_cfg.ini，实际 {len(sensor_files)} 个") parser = None else: parser = configparser.ConfigParser(interpolation=None, strict=False) try: parser.read(sensor_files[0], encoding="utf-8") except configparser.Error as exc: errors.append(f"sensor_cfg.ini 解析失败: {exc}") parser = None sensors = topology.get("sensors") if not isinstance(sensors, list) or not sensors: errors.append("sensors 必须是非空数组") sensors = [] required_sensor_fields = ( "section", "model", "enum_name", "i2c_bus", "i2c_address", "mipi_device", "lane_count", "evidence_sources", "config_expectations", ) seen_sections: set[str] = set() checks: list[dict[str, object]] = [] for index, sensor in enumerate(sensors): if not isinstance(sensor, dict): errors.append(f"sensors[{index}] 必须是对象") continue missing = [field for field in required_sensor_fields if field not in sensor] if missing: errors.append(f"sensors[{index}] 缺少字段: {missing}") continue section = str(sensor["section"]) if section.casefold() in seen_sections: errors.append(f"Sensor section 重复: {section}") seen_sections.add(section.casefold()) evidence_indexes = sensor.get("evidence_sources") if not isinstance(evidence_indexes, list) or not evidence_indexes or any( not isinstance(value, int) or value < 0 or value >= len(sources) for value in evidence_indexes ): errors.append(f"{section}.evidence_sources 必须引用有效 sources 索引") expectations = sensor.get("config_expectations") row_checks: list[dict[str, object]] = [] if not isinstance(expectations, dict) or not expectations: errors.append(f"{section}.config_expectations 必须明确列出配置字段") elif parser is None or not parser.has_section(section): errors.append(f"sensor_cfg.ini 缺少硬件拓扑声明的 section: {section}") else: for key, expected in sorted(expectations.items()): actual = parser.get(section, key, fallback=None) matched = actual is not None and normalized(actual) == normalized(expected) row_checks.append({"key": key, "expected": expected, "actual": actual, "matched": matched}) if not matched: errors.append(f"{section}.{key} 与硬件拓扑不一致：期望 {expected!r}，实际 {actual!r}") checks.append({"section": section, "model": sensor.get("model"), "checks": row_checks}) configured_sections = [] if parser is None else [ name for name in parser.sections() if name.casefold().startswith("sensor") ] if len(configured_sections) != len(sensors): errors.append(f"Sensor 数量不一致：拓扑 {len(sensors)}，配置 {len(configured_sections)}") return { "schema_version": 1, "passed": not errors, "topology_path": str(topology_path), "topology_sha256": sha256_file(topology_path), "psp": str(psp), "sensor_cfg": str(sensor_files[0]) if len(sensor_files) == 1 else None, "sources": source_results, "sensor_checks": checks, "errors": errors, } def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--psp", type=Path, required=True) parser.add_argument("--topology", type=Path, required=True) parser.add_argument("--output", type=Path) args = parser.parse_args() try: result = validate(args.psp, args.topology) if args.output: write_json(args.output, result) print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if result["passed"] else 2 except (OSError, UnicodeError, ValueError, 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())