test(scanner): report Linux descriptor drift ledger (#7692)

Add a read-only descriptor ledger mode to the Scanner/Heal Linux evidence planner so release operators can separate current-head measured descriptors from old-head measured artifacts and case-level inputs before final bundle assembly.

Co-authored-by: zhi22915 <qiuzgang@gmail.com>

scripts/run_scanner_heal_linux_evidence_plan.py

@@ -46,6 +46,10 @@ REQUIRED_GATES = {
 }
 
 
+def is_sha(value: Any) -> bool:
+    return isinstance(value, str) and len(value) == 40 and all(char in "0123456789abcdef" for char in value)
+
+
 def command(*parts: str) -> list[str]:
     return list(parts)
 
@@ -74,6 +78,14 @@ def load_registry() -> dict[str, Any]:
     return registry
 
 
+def read_json_object(path: Path) -> dict[str, Any]:
+    with path.open() as stream:
+        payload = json.load(stream)
+    if not isinstance(payload, dict):
+        raise ValueError(f"expected JSON object: {path}")
+    return payload
+
+
 def validate_registry(registry: dict[str, Any]) -> None:
     gates = {item["gate"] for item in registry.get("release_requirements", [])}
     missing = sorted(REQUIRED_GATES - gates)
@@ -604,6 +616,89 @@ def build_status(plan: dict[str, Any], run_root: Path) -> dict[str, Any]:
     }
 
 
+def descriptor_gate_names(payload: dict[str, Any]) -> list[str]:
+    gates = payload.get("gates")
+    if not isinstance(gates, dict):
+        return []
+    return sorted(gate for gate in gates if isinstance(gate, str))
+
+
+def build_descriptor_ledger(descriptor_paths: list[Path], revision: str) -> dict[str, Any]:
+    entries = []
+    same_head_gates: set[str] = set()
+    old_head_gates: set[str] = set()
+    duplicate_gates: dict[str, list[str]] = {}
+    gate_sources: dict[str, list[str]] = {}
+    for raw_path in descriptor_paths:
+        path = raw_path.resolve()
+        entry: dict[str, Any] = {
+            "path": str(raw_path),
+            "file_name": path.name,
+        }
+        try:
+            if not path.is_file():
+                raise ValueError("descriptor is missing")
+            if path.stat().st_size <= 0:
+                raise ValueError("descriptor is empty")
+            payload = read_json_object(path)
+            evidence = payload.get("evidence")
+            descriptor_revision = payload.get("source_revision")
+            gates = descriptor_gate_names(payload)
+            if evidence != "measured":
+                classification = "case-level only"
+            elif not is_sha(descriptor_revision):
+                classification = "invalid"
+            elif not gates:
+                classification = "case-level only"
+            elif descriptor_revision == revision:
+                classification = "same-head verified"
+                same_head_gates.update(gates)
+            else:
+                classification = "old-head measured, drift-readable"
+                old_head_gates.update(gates)
+            for gate in gates:
+                gate_sources.setdefault(gate, []).append(path.name)
+            entry.update({
+                "status": "present",
+                "classification": classification,
+                "evidence": evidence,
+                "source_revision": descriptor_revision,
+                "gates": gates,
+            })
+        except (ValueError, OSError, json.JSONDecodeError) as error:
+            entry.update({
+                "status": "invalid",
+                "classification": "invalid",
+                "error": str(error),
+                "gates": [],
+            })
+        entries.append(entry)
+    for gate, sources in sorted(gate_sources.items()):
+        if len(sources) > 1:
+            duplicate_gates[gate] = sorted(sources)
+    measured_gates = same_head_gates | old_head_gates
+    return {
+        "schema": 1,
+        "kind": "scanner-heal-descriptor-ledger",
+        "source_revision": revision,
+        "release_approved": False,
+        "entries": entries,
+        "same_head_verified_gates": sorted(same_head_gates),
+        "old_head_measured_gates": sorted(old_head_gates - same_head_gates),
+        "missing_measured_gates": sorted(REQUIRED_GATES - measured_gates),
+        "missing_current_head_gates": sorted(REQUIRED_GATES - same_head_gates),
+        "duplicate_gates": duplicate_gates,
+        "totals": {
+            "descriptors": len(entries),
+            "same_head_verified_gates": len(same_head_gates),
+            "old_head_measured_gates": len(old_head_gates - same_head_gates),
+            "missing_measured_gates": len(REQUIRED_GATES - measured_gates),
+            "missing_current_head_gates": len(REQUIRED_GATES - same_head_gates),
+            "invalid_descriptors": sum(1 for entry in entries if entry["status"] == "invalid"),
+        },
+    }
+
+
 def run_preflight(plan: dict[str, Any]) -> int:
     commands = iter_preflight_commands(plan)
     if not commands:
@@ -672,6 +767,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     parser.add_argument("--format", choices=("text", "json"), default="text")
     parser.add_argument("--run-preflight", action="store_true")
     parser.add_argument("--status-root", type=Path)
+    parser.add_argument("--descriptor-ledger", type=Path, nargs="+")
     parser.add_argument("--self-test", action="store_true")
     return parser.parse_args(argv)
 
@@ -687,9 +783,15 @@ def main(argv: list[str] | None = None) -> int:
         phases = set(args.phase or ["all"])
         if "all" in phases and len(phases) > 1:
             raise ValueError("--phase all cannot be combined with another phase")
-        if args.status_root is not None and (args.write_plan or args.run_preflight):
-            raise ValueError("--status-root cannot be combined with --write-plan or --run-preflight")
-        plan = build_plan(registry, source_revision(args.source_revision), phases)
+        if args.status_root is not None and (args.write_plan or args.run_preflight or args.descriptor_ledger):
+            raise ValueError("--status-root cannot be combined with --write-plan, --run-preflight, or --descriptor-ledger")
+        if args.descriptor_ledger and (args.write_plan or args.run_preflight):
+            raise ValueError("--descriptor-ledger cannot be combined with --write-plan or --run-preflight")
+        revision = source_revision(args.source_revision)
+        if args.descriptor_ledger:
+            print(json.dumps(build_descriptor_ledger(args.descriptor_ledger, revision), indent=2, sort_keys=True))
+            return 0
+        plan = build_plan(registry, revision, phases)
         if args.status_root is not None:
             status = build_status(plan, args.status_root)
             print(json.dumps(status, indent=2, sort_keys=True))

scripts/test_scanner_heal_linux_evidence_plan.sh

@@ -108,6 +108,60 @@ assert status["release_approved"] is False
 assert status["artifact_totals"]["missing"] == 0
 PY
 
+"${RUSTFS_PYTHON_BIN:-python3}" - "$TMP_DIR" <<'PY'
+import json
+import pathlib
+import sys
+
+root = pathlib.Path(sys.argv[1])
+same_head = root / "same-head.json"
+old_head = root / "old-head.json"
+case_level = root / "case-level.json"
+same_head.write_text(json.dumps({
+    "schema": 1,
+    "evidence": "measured",
+    "source_revision": "a" * 40,
+    "gates": {"G01": {}, "G02": {}},
+}) + "\n")
+old_head.write_text(json.dumps({
+    "schema": 1,
+    "evidence": "measured",
+    "source_revision": "b" * 40,
+    "gates": {"G03": {}, "G09": {}},
+}) + "\n")
+case_level.write_text(json.dumps({
+    "schema": 1,
+    "evidence": "case",
+    "source_revision": "a" * 40,
+    "gates": {"G14": {}},
+}) + "\n")
+PY
+
+"${RUSTFS_PYTHON_BIN:-python3}" "$RUNNER" \
+  --source-revision aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa \
+  --descriptor-ledger "$TMP_DIR/same-head.json" "$TMP_DIR/old-head.json" "$TMP_DIR/case-level.json" \
+  "$TMP_DIR/missing.json" >"$TMP_DIR/ledger.json"
+
+"${RUSTFS_PYTHON_BIN:-python3}" - "$TMP_DIR/ledger.json" <<'PY'
+import json
+import pathlib
+import sys
+
+ledger = json.loads(pathlib.Path(sys.argv[1]).read_text())
+assert ledger["kind"] == "scanner-heal-descriptor-ledger"
+assert ledger["release_approved"] is False
+assert ledger["same_head_verified_gates"] == ["G01", "G02"]
+assert ledger["old_head_measured_gates"] == ["G03", "G09"]
+assert "G14" in ledger["missing_measured_gates"]
+assert "G03" in ledger["missing_current_head_gates"]
+assert ledger["totals"]["invalid_descriptors"] == 1
+classifications = {entry["file_name"]: entry["classification"] for entry in ledger["entries"]}
+assert classifications["same-head.json"] == "same-head verified"
+assert classifications["old-head.json"] == "old-head measured, drift-readable"
+assert classifications["case-level.json"] == "case-level only"
+assert classifications["missing.json"] == "invalid"
+PY
+
 if "${RUSTFS_PYTHON_BIN:-python3}" "$RUNNER" \
   --phase performance \
   --run-preflight >/dev/null 2>"$TMP_DIR/no-preflight.err"; then