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

feat(iwooos): add wazuh incident closure contract
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
Details
CD Pipeline / cancel-stale-cd (push) Has been skipped
Details
CD Pipeline / build-and-deploy (push) Has been cancelled
Details
CD Pipeline / post-deploy-checks (push) Has been cancelled
Details
CD Pipeline / tests (push) Has been cancelled
Details

Browse Source
This commit is contained in:
ogt
2026-07-10 20:53:15 +08:00
parent 2d2349a292
commit 98557b01ed
9 changed files with 1030 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -49,6 +49,12 @@ from src.services.iwooos_wazuh_fim_vulnerability_alert_verifier import (
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_incident_response_closure import (
load_latest_iwooos_wazuh_incident_response_closure,
)
from src.services.iwooos_wazuh_incident_response_closure import (
validate_iwooos_wazuh_incident_response_closure_packet as validate_wazuh_incident_response_closure_packet_payload,
)
from src.services.iwooos_wazuh_live_metadata_gate import (
load_latest_iwooos_wazuh_live_metadata_gate,
)
@@ -680,6 +686,75 @@ async def validate_iwooos_wazuh_fim_vulnerability_alert_verifier_packet(
) from exc
@router.get(
"/api/v1/iwooos/wazuh-incident-response-closure-readback",
response_model=dict[str, Any],
summary="取得 Wazuh incident response same-run closure 讀回",
description=(
"彙整 Wazuh FIM / vulnerability alert verifier、controlled apply preflight、"
"allowlisted check-mode dry-run，形成 P0-03 Wazuh incident response same-run "
"closure packet contract。此端點不查 live Wazuh API、不讀主機、不保存 payload、"
"不讀或回傳機密明文、不啟用主動回應、不改 Nginx / Docker / K8s / firewall。"
),
)
async def get_iwooos_wazuh_incident_response_closure_readback() -> dict[str, Any]:
"""回傳 Wazuh incident response same-run closure 公開安全讀回。"""
try:
alert_receipt_readback = await _alertmanager_receipt_readback()
payload = await asyncio.to_thread(
load_latest_iwooos_wazuh_incident_response_closure,
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 incident response closure 讀回無效：{exc}",
) from exc
@router.post(
"/api/v1/iwooos/wazuh-incident-response-closure/validate-incident-response-packet",
response_model=dict[str, Any],
summary="驗證 Wazuh incident response same-run closure 脫敏 packet",
description=(
"針對單次 redacted Wazuh incident response same-run closure packet 進行 no-persist "
"receipt validation，檢查 trace_id、event receipt、forensics、incident case、"
"controlled response candidate、check-mode dry-run、bounded execution receipt、"
"post-verifier、KM / RAG / MCP / PlayBook writeback 與邊界 ack。此端點不保存 payload、"
"不查 live Wazuh API、不讀主機、不讀或回傳機密明文、不啟用主動回應、不改 Nginx / "
"Docker / K8s / firewall，也不更新 runtime gate 總帳。"
),
)
async def validate_iwooos_wazuh_incident_response_closure_packet(
incident_response_packet: dict[str, Any],
) -> dict[str, Any]:
"""回傳單次 Wazuh incident response same-run packet 的公開安全驗證結果。"""
try:
alert_receipt_readback = await _alertmanager_receipt_readback()
payload = await asyncio.to_thread(
validate_wazuh_incident_response_closure_packet_payload,
incident_response_packet,
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 incident response closure packet 驗證器無效：{exc}",
) from exc
@router.get(
"/api/v1/iwooos/wazuh-runtime-gate-owner-review-readback",
response_model=dict[str, Any],

76
apps/api/src/services/iwooos_security_tool_closure_readback.py
View File

@@ -22,6 +22,9 @@ from src.services.iwooos_wazuh_allowlisted_check_mode_dry_run import (
from src.services.iwooos_wazuh_fim_vulnerability_alert_verifier import (
load_latest_iwooos_wazuh_fim_vulnerability_alert_verifier,
)
from src.services.iwooos_wazuh_incident_response_closure import (
load_latest_iwooos_wazuh_incident_response_closure,
)
from src.services.iwooos_wazuh_managed_host_coverage import (
load_latest_iwooos_wazuh_managed_host_coverage,
)
@@ -58,6 +61,11 @@ def load_latest_iwooos_security_tool_closure_readback(
wazuh_verifier = load_latest_iwooos_wazuh_fim_vulnerability_alert_verifier(
alert_receipt_readback=alert_receipt_readback
)
wazuh_incident_response_closure = (
load_latest_iwooos_wazuh_incident_response_closure(
alert_receipt_readback=alert_receipt_readback
)
)
source_payloads = [
coverage,
@@ -68,6 +76,7 @@ def load_latest_iwooos_security_tool_closure_readback(
owner_review,
dry_run,
wazuh_verifier,
wazuh_incident_response_closure,
]
_require_runtime_boundaries(source_payloads)
@@ -80,6 +89,9 @@ def load_latest_iwooos_security_tool_closure_readback(
"owner_review": _summary(owner_review),
"dry_run": _summary(dry_run),
"wazuh_verifier": _summary(wazuh_verifier),
"wazuh_incident_response_closure": _summary(
wazuh_incident_response_closure
),
}
tracks = _tool_tracks(source_summaries)
summary = _summary_rollup(source_summaries, tracks)
@@ -132,6 +144,7 @@ def load_latest_iwooos_security_tool_closure_readback(
*owner_review.get("source_refs", []),
*dry_run.get("source_refs", []),
*wazuh_verifier.get("source_refs", []),
*wazuh_incident_response_closure.get("source_refs", []),
],
}
@@ -145,6 +158,9 @@ def _tool_tracks(source_summaries: dict[str, dict[str, Any]]) -> list[dict[str,
owner_review = source_summaries["owner_review"]
dry_run = source_summaries["dry_run"]
wazuh_verifier = source_summaries["wazuh_verifier"]
wazuh_incident_response_closure = source_summaries[
"wazuh_incident_response_closure"
]
expected_hosts = max(
_int(managed_hosts.get("expected_host_scope_count")),
@@ -385,6 +401,9 @@ def _summary_rollup(
owner_review = source_summaries["owner_review"]
dry_run = source_summaries["dry_run"]
wazuh_verifier = source_summaries["wazuh_verifier"]
wazuh_incident_response_closure = source_summaries[
"wazuh_incident_response_closure"
]
expected_hosts = max(
_int(managed_hosts.get("expected_host_scope_count")),
@@ -460,6 +479,27 @@ def _summary_rollup(
"wazuh_fim_vulnerability_alert_verifier_ready_signal_count": _int(
wazuh_verifier.get("verifier_ready_signal_count")
),
"wazuh_incident_response_closure_validator_available_count": _int(
wazuh_incident_response_closure.get(
"closure_packet_validator_available_count"
)
),
"wazuh_incident_response_same_run_stage_ready_count": _int(
wazuh_incident_response_closure.get("same_run_source_stage_ready_count")
),
"wazuh_incident_response_same_run_required_stage_count": _int(
wazuh_incident_response_closure.get("same_run_required_stage_count")
),
"wazuh_incident_response_same_run_candidate_ready_count": _int(
wazuh_incident_response_closure.get(
"same_run_closure_candidate_ready_count"
)
),
"wazuh_incident_response_runtime_closed_count": _int(
wazuh_incident_response_closure.get(
"wazuh_incident_response_runtime_closed_count"
)
),
"alert_receipt_observed_count": _int(
security_os.get("alert_receipt_observed_count")
),
@@ -500,6 +540,9 @@ def _next_executable_queue(
managed_hosts = source_summaries["managed_hosts"]
registry = source_summaries["registry"]
wazuh_verifier = source_summaries["wazuh_verifier"]
wazuh_incident_response_closure = source_summaries[
"wazuh_incident_response_closure"
]
controlled_apply = source_summaries["controlled_apply"]
dry_run = source_summaries["dry_run"]
expected_hosts = max(
@@ -526,9 +569,22 @@ def _next_executable_queue(
)
> 0
)
same_run_candidate_ready = (
_int(
wazuh_incident_response_closure.get(
"same_run_closure_candidate_ready_count"
)
)
> 0
)
if runtime_closed:
registry_gate = "wazuh_detection_response_runtime_closed_monitor_recurrence"
wazuh_next_action = "monitor_recurrence_and_control_freshness_slo"
elif same_run_candidate_ready:
registry_gate = "wazuh_same_run_closure_candidate_ready_runtime_open"
wazuh_next_action = (
"submit_redacted_same_run_wazuh_incident_response_closure_packet"
)
elif source_verifier_ready:
registry_gate = "wazuh_source_verifier_ready_runtime_closure_open"
wazuh_next_action = (
@@ -558,6 +614,26 @@ def _next_executable_queue(
0,
),
"required_verifier": "wazuh_registry_fim_vulnerability_alert_receipt_post_verifier",
"same_run_required_stage_count": _int(
wazuh_incident_response_closure.get("same_run_required_stage_count")
),
"same_run_source_stage_ready_count": _int(
wazuh_incident_response_closure.get(
"same_run_source_stage_ready_count"
)
),
"same_run_closure_candidate_ready_count": _int(
wazuh_incident_response_closure.get(
"same_run_closure_candidate_ready_count"
)
),
"incident_response_packet_validation_endpoint": (
"/api/v1/iwooos/wazuh-incident-response-closure/"
"validate-incident-response-packet"
),
"next_gate_after_packet_acceptance": (
"route_to_iwooos_controlled_executor_for_bounded_execution_and_post_verifier_writeback"
),
"controlled_apply_policy_allowed": True,
},
{

620
apps/api/src/services/iwooos_wazuh_incident_response_closure.py Normal file
View File

@@ -0,0 +1,620 @@
"""
IwoooS Wazuh incident response closure readback.
This service exposes the P0-03 machine contract between the source-ready Wazuh
verifier and the later controlled executor. It validates redacted same-run
receipt packets and never queries live Wazuh, reads hosts, persists payloads,
collects secrets, or authorizes active response.
"""
from __future__ import annotations
import re
from pathlib import Path
from typing import Any
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_runtime_controlled_apply_preflight import (
load_latest_iwooos_wazuh_runtime_controlled_apply_preflight,
)
_SCHEMA_VERSION = "iwooos_wazuh_incident_response_closure_readback_v1"
_VALIDATION_SCHEMA_VERSION = (
"iwooos_wazuh_incident_response_closure_validation_result_v1"
)
_REQUIRED_SAME_RUN_STAGE_IDS = (
"sensor_source_receipt",
"normalized_asset_identity",
"source_truth_diff",
"ai_decision_candidate_action",
"risk_policy_decision",
"check_mode_dry_run_receipt",
"bounded_execution_receipt",
"independent_post_verifier",
"km_rag_mcp_playbook_write_ack",
)
_REQUIRED_PACKET_FIELDS = (
"trace_id",
"run_id",
"asset_scope_aliases",
"wazuh_event_receipt_ref",
"forensics_evidence_ref",
"incident_case_ref",
"ai_decision_ref",
"candidate_controlled_response_ref",
"source_truth_diff_ref",
"risk_policy_decision_ref",
"check_mode_dry_run_ref",
"bounded_execution_receipt_ref",
"rollback_plan_ref",
"post_verifier_ref",
"km_rag_mcp_playbook_writeback_ref",
"same_run_stage_receipts",
"runtime_boundary_ack",
"host_write_boundary_ack",
"secret_boundary_ack",
"live_wazuh_query_boundary_ack",
)
_EXPECTED_ACKS = {
"runtime_boundary_ack": "runtime_gate_remains_closed_until_executor_post_verifier",
"host_write_boundary_ack": "no_host_write_performed_by_validator",
"secret_boundary_ack": "no_secret_value_collected_or_submitted",
"live_wazuh_query_boundary_ack": "no_live_wazuh_query_performed_by_validator",
}
_BOUNDARY_ACK_FIELD_NAMES = set(_EXPECTED_ACKS)
_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
),
"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",
"secret_value",
"session",
"stored_api_password",
"stdout",
"stderr",
"token",
"unredacted_screenshot",
"wazuh_api_password",
}
_RUNTIME_ACTION_KEYS = {
"active_response_enable",
"agent_reenroll",
"agent_restart",
"ansible_apply",
"ansible_playbook_run",
"apply_now",
"argocd_sync",
"auto_block",
"credentialed_scan",
"database_migration",
"docker_restart",
"execute_now",
"exploit_attempt",
"firewall_change",
"force_push",
"host_write",
"k8s_apply",
"kali_active_scan",
"nginx_reload",
"production_write",
"repo_ref_delete",
"runtime_execution_authorized",
"secret_rotation",
"systemd_restart",
"wazuh_active_response",
"wazuh_agent_reenroll",
"wazuh_agent_restart",
"wazuh_api_live_query",
"wazuh_manager_restart",
"workflow_trigger",
}
def load_latest_iwooos_wazuh_incident_response_closure(
security_dir: Path | None = None,
alert_receipt_readback: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Return the public-safe Wazuh incident response closure contract."""
verifier = load_latest_iwooos_wazuh_fim_vulnerability_alert_verifier(
security_dir=security_dir,
alert_receipt_readback=alert_receipt_readback,
)
controlled_apply = load_latest_iwooos_wazuh_runtime_controlled_apply_preflight(
security_dir
)
dry_run = load_latest_iwooos_wazuh_allowlisted_check_mode_dry_run(security_dir)
verifier_summary = _summary(verifier)
controlled_summary = _summary(controlled_apply)
dry_run_summary = _summary(dry_run)
source_verifier_ready = (
_int(
verifier_summary.get(
"wazuh_fim_vulnerability_alert_source_verifier_ready_count"
)
)
> 0
)
controlled_apply_ready = (
_int(controlled_summary.get("controlled_apply_preflight_ready_count")) > 0
)
dry_run_ready = _int(dry_run_summary.get("dry_run_result_ref_count")) > 0
post_dry_run_ready = _int(dry_run_summary.get("post_dry_run_verifier_count")) > 0
km_writeback_ready = max(
_int(controlled_summary.get("km_playbook_writeback_count")),
_int(dry_run_summary.get("km_playbook_writeback_ready_count")),
) > 0
incident_case_ready = _int(verifier_summary.get("incident_case_accepted_count")) > 0
runtime_closed = (
_int(verifier_summary.get("wazuh_fim_vulnerability_alert_runtime_closed_count"))
> 0
)
candidate_ready = all(
(
source_verifier_ready,
controlled_apply_ready,
dry_run_ready,
post_dry_run_ready,
km_writeback_ready,
incident_case_ready,
)
) and not runtime_closed
ready_stage_count = sum(
(
1 if source_verifier_ready else 0,
1 if _int(verifier_summary.get("wazuh_event_accepted_count")) > 0 else 0,
1 if _int(verifier_summary.get("forensic_evidence_accepted_count")) > 0 else 0,
1 if incident_case_ready else 0,
1 if controlled_apply_ready else 0,
1 if dry_run_ready else 0,
1 if post_dry_run_ready else 0,
1 if km_writeback_ready else 0,
1 if runtime_closed else 0,
)
)
summary = {
"closure_packet_validator_available_count": 1,
"source_verifier_ready_count": 1 if source_verifier_ready else 0,
"controlled_apply_preflight_ready_count": 1 if controlled_apply_ready else 0,
"allowlisted_dry_run_ready_count": 1 if dry_run_ready else 0,
"post_dry_run_verifier_ready_count": 1 if post_dry_run_ready else 0,
"km_rag_mcp_playbook_writeback_ready_count": 1 if km_writeback_ready else 0,
"wazuh_event_accepted_count": _int(
verifier_summary.get("wazuh_event_accepted_count")
),
"forensic_evidence_accepted_count": _int(
verifier_summary.get("forensic_evidence_accepted_count")
),
"incident_case_accepted_count": _int(
verifier_summary.get("incident_case_accepted_count")
),
"same_run_required_stage_count": len(_REQUIRED_SAME_RUN_STAGE_IDS),
"same_run_source_stage_ready_count": ready_stage_count,
"same_run_closure_candidate_ready_count": 1 if candidate_ready else 0,
"same_run_closure_candidate_missing_count": 0 if candidate_ready else 1,
"runtime_execution_performed_count": 1 if runtime_closed else 0,
"wazuh_incident_response_runtime_closed_count": 1 if runtime_closed else 0,
"runtime_gate_count": 0,
"wazuh_api_live_query_authorized_count": 0,
"wazuh_active_response_authorized_count": 0,
"host_write_authorized_count": 0,
"secret_value_collection_allowed_count": 0,
}
return {
"schema_version": _SCHEMA_VERSION,
"status": (
"wazuh_incident_response_same_run_candidate_ready_runtime_open"
if candidate_ready
else "wazuh_incident_response_same_run_candidate_missing_receipts"
),
"mode": "same_run_closure_contract_no_live_wazuh_no_runtime_action",
"summary": summary,
"required_same_run_stage_ids": list(_REQUIRED_SAME_RUN_STAGE_IDS),
"required_incident_response_packet_fields": list(_REQUIRED_PACKET_FIELDS),
"incident_response_packet_validation_endpoint": (
"/api/v1/iwooos/wazuh-incident-response-closure/"
"validate-incident-response-packet"
),
"next_executable_action": (
"submit_redacted_same_run_wazuh_incident_response_closure_packet"
),
"next_gate_after_packet_acceptance": (
"route_to_iwooos_controlled_executor_for_bounded_execution_and_post_verifier_writeback"
),
"boundary_markers": _boundary_markers(summary),
"boundaries": _boundaries(),
"no_false_green_rules": [
"same-run packet review readiness is not runtime closure",
"runtime closure remains zero until independent production readback proves bounded execution and post-verifier in the same trace",
"this validator rejects runtime action requests and never enables Wazuh active response",
"redacted refs are accepted; raw Wazuh payloads, hostnames, IPs, secrets and logs are quarantined",
],
"source_refs": [
*verifier.get("source_refs", []),
*controlled_apply.get("source_refs", []),
*dry_run.get("source_refs", []),
"apps/api/src/services/iwooos_wazuh_incident_response_closure.py",
],
}
def validate_iwooos_wazuh_incident_response_closure_packet(
packet: dict[str, Any],
security_dir: Path | None = None,
alert_receipt_readback: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Validate one redacted same-run incident response packet without side effects."""
contract = load_latest_iwooos_wazuh_incident_response_closure(
security_dir=security_dir,
alert_receipt_readback=alert_receipt_readback,
)
findings: list[dict[str, Any]] = []
if not isinstance(packet, dict):
findings.append(
_finding(
"WIRC-01",
"blocker",
"request_incident_response_packet_supplement",
"incident response closure packet must be a JSON object.",
[],
)
)
return _validation_result(contract, "request_incident_response_packet_supplement", findings)
sensitive_hits = _collect_sensitive_hits(packet)
if sensitive_hits:
findings.append(
_finding(
"WIRC-04",
"critical",
"quarantine_sensitive_payload",
"incident response 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(packet)
if runtime_hits:
findings.append(
_finding(
"WIRC-05",
"critical",
"reject_runtime_action_request",
"incident response packet requested runtime execution; this validator only reviews receipts.",
runtime_hits[:12],
)
)
return _validation_result(contract, "reject_runtime_action_request", findings)
missing_fields = [
field for field in _REQUIRED_PACKET_FIELDS if not _present(packet.get(field))
]
if missing_fields:
findings.append(
_finding(
"WIRC-02",
"blocker",
"request_incident_response_packet_supplement",
"incident response closure packet is missing required fields.",
missing_fields,
)
)
aliases = packet.get("asset_scope_aliases")
if not isinstance(aliases, list) or not all(isinstance(item, str) and item for item in aliases):
findings.append(
_finding(
"WIRC-03",
"blocker",
"request_asset_scope_alias_fix",
"asset_scope_aliases must be a non-empty array of public aliases.",
["asset_scope_aliases"],
)
)
stage_issue = _stage_receipt_issue(packet.get("same_run_stage_receipts"))
if stage_issue:
findings.append(stage_issue)
bad_ack_fields = [
field for field, expected in _EXPECTED_ACKS.items() if packet.get(field) != expected
]
if bad_ack_fields:
findings.append(
_finding(
"WIRC-07",
"blocker",
"request_boundary_ack_fix",
"boundary acks must keep runtime, host write, live Wazuh query and secret collection closed for this validator.",
bad_ack_fields,
)
)
outcome = (
_first_blocking_lane(findings)
or "accepted_for_wazuh_incident_response_closure_review_only"
)
if outcome == "accepted_for_wazuh_incident_response_closure_review_only":
findings.append(
_finding(
"WIRC-08",
"info",
"same_run_incident_response_packet_review_ready",
"same-run incident response packet passed no-persist validation; route to controlled executor for bounded execution and independent post-verifier.",
[
"trace_id",
"same_run_stage_receipts",
"post_verifier_ref",
"km_rag_mcp_playbook_writeback_ref",
],
)
)
return _validation_result(contract, outcome, findings)
def _stage_receipt_issue(value: Any) -> dict[str, Any] | None:
receipts = value if isinstance(value, list) else []
stage_ids = {
str(item.get("stage_id"))
for item in receipts
if isinstance(item, dict) and item.get("stage_id")
}
missing = sorted(set(_REQUIRED_SAME_RUN_STAGE_IDS) - stage_ids)
extra = sorted(stage_ids - set(_REQUIRED_SAME_RUN_STAGE_IDS))
invalid_refs = [
str(item.get("stage_id") or "<missing>")
for item in receipts
if isinstance(item, dict) and not _present(item.get("receipt_ref"))
]
if not receipts or missing or extra or invalid_refs:
return _finding(
"WIRC-06",
"blocker",
"request_same_run_stage_receipts_fix",
"same_run_stage_receipts must cover every required stage once with redacted receipt refs.",
["same_run_stage_receipts"],
{"missing_stage_ids": missing, "extra_stage_ids": extra, "invalid_ref_stage_ids": invalid_refs},
)
return None
def _validation_result(
contract: dict[str, Any],
outcome_lane: str,
findings: list[dict[str, Any]],
) -> dict[str, Any]:
accepted = outcome_lane == "accepted_for_wazuh_incident_response_closure_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": _VALIDATION_SCHEMA_VERSION,
"contract_schema_version": contract["schema_version"],
"status": outcome_lane,
"mode": "no_persist_same_run_incident_response_review_no_runtime_no_secret_collection",
"accepted_for_incident_response_closure_review_only": accepted,
"quarantined": quarantined,
"runtime_action_rejected": rejected_runtime,
"summary": {
"incident_response_packet_received_count": 1,
"incident_response_packet_review_ready_count": 1 if accepted else 0,
"incident_response_packet_supplement_required_count": 1
if supplement_required
else 0,
"incident_response_packet_quarantined_count": 1 if quarantined else 0,
"incident_response_packet_runtime_action_rejected_count": 1
if rejected_runtime
else 0,
"same_run_required_stage_count": len(_REQUIRED_SAME_RUN_STAGE_IDS),
"same_run_packet_stage_ready_count": len(_REQUIRED_SAME_RUN_STAGE_IDS)
if accepted
else 0,
"runtime_execution_performed_count": 0,
"wazuh_incident_response_runtime_closed_count": 0,
"runtime_gate_count": 0,
"wazuh_api_live_query_authorized_count": 0,
"wazuh_active_response_authorized_count": 0,
"host_write_authorized_count": 0,
"secret_value_collection_allowed_count": 0,
"finding_count": len(findings),
},
"validation_findings": findings,
"next_gate": (
"route_to_iwooos_controlled_executor_for_bounded_execution_and_post_verifier_writeback"
if accepted
else "wazuh_incident_response_packet_fix_and_resubmit"
),
"source_contract_status": contract["status"],
"boundary_markers": [
"wazuh_incident_response_packet_validation_received_count=1",
f"wazuh_incident_response_packet_validation_ready_count={1 if accepted else 0}",
f"wazuh_incident_response_packet_validation_quarantined_count={1 if quarantined else 0}",
f"wazuh_incident_response_packet_validation_runtime_action_rejected_count={1 if rejected_runtime else 0}",
"wazuh_incident_response_packet_validation_runtime_closed_count=0",
"wazuh_incident_response_packet_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": _boundaries(),
}
def _summary(payload: dict[str, Any]) -> dict[str, Any]:
summary = payload.get("summary")
return summary if isinstance(summary, dict) else {}
def _int(value: Any) -> int:
return value if isinstance(value, int) else 0
def _present(value: Any) -> bool:
if value is None:
return False
if isinstance(value, str):
return bool(value.strip())
if isinstance(value, list | dict | tuple | set):
return bool(value)
return True
def _first_blocking_lane(findings: list[dict[str, Any]]) -> str | None:
for finding in findings:
if finding.get("severity") in {"blocker", "critical"}:
return str(finding.get("lane") or finding.get("outcome") or "blocked")
return None
def _finding(
check_id: str,
severity: str,
lane: str,
message: str,
field_paths: list[str],
metadata: dict[str, Any] | None = None,
) -> dict[str, Any]:
return {
"check_id": check_id,
"severity": severity,
"lane": lane,
"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():
child_path = f"{path}.{key}"
lowered_key = str(key).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)
if key_text in _BOUNDARY_ACK_FIELD_NAMES:
continue
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_incident_response_closure_validator_available=true",
f"wazuh_incident_response_source_verifier_ready_count={summary['source_verifier_ready_count']}",
f"wazuh_incident_response_same_run_required_stage_count={summary['same_run_required_stage_count']}",
f"wazuh_incident_response_same_run_source_stage_ready_count={summary['same_run_source_stage_ready_count']}",
f"wazuh_incident_response_same_run_closure_candidate_ready_count={summary['same_run_closure_candidate_ready_count']}",
f"wazuh_incident_response_runtime_closed_count={summary['wazuh_incident_response_runtime_closed_count']}",
"wazuh_incident_response_runtime_gate_count=0",
"wazuh_api_live_query_authorized=false",
"wazuh_active_response_authorized=false",
"host_write_authorized=false",
"secret_value_collection_allowed=false",
"not_authorization=true",
]
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,
}

29
apps/api/tests/test_iwooos_security_tool_closure_readback.py
View File

@@ -24,7 +24,7 @@ def test_iwooos_security_tool_closure_readback_rolls_up_tool_tracks() -> None:
assert payload["schema_version"] == "iwooos_security_tool_closure_readback_v2"
assert payload["status"] == "tool_source_readiness_available_runtime_closure_open"
assert payload["summary"]["source_readback_count"] == 8
assert payload["summary"]["source_readback_count"] == 9
assert payload["summary"]["tool_track_count"] == 8
assert payload["summary"]["p0_tool_track_count"] == 3
assert payload["summary"]["wazuh_expected_host_scope_count"] == 6
@@ -43,6 +43,18 @@ def test_iwooos_security_tool_closure_readback_rolls_up_tool_tracks() -> None:
payload["summary"]["wazuh_fim_vulnerability_alert_verifier_ready_signal_count"]
== 5
)
assert (
payload["summary"][
"wazuh_incident_response_closure_validator_available_count"
]
== 1
)
assert (
payload["summary"][
"wazuh_incident_response_same_run_required_stage_count"
]
== 9
)
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
@@ -118,6 +130,11 @@ def test_iwooos_security_tool_closure_queue_has_executable_receipt_contract() ->
assert queue["P0-03"]["track_id"] == "wazuh_detection_response"
assert queue["P0-03"]["required_signal_count"] == 6
assert queue["P0-03"]["missing_signal_count"] >= 0
assert queue["P0-03"]["same_run_required_stage_count"] == 9
assert "incident_response_packet_validation_endpoint" in queue["P0-03"]
assert queue["P0-03"]["incident_response_packet_validation_endpoint"].endswith(
"/validate-incident-response-packet"
)
assert queue["P0-03"]["required_verifier"] == (
"wazuh_registry_fim_vulnerability_alert_receipt_post_verifier"
)
@@ -164,6 +181,15 @@ def test_iwooos_security_tool_closure_readback_accepts_alert_receipt_signal() ->
assert tracks["wazuh_detection_response"]["execution_candidate_ready"] is True
assert tracks["normalized_detection_schema"]["ready_signal_count"] == 1
assert payload["summary"]["controlled_apply_policy_active_count"] == 1
queue = {item["queue_id"]: item for item in payload["next_executable_queue"]}
assert queue["P0-03"]["state"] == (
"wazuh_source_verifier_ready_runtime_closure_open"
)
assert queue["P0-03"]["same_run_closure_candidate_ready_count"] == 0
assert queue["P0-03"]["next_executable_action"] == (
"attach_live_wazuh_event_forensics_incident_case_controlled_response_"
"and_independent_post_verifier_receipts"
)
def test_iwooos_security_tool_closure_readback_api_is_public_safe(monkeypatch) -> None:
@@ -185,6 +211,7 @@ def test_iwooos_security_tool_closure_readback_api_is_public_safe(monkeypatch) -
assert response.status_code == 200
data = response.json()
assert data["schema_version"] == "iwooos_security_tool_closure_readback_v2"
assert data["summary"]["source_readback_count"] == 9
assert data["summary"]["tool_track_count"] == 8
assert data["summary"]["alert_receipt_accepted_count"] == 1
assert data["summary"]["wazuh_fim_vulnerability_alert_closed_count"] == 0

208
apps/api/tests/test_iwooos_wazuh_incident_response_closure.py Normal file
View File

@@ -0,0 +1,208 @@
from __future__ import annotations
import os
os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test")
from fastapi import FastAPI
from fastapi.testclient import TestClient
from src.api.v1.iwooos import router
from src.services.iwooos_wazuh_incident_response_closure import (
load_latest_iwooos_wazuh_incident_response_closure,
validate_iwooos_wazuh_incident_response_closure_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-closure"}],
"total": 1,
"limit": 5,
"source_status": "ready",
}
def _stage_receipts() -> list[dict[str, str]]:
return [
{"stage_id": "sensor_source_receipt", "receipt_ref": "ref-sensor"},
{"stage_id": "normalized_asset_identity", "receipt_ref": "ref-asset"},
{"stage_id": "source_truth_diff", "receipt_ref": "ref-diff"},
{"stage_id": "ai_decision_candidate_action", "receipt_ref": "ref-ai"},
{"stage_id": "risk_policy_decision", "receipt_ref": "ref-policy"},
{"stage_id": "check_mode_dry_run_receipt", "receipt_ref": "ref-dry-run"},
{"stage_id": "bounded_execution_receipt", "receipt_ref": "ref-bounded"},
{"stage_id": "independent_post_verifier", "receipt_ref": "ref-post"},
{
"stage_id": "km_rag_mcp_playbook_write_ack",
"receipt_ref": "ref-km-rag-mcp-playbook",
},
]
def _valid_packet() -> dict[str, object]:
return {
"trace_id": "iwooos-wazuh-p0-03-trace-001",
"run_id": "iwooos-wazuh-p0-03-run-001",
"asset_scope_aliases": ["managed_core_node_a", "managed_core_node_b"],
"wazuh_event_receipt_ref": "ref-wazuh-event",
"forensics_evidence_ref": "ref-forensics",
"incident_case_ref": "ref-incident-case",
"ai_decision_ref": "ref-ai-decision",
"candidate_controlled_response_ref": "ref-controlled-response-candidate",
"source_truth_diff_ref": "ref-source-truth-diff",
"risk_policy_decision_ref": "ref-risk-policy",
"check_mode_dry_run_ref": "ref-check-mode-dry-run",
"bounded_execution_receipt_ref": "ref-bounded-execution",
"rollback_plan_ref": "ref-rollback-plan",
"post_verifier_ref": "ref-independent-post-verifier",
"km_rag_mcp_playbook_writeback_ref": "ref-km-rag-mcp-playbook",
"same_run_stage_receipts": _stage_receipts(),
"runtime_boundary_ack": (
"runtime_gate_remains_closed_until_executor_post_verifier"
),
"host_write_boundary_ack": "no_host_write_performed_by_validator",
"secret_boundary_ack": "no_secret_value_collected_or_submitted",
"live_wazuh_query_boundary_ack": (
"no_live_wazuh_query_performed_by_validator"
),
}
def test_wazuh_incident_response_closure_readback_exposes_same_run_contract() -> None:
payload = load_latest_iwooos_wazuh_incident_response_closure(
alert_receipt_readback=_alert_receipt()
)
assert (
payload["schema_version"]
== "iwooos_wazuh_incident_response_closure_readback_v1"
)
assert (
payload["status"]
== "wazuh_incident_response_same_run_candidate_missing_receipts"
)
assert payload["summary"]["closure_packet_validator_available_count"] == 1
assert payload["summary"]["source_verifier_ready_count"] == 1
assert payload["summary"]["controlled_apply_preflight_ready_count"] == 1
assert payload["summary"]["allowlisted_dry_run_ready_count"] == 1
assert payload["summary"]["same_run_required_stage_count"] == 9
assert payload["summary"]["wazuh_event_accepted_count"] == 0
assert payload["summary"]["forensic_evidence_accepted_count"] == 0
assert payload["summary"]["incident_case_accepted_count"] == 0
assert payload["summary"]["same_run_source_stage_ready_count"] == 5
assert payload["summary"]["same_run_closure_candidate_ready_count"] == 0
assert payload["summary"]["same_run_closure_candidate_missing_count"] == 1
assert payload["summary"]["wazuh_incident_response_runtime_closed_count"] == 0
assert payload["summary"]["wazuh_active_response_authorized_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_wazuh_incident_response_closure_packet_accepts_redacted_same_run_receipts() -> None:
payload = validate_iwooos_wazuh_incident_response_closure_packet(
_valid_packet(),
alert_receipt_readback=_alert_receipt(),
)
assert (
payload["schema_version"]
== "iwooos_wazuh_incident_response_closure_validation_result_v1"
)
assert (
payload["status"]
== "accepted_for_wazuh_incident_response_closure_review_only"
)
assert payload["accepted_for_incident_response_closure_review_only"] is True
assert payload["summary"]["incident_response_packet_review_ready_count"] == 1
assert payload["summary"]["same_run_packet_stage_ready_count"] == 9
assert payload["summary"]["wazuh_incident_response_runtime_closed_count"] == 0
assert payload["summary"]["runtime_gate_count"] == 0
assert payload["boundaries"]["payload_persisted"] is False
assert payload["boundaries"]["runtime_execution_authorized"] is False
def test_wazuh_incident_response_closure_packet_requests_missing_stage_receipts() -> None:
packet = _valid_packet()
packet["same_run_stage_receipts"] = _stage_receipts()[:-1]
payload = validate_iwooos_wazuh_incident_response_closure_packet(
packet,
alert_receipt_readback=_alert_receipt(),
)
assert payload["status"] == "request_same_run_stage_receipts_fix"
assert payload["summary"]["incident_response_packet_supplement_required_count"] == 1
assert any(
finding["check_id"] == "WIRC-06"
and "km_rag_mcp_playbook_write_ack"
in finding["metadata"]["missing_stage_ids"]
for finding in payload["validation_findings"]
)
def test_wazuh_incident_response_closure_packet_quarantines_sensitive_payload() -> None:
packet = _valid_packet()
packet["wazuh_event_receipt_ref"] = "ref accidentally included 10.10.10.10"
payload = validate_iwooos_wazuh_incident_response_closure_packet(
packet,
alert_receipt_readback=_alert_receipt(),
)
assert payload["status"] == "quarantine_sensitive_payload"
assert payload["quarantined"] is True
assert payload["summary"]["incident_response_packet_quarantined_count"] == 1
assert "10.10.10.10" not in str(payload)
def test_wazuh_incident_response_closure_packet_rejects_runtime_action() -> None:
packet = _valid_packet()
packet["requested_actions"] = ["wazuh_active_response"]
payload = validate_iwooos_wazuh_incident_response_closure_packet(
packet,
alert_receipt_readback=_alert_receipt(),
)
assert payload["status"] == "reject_runtime_action_request"
assert payload["runtime_action_rejected"] is True
assert payload["summary"]["incident_response_packet_runtime_action_rejected_count"] == 1
assert payload["summary"]["wazuh_active_response_authorized_count"] == 0
def test_wazuh_incident_response_closure_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,
)
client = _client()
readback = client.get("/api/v1/iwooos/wazuh-incident-response-closure-readback")
validation = client.post(
"/api/v1/iwooos/wazuh-incident-response-closure/validate-incident-response-packet",
json=_valid_packet(),
)
assert readback.status_code == 200
assert validation.status_code == 200
assert readback.json()["summary"]["same_run_closure_candidate_ready_count"] == 0
assert readback.json()["summary"]["same_run_closure_candidate_missing_count"] == 1
assert (
validation.json()["status"]
== "accepted_for_wazuh_incident_response_closure_review_only"
)
assert "192.168.0." not in readback.text
assert "192.168.0." not in validation.text
assert "WAZUH_API_PASSWORD" not in validation.text
assert "批准！繼續" not in validation.text