df42c32b22
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / build-and-deploy (push) Has been cancelled
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / tests (push) Has been cancelled
208 lines
8.2 KiB
Python
208 lines
8.2 KiB
Python
"""Windows99 VMX source repair check-mode package readback.
|
|
|
|
This API-facing wrapper reuses the reboot SLO scorecard's canonical package
|
|
builder, then adds compact count/metadata fields for manager UI consumption.
|
|
It does not read secrets, start VMs, write Windows state, restart services, or
|
|
reboot hosts.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from src.services.reboot_auto_recovery_slo_scorecard import (
|
|
_build_windows99_vmx_source_repair_package,
|
|
_taipei_now_iso,
|
|
load_latest_reboot_auto_recovery_slo_scorecard,
|
|
)
|
|
|
|
|
|
def _dict(value: Any) -> dict[str, Any]:
|
|
return value if isinstance(value, dict) else {}
|
|
|
|
|
|
def _strings(value: Any) -> list[str]:
|
|
if isinstance(value, list):
|
|
return [str(item).strip() for item in value if str(item).strip()]
|
|
if isinstance(value, str):
|
|
return [part.strip() for part in value.split(",") if part.strip()]
|
|
return []
|
|
|
|
|
|
def _with_ui_counts(
|
|
package: dict[str, Any],
|
|
scorecard: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
payload = dict(package)
|
|
missing_vmx_aliases = _strings(payload.get("missing_vmx_aliases"))
|
|
powered_off_aliases = _strings(payload.get("powered_off_aliases"))
|
|
readback = _dict(scorecard.get("readback"))
|
|
windows99_management = _dict(scorecard.get("windows99_management_channel"))
|
|
ssh_batch = _dict(windows99_management.get("ssh_batch"))
|
|
payload["controlled_restore_required_count"] = len(missing_vmx_aliases)
|
|
payload["powered_off_count"] = len(powered_off_aliases)
|
|
payload["scorecard_status"] = str(scorecard.get("status") or "unknown")
|
|
payload["scorecard_generated_at"] = str(scorecard.get("generated_at") or "")
|
|
payload["collector_status"] = str(
|
|
readback.get("windows99_no_secret_collector_status") or "unknown"
|
|
)
|
|
payload["collector_readback_present"] = (
|
|
readback.get("windows99_no_secret_collector_readback_present") is True
|
|
)
|
|
payload["collector_ssh_batchmode_auth_ready"] = (
|
|
readback.get("windows99_no_secret_collector_ssh_batchmode_auth_ready")
|
|
is True
|
|
)
|
|
payload["collector_remote_verify_attempted"] = (
|
|
readback.get("windows99_no_secret_collector_remote_verify_attempted")
|
|
is True
|
|
)
|
|
payload["collector_safe_next_step"] = str(
|
|
readback.get("windows99_no_secret_collector_safe_next_step") or ""
|
|
)
|
|
payload["management_channel_ready"] = (
|
|
windows99_management.get("remote_execution_channel_ready") is True
|
|
)
|
|
payload["can_collect_vmware_verify_without_secret"] = (
|
|
windows99_management.get("can_collect_vmware_verify_without_secret") is True
|
|
)
|
|
payload["ssh_batch_ready"] = ssh_batch.get("ready") is True
|
|
payload["ssh_batch_status"] = str(ssh_batch.get("status") or "unknown")
|
|
payload["winrm_http_open"] = windows99_management.get("winrm_http_open") is True
|
|
payload["winrm_https_open"] = windows99_management.get("winrm_https_open") is True
|
|
payload["local_console_channel_reachable"] = (
|
|
windows99_management.get("local_console_channel_reachable") is True
|
|
)
|
|
payload["console_collection_channels"] = _strings(
|
|
windows99_management.get("console_collection_channels")
|
|
)
|
|
payload["collector_management_channel_conflict"] = bool(
|
|
payload["collector_ssh_batchmode_auth_ready"]
|
|
and not payload["management_channel_ready"]
|
|
)
|
|
payload["effective_check_mode_execution_ready"] = bool(
|
|
payload.get("package_ready") is True
|
|
and payload.get("check_mode_allowed_by_this_package") is True
|
|
and payload["management_channel_ready"]
|
|
and not payload["collector_management_channel_conflict"]
|
|
)
|
|
return payload
|
|
|
|
|
|
def _mark_package_public_redaction_boundary(
|
|
package: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
payload = dict(package)
|
|
payload["public_lan_topology_redacted"] = True
|
|
payload["redacted_public_display_only"] = True
|
|
payload["check_mode_probe_commands_executable"] = False
|
|
payload["internal_console_package_required_for_execution"] = True
|
|
payload["public_execution_note"] = (
|
|
"public_endpoint_redacts_lan_like_vmx_path_segments_use_authorized_"
|
|
"internal_console_package_before_executing_probe_commands"
|
|
)
|
|
|
|
vm_rows = payload.get("vm_rows")
|
|
if isinstance(vm_rows, list):
|
|
redacted_rows: list[Any] = []
|
|
for row in vm_rows:
|
|
if isinstance(row, dict):
|
|
redacted_row = dict(row)
|
|
redacted_row["public_lan_topology_redacted"] = True
|
|
redacted_row["check_mode_probe_commands_executable"] = False
|
|
redacted_rows.append(redacted_row)
|
|
else:
|
|
redacted_rows.append(row)
|
|
payload["vm_rows"] = redacted_rows
|
|
|
|
action_chain = payload.get("action_chain")
|
|
if isinstance(action_chain, dict):
|
|
redacted_action_chain = dict(action_chain)
|
|
redacted_action_chain["public_lan_topology_redacted"] = True
|
|
redacted_action_chain["redacted_public_display_only"] = True
|
|
check_mode = redacted_action_chain.get("check_mode")
|
|
if isinstance(check_mode, dict):
|
|
redacted_check_mode = dict(check_mode)
|
|
redacted_check_mode["probe_commands_executable"] = False
|
|
redacted_action_chain["check_mode"] = redacted_check_mode
|
|
source_diffs = redacted_action_chain.get("source_of_truth_diff")
|
|
if isinstance(source_diffs, list):
|
|
redacted_diffs: list[Any] = []
|
|
for diff in source_diffs:
|
|
if isinstance(diff, dict):
|
|
redacted_diff = dict(diff)
|
|
redacted_diff["public_lan_topology_redacted"] = True
|
|
redacted_diff["redacted_public_display_only"] = True
|
|
redacted_diffs.append(redacted_diff)
|
|
else:
|
|
redacted_diffs.append(diff)
|
|
redacted_action_chain["source_of_truth_diff"] = redacted_diffs
|
|
payload["action_chain"] = redacted_action_chain
|
|
return payload
|
|
|
|
|
|
def mark_windows99_vmx_source_repair_public_redaction_boundary(
|
|
payload: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
"""Mark public-redacted VMX package fields as display-only evidence."""
|
|
if payload.get("schema_version") == (
|
|
"windows99_vmx_source_repair_check_mode_package_v1"
|
|
):
|
|
return _mark_package_public_redaction_boundary(payload)
|
|
|
|
package = payload.get("windows99_vmx_source_repair_package")
|
|
if isinstance(package, dict):
|
|
redacted_payload = dict(payload)
|
|
redacted_payload["windows99_vmx_source_repair_package"] = (
|
|
_mark_package_public_redaction_boundary(package)
|
|
)
|
|
return redacted_payload
|
|
return payload
|
|
|
|
|
|
def _alias_lists_match(package: dict[str, Any], windows99: dict[str, Any]) -> bool:
|
|
return _strings(package.get("missing_vmx_aliases")) == _strings(
|
|
windows99.get("missing_vmx_aliases")
|
|
) and _strings(package.get("powered_off_aliases")) == _strings(
|
|
windows99.get("powered_off_aliases")
|
|
)
|
|
|
|
|
|
def _package_has_action_chain(package: dict[str, Any]) -> bool:
|
|
action_chain = _dict(package.get("action_chain"))
|
|
target_selector = _dict(action_chain.get("target_selector"))
|
|
return bool(
|
|
action_chain.get("work_order_id")
|
|
and action_chain.get("stage")
|
|
and action_chain.get("next_executable_step")
|
|
and target_selector.get("host_alias")
|
|
)
|
|
|
|
|
|
def build_windows99_vmx_source_repair_package(
|
|
scorecard: dict[str, Any],
|
|
*,
|
|
generated_at: str | None = None,
|
|
) -> dict[str, Any]:
|
|
existing_package = _dict(scorecard.get("windows99_vmx_source_repair_package"))
|
|
windows99 = _dict(scorecard.get("windows99_vmware_autostart"))
|
|
package = (
|
|
dict(existing_package)
|
|
if existing_package
|
|
and generated_at is None
|
|
and _alias_lists_match(existing_package, windows99)
|
|
and _package_has_action_chain(existing_package)
|
|
else _build_windows99_vmx_source_repair_package(
|
|
windows99,
|
|
generated_at=generated_at
|
|
or str(scorecard.get("generated_at") or _taipei_now_iso()),
|
|
)
|
|
)
|
|
return _with_ui_counts(package, scorecard)
|
|
|
|
|
|
def load_latest_windows99_vmx_source_repair_package() -> dict[str, Any]:
|
|
return build_windows99_vmx_source_repair_package(
|
|
load_latest_reboot_auto_recovery_slo_scorecard()
|
|
)
|