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

Merge remote-tracking branch 'gitea/main' into codex/github-redacted-evidence-validator-20260627

Browse Source
This commit is contained in:
Your Name
2026-06-27 23:01:11 +08:00
parent 82e9d780bb 48b192fab7
commit d4fcce4170
11 changed files with 583 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
apps/api/src/api/v1/iwooos.py
View File

@@ -37,6 +37,7 @@ from src.services.iwooos_wazuh_managed_host_coverage import (
)
from src.services.iwooos_wazuh_manager_registry_reviewer_validation import (
load_latest_iwooos_wazuh_manager_registry_reviewer_validation,
validate_iwooos_wazuh_manager_registry_acceptance_evidence as validate_wazuh_manager_registry_acceptance_evidence_payload,
validate_iwooos_wazuh_manager_registry_owner_export as validate_wazuh_manager_registry_owner_export_payload,
)
from src.services.iwooos_wazuh_owner_evidence_preflight import (
@@ -211,6 +212,40 @@ async def validate_iwooos_wazuh_manager_registry_owner_export(owner_export: dict
) from exc
@router.post(
"/api/v1/iwooos/wazuh-manager-registry-reviewer-validation/validate-manager-registry-acceptance",
response_model=dict[str, Any],
summary="驗證 Wazuh manager registry accepted 脫敏 evidence packet",
description=(
"針對單次 owner / reviewer 提供的 redacted Wazuh manager registry acceptance evidence "
"packet 進行 no-persist review readiness validation，回傳 accepted-for-review / needs supplement / "
"quarantined / rejected runtime action 分流。此端點不保存 payload、不查 Wazuh API、不讀主機、"
"不重新註冊 agent、不重啟服務、不讀或回傳機密明文、不啟用主動回應、不改 Nginx / Docker / "
"K8s / firewall，也不更新 manager registry accepted 總帳。"
),
)
async def validate_iwooos_wazuh_manager_registry_acceptance_evidence(
acceptance_evidence: dict[str, Any],
) -> dict[str, Any]:
"""回傳單次 Wazuh manager registry accepted evidence 的公開安全驗證結果。"""
try:
payload = await asyncio.to_thread(
validate_wazuh_manager_registry_acceptance_evidence_payload,
acceptance_evidence,
)
return redact_public_lan_topology(payload)
except FileNotFoundError as exc:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=str(exc),
) from exc
except (json.JSONDecodeError, ValueError) as exc:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"IwoooS Wazuh manager registry acceptance evidence 驗證器無效：{exc}",
) from exc
@router.get(
"/api/v1/iwooos/runtime-security-readback",
response_model=dict[str, Any],

270
apps/api/src/services/iwooos_wazuh_manager_registry_reviewer_validation.py
View File

@@ -111,6 +111,24 @@ _ACCOUNTABILITY_REQUIRED_FIELDS = {
"postcheck_plan",
}
_ACCEPTANCE_EVIDENCE_REQUIRED_FIELDS = {
"acceptance_reviewer_role",
"acceptance_decision",
"acceptance_decision_reason",
"accepted_scope_aliases",
"accepted_agent_total",
"accepted_agent_active",
"accepted_agent_disconnected",
"accepted_agent_never_connected",
"manager_registry_acceptance_evidence_refs",
"post_enable_readback_ref",
"product_scope_parity_ref",
"reviewed_at",
"followup_owner",
"rollback_owner",
"post_acceptance_verifier_plan",
}
def load_latest_iwooos_wazuh_manager_registry_reviewer_validation(
security_dir: Path | None = None,
@@ -137,6 +155,15 @@ def load_latest_iwooos_wazuh_manager_registry_reviewer_validation(
"reviewer_validation_failed_count": _int(summary.get("reviewer_validation_failed_count")),
"reviewer_validation_quarantined_count": _int(summary.get("reviewer_validation_quarantined_count")),
"manager_registry_accepted_count": _int(summary.get("manager_registry_accepted_count")),
"manager_registry_acceptance_intake_endpoint_available_count": _int(
summary.get("manager_registry_acceptance_intake_endpoint_available_count")
),
"manager_registry_acceptance_evidence_received_count": _int(
summary.get("manager_registry_acceptance_evidence_received_count")
),
"manager_registry_acceptance_evidence_review_ready_count": _int(
summary.get("manager_registry_acceptance_evidence_review_ready_count")
),
"post_enable_readback_passed_count": _int(summary.get("post_enable_readback_passed_count")),
"runtime_gate_count": _int(summary.get("runtime_gate_count")),
"host_write_authorized_count": _int(summary.get("host_write_authorized_count")),
@@ -157,6 +184,10 @@ def load_latest_iwooos_wazuh_manager_registry_reviewer_validation(
"/api/v1/iwooos/wazuh-manager-registry-reviewer-validation/validate-owner-export"
),
"owner_export_validation_mode": "no_persist_validation_no_runtime_action",
"manager_registry_acceptance_validation_endpoint": (
"/api/v1/iwooos/wazuh-manager-registry-reviewer-validation/validate-manager-registry-acceptance"
),
"manager_registry_acceptance_validation_mode": "no_persist_acceptance_evidence_review_no_runtime_action",
"summary": merged_summary,
"expected_scope_aliases": _strings(snapshot.get("expected_scope_aliases")),
"reviewer_validation_checks": _checks(snapshot.get("reviewer_validation_checks")),
@@ -267,6 +298,9 @@ def _boundary_markers(summary: dict[str, int]) -> list[str]:
f"wazuh_manager_registry_reviewer_validation_passed_count={summary['reviewer_validation_passed_count']}",
f"wazuh_manager_registry_reviewer_validation_quarantined_count={summary['reviewer_validation_quarantined_count']}",
f"wazuh_manager_registry_reviewer_validation_manager_registry_accepted_count={summary['manager_registry_accepted_count']}",
"wazuh_manager_registry_acceptance_validation_api_available=true",
f"wazuh_manager_registry_acceptance_evidence_received_count={summary['manager_registry_acceptance_evidence_received_count']}",
f"wazuh_manager_registry_acceptance_evidence_review_ready_count={summary['manager_registry_acceptance_evidence_review_ready_count']}",
f"wazuh_manager_registry_reviewer_validation_post_enable_readback_passed_count={summary['post_enable_readback_passed_count']}",
f"wazuh_manager_registry_reviewer_validation_runtime_gate_count={summary['runtime_gate_count']}",
"wazuh_api_live_query_authorized=false",
@@ -299,7 +333,24 @@ def _require_boundaries(payload: dict[str, Any]) -> None:
failed = _int(summary.get("reviewer_validation_failed_count"))
quarantined = _int(summary.get("reviewer_validation_quarantined_count"))
post_enable = _int(summary.get("post_enable_readback_passed_count"))
if any(value < 0 for value in (received, accepted, ready, passed, failed, quarantined, post_enable)):
acceptance_endpoint = _int(summary.get("manager_registry_acceptance_intake_endpoint_available_count"))
acceptance_received = _int(summary.get("manager_registry_acceptance_evidence_received_count"))
acceptance_ready = _int(summary.get("manager_registry_acceptance_evidence_review_ready_count"))
if any(
value < 0
for value in (
received,
accepted,
ready,
passed,
failed,
quarantined,
post_enable,
acceptance_endpoint,
acceptance_received,
acceptance_ready,
)
):
raise ValueError("Wazuh manager registry reviewer validation counters 不得為負數")
if accepted > received:
raise ValueError("owner_registry_export_accepted_count 不得大於 received_count")
@@ -309,6 +360,14 @@ def _require_boundaries(payload: dict[str, Any]) -> None:
raise ValueError("reviewer_validation_passed_count 不得大於 accepted_count")
if post_enable > passed:
raise ValueError("post_enable_readback_passed_count 不得大於 reviewer_validation_passed_count")
if acceptance_endpoint > 1:
raise ValueError("manager_registry_acceptance_intake_endpoint_available_count 不得大於 1")
if acceptance_ready > acceptance_received:
raise ValueError(
"manager_registry_acceptance_evidence_review_ready_count 不得大於 received_count"
)
if acceptance_ready:
raise ValueError("manager_registry_acceptance_evidence_review_ready_count 目前不得在全域讀回中自動上修")
if failed and passed:
raise ValueError("reviewer_validation_failed_count 與 passed_count 不得同時為正")
if quarantined and accepted:
@@ -477,6 +536,124 @@ def validate_iwooos_wazuh_manager_registry_owner_export(
return _validation_result(contract, outcome, findings, evidence_status)
def validate_iwooos_wazuh_manager_registry_acceptance_evidence(
acceptance_evidence: dict[str, Any],
security_dir: Path | None = None,
) -> dict[str, Any]:
"""Validate one redacted manager-registry acceptance packet without changing runtime truth."""
contract = load_latest_iwooos_wazuh_manager_registry_reviewer_validation(security_dir)
snapshot = _load_snapshot(security_dir or _DEFAULT_SECURITY_DIR)
expected_aliases = set(_strings(snapshot.get("expected_scope_aliases")))
findings: list[dict[str, Any]] = []
if not isinstance(acceptance_evidence, dict):
findings.append(
_finding(
"MRA-01",
"blocker",
"request_manager_registry_acceptance_evidence_supplement",
"manager registry acceptance evidence 必須是 JSON object。",
[],
)
)
return _acceptance_validation_result(contract, "request_manager_registry_acceptance_evidence_supplement", findings)
sensitive_hits = _collect_sensitive_hits(acceptance_evidence)
if sensitive_hits:
findings.append(
_finding(
"MRA-04",
"critical",
"quarantine_sensitive_payload",
"acceptance evidence 含禁止內容或疑似未脫敏內容，已進隔離分流；回應不回傳原始值。",
[hit["path"] for hit in sensitive_hits[:12]],
{"categories": sorted({hit["category"] for hit in sensitive_hits})},
)
)
return _acceptance_validation_result(contract, "quarantine_sensitive_payload", findings)
runtime_hits = _collect_runtime_action_hits(acceptance_evidence)
if runtime_hits:
findings.append(
_finding(
"MRA-05",
"critical",
"reject_runtime_action_request",
"acceptance evidence 夾帶 runtime action request；此 validator 只允許脫敏 acceptance refs，不授權 active response、restart、host write 或掃描。",
runtime_hits[:12],
)
)
return _acceptance_validation_result(contract, "reject_runtime_action_request", findings)
missing_fields = [
field for field in sorted(_ACCEPTANCE_EVIDENCE_REQUIRED_FIELDS) if not _present(acceptance_evidence.get(field))
]
if missing_fields:
findings.append(
_finding(
"MRA-01",
"blocker",
"request_manager_registry_acceptance_evidence_supplement",
"manager registry acceptance evidence 欄位不足，需要補齊後再驗證。",
missing_fields,
)
)
count_issue = _validate_acceptance_counts(acceptance_evidence)
if count_issue:
findings.append(
_finding(
"MRA-02",
"blocker",
"request_manager_registry_acceptance_counts_fix",
count_issue,
[
"accepted_agent_total",
"accepted_agent_active",
"accepted_agent_disconnected",
"accepted_agent_never_connected",
],
)
)
alias_issue = _validate_aliases(acceptance_evidence.get("accepted_scope_aliases"), expected_aliases)
if alias_issue:
findings.append(
_finding(
"MRA-03",
"blocker",
"request_manager_registry_acceptance_scope_parity_fix",
alias_issue,
["accepted_scope_aliases"],
)
)
decision = acceptance_evidence.get("acceptance_decision")
if decision != "accept_manager_registry_evidence_for_review":
findings.append(
_finding(
"MRA-06",
"blocker",
"request_manager_registry_acceptance_decision_fix",
"acceptance_decision 必須明確為 accept_manager_registry_evidence_for_review；不可宣稱已更新 manager registry accepted 總帳。",
["acceptance_decision"],
)
)
outcome = _first_blocking_lane(findings) or "accepted_for_manager_registry_acceptance_review_only"
if outcome == "accepted_for_manager_registry_acceptance_review_only":
findings.append(
_finding(
"MRA-07",
"info",
"manager_registry_acceptance_evidence_review_ready",
"acceptance evidence 已通過 no-persist review readiness；仍不更新全域 manager_registry_accepted_count，不開 runtime gate。",
["manager_registry_acceptance_evidence_refs"],
)
)
return _acceptance_validation_result(contract, outcome, findings)
def _validation_result(
contract: dict[str, Any],
outcome_lane: str,
@@ -543,6 +720,70 @@ def _validation_result(
}
def _acceptance_validation_result(
contract: dict[str, Any],
outcome_lane: str,
findings: list[dict[str, Any]],
) -> dict[str, Any]:
accepted = outcome_lane == "accepted_for_manager_registry_acceptance_review_only"
quarantined = outcome_lane == "quarantine_sensitive_payload"
rejected_runtime = outcome_lane == "reject_runtime_action_request"
supplement_required = not accepted and not quarantined and not rejected_runtime
return {
"schema_version": "iwooos_wazuh_manager_registry_acceptance_evidence_validation_result_v1",
"contract_schema_version": contract["schema_version"],
"status": outcome_lane,
"mode": "no_persist_acceptance_evidence_review_no_runtime_no_secret_collection",
"outcome_lane": outcome_lane,
"accepted_for_manager_registry_acceptance_review_only": accepted,
"quarantined": quarantined,
"runtime_action_rejected": rejected_runtime,
"summary": {
"manager_registry_acceptance_evidence_received_count": 1,
"manager_registry_acceptance_evidence_review_ready_count": 1 if accepted else 0,
"manager_registry_acceptance_evidence_supplement_required_count": 1 if supplement_required else 0,
"manager_registry_acceptance_evidence_quarantined_count": 1 if quarantined else 0,
"manager_registry_acceptance_runtime_action_rejected_count": 1 if rejected_runtime else 0,
"manager_registry_accepted_count": 0,
"runtime_gate_count": 0,
"host_write_authorized_count": 0,
"active_response_authorized_count": 0,
"secret_value_collection_allowed_count": 0,
"finding_count": len(findings),
},
"validation_findings": findings,
"boundary_markers": [
"wazuh_manager_registry_acceptance_validation_received_count=1",
f"wazuh_manager_registry_acceptance_validation_review_ready_count={1 if accepted else 0}",
f"wazuh_manager_registry_acceptance_validation_quarantined_count={1 if quarantined else 0}",
"wazuh_manager_registry_acceptance_validation_manager_registry_accepted_count=0",
"wazuh_manager_registry_acceptance_validation_runtime_gate_count=0",
"wazuh_manager_registry_acceptance_validation_no_persist=true",
"wazuh_api_live_query_authorized=false",
"wazuh_active_response_authorized=false",
"host_write_authorized=false",
"secret_value_collection_allowed=false",
"not_authorization=true",
],
"boundaries": {
"payload_persisted": False,
"wazuh_api_live_query_authorized": False,
"wazuh_agent_reenroll_authorized": False,
"wazuh_agent_restart_authorized": False,
"wazuh_manager_restart_authorized": False,
"wazuh_active_response_authorized": False,
"host_write_authorized": False,
"secret_value_collection_allowed": False,
"raw_wazuh_payload_storage_allowed": False,
"kali_active_scan_authorized": False,
"runtime_execution_authorized": False,
"manager_registry_accepted_updated": False,
"not_authorization": True,
},
"next_gate": "commit_manager_registry_accepted_readback_evidence" if accepted else "manager_registry_acceptance_fix_and_resubmit",
}
def _finding(
check_id: str,
severity: str,
@@ -614,6 +855,29 @@ def _validate_counts(owner_export: dict[str, Any]) -> str | None:
return None
def _validate_acceptance_counts(acceptance_evidence: dict[str, Any]) -> str | None:
fields = [
"accepted_agent_total",
"accepted_agent_active",
"accepted_agent_disconnected",
"accepted_agent_never_connected",
]
counts = {field: _int_or_none(acceptance_evidence.get(field)) for field in fields}
if any(value is None for value in counts.values()):
return "accepted agent count 欄位必須是非負整數。"
if any(value is not None and value < 0 for value in counts.values()):
return "accepted agent count 欄位不得為負數。"
total = counts["accepted_agent_total"]
active = counts["accepted_agent_active"]
disconnected = counts["accepted_agent_disconnected"]
never_connected = counts["accepted_agent_never_connected"]
if total is None or active is None or disconnected is None or never_connected is None:
return "accepted agent count 欄位必須是非負整數。"
if total < active + disconnected + never_connected:
return "accepted_agent_total 不得小於 active + disconnected + never_connected。"
return None
def _validate_aliases(value: Any, expected_aliases: set[str]) -> str | None:
aliases = value if isinstance(value, list) else []
if not aliases or not all(isinstance(item, str) for item in aliases):
@@ -709,6 +973,10 @@ def _first_blocking_lane(findings: list[dict[str, Any]]) -> str | None:
"request_dashboard_api_repair_postcheck",
"request_readonly_credential_metadata",
"request_owner_accountability_supplement",
"request_manager_registry_acceptance_evidence_supplement",
"request_manager_registry_acceptance_counts_fix",
"request_manager_registry_acceptance_scope_parity_fix",
"request_manager_registry_acceptance_decision_fix",
):
if any(item.get("lane") == lane for item in findings):
return lane

132
apps/api/tests/test_iwooos_wazuh_manager_registry_reviewer_validation.py
View File

@@ -6,6 +6,7 @@ from fastapi.testclient import TestClient
from src.api.v1.iwooos import router
from src.services.iwooos_wazuh_manager_registry_reviewer_validation import (
load_latest_iwooos_wazuh_manager_registry_reviewer_validation,
validate_iwooos_wazuh_manager_registry_acceptance_evidence,
validate_iwooos_wazuh_manager_registry_owner_export,
)
@@ -78,18 +79,42 @@ def _valid_owner_export() -> dict:
}
def _valid_acceptance_evidence() -> dict:
return {
"acceptance_reviewer_role": "IwoooS reviewer",
"acceptance_decision": "accept_manager_registry_evidence_for_review",
"acceptance_decision_reason": "脫敏 manager registry evidence 與六個公開別名 scope parity 一致，可進 commit review。",
"accepted_scope_aliases": EXPECTED_ALIASES,
"accepted_agent_total": 6,
"accepted_agent_active": 2,
"accepted_agent_disconnected": 3,
"accepted_agent_never_connected": 1,
"manager_registry_acceptance_evidence_refs": [
"evidence-ref-registry-summary",
"evidence-ref-scope-parity",
"evidence-ref-post-enable-readback",
],
"post_enable_readback_ref": "evidence-ref-post-enable-readback",
"product_scope_parity_ref": "evidence-ref-scope-parity",
"reviewed_at": "2026-06-27T22:10:00+08:00",
"followup_owner": "IwoooS reviewer",
"rollback_owner": "IwoooS runtime owner",
"post_acceptance_verifier_plan": "commit_readback_without_runtime_action",
}
def test_iwooos_wazuh_manager_registry_reviewer_validation_contract_has_passed_redacted_export() -> None:
payload = load_latest_iwooos_wazuh_manager_registry_reviewer_validation()
assert payload["schema_version"] == "iwooos_wazuh_manager_registry_reviewer_validation_readback_v1"
assert payload["source_schema_version"] == "wazuh_manager_registry_reviewer_validation_v1"
assert payload["status"] == "post_enable_iwooos_readback_passed_no_runtime_no_secret_collection"
assert payload["mode"] == "committed_post_enable_iwooos_readback_passed_no_runtime_no_secret_collection"
assert payload["status"] == "manager_registry_acceptance_evidence_intake_ready_no_runtime_no_secret_collection"
assert payload["mode"] == "committed_manager_registry_acceptance_evidence_intake_ready_no_runtime_no_secret_collection"
assert payload["summary"]["expected_scope_alias_count"] == 6
assert payload["summary"]["required_owner_field_count"] == 28
assert payload["summary"]["per_host_required_field_count"] == 9
assert payload["summary"]["reviewer_validation_check_count"] == 10
assert payload["summary"]["outcome_lane_count"] == 14
assert payload["summary"]["reviewer_validation_check_count"] == 11
assert payload["summary"]["outcome_lane_count"] == 17
assert payload["summary"]["evidence_slot_count"] == 6
assert payload["summary"]["forbidden_payload_count"] == 27
assert payload["summary"]["owner_registry_export_received_count"] == 1
@@ -99,6 +124,9 @@ def test_iwooos_wazuh_manager_registry_reviewer_validation_contract_has_passed_r
assert payload["summary"]["post_enable_readback_passed_count"] == 1
assert payload["summary"]["reviewer_validation_quarantined_count"] == 0
assert payload["summary"]["manager_registry_accepted_count"] == 0
assert payload["summary"]["manager_registry_acceptance_intake_endpoint_available_count"] == 1
assert payload["summary"]["manager_registry_acceptance_evidence_received_count"] == 0
assert payload["summary"]["manager_registry_acceptance_evidence_review_ready_count"] == 0
assert payload["summary"]["runtime_gate_count"] == 0
@@ -132,8 +160,11 @@ def test_iwooos_wazuh_manager_registry_reviewer_validation_api_is_public_safe()
assert data["summary"]["reviewer_validation_passed_count"] == 1
assert data["summary"]["post_enable_readback_passed_count"] == 1
assert data["summary"]["manager_registry_accepted_count"] == 0
assert data["summary"]["manager_registry_acceptance_intake_endpoint_available_count"] == 1
assert data["summary"]["manager_registry_acceptance_evidence_received_count"] == 0
assert data["summary"]["manager_registry_acceptance_evidence_review_ready_count"] == 0
assert data["summary"]["runtime_gate_count"] == 0
assert len(data["reviewer_validation_checks"]) == 10
assert len(data["reviewer_validation_checks"]) == 11
assert len(data["evidence_slots"]) == 6
assert any(
marker == "wazuh_manager_registry_reviewer_validation_owner_registry_export_received_count=1"
@@ -155,6 +186,18 @@ def test_iwooos_wazuh_manager_registry_reviewer_validation_api_is_public_safe()
marker == "wazuh_manager_registry_reviewer_validation_manager_registry_accepted_count=0"
for marker in data["boundary_markers"]
)
assert any(
marker == "wazuh_manager_registry_acceptance_validation_api_available=true"
for marker in data["boundary_markers"]
)
assert any(
marker == "wazuh_manager_registry_acceptance_evidence_received_count=0"
for marker in data["boundary_markers"]
)
assert any(
marker == "wazuh_manager_registry_acceptance_evidence_review_ready_count=0"
for marker in data["boundary_markers"]
)
assert any(
marker == "wazuh_manager_registry_reviewer_validation_runtime_gate_count=0"
for marker in data["boundary_markers"]
@@ -251,3 +294,82 @@ def test_iwooos_wazuh_manager_registry_owner_export_validation_rejects_runtime_a
assert payload["summary"]["owner_registry_export_accepted_count"] == 0
assert payload["summary"]["runtime_gate_count"] == 0
assert any(finding["check_id"] == "RV-09" for finding in payload["validation_findings"])
def test_iwooos_wazuh_manager_registry_acceptance_evidence_validation_accepts_redacted_packet_for_review_only() -> None:
payload = validate_iwooos_wazuh_manager_registry_acceptance_evidence(_valid_acceptance_evidence())
assert payload["schema_version"] == "iwooos_wazuh_manager_registry_acceptance_evidence_validation_result_v1"
assert payload["status"] == "accepted_for_manager_registry_acceptance_review_only"
assert payload["mode"] == "no_persist_acceptance_evidence_review_no_runtime_no_secret_collection"
assert payload["accepted_for_manager_registry_acceptance_review_only"] is True
assert payload["summary"]["manager_registry_acceptance_evidence_received_count"] == 1
assert payload["summary"]["manager_registry_acceptance_evidence_review_ready_count"] == 1
assert payload["summary"]["manager_registry_accepted_count"] == 0
assert payload["summary"]["runtime_gate_count"] == 0
assert payload["boundaries"]["payload_persisted"] is False
assert payload["boundaries"]["manager_registry_accepted_updated"] is False
assert payload["boundaries"]["runtime_execution_authorized"] is False
assert payload["next_gate"] == "commit_manager_registry_accepted_readback_evidence"
def test_iwooos_wazuh_manager_registry_acceptance_evidence_api_does_not_persist_or_open_runtime() -> None:
client = _client()
response = client.post(
"/api/v1/iwooos/wazuh-manager-registry-reviewer-validation/validate-manager-registry-acceptance",
json=_valid_acceptance_evidence(),
)
assert response.status_code == 200
result = response.json()
assert result["status"] == "accepted_for_manager_registry_acceptance_review_only"
assert result["summary"]["manager_registry_acceptance_evidence_review_ready_count"] == 1
assert result["summary"]["manager_registry_accepted_count"] == 0
assert result["summary"]["runtime_gate_count"] == 0
readback = client.get("/api/v1/iwooos/wazuh-manager-registry-reviewer-validation").json()
assert readback["summary"]["manager_registry_acceptance_evidence_received_count"] == 0
assert readback["summary"]["manager_registry_acceptance_evidence_review_ready_count"] == 0
assert readback["summary"]["manager_registry_accepted_count"] == 0
assert readback["summary"]["runtime_gate_count"] == 0
def test_iwooos_wazuh_manager_registry_acceptance_evidence_validation_requests_missing_fields() -> None:
candidate = _valid_acceptance_evidence()
candidate.pop("product_scope_parity_ref")
payload = validate_iwooos_wazuh_manager_registry_acceptance_evidence(candidate)
assert payload["status"] == "request_manager_registry_acceptance_evidence_supplement"
assert payload["accepted_for_manager_registry_acceptance_review_only"] is False
assert payload["summary"]["manager_registry_acceptance_evidence_supplement_required_count"] == 1
assert payload["summary"]["manager_registry_accepted_count"] == 0
assert any("product_scope_parity_ref" in finding["field_paths"] for finding in payload["validation_findings"])
def test_iwooos_wazuh_manager_registry_acceptance_evidence_validation_quarantines_sensitive_payload() -> None:
candidate = _valid_acceptance_evidence()
candidate["post_enable_readback_ref"] = "post enable accidentally included 10.250.250.250"
payload = validate_iwooos_wazuh_manager_registry_acceptance_evidence(candidate)
assert payload["status"] == "quarantine_sensitive_payload"
assert payload["quarantined"] is True
assert payload["summary"]["manager_registry_acceptance_evidence_quarantined_count"] == 1
assert payload["summary"]["manager_registry_accepted_count"] == 0
assert "10.250.250.250" not in str(payload)
assert any(finding["check_id"] == "MRA-04" for finding in payload["validation_findings"])
def test_iwooos_wazuh_manager_registry_acceptance_evidence_validation_rejects_runtime_action_request() -> None:
candidate = _valid_acceptance_evidence()
candidate["requested_actions"] = ["wazuh_active_response"]
payload = validate_iwooos_wazuh_manager_registry_acceptance_evidence(candidate)
assert payload["status"] == "reject_runtime_action_request"
assert payload["runtime_action_rejected"] is True
assert payload["summary"]["manager_registry_acceptance_runtime_action_rejected_count"] == 1
assert payload["summary"]["manager_registry_accepted_count"] == 0
assert payload["summary"]["runtime_gate_count"] == 0
assert any(finding["check_id"] == "MRA-05" for finding in payload["validation_findings"])