#!/usr/bin/env python3 """Verify a transferred legacy Windows lab kit ZIP before extraction.""" from __future__ import annotations import argparse import hashlib import json import sys import zipfile from pathlib import Path from typing import Any TOOL_NAME = "legacy-windows-lab-kit-verify" KIT_SCHEMA = "legacy-windows-lab-kit" REQUIRED_ROLES = { "agent_windows", "recover_legacy", "recover_legacy_version", "checksums", "build_manifest", "legacy_agent_script", "legacy_watchdog_script", "legacy_tray_script", "lab_smoke_script", "lab_collect_script", "lab_run_script", "win10_bridge_script", "lab_evidence_checker", "lab_intake_helper", "evidence_bundle_helper", "readiness_gate", "first_iteration_finalizer", "pilot_approval_helper", "pilot_readiness_helper", "lab_kit_verify_helper", "lab_kit_builder", "runbook", "ship_note", } BUILD_MANIFEST_REQUIRED_ARTIFACTS = ( "bin/sk-agent.exe", "bin/sk-recover-legacy.exe", "bin/sk-recover-legacy.exe.version", "bin/sk-agent-checksums.txt", "client/sk-agent-legacy.ps1", "client/sk-watchdog-legacy.ps1", "client/sk-tray-legacy.ps1", "client/legacy-windows-lab-smoke.ps1", "client/legacy-windows-lab-collect.ps1", "client/legacy-windows-lab-run.ps1", "client/upgrade_win10_media.ps1", "scripts/legacy-windows-lab-evidence-check.py", "scripts/legacy-windows-lab-intake.py", "scripts/legacy-windows-first-iteration-evidence-bundle.py", "scripts/legacy-windows-first-iteration-readiness.py", "scripts/legacy-windows-first-iteration-finalize.py", "scripts/legacy-windows-first-iteration-pilot-approval.py", "scripts/legacy-windows-first-iteration-pilot-readiness.py", "scripts/legacy-windows-lab-kit-verify.py", "scripts/legacy-windows-lab-kit.py", "docs/legacy-windows-first-iteration-runbook.md", "docs/ship-notes/2026-07-08-legacy-windows-first-iteration.md", ) REQUIRED_ROLE_SNIPPETS = { "lab_smoke_script": ( ( '"os_kind"', "missing os_kind smoke evidence field", ), ( "unsupported OS; expected Windows 7 SP1 or Windows Server 2008 R2 SP1", "missing unsupported OS identity guard", ), ( "fresh restart-agent.flag observed", "missing fresh peer recover restart-agent.flag proof", ), ( "fresh restart-agent.flag and update-flag.txt observed", "missing fresh self-rescue restart-agent.flag/update-flag.txt proof", ), ), "lab_collect_script": ( ( "function Get-SmokeJsonString", "missing collector smoke JSON reader", ), ( '$SmokeOsKind = Get-SmokeJsonString $SmokeJson "os_kind"', "missing collector os_kind propagation from smoke", ), ( '"os_kind": "', "missing collector manifest os_kind field", ), ( "function Get-Sha256Hex", "missing build manifest SHA-256 helper", ), ( "[Security.Cryptography.SHA256]::Create()", "missing .NET SHA-256 implementation", ), ( "function Resolve-PayloadArtifact", "missing payload artifact resolver", ), ( '"legacy-windows-first-iteration-build-manifest.json"', "missing build manifest artifact lookup", ), ( '"build_manifest_sha256": "', "missing collector build manifest provenance field", ), ( '"build_manifest_path": "', "missing collector build manifest path field", ), ), "lab_evidence_checker": ( ( "peer_recover detail does not prove fresh restart-agent.flag", "missing retained peer recover fresh-flag gate", ), ( "self_rescue detail does not prove fresh restart-agent.flag/update-flag.txt", "missing retained self-rescue fresh-flag gate", ), ), "readiness_gate": ( ( "peer_recover detail does not prove fresh restart-agent.flag", "missing readiness peer recover fresh-flag gate", ), ( "self_rescue detail does not prove fresh restart-agent.flag/update-flag.txt", "missing readiness self-rescue fresh-flag gate", ), ), } def _sha256(path: Path) -> str: h = hashlib.sha256() with path.open("rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() def _sha256_bytes(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def _is_sha256_hex(value: str) -> bool: if len(value) != 64: return False try: int(value, 16) except ValueError: return False return True def _validate_sidecar(path: Path) -> list[str]: sidecar = path.with_suffix(path.suffix + ".sha256") if not sidecar.exists(): return [f"lab kit checksum sidecar is missing: {sidecar}"] try: parts = sidecar.read_text(encoding="utf-8").strip().split() except Exception as exc: return [f"lab kit checksum sidecar cannot be read: {exc}"] if not parts: return ["lab kit checksum sidecar cannot be read: empty sidecar"] expected = parts[0] if len(parts) < 2: return ["lab kit checksum sidecar filename is missing"] if len(parts) > 2: return ["lab kit checksum sidecar must contain exactly sha256 and filename"] if parts[1] != path.name: return ["lab kit checksum sidecar filename mismatch"] if not _is_sha256_hex(expected): return ["lab kit checksum sidecar SHA-256 is malformed"] actual = _sha256(path) if expected != actual: return ["lab kit checksum sidecar mismatch"] return [] def _validate_verifier_sidecar(required: bool = False) -> tuple[bool, list[str]]: verifier = Path(__file__).resolve() sidecar = verifier.with_suffix(verifier.suffix + ".sha256") if not sidecar.exists(): if required: return False, [f"verifier checksum sidecar is missing: {sidecar}"] return False, [] try: parts = sidecar.read_text(encoding="utf-8").strip().split() except Exception as exc: return True, [f"verifier checksum sidecar cannot be read: {exc}"] if not parts: return True, ["verifier checksum sidecar cannot be read: empty sidecar"] expected = parts[0] if len(parts) < 2: return True, ["verifier checksum sidecar filename is missing"] if len(parts) > 2: return True, ["verifier checksum sidecar must contain exactly sha256 and filename"] if parts[1] != verifier.name: return True, ["verifier checksum sidecar filename mismatch"] if not _is_sha256_hex(expected): return True, ["verifier checksum sidecar SHA-256 is malformed"] actual = _sha256(verifier) if expected != actual: return True, ["verifier checksum sidecar mismatch"] return True, [] def _read_manifest(zf: zipfile.ZipFile) -> tuple[dict[str, Any] | None, list[str]]: try: payload = json.loads(zf.read("manifest.json").decode("utf-8")) except Exception as exc: return None, [f"manifest.json: cannot read JSON: {exc}"] if not isinstance(payload, dict): return None, ["manifest.json: root JSON value must be an object"] return payload, [] def _validate_build_manifest_payload( payload: dict[str, Any], zf: zipfile.ZipFile | None = None, ) -> list[str]: errors: list[str] = [] if payload.get("schema") != "legacy-windows-first-iteration-build-manifest": errors.append("build manifest schema mismatch") if "go1.20" not in str(payload.get("legacy_go_version") or ""): errors.append("build manifest legacy_go_version must be Go 1.20") host_ips = payload.get("host_ips") if ( not isinstance(host_ips, list) or not host_ips or any(not isinstance(item, str) or not item.strip() for item in host_ips) ): errors.append("build manifest host_ips must be a non-empty string list") elif "10.0.10.41" not in host_ips: errors.append("build manifest host_ips must include Mac-Studio 10.0.10.41") raw_artifacts = payload.get("artifacts") artifacts: dict[str, dict[str, Any]] = {} if not isinstance(raw_artifacts, list): errors.append("build manifest artifacts must be a list") else: seen_artifacts: set[str] = set() for item in raw_artifacts: if not isinstance(item, dict): errors.append("build manifest artifact entries must be objects") continue artifact_path = str(item.get("path") or "") if artifact_path in seen_artifacts: errors.append(f"build manifest duplicate artifact path: {artifact_path}") seen_artifacts.add(artifact_path) artifacts[artifact_path] = item for required in BUILD_MANIFEST_REQUIRED_ARTIFACTS: item = artifacts.get(required) if not item: errors.append(f"build manifest missing artifact metadata: {required}") continue raw_sha = item.get("sha256") sha = raw_sha.strip() if isinstance(raw_sha, str) else "" if not _is_sha256_hex(sha): errors.append(f"build manifest artifact sha256 must be a 64-character hex string: {required}") raw_size = item.get("size_bytes") if isinstance(raw_size, bool) or not isinstance(raw_size, int) or raw_size <= 0: errors.append(f"build manifest artifact size_bytes must be a positive integer: {required}") if zf is None: continue archive_path = f"payload/{required}" try: data = zf.read(archive_path) except KeyError: errors.append(f"build manifest artifact missing from archive payload: {required}") continue if _is_sha256_hex(sha) and sha != _sha256_bytes(data): errors.append(f"build manifest artifact sha256 mismatch: {required}") if isinstance(raw_size, int) and not isinstance(raw_size, bool) and raw_size != len(data): errors.append(f"build manifest artifact size_bytes mismatch: {required}") return errors def verify(path: Path) -> tuple[dict[str, Any] | None, list[str]]: errors = _validate_sidecar(path) try: with zipfile.ZipFile(path) as zf: names = zf.namelist() name_set = set(names) duplicates = sorted({name for name in names if names.count(name) > 1}) if duplicates: errors.append(f"duplicate archive entries are not allowed: {', '.join(duplicates)}") if "README-FIRST.txt" not in name_set: errors.append("missing README-FIRST.txt") manifest, manifest_errors = _read_manifest(zf) errors.extend(manifest_errors) if manifest is None: return None, errors if manifest.get("schema") != KIT_SCHEMA: errors.append(f"manifest schema is not {KIT_SCHEMA}") artifacts = manifest.get("artifacts") if not isinstance(artifacts, list): artifacts = [] errors.append("manifest artifacts must be a list") roles: dict[str, dict[str, Any]] = {} seen_archive_paths: set[str] = set() for item in artifacts: if not isinstance(item, dict): errors.append("manifest artifact entries must be objects") continue role = str(item.get("role") or "") if not role: errors.append("manifest artifact role is required") continue if role in roles: errors.append(f"duplicate manifest artifact role: {role}") roles[role] = item archive_path = str(item.get("archive_path") or "") if archive_path: if archive_path in seen_archive_paths: errors.append(f"duplicate manifest artifact archive_path: {archive_path}") seen_archive_paths.add(archive_path) if archive_path not in name_set: errors.append(f"{role}: missing archive path: {archive_path}") continue try: data = zf.read(archive_path) except Exception as exc: errors.append(f"{role}: cannot read archive path {archive_path}: {exc}") continue raw_size = item.get("size_bytes") expected_size = raw_size if isinstance(raw_size, int) and not isinstance(raw_size, bool) else 0 raw_sha = item.get("sha256") expected_sha = raw_sha.strip() if isinstance(raw_sha, str) else "" if expected_size <= 0 or not _is_sha256_hex(expected_sha): errors.append(f"{role}: invalid manifest metadata") if expected_size != len(data): errors.append(f"{role}: artifact size mismatch") if _is_sha256_hex(expected_sha) and expected_sha != _sha256_bytes(data): errors.append(f"{role}: artifact hash mismatch") for snippet, message in REQUIRED_ROLE_SNIPPETS.get(role, ()): if snippet not in data.decode("utf-8", errors="ignore"): errors.append(f"{role}: {message}") missing_roles = sorted(REQUIRED_ROLES - set(roles)) if missing_roles: errors.append(f"missing required lab kit role(s): {', '.join(missing_roles)}") build_manifest_role = roles.get("build_manifest") build_manifest_path = str((build_manifest_role or {}).get("archive_path") or "") if build_manifest_path: try: build_manifest = json.loads(zf.read(build_manifest_path).decode("utf-8")) except Exception as exc: errors.append(f"build_manifest: cannot read JSON: {exc}") else: if not isinstance(build_manifest, dict): errors.append("build_manifest: root JSON value must be an object") else: for error in _validate_build_manifest_payload(build_manifest, zf): errors.append(f"build_manifest: {error}") return { "artifact_count": len(names), "roles": sorted(roles), "sha256": _sha256(path), }, errors except zipfile.BadZipFile as exc: return None, [f"cannot open lab kit ZIP: {exc}"] def main(argv: list[str]) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--kit", required=True, type=Path, help="legacy-windows-lab-kit.zip") parser.add_argument( "--require-verifier-sidecar", action="store_true", help="Fail if legacy-windows-lab-kit-verify.py.sha256 is not beside this verifier", ) args = parser.parse_args(argv) verifier_checked, self_errors = _validate_verifier_sidecar( required=args.require_verifier_sidecar ) summary, errors = verify(args.kit) errors = self_errors + errors if errors: print("legacy Windows lab kit verification failed", file=sys.stderr) for error in errors: print(f"- {error}", file=sys.stderr) return 2 assert summary is not None print("legacy Windows lab kit verification passed") if verifier_checked: print("verifier self-check: passed") print(f"artifact count: {summary['artifact_count']}") print(f"kit sha256: {summary['sha256']}") return 0 if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))