fix(reboot): automate windows99 backup search receipt

scripts/reboot-recovery/build-windows99-vmx-source-backup-search-receipt.py

@@ -0,0 +1,244 @@
+#!/usr/bin/env python3
+"""Build a public-safe Windows99 VMX backup-search receipt.
+
+The collector emits no-secret, path-fingerprint-only text. This builder turns
+that stdout into the JSON receipt consumed by the AWOOOI runtime readback API.
+It intentionally keeps raw LAN topology, filesystem paths, and fingerprint
+values out of the public receipt.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import sys
+from datetime import datetime, timezone, timedelta
+from pathlib import Path
+from typing import Any
+
+SCHEMA_VERSION = "windows99_vmx_source_backup_search_receipt_v1"
+COLLECTOR_SCHEMA_VERSION = "windows99_vmx_source_backup_search_collector_v1"
+DEFAULT_OUTPUT = Path("docs/operations/windows99-vmx-source-backup-search-receipt.snapshot.json")
+FORBIDDEN_FRAGMENTS = ("192.168.0.", "D:\\", "C:\\", "E:\\", "/Users/", "/Volumes/")
+
+
+def taipei_now_iso() -> str:
+    return datetime.now(timezone(timedelta(hours=8))).replace(microsecond=0).isoformat()
+
+
+def safe_alias(value: str) -> str:
+    text = value.strip()
+    if re.fullmatch(r"[A-Za-z0-9_-]{1,32}", text):
+        return text
+    return "redacted"
+
+
+def parse_kv_line(line: str) -> dict[str, str]:
+    values: dict[str, str] = {}
+    for token in line.strip().split():
+        if "=" not in token:
+            continue
+        key, value = token.split("=", 1)
+        values[key] = value
+    return values
+
+
+def parse_collector_stdout(text: str) -> tuple[dict[str, str], list[dict[str, Any]]]:
+    globals_: dict[str, str] = {}
+    rows: list[dict[str, Any]] = []
+    for raw_line in text.splitlines():
+        line = raw_line.strip()
+        if not line:
+            continue
+        if line.startswith("HOST_SEARCH "):
+            kv = parse_kv_line(line[len("HOST_SEARCH ") :])
+            alias = safe_alias(kv.get("alias", "unknown"))
+            status = kv.get("status", "unknown").strip() or "unknown"
+            try:
+                count = int(kv.get("candidate_source_count", "0"))
+            except ValueError:
+                count = 0
+            fingerprints = [
+                part
+                for part in kv.get("path_fingerprints", "").split(",")
+                if re.fullmatch(r"[A-Fa-f0-9]{8,128}|fp[-A-Za-z0-9_]+", part)
+            ]
+            known = status in {"searched", "not_matched"}
+            rows.append(
+                {
+                    "host_alias": alias,
+                    "search_role": "linux_backup_roots",
+                    "status": status,
+                    "candidate_source_count": max(count, 0),
+                    "candidate_source_count_known": known,
+                    "candidate_source_fingerprint_count": len(fingerprints),
+                    "collector": "collect-windows99-vmx-source-backup-search.sh",
+                }
+            )
+            continue
+        if "=" in line:
+            key, value = line.split("=", 1)
+            globals_[key.strip()] = value.strip()
+    return globals_, rows
+
+
+def locator_row(locator_receipt: dict[str, Any]) -> dict[str, Any]:
+    status = str(locator_receipt.get("locate_status") or "not_collected")
+    known = locator_receipt.get("candidate_source_count_known") is True
+    candidate_count = int(locator_receipt.get("candidate_source_count") or 0) if known else 0
+    fingerprint_count = int(locator_receipt.get("candidate_source_fingerprint_count") or 0)
+    return {
+        "host_alias": "99",
+        "search_role": "windows_vmware_roots_via_locator",
+        "status": status,
+        "candidate_source_count": candidate_count,
+        "candidate_source_count_known": known,
+        "candidate_source_fingerprint_count": fingerprint_count,
+        "collector": "windows99-vmx-source-locator-readback",
+    }
+
+
+def load_locator_receipt(path: Path | None) -> dict[str, Any]:
+    if path is None or not path.exists():
+        return {}
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    return payload if isinstance(payload, dict) else {}
+
+
+def build_receipt(
+    collector_stdout: str,
+    *,
+    generated_at: str,
+    locator_receipt: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    globals_, rows = parse_collector_stdout(collector_stdout)
+    if globals_.get("schema_version") != COLLECTOR_SCHEMA_VERSION:
+        raise ValueError("collector stdout missing expected schema_version")
+
+    host_rows: list[dict[str, Any]] = []
+    if locator_receipt:
+        host_rows.append(locator_row(locator_receipt))
+    host_rows.extend(rows)
+
+    searched_count = sum(
+        1 for row in host_rows if row.get("status") in {"searched", "not_matched"}
+    )
+    blocked_count = sum(
+        1
+        for row in host_rows
+        if str(row.get("status") or "").startswith("blocked_")
+        or row.get("status") == "transport_intermittent"
+    )
+    candidate_count = sum(int(row.get("candidate_source_count") or 0) for row in host_rows)
+    fingerprint_count = sum(
+        int(row.get("candidate_source_fingerprint_count") or 0) for row in host_rows
+    )
+
+    if candidate_count > 0:
+        search_status = "candidate_fingerprint_found_requires_internal_path_resolution"
+        next_step = "resolve_candidate_fingerprint_to_authorized_internal_path"
+        safe_next = (
+            "resolve_candidate_fingerprint_to_authorized_internal_path_then_"
+            "build_scoped_relink_dry_run_no_write"
+        )
+    elif blocked_count > 0:
+        search_status = "blocked_no_verified_vmx_source_found_blocked_hosts_unavailable"
+        next_step = "continue_verified_backup_inventory_search_or_authorized_console_source_path"
+        safe_next = (
+            "continue_verified_backup_inventory_search_or_authorized_console_source_path_"
+            "then_scoped_dry_run_no_write"
+        )
+    else:
+        search_status = "blocked_no_verified_vmx_source_found"
+        next_step = "continue_verified_backup_inventory_search_or_console_source_path"
+        safe_next = (
+            "recover_verified_external_backup_artifact_or_authorized_console_path_"
+            "then_build_scoped_relink_dry_run_no_write"
+        )
+
+    receipt = {
+        "schema_version": SCHEMA_VERSION,
+        "generated_at": generated_at,
+        "receipt_ref": "log://awoooi/p0-006/windows99-vmx-source-backup-search-111",
+        "source": "live_no_secret_filename_search",
+        "collector_readback_present": True,
+        "collector_schema_version": COLLECTOR_SCHEMA_VERSION,
+        "target_host_alias": "99",
+        "target_vm_aliases": ["111"],
+        "search_mode": "filename_path_fingerprint_only",
+        "host_rows": host_rows,
+        "searched_host_count": searched_count,
+        "blocked_host_count": blocked_count,
+        "candidate_source_count": candidate_count,
+        "candidate_source_fingerprint_count": fingerprint_count,
+        "search_status": search_status,
+        "next_executable_step": next_step,
+        "safe_next_step": safe_next,
+        "public_lan_topology_redacted": True,
+        "redacted_public_display_only": True,
+        "raw_path_output": False,
+        "path_fingerprint_only": True,
+        "file_content_read": False,
+        "secret_value_read": False,
+        "password_prompt_allowed": False,
+        "remote_write_performed": False,
+        "host_reboot_performed": False,
+        "vm_power_change_performed": False,
+        "windows_service_restart_performed": False,
+        "windows_registry_apply_performed": False,
+        "scheduled_task_write_performed": False,
+        "km_writeback_ref": "km://awoooi/reboot-slo/windows99-vmx-source-backup-search-111",
+        "playbook_trust_writeback_ref": (
+            "playbook://awoooi/windows99-vmx-source-backup-search/"
+            "windows99-vmx-source-backup-search-111"
+        ),
+        "post_search_verifier": [
+            "GET /api/v1/agents/windows99-vmx-source-backup-search-package",
+            "GET /api/v1/agents/windows99-vmx-source-locator-readback",
+            "GET /api/v1/agents/reboot-auto-recovery-slo-scorecard",
+            "GET /api/v1/agents/awoooi-priority-work-order-readback",
+        ],
+    }
+    serialized = json.dumps(receipt, ensure_ascii=False)
+    leaked = [fragment for fragment in FORBIDDEN_FRAGMENTS if fragment in serialized]
+    if leaked:
+        raise ValueError(f"unsafe receipt fragments: {leaked}")
+    return receipt
+
+
+def parse_args(argv: list[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Build the public-safe Windows99 VMX backup-search receipt."
+    )
+    parser.add_argument("--collector-output", type=Path)
+    parser.add_argument("--locator-receipt", type=Path)
+    parser.add_argument("--generated-at", default=taipei_now_iso())
+    parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
+    parser.add_argument("--stdout", action="store_true")
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str]) -> int:
+    args = parse_args(argv)
+    text = (
+        args.collector_output.read_text(encoding="utf-8")
+        if args.collector_output
+        else sys.stdin.read()
+    )
+    receipt = build_receipt(
+        text,
+        generated_at=args.generated_at,
+        locator_receipt=load_locator_receipt(args.locator_receipt),
+    )
+    rendered = json.dumps(receipt, ensure_ascii=False, indent=2) + "\n"
+    if args.stdout:
+        sys.stdout.write(rendered)
+    else:
+        args.output.parent.mkdir(parents=True, exist_ok=True)
+        args.output.write_text(rendered, encoding="utf-8")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv[1:]))

scripts/reboot-recovery/tests/test_windows99_vmx_source_backup_search.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import os
+import json
 import subprocess
 import textwrap
 from pathlib import Path
@@ -13,6 +14,12 @@ SCRIPT = (
     / "reboot-recovery"
     / "collect-windows99-vmx-source-backup-search.sh"
 )
+RECEIPT_BUILDER = (
+    ROOT
+    / "scripts"
+    / "reboot-recovery"
+    / "build-windows99-vmx-source-backup-search-receipt.py"
+)
 
 
 def _write_executable(path: Path, text: str) -> None:
@@ -171,3 +178,78 @@ def test_backup_search_collector_resolves_bare_aliases_and_blocks_unknown(
     assert "HOST_SEARCH alias=missing status=blocked_host_alias_unresolved" in (
         result.stdout
     )
+
+
+def test_backup_search_receipt_builder_projects_collector_stdout_safely(
+    tmp_path: Path,
+) -> None:
+    collector_output = tmp_path / "collector.txt"
+    collector_output.write_text(
+        textwrap.dedent(
+            """
+            schema_version=windows99_vmx_source_backup_search_collector_v1
+            search_hosts=110=wooo@192.168.0.110,120=wooo@192.168.0.120
+            secret_value_read=false
+            file_content_read=false
+            raw_path_output=false
+            HOST_SEARCH alias=110 status=blocked_ssh_batchmode_unavailable candidate_source_count=0 path_fingerprints=
+            HOST_SEARCH alias=120 status=searched candidate_source_count=2 path_fingerprints=abc12345,def67890
+            candidate_source_count=2
+            search_status=candidate_fingerprint_found_requires_internal_path_resolution
+            """
+        ).strip()
+        + "\n",
+        encoding="utf-8",
+    )
+    locator_receipt = tmp_path / "locator.json"
+    locator_receipt.write_text(
+        json.dumps(
+            {
+                "schema_version": "windows99_vmx_source_locator_receipt_v1",
+                "collector_readback_present": True,
+                "locate_status": "blocked_missing_expected_vmx_source_no_candidate_found",
+                "candidate_source_count_known": True,
+                "candidate_source_count": 0,
+                "candidate_source_fingerprint_count": 0,
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    result = subprocess.run(
+        [
+            "python3",
+            str(RECEIPT_BUILDER),
+            "--collector-output",
+            str(collector_output),
+            "--locator-receipt",
+            str(locator_receipt),
+            "--generated-at",
+            "2026-07-09T17:10:00+08:00",
+            "--stdout",
+        ],
+        cwd=ROOT,
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+
+    payload = json.loads(result.stdout)
+    assert payload["schema_version"] == "windows99_vmx_source_backup_search_receipt_v1"
+    assert payload["collector_readback_present"] is True
+    assert payload["searched_host_count"] == 1
+    assert payload["blocked_host_count"] == 2
+    assert payload["candidate_source_count"] == 2
+    assert payload["candidate_source_fingerprint_count"] == 2
+    assert payload["search_status"] == (
+        "candidate_fingerprint_found_requires_internal_path_resolution"
+    )
+    assert payload["raw_path_output"] is False
+    assert payload["path_fingerprint_only"] is True
+    assert payload["file_content_read"] is False
+    assert payload["secret_value_read"] is False
+    assert payload["remote_write_performed"] is False
+    assert payload["vm_power_change_performed"] is False
+    assert "192.168.0." not in result.stdout
+    assert "D:\\" not in result.stdout
+    assert "/Users/" not in result.stdout