diff --git a/scripts/reboot-recovery/build-windows99-vmx-source-backup-search-receipt.py b/scripts/reboot-recovery/build-windows99-vmx-source-backup-search-receipt.py new file mode 100755 index 000000000..5cbaa54eb --- /dev/null +++ b/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:])) diff --git a/scripts/reboot-recovery/tests/test_windows99_vmx_source_backup_search.py b/scripts/reboot-recovery/tests/test_windows99_vmx_source_backup_search.py index 33248502f..987411226 100644 --- a/scripts/reboot-recovery/tests/test_windows99_vmx_source_backup_search.py +++ b/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