wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

fix(windows99): validate vmx candidate confirmation
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 1s
Details
CD Pipeline / cancel-stale-cd (push) Has been skipped
Details
CD Pipeline / tests (push) Successful in 3m16s
Details
AWOOOI Harbor 110 Local Repair / workflow-shape (push) Successful in 1s
Details
AWOOOI Harbor 110 Local Repair / harbor-110-local-repair (push) Successful in 35s
Details
CD Pipeline / build-and-deploy (push) Successful in 5m3s
Details
CD Pipeline / post-deploy-checks (push) Successful in 2m0s
Details

Browse Source
This commit is contained in:
ogt
2026-07-10 00:15:31 +08:00
parent a1ec18b9ba
commit e150a85a6d
5 changed files with 454 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

210
apps/api/src/services/windows99_vmx_source_locator_readback.py
View File

@@ -29,9 +29,67 @@ _RECEIPT_SCHEMA_VERSION = "windows99_vmx_source_locator_receipt_v1"
_CANDIDATE_CONFIRMATION_SCHEMA_VERSION = (
"windows99_vmx_source_candidate_confirmation_gate_v1"
)
_CANDIDATE_CONFIRMATION_VALIDATION_SCHEMA_VERSION = (
"windows99_vmx_source_candidate_confirmation_validation_v1"
)
_CANDIDATE_CONFIRMATION_PACKET_SCHEMA_VERSION = (
"windows99_vmx_source_candidate_confirmation_packet_v1"
)
_RECEIPT_FILE = "windows99-vmx-source-locator-receipt.snapshot.json"
_COLLECTOR_SCRIPT_PATH = "scripts/reboot-recovery/locate-windows99-vmx-source.sh"
_LOCATOR_SCRIPT_PATH = "scripts/reboot-recovery/windows99-vmx-source-locator.ps1"
_CANDIDATE_CONFIRMATION_REQUIRED_FIELDS = [
"schema_version",
"target_host_alias",
"target_vm_aliases",
"candidate_identity_kind",
"candidate_identity_evidence_ref",
"candidate_match_count",
"redacted_metadata_only",
"bounded_identity_metadata_only",
"path_fingerprint_only",
"raw_path_output",
"secret_value_read",
"remote_write_performed",
"host_reboot_performed",
"vm_power_change_performed",
"windows_registry_apply_performed",
"scheduled_task_write_performed",
"apply_requested",
"vm_power_change_requested",
]
_CANDIDATE_CONFIRMATION_ALLOWED_KINDS = {
"target_alias_hint",
"authorized_console_identity",
"verified_backup_path_fingerprint",
}
_CANDIDATE_CONFIRMATION_FORBIDDEN_ACTIONS = [
"raw_vmx_path_output",
"vmx_file_content_read",
"secret_value_read",
"password_or_token_collection",
"remote_write",
"vm_power_change",
"windows_registry_apply",
"scheduled_task_write",
"host_reboot",
"runtime_apply",
]
_CANDIDATE_CONFIRMATION_FORBIDDEN_FRAGMENTS = [
"192.168.0.",
"D:\\",
"C:\\",
"E:\\",
"/Users/",
"/Volumes/",
"uuid.",
"generatedAddress",
"BEGIN PRIVATE KEY",
"Authorization:",
"password=",
"token=",
"secret_value",
]
def _dict(value: Any) -> dict[str, Any]:
@@ -60,6 +118,31 @@ def _int(value: Any, default: int | None = None) -> int | None:
return default
def _walk_strings(value: Any) -> list[str]:
if isinstance(value, dict):
values: list[str] = []
for nested in value.values():
values.extend(_walk_strings(nested))
return values
if isinstance(value, list):
values = []
for item in value:
values.extend(_walk_strings(item))
return values
if isinstance(value, str):
return [value]
return []
def _forbidden_fragments(value: dict[str, Any]) -> list[str]:
text_values = _walk_strings(value)
hits = []
for fragment in _CANDIDATE_CONFIRMATION_FORBIDDEN_FRAGMENTS:
if any(fragment in text for text in text_values):
hits.append(fragment)
return hits
def _load_locator_receipt(operations_dir: Path | None = None) -> dict[str, Any]:
path = (operations_dir or _DEFAULT_OPERATIONS_DIR) / _RECEIPT_FILE
if not path.exists():
@@ -206,6 +289,133 @@ def _build_candidate_confirmation_gate(
}
def validate_windows99_vmx_source_candidate_confirmation_packet(
packet: dict[str, Any],
*,
locator_readback: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Validate one redacted Windows99 candidate-confirmation packet.
This is a no-persist validator. It never reads VMX contents, exposes raw
paths, writes Windows state, starts/stops VMs, reboots hosts, or opens an
apply gate.
"""
locator = locator_readback or load_latest_windows99_vmx_source_locator_readback()
current_target_aliases = _strings(locator.get("target_vm_aliases"))
submitted_aliases = _strings(packet.get("target_vm_aliases"))
missing_fields = [
field
for field in _CANDIDATE_CONFIRMATION_REQUIRED_FIELDS
if field not in packet
]
rejected_fields: list[str] = []
forbidden_fragments = _forbidden_fragments(packet)
if packet.get("schema_version") != _CANDIDATE_CONFIRMATION_PACKET_SCHEMA_VERSION:
rejected_fields.append("schema_version")
if str(packet.get("target_host_alias") or "") != "99":
rejected_fields.append("target_host_alias")
if not submitted_aliases or submitted_aliases != current_target_aliases:
rejected_fields.append("target_vm_aliases")
if str(packet.get("candidate_identity_kind") or "") not in (
_CANDIDATE_CONFIRMATION_ALLOWED_KINDS
):
rejected_fields.append("candidate_identity_kind")
if str(packet.get("candidate_identity_evidence_ref") or "").strip() == "":
rejected_fields.append("candidate_identity_evidence_ref")
if _int(packet.get("candidate_match_count"), 0) != 1:
rejected_fields.append("candidate_match_count")
required_true_flags = [
"redacted_metadata_only",
"bounded_identity_metadata_only",
"path_fingerprint_only",
]
required_false_flags = [
"raw_path_output",
"secret_value_read",
"remote_write_performed",
"host_reboot_performed",
"vm_power_change_performed",
"windows_registry_apply_performed",
"scheduled_task_write_performed",
"apply_requested",
"vm_power_change_requested",
]
for field in required_true_flags:
if packet.get(field) is not True:
rejected_fields.append(field)
for field in required_false_flags:
if packet.get(field, False) is not False:
rejected_fields.append(field)
if forbidden_fragments:
rejected_fields.append("forbidden_fragments")
accepted = not missing_fields and not rejected_fields
status = (
"accepted_for_scoped_relink_dry_run_preparation_only"
if accepted
else "rejected_candidate_confirmation_packet"
)
next_safe_step = (
"build_scoped_relink_dry_run_no_write_then_post_verifier"
if accepted
else "submit_redacted_single_candidate_identity_evidence_without_raw_paths"
)
return {
"schema_version": _CANDIDATE_CONFIRMATION_VALIDATION_SCHEMA_VERSION,
"status": status,
"accepted": accepted,
"target_host_alias": "99",
"target_vm_aliases": current_target_aliases,
"current_locator_status": str(locator.get("status") or ""),
"current_candidate_confirmation_status": str(
locator.get("candidate_confirmation_status") or ""
),
"submitted_candidate_identity_kind": str(
packet.get("candidate_identity_kind") or ""
),
"submitted_candidate_match_count": _int(
packet.get("candidate_match_count"), 0
),
"missing_fields": missing_fields,
"rejected_fields": rejected_fields,
"forbidden_fragment_count": len(forbidden_fragments),
"forbidden_actions": _CANDIDATE_CONFIRMATION_FORBIDDEN_ACTIONS,
"projected_confirmation_gate": {
"schema_version": _CANDIDATE_CONFIRMATION_SCHEMA_VERSION,
"status": (
"candidate_identity_confirmed_ready_for_scoped_relink_dry_run"
if accepted
else "blocked_candidate_identity_evidence_not_accepted"
),
"can_prepare_scoped_relink_dry_run": accepted,
"apply_allowed_by_confirmation_gate": False,
"vm_power_change_allowed": False,
"remote_write_performed": False,
"vm_power_change_performed": False,
"secret_value_read": False,
"safe_next_step": next_safe_step,
},
"operation_boundaries": {
"no_persist_validator": True,
"request_payload_saved": False,
"secret_value_read": False,
"raw_path_output": False,
"file_content_read": False,
"password_prompt_allowed": False,
"remote_write_performed": False,
"host_reboot_performed": False,
"vm_power_change_performed": False,
"windows_registry_apply_performed": False,
"scheduled_task_write_performed": False,
"apply_gate_opened": False,
},
"next_safe_step": next_safe_step,
}
def build_windows99_vmx_source_locator_readback(
scorecard: dict[str, Any],
*,