358 lines
12 KiB
Python
358 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import inspect
|
|
|
|
from src.services.awooop_ansible_audit_service import (
|
|
build_typed_target_route_generation,
|
|
get_ansible_catalog_item,
|
|
)
|
|
from src.services.awooop_ansible_check_mode_service import (
|
|
_apply_ansible_playbook_trust_gate,
|
|
_revalidate_claim_playbook_trust,
|
|
build_ansible_check_mode_claim_input,
|
|
run_controlled_apply_for_claim,
|
|
)
|
|
|
|
|
|
def _candidate(
|
|
catalog_id: str,
|
|
*,
|
|
catalog_revision: str = "test-revision-v1",
|
|
) -> dict[str, object]:
|
|
catalog_item = get_ansible_catalog_item(catalog_id) or {}
|
|
return {
|
|
"catalog_id": catalog_id,
|
|
"playbook_path": f"infra/ansible/playbooks/{catalog_id.split(':', 1)[1]}.yml",
|
|
"inventory_hosts": ["host_188"],
|
|
"risk_level": "medium",
|
|
"catalog_revision": (
|
|
catalog_item.get("catalog_revision") or catalog_revision
|
|
),
|
|
}
|
|
|
|
|
|
def test_low_trust_failed_playbook_opens_claim_circuit() -> None:
|
|
candidate = _candidate("ansible:nginx-sync")
|
|
guarded, gate = _apply_ansible_playbook_trust_gate(
|
|
{"executor_candidates": [candidate]},
|
|
[
|
|
{
|
|
"playbook_id": "PB-ANSIBLE-NGINX-SYNC",
|
|
"status": "approved",
|
|
"trust_score": 0.001417,
|
|
"success_count": 0,
|
|
"failure_count": 24,
|
|
"review_required": False,
|
|
"symptom_pattern": {
|
|
"catalog_revision": candidate["catalog_revision"],
|
|
},
|
|
}
|
|
],
|
|
)
|
|
|
|
assert guarded["executor_candidates"] == []
|
|
assert gate["status"] == "circuit_open"
|
|
assert gate["blocked_candidate_count"] == 1
|
|
assert gate["decisions"][0]["reason"] == (
|
|
"playbook_trust_below_archive_threshold"
|
|
)
|
|
assert gate["next_controlled_action"] == (
|
|
"generate_or_verify_playbook_repair_candidate"
|
|
)
|
|
|
|
|
|
def test_low_trust_candidate_is_filtered_without_blocking_healthy_candidate() -> None:
|
|
failed = _candidate("ansible:nginx-sync")
|
|
healthy = _candidate("ansible:188-momo-backup-user")
|
|
guarded, gate = _apply_ansible_playbook_trust_gate(
|
|
{
|
|
"executor_candidates": [
|
|
failed,
|
|
healthy,
|
|
]
|
|
},
|
|
[
|
|
{
|
|
"playbook_id": "PB-ANSIBLE-NGINX-SYNC",
|
|
"status": "approved",
|
|
"trust_score": 0.01,
|
|
"success_count": 0,
|
|
"failure_count": 15,
|
|
"review_required": True,
|
|
"symptom_pattern": {
|
|
"catalog_revision": failed["catalog_revision"],
|
|
},
|
|
},
|
|
{
|
|
"playbook_id": "PB-ANSIBLE-188-MOMO-BACKUP-USER",
|
|
"status": "approved",
|
|
"trust_score": 0.65,
|
|
"success_count": 7,
|
|
"failure_count": 1,
|
|
"review_required": False,
|
|
"symptom_pattern": {
|
|
"catalog_revision": healthy["catalog_revision"],
|
|
},
|
|
},
|
|
],
|
|
)
|
|
|
|
assert guarded["executor_candidates"] == [healthy]
|
|
assert gate["status"] == "degraded_candidates_filtered"
|
|
assert gate["eligible_candidate_count"] == 1
|
|
assert gate["blocked_candidate_count"] == 1
|
|
|
|
|
|
def test_mixed_trust_filter_preserves_original_typed_generation() -> None:
|
|
canonical_asset_id = "service:host188-maintenance"
|
|
typed_domain = "host_systemd"
|
|
|
|
def typed_candidate(catalog_id: str) -> dict[str, object]:
|
|
catalog_item = get_ansible_catalog_item(catalog_id) or {}
|
|
playbook_path = str(catalog_item.get("playbook_path") or "")
|
|
return {
|
|
"catalog_id": catalog_id,
|
|
"catalog_revision": str(
|
|
catalog_item.get("catalog_revision") or ""
|
|
),
|
|
"playbook_path": playbook_path,
|
|
"check_mode_playbook_path": str(
|
|
catalog_item.get("check_mode_playbook_path")
|
|
or playbook_path
|
|
),
|
|
"inventory_hosts": list(
|
|
catalog_item.get("inventory_hosts") or []
|
|
),
|
|
"risk_level": str(catalog_item.get("risk_level") or ""),
|
|
"canonical_asset_id": canonical_asset_id,
|
|
"typed_domain": typed_domain,
|
|
"target_scope_validated": True,
|
|
}
|
|
|
|
failed = typed_candidate("ansible:nginx-sync")
|
|
healthy = typed_candidate("ansible:188-momo-backup-user")
|
|
typed_target_route = {
|
|
"schema_version": "typed_domain_target_route_v2",
|
|
"route_id": "typed:host_systemd:host188-maintenance",
|
|
"resolution_status": "resolved",
|
|
"executor": "host_ansible_executor",
|
|
"verifier": "host_runtime_independent_verifier",
|
|
"cross_domain_fallback_allowed": False,
|
|
"allowed_catalog_ids": [
|
|
"ansible:nginx-sync",
|
|
"ansible:188-momo-backup-user",
|
|
],
|
|
"allowed_inventory_hosts": ["host_188"],
|
|
"canonical_asset_id": canonical_asset_id,
|
|
"target_kind": typed_domain,
|
|
"risk_class": "high",
|
|
"controlled_apply_allowed": True,
|
|
"critical_break_glass_required": False,
|
|
}
|
|
candidates = [failed, healthy]
|
|
original_generation = build_typed_target_route_generation(
|
|
typed_target_route=typed_target_route,
|
|
executor_candidates=candidates,
|
|
)
|
|
guarded, gate = _apply_ansible_playbook_trust_gate(
|
|
{
|
|
"incident_id": "INC-MIXED-TRUST-TYPED-GENERATION",
|
|
"candidate_catalog_schema": "typed_domain_router_v2",
|
|
"target_route_generation": original_generation,
|
|
"typed_target_route": typed_target_route,
|
|
"single_writer_executor": "awoooi-ansible-executor-broker",
|
|
"executor_candidates": candidates,
|
|
},
|
|
[
|
|
{
|
|
"playbook_id": "PB-ANSIBLE-NGINX-SYNC",
|
|
"status": "approved",
|
|
"trust_score": 0.01,
|
|
"success_count": 0,
|
|
"failure_count": 15,
|
|
"review_required": False,
|
|
"symptom_pattern": {
|
|
"catalog_revision": failed["catalog_revision"],
|
|
},
|
|
},
|
|
{
|
|
"playbook_id": "PB-ANSIBLE-188-MOMO-BACKUP-USER",
|
|
"status": "approved",
|
|
"trust_score": 0.65,
|
|
"success_count": 7,
|
|
"failure_count": 1,
|
|
"review_required": False,
|
|
"symptom_pattern": {
|
|
"catalog_revision": healthy["catalog_revision"],
|
|
},
|
|
},
|
|
],
|
|
)
|
|
|
|
claim = build_ansible_check_mode_claim_input(
|
|
source_candidate_op_id="00000000-0000-0000-0000-000000000501",
|
|
candidate_input=guarded,
|
|
)
|
|
|
|
assert gate["status"] == "degraded_candidates_filtered"
|
|
assert claim["catalog_id"] == "ansible:188-momo-backup-user"
|
|
assert claim["target_route_generation"] == original_generation
|
|
assert len(claim["typed_generation_candidates"]) == 2
|
|
|
|
|
|
def test_repaired_catalog_revision_gets_one_bounded_canary() -> None:
|
|
candidate = _candidate(
|
|
"ansible:110-devops",
|
|
catalog_revision="host110-route-v3",
|
|
)
|
|
current_revision = str(candidate["catalog_revision"])
|
|
guarded, gate = _apply_ansible_playbook_trust_gate(
|
|
{"executor_candidates": [candidate]},
|
|
[
|
|
{
|
|
"playbook_id": "PB-ANSIBLE-110-DEVOPS",
|
|
"status": "approved",
|
|
"trust_score": 0.024,
|
|
"success_count": 0,
|
|
"failure_count": 1,
|
|
"review_required": False,
|
|
"symptom_pattern": {
|
|
"catalog_revision": f"{current_revision}-previous",
|
|
},
|
|
}
|
|
],
|
|
)
|
|
|
|
assert guarded["executor_candidates"] == [candidate]
|
|
assert gate["status"] == "closed"
|
|
decision = gate["decisions"][0]
|
|
assert decision["circuit_open"] is False
|
|
assert decision["bounded_repair_canary"] is True
|
|
assert decision["reason"] == "repaired_catalog_revision_canary_allowed"
|
|
assert decision["current_catalog_revision"] == current_revision
|
|
assert decision["catalog_revision_source"] == "allowlisted_catalog"
|
|
assert decision["observed_catalog_revision"] == (
|
|
f"{current_revision}-previous"
|
|
)
|
|
|
|
|
|
def test_failed_repaired_revision_reopens_circuit_for_same_revision() -> None:
|
|
candidate = _candidate(
|
|
"ansible:110-devops",
|
|
catalog_revision="host110-route-v3",
|
|
)
|
|
current_revision = str(candidate["catalog_revision"])
|
|
guarded, gate = _apply_ansible_playbook_trust_gate(
|
|
{"executor_candidates": [candidate]},
|
|
[
|
|
{
|
|
"playbook_id": "PB-ANSIBLE-110-DEVOPS",
|
|
"status": "approved",
|
|
"trust_score": 0.0192,
|
|
"success_count": 0,
|
|
"failure_count": 2,
|
|
"review_required": False,
|
|
"symptom_pattern": {
|
|
"catalog_revision": current_revision,
|
|
},
|
|
}
|
|
],
|
|
)
|
|
|
|
assert guarded["executor_candidates"] == []
|
|
assert gate["status"] == "circuit_open"
|
|
decision = gate["decisions"][0]
|
|
assert decision["circuit_open"] is True
|
|
assert decision["bounded_repair_canary"] is False
|
|
assert decision["reason"] == "playbook_trust_below_archive_threshold"
|
|
|
|
|
|
def test_legacy_failed_record_without_revision_gets_migration_canary() -> None:
|
|
candidate = _candidate(
|
|
"ansible:110-devops",
|
|
catalog_revision="host110-route-v3",
|
|
)
|
|
guarded, gate = _apply_ansible_playbook_trust_gate(
|
|
{"executor_candidates": [candidate]},
|
|
[
|
|
{
|
|
"playbook_id": "PB-ANSIBLE-110-DEVOPS",
|
|
"status": "approved",
|
|
"trust_score": 0.02,
|
|
"success_count": 0,
|
|
"failure_count": 3,
|
|
"review_required": False,
|
|
"symptom_pattern": {},
|
|
}
|
|
],
|
|
)
|
|
|
|
assert guarded["executor_candidates"] == [candidate]
|
|
decision = gate["decisions"][0]
|
|
assert decision["bounded_repair_canary"] is True
|
|
assert decision["observed_catalog_revision"] is None
|
|
|
|
|
|
def test_candidate_cannot_forge_revision_to_bypass_same_revision_circuit() -> None:
|
|
catalog_id = "ansible:110-devops"
|
|
current_revision = str(
|
|
(get_ansible_catalog_item(catalog_id) or {}).get("catalog_revision")
|
|
or ""
|
|
)
|
|
candidate = _candidate(catalog_id)
|
|
candidate["catalog_revision"] = "forged-revision-v999"
|
|
|
|
guarded, gate = _apply_ansible_playbook_trust_gate(
|
|
{"executor_candidates": [candidate]},
|
|
[
|
|
{
|
|
"playbook_id": "PB-ANSIBLE-110-DEVOPS",
|
|
"status": "approved",
|
|
"trust_score": 0.02,
|
|
"success_count": 0,
|
|
"failure_count": 3,
|
|
"review_required": False,
|
|
"symptom_pattern": {
|
|
"catalog_revision": current_revision,
|
|
},
|
|
}
|
|
],
|
|
)
|
|
|
|
assert guarded["executor_candidates"] == []
|
|
decision = gate["decisions"][0]
|
|
assert decision["bounded_repair_canary"] is False
|
|
assert decision["candidate_catalog_revision_mismatch"] is True
|
|
assert decision["current_catalog_revision"] == current_revision
|
|
|
|
|
|
def test_owner_review_flag_does_not_become_low_medium_high_terminal_gate() -> None:
|
|
candidate = _candidate("ansible:188-momo-backup-user")
|
|
guarded, gate = _apply_ansible_playbook_trust_gate(
|
|
{"executor_candidates": [candidate]},
|
|
[
|
|
{
|
|
"playbook_id": "PB-ANSIBLE-188-MOMO-BACKUP-USER",
|
|
"status": "approved",
|
|
"trust_score": 0.4,
|
|
"success_count": 2,
|
|
"failure_count": 1,
|
|
"review_required": True,
|
|
}
|
|
],
|
|
)
|
|
|
|
assert guarded["executor_candidates"] == [candidate]
|
|
assert gate["status"] == "closed"
|
|
assert gate["owner_review_used_as_terminal"] is False
|
|
|
|
|
|
def test_controlled_apply_revalidates_trust_before_runtime_write() -> None:
|
|
apply_source = inspect.getsource(run_controlled_apply_for_claim)
|
|
revalidate_source = inspect.getsource(_revalidate_claim_playbook_trust)
|
|
|
|
assert "_revalidate_claim_playbook_trust" in apply_source
|
|
assert "ansible_controlled_apply_blocked_by_playbook_trust" in apply_source
|
|
assert "playbook_trust_circuit_open" in revalidate_source
|
|
assert "UPDATE automation_operation_log" in revalidate_source
|