fix(iwooos): add wazuh alert verifier readback
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

This commit is contained in:
ogt
2026-07-10 17:48:14 +08:00
parent 05033c5a29
commit aceee39b4a
5 changed files with 1004 additions and 15 deletions

View File

@@ -25,15 +25,15 @@ from src.services.iwooos_owner_evidence_intake_preflight import (
from src.services.iwooos_runtime_security_readback import (
load_latest_iwooos_runtime_security_readback,
)
from src.services.iwooos_security_control_coverage import (
load_latest_iwooos_security_control_coverage,
)
from src.services.iwooos_security_operating_system import (
load_latest_iwooos_security_operating_system,
)
from src.services.iwooos_security_operating_system import (
validate_iwooos_security_operation_packet as validate_iwooos_security_operation_packet_payload,
)
from src.services.iwooos_security_control_coverage import (
load_latest_iwooos_security_control_coverage,
)
from src.services.iwooos_security_tool_closure_readback import (
load_latest_iwooos_security_tool_closure_readback,
)
@@ -43,6 +43,12 @@ from src.services.iwooos_wazuh_allowlisted_check_mode_dry_run import (
from src.services.iwooos_wazuh_allowlisted_check_mode_dry_run import (
validate_iwooos_wazuh_allowlisted_check_mode_dry_run_packet as validate_wazuh_allowlisted_check_mode_dry_run_packet_payload,
)
from src.services.iwooos_wazuh_fim_vulnerability_alert_verifier import (
load_latest_iwooos_wazuh_fim_vulnerability_alert_verifier,
)
from src.services.iwooos_wazuh_fim_vulnerability_alert_verifier import (
validate_iwooos_wazuh_fim_vulnerability_alert_verifier_packet as validate_wazuh_fim_vulnerability_alert_verifier_packet_payload,
)
from src.services.iwooos_wazuh_live_metadata_gate import (
load_latest_iwooos_wazuh_live_metadata_gate,
)
@@ -607,6 +613,73 @@ async def validate_iwooos_wazuh_allowlisted_check_mode_dry_run_packet(
) from exc
@router.get(
"/api/v1/iwooos/wazuh-fim-vulnerability-alert-verifier",
response_model=dict[str, Any],
summary="取得 Wazuh FIM / vulnerability alert verifier 只讀讀回",
description=(
"彙整已提交的 Wazuh manager registry、FIM / persistence normalization、"
"CVE / KEV prioritization、SIEM correlation 與 Alertmanager receipt gate"
"形成 P0 Wazuh 偵測閉環 post-verifier 讀回。此端點不查 live Wazuh API、"
"不讀主機、不執行 scanner、不保存 payload、不啟用 active response、不改 Nginx / "
"Docker / K8s / firewall。"
),
)
async def get_iwooos_wazuh_fim_vulnerability_alert_verifier() -> dict[str, Any]:
"""回傳 Wazuh FIM / vulnerability alert verifier 公開安全只讀狀態。"""
try:
alert_receipt_readback = await _alertmanager_receipt_readback()
payload = await asyncio.to_thread(
load_latest_iwooos_wazuh_fim_vulnerability_alert_verifier,
alert_receipt_readback=alert_receipt_readback,
)
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 FIM / vulnerability alert verifier 無效:{exc}",
) from exc
@router.post(
"/api/v1/iwooos/wazuh-fim-vulnerability-alert-verifier/validate-verifier-packet",
response_model=dict[str, Any],
summary="驗證 Wazuh FIM / vulnerability alert verifier 脫敏 packet",
description=(
"針對單次 owner / reviewer 提供的 redacted Wazuh FIM / vulnerability alert verifier "
"packet 進行 no-persist review validation回傳 accepted-for-review / needs supplement / "
"quarantined / rejected runtime action 分流。此端點不保存 payload、不查 Wazuh API、"
"不讀主機、不執行 scanner、不讀或回傳機密明文、不啟用主動回應、不改 Nginx / "
"Docker / K8s / firewall也不更新 verifier 總帳。"
),
)
async def validate_iwooos_wazuh_fim_vulnerability_alert_verifier_packet(
verifier_packet: dict[str, Any],
) -> dict[str, Any]:
"""回傳單次 Wazuh FIM / vulnerability verifier packet 的公開安全驗證結果。"""
try:
payload = await asyncio.to_thread(
validate_wazuh_fim_vulnerability_alert_verifier_packet_payload,
verifier_packet,
)
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 FIM / vulnerability alert verifier packet 驗證器無效:{exc}",
) from exc
@router.get(
"/api/v1/iwooos/wazuh-runtime-gate-owner-review-readback",
response_model=dict[str, Any],

View File

@@ -19,6 +19,9 @@ from src.services.iwooos_security_operating_system import (
from src.services.iwooos_wazuh_allowlisted_check_mode_dry_run import (
load_latest_iwooos_wazuh_allowlisted_check_mode_dry_run,
)
from src.services.iwooos_wazuh_fim_vulnerability_alert_verifier import (
load_latest_iwooos_wazuh_fim_vulnerability_alert_verifier,
)
from src.services.iwooos_wazuh_managed_host_coverage import (
load_latest_iwooos_wazuh_managed_host_coverage,
)
@@ -52,6 +55,9 @@ def load_latest_iwooos_security_tool_closure_readback(
controlled_apply = load_latest_iwooos_wazuh_runtime_controlled_apply_preflight()
owner_review = load_latest_iwooos_wazuh_runtime_gate_owner_review_readback()
dry_run = load_latest_iwooos_wazuh_allowlisted_check_mode_dry_run()
wazuh_verifier = load_latest_iwooos_wazuh_fim_vulnerability_alert_verifier(
alert_receipt_readback=alert_receipt_readback
)
source_payloads = [
coverage,
@@ -61,6 +67,7 @@ def load_latest_iwooos_security_tool_closure_readback(
controlled_apply,
owner_review,
dry_run,
wazuh_verifier,
]
_require_runtime_boundaries(source_payloads)
@@ -72,6 +79,7 @@ def load_latest_iwooos_security_tool_closure_readback(
"controlled_apply": _summary(controlled_apply),
"owner_review": _summary(owner_review),
"dry_run": _summary(dry_run),
"wazuh_verifier": _summary(wazuh_verifier),
}
tracks = _tool_tracks(source_summaries)
summary = _summary_rollup(source_summaries, tracks)
@@ -122,6 +130,7 @@ def load_latest_iwooos_security_tool_closure_readback(
*controlled_apply.get("source_refs", []),
*owner_review.get("source_refs", []),
*dry_run.get("source_refs", []),
*wazuh_verifier.get("source_refs", []),
],
}
@@ -134,6 +143,7 @@ def _tool_tracks(source_summaries: dict[str, dict[str, Any]]) -> list[dict[str,
controlled_apply = source_summaries["controlled_apply"]
owner_review = source_summaries["owner_review"]
dry_run = source_summaries["dry_run"]
wazuh_verifier = source_summaries["wazuh_verifier"]
expected_hosts = max(
_int(managed_hosts.get("expected_host_scope_count")),
@@ -148,6 +158,9 @@ def _tool_tracks(source_summaries: dict[str, dict[str, Any]]) -> list[dict[str,
)
registry_ready = expected_hosts > 0 and accepted_hosts >= expected_hosts
alert_ready = _int(security_os.get("alert_receipt_accepted_count")) > 0
wazuh_verifier_ready_signals = _int(
wazuh_verifier.get("verifier_ready_signal_count")
)
controlled_apply_ready_signals = sum(
1
@@ -196,13 +209,17 @@ def _tool_tracks(source_summaries: dict[str, dict[str, Any]]) -> list[dict[str,
priority="P0",
label="Wazuh detection and response",
tool_ids=["wazuh"],
ready_signal_count=(1 if registry_ready else 0) + (1 if alert_ready else 0),
ready_signal_count=max(
(1 if registry_ready else 0) + (1 if alert_ready else 0),
wazuh_verifier_ready_signals,
),
required_signal_count=6,
next_executable_action="close_fim_vulnerability_alert_receipt_then_keep_active_response_gated",
next_executable_action="attach_incident_case_and_active_response_dry_run_receipts_without_runtime_gate",
required_verifier="wazuh_registry_fim_vulnerability_alert_receipt_post_verifier",
source_refs=[
"docs/security/wazuh-managed-host-coverage-gate.snapshot.json",
"docs/security/iwooos-security-operating-system.snapshot.json",
"apps/api/src/services/iwooos_wazuh_fim_vulnerability_alert_verifier.py",
],
),
_track(
@@ -348,6 +365,7 @@ def _summary_rollup(
controlled_apply = source_summaries["controlled_apply"]
owner_review = source_summaries["owner_review"]
dry_run = source_summaries["dry_run"]
wazuh_verifier = source_summaries["wazuh_verifier"]
expected_hosts = max(
_int(managed_hosts.get("expected_host_scope_count")),
@@ -387,7 +405,15 @@ def _summary_rollup(
"wazuh_expected_host_scope_count": expected_hosts,
"wazuh_manager_registry_accepted_count": accepted_hosts,
"wazuh_registry_complete_count": 1 if expected_hosts > 0 and accepted_hosts >= expected_hosts else 0,
"wazuh_fim_vulnerability_alert_closed_count": 0,
"wazuh_fim_vulnerability_alert_closed_count": _int(
wazuh_verifier.get("wazuh_fim_vulnerability_alert_closed_count")
),
"wazuh_fim_vulnerability_alert_verifier_ready_count": _int(
wazuh_verifier.get("wazuh_fim_vulnerability_alert_verifier_ready_count")
),
"wazuh_fim_vulnerability_alert_verifier_ready_signal_count": _int(
wazuh_verifier.get("verifier_ready_signal_count")
),
"alert_receipt_observed_count": _int(
security_os.get("alert_receipt_observed_count")
),
@@ -425,6 +451,7 @@ def _next_executable_queue(
) -> list[dict[str, Any]]:
managed_hosts = source_summaries["managed_hosts"]
registry = source_summaries["registry"]
wazuh_verifier = source_summaries["wazuh_verifier"]
controlled_apply = source_summaries["controlled_apply"]
dry_run = source_summaries["dry_run"]
expected_hosts = max(
@@ -435,18 +462,28 @@ def _next_executable_queue(
_int(managed_hosts.get("manager_registry_accepted_count")),
_int(registry.get("manager_registry_accepted_count")),
)
registry_gate = (
"wazuh_registry_complete_move_to_fim_vulnerability_alert"
if expected_hosts > 0 and accepted_hosts >= expected_hosts
else "complete_wazuh_manager_registry_alias_matrix"
verifier_closed = (
_int(wazuh_verifier.get("wazuh_fim_vulnerability_alert_closed_count")) > 0
)
if verifier_closed:
registry_gate = "wazuh_fim_vulnerability_alert_verifier_ready_keep_runtime_gated"
wazuh_next_action = "attach_incident_case_and_active_response_dry_run_receipts_without_runtime_gate"
elif expected_hosts > 0 and accepted_hosts >= expected_hosts:
registry_gate = "wazuh_registry_complete_move_to_fim_vulnerability_alert"
wazuh_next_action = "attach_fim_vulnerability_alert_receipt_post_verifier"
else:
registry_gate = "complete_wazuh_manager_registry_alias_matrix"
wazuh_next_action = "complete_wazuh_manager_registry_alias_matrix"
return [
{
"queue_id": "P0-03",
"track_id": "wazuh_detection_response",
"state": registry_gate,
"next_action": "attach_fim_vulnerability_alert_receipt_post_verifier",
"next_action": wazuh_next_action,
"ready_signal_count": _int(
wazuh_verifier.get("verifier_ready_signal_count")
),
"controlled_apply_policy_allowed": True,
},
{

View File

@@ -0,0 +1,675 @@
"""
IwoooS Wazuh FIM / vulnerability alert verifier readback.
This service exposes the next Wazuh P0 closure slice as a public-safe verifier:
registry coverage, FIM/persistence normalization, CVE/KEV prioritization and
Alertmanager receipt readiness. It never queries live Wazuh, reads hosts,
persists payloads, runs scanners, enables active response, or opens runtime
gates.
"""
from __future__ import annotations
import json
import re
from pathlib import Path
from typing import Any
from src.services.iwooos_wazuh_manager_registry_reviewer_validation import (
load_latest_iwooos_wazuh_manager_registry_reviewer_validation,
)
from src.services.snapshot_paths import default_security_dir
_DEFAULT_SECURITY_DIR = default_security_dir(Path(__file__))
_INTRUSION_FILE = "wazuh-iwooos-intrusion-readback-plan.snapshot.json"
_SOC_FILE = "soc-siem-kali-wazuh-integration-control.snapshot.json"
_CONVERGENCE_FILE = "iwooos-p0-security-incident-convergence-gate.snapshot.json"
_EXPECTED_INTRUSION_SCHEMA = "wazuh_iwooos_intrusion_readback_plan_v1"
_EXPECTED_SOC_SCHEMA = "soc_siem_kali_wazuh_integration_control_v1"
_EXPECTED_CONVERGENCE_SCHEMA = "iwooos_p0_security_incident_convergence_gate_v1"
_REQUIRED_FALSE_BOUNDARIES = {
"active_response_authorized",
"active_scan_authorized",
"auto_block_authorized",
"firewall_change_authorized",
"host_write_authorized",
"kali_execute_authorized",
"kali_scan_authorized",
"nginx_reload_authorized",
"production_write_authorized",
"raw_log_storage_allowed",
"raw_payload_storage_allowed",
"raw_wazuh_payload_storage_allowed",
"runtime_execution_authorized",
"runtime_gate_open",
"secret_value_collection_allowed",
"soar_action_authorized",
"telegram_send_authorized",
"wazuh_active_response_authorized",
"wazuh_api_live_query_authorized",
}
_RUNTIME_ACTION_KEYS = {
"active_response_enable",
"agent_reenroll",
"agent_restart",
"ansible_apply",
"apply_now",
"auto_block",
"credentialed_scan",
"docker_restart",
"execute_now",
"exploit_attempt",
"firewall_change",
"force_push",
"host_write",
"kali_active_scan",
"nginx_reload",
"production_write",
"runtime_execution_authorized",
"secret_rotation",
"soar_action",
"systemd_restart",
"telegram_live_send",
"wazuh_active_response",
"wazuh_agent_reenroll",
"wazuh_agent_restart",
"wazuh_api_live_query",
"wazuh_manager_restart",
}
_SENSITIVE_TEXT_PATTERNS = {
"internal_ip": re.compile(
r"\b(?:10|127|172\.(?:1[6-9]|2\d|3[01])|192\.168)\.\d{1,3}\.\d{1,3}\b"
),
"authorization_header": re.compile(r"Authorization\s*:", re.IGNORECASE),
"bearer_token": re.compile(r"Bearer\s+[A-Za-z0-9._-]{10,}", re.IGNORECASE),
"basic_auth": re.compile(r"Basic\s+[A-Za-z0-9+/=]{10,}", re.IGNORECASE),
"password_assignment": re.compile(
r"password\s*[:=]\s*['\"][^'\"]+['\"]", re.IGNORECASE
),
"token_assignment": re.compile(r"token\s*[:=]\s*['\"][^'\"]+['\"]", re.IGNORECASE),
"cookie_assignment": re.compile(
r"cookie\s*[:=]\s*['\"][^'\"]+['\"]", re.IGNORECASE
),
"client_keys": re.compile(r"client\.keys", re.IGNORECASE),
"private_key": re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"),
"raw_session_text": re.compile(
r"(工作視窗|批准!繼續|source_thread_id|raw session)", re.IGNORECASE
),
}
_FORBIDDEN_KEY_FRAGMENTS = {
"authorization_header",
"basic_auth",
"bearer_token",
"client_keys",
"cookie",
"env_file",
"full_cli_output",
"full_journal",
"hostname",
"internal_ip",
"password",
"private_key",
"raw_agent_identity",
"raw_alert_payload",
"raw_env",
"raw_hostname",
"raw_log",
"raw_runtime_volume",
"raw_session",
"raw_wazuh_payload",
"session",
"stored_api_password",
"token",
"unredacted_screenshot",
"wazuh_api_password",
}
_REQUIRED_PACKET_FIELDS = (
"verifier_role",
"decision",
"decision_reason",
"scope_aliases",
"fim_persistence_evidence_refs",
"vulnerability_kev_evidence_refs",
"alert_receipt_ref",
"source_snapshot_refs",
"post_verifier_ref",
"reviewed_at",
"followup_owner",
"rollback_owner",
)
def load_latest_iwooos_wazuh_fim_vulnerability_alert_verifier(
security_dir: Path | None = None,
alert_receipt_readback: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Load the public-safe Wazuh FIM / vulnerability alert verifier readback."""
directory = security_dir or _DEFAULT_SECURITY_DIR
intrusion = _load_snapshot(directory / _INTRUSION_FILE, _EXPECTED_INTRUSION_SCHEMA)
soc = _load_snapshot(directory / _SOC_FILE, _EXPECTED_SOC_SCHEMA)
convergence = _load_snapshot(
directory / _CONVERGENCE_FILE, _EXPECTED_CONVERGENCE_SCHEMA
)
registry = load_latest_iwooos_wazuh_manager_registry_reviewer_validation(directory)
_require_boundaries([intrusion, soc, convergence])
registry_summary = _summary(registry)
intrusion_summary = _summary(intrusion)
soc_summary = _summary(soc)
convergence_summary = _summary(convergence)
alert_summary = _alert_receipt_summary(alert_receipt_readback)
expected_scope_alias_count = _int(registry_summary.get("expected_scope_alias_count"))
manager_registry_accepted_count = _int(
registry_summary.get("manager_registry_accepted_count")
)
registry_complete = (
expected_scope_alias_count > 0
and manager_registry_accepted_count >= expected_scope_alias_count
)
fim_domain = _domain_ready(soc, "fim_persistence_detection")
vulnerability_domain = _domain_ready(soc, "vulnerability_kev_prioritization")
wazuh_correlation_domain = _domain_ready(soc, "wazuh_siem_correlation")
alert_route_gate = _validation_gate_ready(soc, "alert_route_receipt_available")
fim_review = _review_check_ready(soc, "soc_review_14")
vulnerability_review = _review_check_ready(soc, "soc_review_18")
dedupe_review = _review_check_ready(soc, "soc_review_20")
intrusion_candidates = _candidate_count(
intrusion,
{
"wazuh_intrusion_readback:host110_intrusion_signal",
"wazuh_intrusion_readback:host188_intrusion_signal",
"wazuh_intrusion_readback:containment_recovery_postcheck",
},
)
convergence_lanes = _lane_count(
convergence,
{"host_intrusion_forensics", "monitoring_alert_receipt"},
)
fim_signal_ready = fim_domain and fim_review and intrusion_candidates >= 2
vulnerability_signal_ready = vulnerability_domain and vulnerability_review
alert_receipt_ready = (
alert_summary["accepted_count"] > 0
and alert_route_gate
and convergence_lanes >= 2
)
verifier_ready = (
registry_complete
and fim_signal_ready
and vulnerability_signal_ready
and wazuh_correlation_domain
and alert_receipt_ready
and dedupe_review
)
summary = {
"expected_scope_alias_count": expected_scope_alias_count,
"manager_registry_accepted_count": manager_registry_accepted_count,
"wazuh_registry_complete_count": 1 if registry_complete else 0,
"fim_persistence_domain_ready_count": 1 if fim_domain else 0,
"fim_persistence_review_rule_count": 1 if fim_review else 0,
"fim_persistence_signal_ready_count": 1 if fim_signal_ready else 0,
"vulnerability_kev_domain_ready_count": 1 if vulnerability_domain else 0,
"vulnerability_kev_review_rule_count": 1 if vulnerability_review else 0,
"vulnerability_kev_signal_ready_count": 1 if vulnerability_signal_ready else 0,
"wazuh_siem_correlation_domain_ready_count": 1 if wazuh_correlation_domain else 0,
"stable_dedupe_fingerprint_rule_count": 1 if dedupe_review else 0,
"alert_route_receipt_gate_count": 1 if alert_route_gate else 0,
"alert_receipt_observed_count": alert_summary["observed_count"],
"alert_receipt_accepted_count": alert_summary["accepted_count"],
"alert_receipt_source_status": alert_summary["source_status"],
"alert_receipt_ready_count": 1 if alert_receipt_ready else 0,
"intrusion_signal_candidate_count": intrusion_candidates,
"p0_convergence_lane_count": convergence_lanes,
"verifier_required_signal_count": 6,
"verifier_ready_signal_count": sum(
(
1 if registry_complete else 0,
1 if fim_signal_ready else 0,
1 if vulnerability_signal_ready else 0,
1 if wazuh_correlation_domain else 0,
1 if alert_receipt_ready else 0,
1 if dedupe_review else 0,
)
),
"wazuh_fim_vulnerability_alert_verifier_ready_count": 1
if verifier_ready
else 0,
"wazuh_fim_vulnerability_alert_closed_count": 1 if verifier_ready else 0,
"wazuh_event_accepted_count": _int(
intrusion_summary.get("wazuh_event_accepted_count")
)
+ _int(soc_summary.get("wazuh_event_ref_received_count")),
"forensic_evidence_accepted_count": _int(
soc_summary.get("forensic_evidence_accepted_count")
)
+ _int(intrusion_summary.get("host_forensics_accepted_count")),
"incident_case_accepted_count": _int(
soc_summary.get("incident_case_accepted_count")
)
+ _int(convergence_summary.get("incident_case_accepted_count")),
"active_response_authorized_count": 0,
"runtime_gate_count": 0,
"host_write_authorized_count": 0,
"secret_value_collection_allowed_count": 0,
}
return {
"schema_version": "iwooos_wazuh_fim_vulnerability_alert_verifier_readback_v1",
"status": (
"wazuh_fim_vulnerability_alert_verifier_ready_no_runtime_action"
if verifier_ready
else "wazuh_fim_vulnerability_alert_verifier_waiting_alert_receipt_or_refs"
),
"mode": "committed_verifier_readback_with_alert_receipt_no_live_wazuh_no_runtime",
"summary": summary,
"verifier_items": _verifier_items(summary),
"verifier_packet_validation_endpoint": (
"/api/v1/iwooos/wazuh-fim-vulnerability-alert-verifier/validate-verifier-packet"
),
"verifier_packet_validation_mode": (
"no_persist_verifier_packet_review_no_runtime_no_secret_collection"
),
"required_verifier_packet_fields": list(_REQUIRED_PACKET_FIELDS),
"boundary_markers": _boundary_markers(summary),
"boundaries": _boundaries(),
"no_false_green_rules": [
"verifier ready means committed redacted readback and alert receipt can be reviewed; it does not prove live incident eradication",
"wazuh event accepted, forensic evidence accepted and incident case accepted remain separate counters",
"active response, host containment, firewall changes and scanner execution remain closed by this endpoint",
"readback endpoint never queries live Wazuh and never persists submitted verifier packets",
],
"source_refs": [
f"docs/security/{_INTRUSION_FILE}",
f"docs/security/{_SOC_FILE}",
f"docs/security/{_CONVERGENCE_FILE}",
"apps/api/src/services/iwooos_wazuh_fim_vulnerability_alert_verifier.py",
],
}
def validate_iwooos_wazuh_fim_vulnerability_alert_verifier_packet(
verifier_packet: dict[str, Any],
security_dir: Path | None = None,
) -> dict[str, Any]:
"""Validate one redacted verifier packet without persisting or executing it."""
contract = load_latest_iwooos_wazuh_fim_vulnerability_alert_verifier(security_dir)
findings: list[dict[str, Any]] = []
if not isinstance(verifier_packet, dict):
findings.append(
_finding(
"WVA-01",
"blocker",
"request_verifier_packet_supplement",
"verifier packet must be a JSON object.",
[],
)
)
return _validation_result(contract, "request_verifier_packet_supplement", findings)
sensitive_hits = _collect_sensitive_hits(verifier_packet)
if sensitive_hits:
findings.append(
_finding(
"WVA-04",
"critical",
"quarantine_sensitive_payload",
"verifier packet contains forbidden or likely unredacted content; response omits raw values.",
[hit["path"] for hit in sensitive_hits[:12]],
{"categories": sorted({hit["category"] for hit in sensitive_hits})},
)
)
return _validation_result(contract, "quarantine_sensitive_payload", findings)
runtime_hits = _collect_runtime_action_hits(verifier_packet)
if runtime_hits:
findings.append(
_finding(
"WVA-05",
"critical",
"reject_runtime_action_request",
"verifier packet requested runtime or destructive action; this validator is readback-only.",
runtime_hits[:12],
)
)
return _validation_result(contract, "reject_runtime_action_request", findings)
missing_fields = [
field
for field in _REQUIRED_PACKET_FIELDS
if field not in verifier_packet or verifier_packet.get(field) in (None, "", [])
]
if missing_fields:
findings.append(
_finding(
"WVA-02",
"blocker",
"request_verifier_packet_supplement",
"required verifier packet fields are missing.",
missing_fields,
)
)
return _validation_result(contract, "request_verifier_packet_supplement", findings)
list_fields = (
"scope_aliases",
"fim_persistence_evidence_refs",
"vulnerability_kev_evidence_refs",
"source_snapshot_refs",
)
invalid_lists = [
field
for field in list_fields
if not _strings(verifier_packet.get(field))
]
if invalid_lists:
findings.append(
_finding(
"WVA-03",
"blocker",
"request_verifier_packet_supplement",
"list fields must contain at least one redacted reference.",
invalid_lists,
)
)
return _validation_result(contract, "request_verifier_packet_supplement", findings)
return _validation_result(
contract,
"accepted_for_wazuh_fim_vulnerability_alert_verifier_review_only",
findings,
)
def _load_snapshot(path: Path, expected_schema: str) -> dict[str, Any]:
if not path.is_file():
raise FileNotFoundError(f"{path}: Wazuh verifier source snapshot does not exist")
with path.open(encoding="utf-8") as handle:
payload = json.load(handle)
if not isinstance(payload, dict):
raise ValueError(f"{path}: expected JSON object")
if payload.get("schema_version") != expected_schema:
raise ValueError(f"{path}: expected schema_version={expected_schema}")
return payload
def _require_boundaries(payloads: list[dict[str, Any]]) -> None:
for payload in payloads:
boundaries = payload.get("execution_boundaries")
if not isinstance(boundaries, dict):
raise ValueError(f"{payload.get('schema_version')}: execution_boundaries missing")
for key in _REQUIRED_FALSE_BOUNDARIES:
if key in boundaries and boundaries.get(key) is not False:
raise ValueError(
f"{payload.get('schema_version')}: execution_boundaries.{key} must remain false"
)
def _summary(payload: dict[str, Any]) -> dict[str, Any]:
summary = payload.get("summary")
return summary if isinstance(summary, dict) else {}
def _domain_ready(payload: dict[str, Any], domain_id: str) -> bool:
for item in _dicts(payload.get("control_domains")):
if item.get("domain_id") == domain_id:
return item.get("runtime_gate_open") is False
return False
def _validation_gate_ready(payload: dict[str, Any], gate_id: str) -> bool:
for item in _dicts(payload.get("validation_gates")):
if item.get("gate_id") == gate_id:
return item.get("runtime_gate_open") is False
return False
def _review_check_ready(payload: dict[str, Any], check_id: str) -> bool:
return any(item.get("check_id") == check_id for item in _dicts(payload.get("reviewer_checks")))
def _candidate_count(payload: dict[str, Any], candidate_ids: set[str]) -> int:
return sum(
1
for item in _dicts(payload.get("readback_candidates"))
if item.get("readback_candidate_id") in candidate_ids
)
def _lane_count(payload: dict[str, Any], lane_ids: set[str]) -> int:
return sum(
1
for item in _dicts(payload.get("p0_lanes"))
if item.get("lane_id") in lane_ids
)
def _alert_receipt_summary(readback: dict[str, Any] | None) -> dict[str, Any]:
if not isinstance(readback, dict):
return {
"accepted_count": 0,
"observed_count": 0,
"source_status": "not_connected",
"source_error": "",
}
events = readback.get("events")
observed_count = len(events) if isinstance(events, list) else _int(readback.get("total"))
source_status = str(readback.get("source_status") or "not_connected")
return {
"accepted_count": 1 if observed_count > 0 and source_status == "ready" else 0,
"observed_count": observed_count,
"source_status": source_status,
"source_error": str(readback.get("source_error") or ""),
}
def _verifier_items(summary: dict[str, Any]) -> list[dict[str, Any]]:
return [
_verifier_item(
"manager_registry_complete",
summary["wazuh_registry_complete_count"],
"manager registry accepted for expected aliases",
),
_verifier_item(
"fim_persistence_signal_normalized",
summary["fim_persistence_signal_ready_count"],
"FIM / persistence domain and reviewer rule are present",
),
_verifier_item(
"vulnerability_kev_prioritized",
summary["vulnerability_kev_signal_ready_count"],
"CVE / KEV domain and owner/SLA reviewer rule are present",
),
_verifier_item(
"wazuh_siem_correlation_ready",
summary["wazuh_siem_correlation_domain_ready_count"],
"Wazuh SIEM correlation domain is present",
),
_verifier_item(
"alert_receipt_ready",
summary["alert_receipt_ready_count"],
"Alertmanager receipt is readable and tied to the convergence gate",
),
_verifier_item(
"stable_dedupe_fingerprint_ready",
summary["stable_dedupe_fingerprint_rule_count"],
"stable dedupe fingerprint reviewer rule is present",
),
]
def _verifier_item(item_id: str, ready_count: int, label: str) -> dict[str, Any]:
return {
"item_id": item_id,
"label": label,
"ready": ready_count > 0,
"ready_signal_count": ready_count,
"required_signal_count": 1,
}
def _validation_result(
contract: dict[str, Any],
status: str,
findings: list[dict[str, Any]],
) -> dict[str, Any]:
accepted = status == "accepted_for_wazuh_fim_vulnerability_alert_verifier_review_only"
quarantined = status == "quarantine_sensitive_payload"
runtime_rejected = status == "reject_runtime_action_request"
return {
"schema_version": "iwooos_wazuh_fim_vulnerability_alert_verifier_validation_result_v1",
"status": status,
"mode": "no_persist_verifier_packet_review_no_runtime_no_secret_collection",
"accepted_for_verifier_review_only": accepted,
"quarantined": quarantined,
"runtime_action_rejected": runtime_rejected,
"summary": {
"verifier_packet_received_count": 1,
"verifier_packet_review_ready_count": 1 if accepted else 0,
"verifier_packet_supplement_required_count": 1
if status == "request_verifier_packet_supplement"
else 0,
"verifier_packet_quarantined_count": 1 if quarantined else 0,
"verifier_packet_runtime_action_rejected_count": 1 if runtime_rejected else 0,
"wazuh_fim_vulnerability_alert_closed_count": 0,
"runtime_gate_count": 0,
"secret_value_collection_allowed_count": 0,
},
"validation_findings": findings,
"next_gate": (
"commit_wazuh_fim_vulnerability_alert_verifier_readback"
if accepted
else status
),
"source_contract_status": contract["status"],
"boundaries": {
"payload_persisted": False,
"runtime_execution_authorized": False,
"wazuh_api_live_query_authorized": False,
"wazuh_active_response_authorized": False,
"host_write_authorized": False,
"secret_value_collection_allowed": False,
"verifier_readback_updated": False,
"not_authorization": True,
},
}
def _finding(
check_id: str,
severity: str,
outcome: str,
message: str,
field_paths: list[str],
metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
return {
"check_id": check_id,
"severity": severity,
"outcome": outcome,
"message": message,
"field_paths": field_paths,
"metadata": metadata or {},
}
def _collect_sensitive_hits(value: Any, path: str = "$") -> list[dict[str, str]]:
hits: list[dict[str, str]] = []
if isinstance(value, dict):
for key, item in value.items():
key_text = str(key)
child_path = f"{path}.{key_text}"
lowered_key = key_text.lower()
for fragment in _FORBIDDEN_KEY_FRAGMENTS:
if fragment in lowered_key:
hits.append({"path": child_path, "category": "forbidden_key"})
break
hits.extend(_collect_sensitive_hits(item, child_path))
return hits
if isinstance(value, list):
for index, item in enumerate(value):
hits.extend(_collect_sensitive_hits(item, f"{path}[{index}]"))
return hits
if isinstance(value, str):
for category, pattern in _SENSITIVE_TEXT_PATTERNS.items():
if pattern.search(value):
hits.append({"path": path, "category": category})
return hits
def _collect_runtime_action_hits(value: Any, path: str = "$") -> list[str]:
hits: list[str] = []
if isinstance(value, dict):
for key, item in value.items():
key_text = str(key)
lowered_key = key_text.lower()
if any(action in lowered_key for action in _RUNTIME_ACTION_KEYS):
hits.append(f"{path}.{key_text}")
hits.extend(_collect_runtime_action_hits(item, f"{path}.{key_text}"))
return hits
if isinstance(value, list):
for index, item in enumerate(value):
hits.extend(_collect_runtime_action_hits(item, f"{path}[{index}]"))
return hits
if isinstance(value, str):
lowered_value = value.lower()
if any(action in lowered_value for action in _RUNTIME_ACTION_KEYS):
hits.append(path)
return hits
def _boundary_markers(summary: dict[str, Any]) -> list[str]:
return [
"wazuh_fim_vulnerability_alert_verifier_visible=true",
f"wazuh_fim_vulnerability_alert_verifier_ready_signal_count={summary['verifier_ready_signal_count']}",
f"wazuh_fim_vulnerability_alert_verifier_required_signal_count={summary['verifier_required_signal_count']}",
f"wazuh_fim_vulnerability_alert_closed_count={summary['wazuh_fim_vulnerability_alert_closed_count']}",
f"wazuh_fim_persistence_signal_ready_count={summary['fim_persistence_signal_ready_count']}",
f"wazuh_vulnerability_kev_signal_ready_count={summary['vulnerability_kev_signal_ready_count']}",
f"wazuh_alert_receipt_ready_count={summary['alert_receipt_ready_count']}",
"wazuh_fim_vulnerability_alert_runtime_gate_count=0",
"wazuh_fim_vulnerability_alert_active_response_authorized_count=0",
"wazuh_fim_vulnerability_alert_secret_value_collection_allowed_count=0",
]
def _boundaries() -> dict[str, Any]:
return {
"payload_persisted": False,
"live_host_query_performed": False,
"live_wazuh_query_performed": False,
"scanner_execution_performed": False,
"runtime_execution_performed": False,
"runtime_execution_authorized": False,
"runtime_gate_open": False,
"wazuh_api_live_query_authorized": False,
"wazuh_active_response_authorized": False,
"host_write_authorized": False,
"firewall_change_authorized": False,
"secret_value_collection_allowed": False,
"readback_endpoint_is_executor": False,
"not_authorization": True,
}
def _dicts(value: Any) -> list[dict[str, Any]]:
if not isinstance(value, list):
return []
return [item for item in value if isinstance(item, dict)]
def _strings(value: Any) -> list[str]:
if not isinstance(value, list):
return []
return [item for item in value if isinstance(item, str) and item]
def _int(value: Any) -> int:
return value if isinstance(value, int) else 0

View File

@@ -22,13 +22,21 @@ def test_iwooos_security_tool_closure_readback_rolls_up_tool_tracks() -> None:
assert payload["status"] == (
"tool_closure_readback_ready_controlled_apply_policy_active"
)
assert payload["summary"]["source_readback_count"] == 7
assert payload["summary"]["source_readback_count"] == 8
assert payload["summary"]["tool_track_count"] == 8
assert payload["summary"]["p0_tool_track_count"] == 3
assert payload["summary"]["wazuh_expected_host_scope_count"] == 6
assert payload["summary"]["wazuh_manager_registry_accepted_count"] == 6
assert payload["summary"]["wazuh_registry_complete_count"] == 1
assert payload["summary"]["wazuh_fim_vulnerability_alert_closed_count"] == 0
assert (
payload["summary"]["wazuh_fim_vulnerability_alert_verifier_ready_count"]
== 0
)
assert (
payload["summary"]["wazuh_fim_vulnerability_alert_verifier_ready_signal_count"]
== 5
)
assert payload["summary"]["supply_chain_scan_closed_count"] == 0
assert payload["summary"]["runtime_detection_closed_count"] == 0
assert payload["summary"]["web_api_dast_closed_count"] == 0
@@ -85,8 +93,8 @@ def test_iwooos_security_tool_closure_readback_marks_partial_vs_missing_tracks()
assert tracks["wazuh_detection_response"]["closure_state"] == (
"partial_readback_missing_verifier"
)
assert tracks["wazuh_detection_response"]["ready_signal_count"] == 1
assert tracks["wazuh_detection_response"]["missing_signal_count"] == 5
assert tracks["wazuh_detection_response"]["ready_signal_count"] == 5
assert tracks["wazuh_detection_response"]["missing_signal_count"] == 1
assert tracks["supply_chain_sbom_scan"]["closure_state"] == "required_not_closed"
assert tracks["runtime_threat_detection"]["tool_ids"] == ["falco"]
assert tracks["web_api_dast"]["tool_ids"] == ["owasp_zap_or_equivalent_dast"]
@@ -107,7 +115,16 @@ def test_iwooos_security_tool_closure_readback_accepts_alert_receipt_signal() ->
assert payload["summary"]["alert_receipt_accepted_count"] == 1
assert payload["summary"]["alert_receipt_observed_count"] == 1
assert tracks["wazuh_detection_response"]["ready_signal_count"] == 2
assert payload["summary"]["wazuh_fim_vulnerability_alert_closed_count"] == 1
assert (
payload["summary"]["wazuh_fim_vulnerability_alert_verifier_ready_count"]
== 1
)
assert tracks["wazuh_detection_response"]["ready_signal_count"] == 6
assert tracks["wazuh_detection_response"]["missing_signal_count"] == 0
assert tracks["wazuh_detection_response"]["closure_state"] == (
"ready_for_controlled_apply_candidate"
)
assert tracks["normalized_detection_schema"]["ready_signal_count"] == 1
assert payload["summary"]["controlled_apply_policy_active_count"] == 1
@@ -133,6 +150,7 @@ def test_iwooos_security_tool_closure_readback_api_is_public_safe(monkeypatch) -
assert data["schema_version"] == "iwooos_security_tool_closure_readback_v1"
assert data["summary"]["tool_track_count"] == 8
assert data["summary"]["alert_receipt_accepted_count"] == 1
assert data["summary"]["wazuh_fim_vulnerability_alert_closed_count"] == 1
assert data["summary"]["controlled_apply_policy_active_count"] == 1
assert data["boundaries"]["controlled_apply_policy_active"] is True
assert any(

View File

@@ -0,0 +1,186 @@
from __future__ import annotations
from fastapi import FastAPI
from fastapi.testclient import TestClient
from src.api.v1.iwooos import router
from src.services.iwooos_wazuh_fim_vulnerability_alert_verifier import (
load_latest_iwooos_wazuh_fim_vulnerability_alert_verifier,
validate_iwooos_wazuh_fim_vulnerability_alert_verifier_packet,
)
def _client() -> TestClient:
app = FastAPI()
app.include_router(router)
return TestClient(app)
def _alert_receipt() -> dict[str, object]:
return {
"events": [{"provider_event_id": "alertmanager:received:wazuh-verifier"}],
"total": 1,
"limit": 5,
"source_status": "ready",
}
def _valid_verifier_packet() -> dict[str, object]:
return {
"verifier_role": "IwoooS reviewer",
"decision": "accept_wazuh_fim_vulnerability_alert_verifier_for_review",
"decision_reason": "FIM persistence, CVE KEV and alert receipt refs are redacted and reviewable.",
"scope_aliases": ["managed_core_node_a", "managed_core_node_b"],
"fim_persistence_evidence_refs": ["evidence-ref-fim-persistence"],
"vulnerability_kev_evidence_refs": ["evidence-ref-kev-owner-sla"],
"alert_receipt_ref": "evidence-ref-alertmanager-receipt",
"source_snapshot_refs": [
"docs/security/wazuh-iwooos-intrusion-readback-plan.snapshot.json",
"docs/security/soc-siem-kali-wazuh-integration-control.snapshot.json",
],
"post_verifier_ref": "evidence-ref-wazuh-post-verifier",
"reviewed_at": "2026-07-10T16:20:00+08:00",
"followup_owner": "IwoooS reviewer",
"rollback_owner": "IwoooS runtime owner",
}
def test_iwooos_wazuh_fim_vulnerability_alert_verifier_waits_for_alert_receipt() -> None:
payload = load_latest_iwooos_wazuh_fim_vulnerability_alert_verifier()
assert (
payload["schema_version"]
== "iwooos_wazuh_fim_vulnerability_alert_verifier_readback_v1"
)
assert (
payload["status"]
== "wazuh_fim_vulnerability_alert_verifier_waiting_alert_receipt_or_refs"
)
assert payload["summary"]["wazuh_registry_complete_count"] == 1
assert payload["summary"]["fim_persistence_signal_ready_count"] == 1
assert payload["summary"]["vulnerability_kev_signal_ready_count"] == 1
assert payload["summary"]["wazuh_siem_correlation_domain_ready_count"] == 1
assert payload["summary"]["stable_dedupe_fingerprint_rule_count"] == 1
assert payload["summary"]["alert_receipt_ready_count"] == 0
assert payload["summary"]["verifier_ready_signal_count"] == 5
assert payload["summary"]["wazuh_fim_vulnerability_alert_closed_count"] == 0
assert payload["summary"]["runtime_gate_count"] == 0
assert payload["boundaries"]["runtime_execution_authorized"] is False
assert payload["boundaries"]["wazuh_active_response_authorized"] is False
assert payload["boundaries"]["secret_value_collection_allowed"] is False
def test_iwooos_wazuh_fim_vulnerability_alert_verifier_accepts_alert_receipt_signal() -> None:
payload = load_latest_iwooos_wazuh_fim_vulnerability_alert_verifier(
alert_receipt_readback=_alert_receipt()
)
assert (
payload["status"]
== "wazuh_fim_vulnerability_alert_verifier_ready_no_runtime_action"
)
assert payload["summary"]["alert_receipt_accepted_count"] == 1
assert payload["summary"]["alert_receipt_ready_count"] == 1
assert payload["summary"]["verifier_ready_signal_count"] == 6
assert (
payload["summary"]["wazuh_fim_vulnerability_alert_verifier_ready_count"]
== 1
)
assert payload["summary"]["wazuh_fim_vulnerability_alert_closed_count"] == 1
assert all(item["ready"] is True for item in payload["verifier_items"])
def test_iwooos_wazuh_fim_vulnerability_alert_verifier_api_is_public_safe(
monkeypatch,
) -> None:
async def fake_recent_events(**_: object) -> dict[str, object]:
return _alert_receipt()
monkeypatch.setattr(
"src.api.v1.iwooos.list_recent_channel_events",
fake_recent_events,
)
response = _client().get("/api/v1/iwooos/wazuh-fim-vulnerability-alert-verifier")
assert response.status_code == 200
data = response.json()
assert (
data["schema_version"]
== "iwooos_wazuh_fim_vulnerability_alert_verifier_readback_v1"
)
assert data["summary"]["wazuh_fim_vulnerability_alert_closed_count"] == 1
assert data["summary"]["active_response_authorized_count"] == 0
assert data["summary"]["runtime_gate_count"] == 0
assert data["boundaries"]["readback_endpoint_is_executor"] is False
assert any(
marker == "wazuh_fim_vulnerability_alert_closed_count=1"
for marker in data["boundary_markers"]
)
assert "192.168.0." not in response.text
assert "工作視窗" not in response.text
assert "批准!繼續" not in response.text
assert "source_thread_id" not in response.text
assert "WAZUH_API_PASSWORD" not in response.text
def test_iwooos_wazuh_fim_vulnerability_alert_verifier_packet_validation_accepts_redacted_packet() -> None:
payload = validate_iwooos_wazuh_fim_vulnerability_alert_verifier_packet(
_valid_verifier_packet()
)
assert (
payload["schema_version"]
== "iwooos_wazuh_fim_vulnerability_alert_verifier_validation_result_v1"
)
assert payload["status"] == (
"accepted_for_wazuh_fim_vulnerability_alert_verifier_review_only"
)
assert payload["accepted_for_verifier_review_only"] is True
assert payload["summary"]["verifier_packet_review_ready_count"] == 1
assert payload["summary"]["wazuh_fim_vulnerability_alert_closed_count"] == 0
assert payload["summary"]["runtime_gate_count"] == 0
assert payload["boundaries"]["payload_persisted"] is False
assert payload["boundaries"]["verifier_readback_updated"] is False
assert payload["boundaries"]["runtime_execution_authorized"] is False
def test_iwooos_wazuh_fim_vulnerability_alert_verifier_packet_validation_requests_missing_fields() -> None:
candidate = _valid_verifier_packet()
candidate.pop("post_verifier_ref")
payload = validate_iwooos_wazuh_fim_vulnerability_alert_verifier_packet(candidate)
assert payload["status"] == "request_verifier_packet_supplement"
assert payload["accepted_for_verifier_review_only"] is False
assert payload["summary"]["verifier_packet_supplement_required_count"] == 1
assert any(
"post_verifier_ref" in finding["field_paths"]
for finding in payload["validation_findings"]
)
def test_iwooos_wazuh_fim_vulnerability_alert_verifier_packet_validation_quarantines_sensitive_payload() -> None:
candidate = _valid_verifier_packet()
candidate["alert_receipt_ref"] = "receipt accidentally included 10.250.250.250"
payload = validate_iwooos_wazuh_fim_vulnerability_alert_verifier_packet(candidate)
assert payload["status"] == "quarantine_sensitive_payload"
assert payload["quarantined"] is True
assert payload["summary"]["verifier_packet_quarantined_count"] == 1
assert "10.250.250.250" not in str(payload)
assert any(finding["check_id"] == "WVA-04" for finding in payload["validation_findings"])
def test_iwooos_wazuh_fim_vulnerability_alert_verifier_packet_validation_rejects_runtime_action_request() -> None:
candidate = _valid_verifier_packet()
candidate["requested_actions"] = ["wazuh_active_response"]
payload = validate_iwooos_wazuh_fim_vulnerability_alert_verifier_packet(candidate)
assert payload["status"] == "reject_runtime_action_request"
assert payload["runtime_action_rejected"] is True
assert payload["summary"]["verifier_packet_runtime_action_rejected_count"] == 1
assert payload["summary"]["runtime_gate_count"] == 0
assert any(finding["check_id"] == "WVA-05" for finding in payload["validation_findings"])