fix(automation): fail closed unresolved VMware targets

This commit is contained in:
ogt
2026-07-15 12:17:42 +08:00
parent 0dc83b64d1
commit eed3f493d5
2 changed files with 58 additions and 1 deletions

View File

@@ -291,7 +291,28 @@ def resolve_typed_alert_target(
str(labels.get("instance") or ""),
target_resource,
alertname,
) or _HOSTS["99"]
)
explicit_agent99_target = any(
marker in _text_token(target_resource)
for marker in ("agent99", "windows99", "windows 99")
)
if host_scope is None and explicit_agent99_target:
host_scope = _HOSTS["99"]
if host_scope is None:
identity = registry.resolve_identity(target_resource or alertname)
return _typed_route(
resolution_status="asset_identity_unresolved",
target_kind="unknown",
target_resource=target_resource,
namespace=namespace,
canonical_asset_id=None,
host=None,
executor=None,
verifier=None,
risk_class="high",
drift_work_item_id=identity["drift_work_item_id"],
stateful_level=StatefulLevel.UNRESOLVED.value,
)
return _typed_route(
resolution_status="resolved",
target_kind="windows_vmware",

View File

@@ -234,6 +234,42 @@ def test_host112_guest_recovery_is_an_exact_agent99_domain_route() -> None:
assert payload["routing"]["crossDomainFallbackAllowed"] is False
def test_unresolved_vmware_target_cannot_default_to_host99() -> None:
route = resolve_typed_alert_target(
alertname="VMwareGuestHealthDegraded",
target_resource="unregistered-vmware-guest",
namespace="virtualization",
labels={},
alert_category="infrastructure",
registry=ServiceRegistryClient(REGISTRY),
)
assert route["resolution_status"] == "asset_identity_unresolved"
assert route["target_kind"] == "unknown"
assert route["executor"] is None
assert route["controlled_apply_allowed"] is False
assert route["agent99_bridge"]["dispatch_allowed"] is False
assert route["drift_work_item_id"].startswith("AIA-ASSET-DRIFT-")
def test_explicit_agent99_target_resolves_to_host99_executor() -> None:
route = resolve_typed_alert_target(
alertname="Agent99ServiceUnhealthy",
target_resource="agent99",
namespace="control-plane",
labels={},
alert_category="infrastructure",
registry=ServiceRegistryClient(REGISTRY),
)
assert route["resolution_status"] == "resolved"
assert route["target_kind"] == "windows_vmware"
assert route["canonical_asset_id"] == "windows-vmware:host_99"
assert route["executor"] == "Agent99"
assert route["allowed_inventory_hosts"] == ["host_99"]
assert route["agent99_bridge"]["dispatch_allowed"] is True
@pytest.mark.asyncio
async def test_agent99_rejects_cross_domain_mutating_dispatch_before_transport(
monkeypatch: pytest.MonkeyPatch,