diff --git a/apps/api/src/api/v1/iwooos.py b/apps/api/src/api/v1/iwooos.py index 9a1124bda..25e5f0c37 100644 --- a/apps/api/src/api/v1/iwooos.py +++ b/apps/api/src/api/v1/iwooos.py @@ -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], diff --git a/apps/api/src/services/iwooos_security_tool_closure_readback.py b/apps/api/src/services/iwooos_security_tool_closure_readback.py index 591f949fe..9d83d35f9 100644 --- a/apps/api/src/services/iwooos_security_tool_closure_readback.py +++ b/apps/api/src/services/iwooos_security_tool_closure_readback.py @@ -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, }, { diff --git a/apps/api/src/services/iwooos_wazuh_fim_vulnerability_alert_verifier.py b/apps/api/src/services/iwooos_wazuh_fim_vulnerability_alert_verifier.py new file mode 100644 index 000000000..8ad0cb1cf --- /dev/null +++ b/apps/api/src/services/iwooos_wazuh_fim_vulnerability_alert_verifier.py @@ -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 diff --git a/apps/api/tests/test_iwooos_security_tool_closure_readback.py b/apps/api/tests/test_iwooos_security_tool_closure_readback.py index ca15f605b..b580ddfb4 100644 --- a/apps/api/tests/test_iwooos_security_tool_closure_readback.py +++ b/apps/api/tests/test_iwooos_security_tool_closure_readback.py @@ -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( diff --git a/apps/api/tests/test_iwooos_wazuh_fim_vulnerability_alert_verifier.py b/apps/api/tests/test_iwooos_wazuh_fim_vulnerability_alert_verifier.py new file mode 100644 index 000000000..45ca000c1 --- /dev/null +++ b/apps/api/tests/test_iwooos_wazuh_fim_vulnerability_alert_verifier.py @@ -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"])