diff --git a/apps/api/src/api/v1/iwooos.py b/apps/api/src/api/v1/iwooos.py index bcb37af8f..ed92e61b5 100644 --- a/apps/api/src/api/v1/iwooos.py +++ b/apps/api/src/api/v1/iwooos.py @@ -23,6 +23,12 @@ 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_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, ) @@ -88,6 +94,68 @@ async def get_iwooos_wazuh_readonly_status_v1() -> JSONResponse: return await _wazuh_readonly_status() +@router.get( + "/api/v1/iwooos/security-operating-system", + response_model=dict[str, Any], + summary="取得 IwoooS 資安作業系統總入口", + description=( + "讀取已提交的 IwoooS security operating system contract，回傳資安框架、角色、" + "severity lane、P0/P1/P2 workstream、告警訊息合約、自動化 loop、驗證節點、" + "no-false-green 規則與 operation packet validator。此端點不查 live host、不讀 secret、" + "不送告警、不啟動 active response、不改 Nginx / Docker / K8s / firewall。" + ), +) +async def get_iwooos_security_operating_system() -> dict[str, Any]: + """回傳 IwoooS 資安作業系統公開安全總入口。""" + try: + payload = await asyncio.to_thread(load_latest_iwooos_security_operating_system) + 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 security operating system 無效：{exc}", + ) from exc + + +@router.post( + "/api/v1/iwooos/security-operating-system/validate-operation-packet", + response_model=dict[str, Any], + summary="驗證 IwoooS 資安作業脫敏 operation packet", + description=( + "針對單次 redacted security operation packet 進行 no-persist review validation，" + "檢查告警訊息合約、severity lane、target selector、source-of-truth diff、check-mode / dry-run、" + "rollback、post-apply verifier、KM / PlayBook writeback 與邊界 ack。此端點不保存 payload、" + "不查 live host / Wazuh / Kali、不讀或回傳機密明文、不送 Telegram、不啟用 SOAR / active response、" + "不改 Nginx / Docker / K8s / firewall，也不更新 runtime gate 總帳。" + ), +) +async def validate_iwooos_security_operation_packet( + operation_packet: dict[str, Any], +) -> dict[str, Any]: + """回傳單次 IwoooS security operation packet 的公開安全驗證結果。""" + try: + payload = await asyncio.to_thread( + validate_iwooos_security_operation_packet_payload, + operation_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 security operation packet 驗證器無效：{exc}", + ) from exc + + @router.get( "/api/v1/iwooos/wazuh-live-metadata-gate", response_model=dict[str, Any], diff --git a/apps/api/src/services/iwooos_security_operating_system.py b/apps/api/src/services/iwooos_security_operating_system.py new file mode 100644 index 000000000..a1b9c0b07 --- /dev/null +++ b/apps/api/src/services/iwooos_security_operating_system.py @@ -0,0 +1,839 @@ +""" +IwoooS security operating system readback and operation-packet validator. + +The validator is the public-safe intake point for a redacted security operation +packet. It checks the security loop evidence contract without persisting the +payload, reading secrets, querying live hosts, or authorizing runtime actions. +""" + +from __future__ import annotations + +import json +import re +from pathlib import Path +from typing import Any + +from src.services.snapshot_paths import default_security_dir + +_DEFAULT_SECURITY_DIR = default_security_dir(Path(__file__)) +_SNAPSHOT_FILE = "iwooos-security-operating-system.snapshot.json" +_EXPECTED_SCHEMA = "iwooos_security_operating_system_v1" + +_REQUIRED_FALSE_BOUNDARIES = { + "auto_block_authorized", + "firewall_change_authorized", + "host_write_authorized", + "kali_active_scan_authorized", + "kali_execute_authorized", + "nginx_reload_authorized", + "production_write_authorized", + "runtime_execution_authorized", + "secret_value_collection_allowed", + "soar_action_authorized", + "telegram_live_send_authorized", + "wazuh_active_response_authorized", +} + +_REQUIRED_OPERATION_PACKET_FIELDS = [ + "operation_intent", + "event_title", + "severity", + "confidence", + "asset_aliases", + "asset_scope", + "what_happened_plain_language", + "why_it_matters", + "redacted_evidence_refs", + "ai_triage_lane", + "candidate_action", + "owner_gate_and_verification", + "target_selector_aliases", + "source_of_truth_diff_ref", + "check_mode_plan_ref", + "dry_run_evidence_ref", + "rollback_plan_ref", + "rollback_owner", + "post_apply_verifier_ref", + "km_playbook_writeback_ref", + "audit_receipt_ref", + "runtime_boundary_ack", + "host_write_boundary_ack", + "secret_boundary_ack", +] + +_EXPECTED_ACKS = { + "runtime_boundary_ack": "runtime_gate_remains_closed_until_post_verifier_passes", + "host_write_boundary_ack": "no_host_write_performed_by_packet_validator", + "secret_boundary_ack": "no_secret_value_collected_or_submitted", +} + +_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_dashboard_request", + "raw_env", + "raw_hostname", + "raw_log", + "raw_runtime_volume", + "raw_session", + "raw_wazuh_payload", + "secret_value", + "session", + "stored_api_password", + "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", + "soar_action", + "systemd_restart", + "telegram_live_send", + "wazuh_active_response", + "wazuh_agent_reenroll", + "wazuh_agent_restart", + "wazuh_api_live_query", + "wazuh_manager_restart", + "workflow_trigger", +} + + +def load_latest_iwooos_security_operating_system( + security_dir: Path | None = None, +) -> dict[str, Any]: + """Load the committed IwoooS security operating system contract.""" + directory = security_dir or _DEFAULT_SECURITY_DIR + snapshot = _load_snapshot(directory) + _require_boundaries(snapshot) + + summary = _summary(snapshot) + merged_summary = { + "reference_framework_count": _int(summary.get("reference_framework_count")), + "operating_role_count": _int(summary.get("operating_role_count")), + "severity_lane_count": _int(summary.get("severity_lane_count")), + "workstream_count": _int(summary.get("workstream_count")), + "p0_workstream_count": _int(summary.get("p0_workstream_count")), + "p1_workstream_count": _int(summary.get("p1_workstream_count")), + "p2_workstream_count": _int(summary.get("p2_workstream_count")), + "alert_contract_field_count": _int(summary.get("alert_contract_field_count")), + "automation_loop_stage_count": _int(summary.get("automation_loop_stage_count")), + "verification_stage_count": _int(summary.get("verification_stage_count")), + "no_false_green_rule_count": _int(summary.get("no_false_green_rule_count")), + "cross_session_sync_checkpoint_count": _int( + summary.get("cross_session_sync_checkpoint_count") + ), + "blocked_action_count": _int(summary.get("blocked_action_count")), + "source_control_artifact_percent": _int( + summary.get("source_control_artifact_percent") + ), + "soc_siem_framework_percent": _int(summary.get("soc_siem_framework_percent")), + "evidence_weighted_security_operating_system_percent": _int( + summary.get("evidence_weighted_security_operating_system_percent") + ), + "wazuh_manager_registry_acceptance_percent": _int( + summary.get("wazuh_manager_registry_acceptance_percent") + ), + "wazuh_registry_accepted_count": _int( + summary.get("wazuh_registry_accepted_count") + ), + "alert_receipt_accepted_count": _int( + summary.get("alert_receipt_accepted_count") + ), + "incident_case_accepted_count": _int( + summary.get("incident_case_accepted_count") + ), + "host_forensics_accepted_count": _int( + summary.get("host_forensics_accepted_count") + ), + "kali_scope_accepted_count": _int(summary.get("kali_scope_accepted_count")), + "owner_response_received_count": _int( + summary.get("owner_response_received_count") + ), + "owner_response_accepted_count": _int( + summary.get("owner_response_accepted_count") + ), + "runtime_response_percent": _int(summary.get("runtime_response_percent")), + "runtime_gate_count": _int(summary.get("runtime_gate_count")), + "action_button_count": _int(summary.get("action_button_count")), + "operation_packet_validator_available_count": 1, + "operation_packet_required_field_count": len(_REQUIRED_OPERATION_PACKET_FIELDS), + } + + return { + "schema_version": "iwooos_security_operating_system_readback_v1", + "source_schema_version": snapshot["schema_version"], + "status": snapshot.get( + "status", "iwooos_security_operating_system_ready_no_runtime_action" + ), + "mode": "committed_security_operating_system_readback_with_no_persist_operation_packet_validator", + "source_refs": [ + f"docs/security/{_SNAPSHOT_FILE}", + "scripts/security/iwooos-security-operating-system.py", + ], + "operation_packet_validation_endpoint": ( + "/api/v1/iwooos/security-operating-system/validate-operation-packet" + ), + "operation_packet_validation_mode": ( + "no_persist_security_operation_review_no_runtime_no_secret_collection" + ), + "summary": merged_summary, + "reference_frameworks": _reference_frameworks(snapshot.get("reference_frameworks")), + "operating_roles": _operating_roles(snapshot.get("operating_roles")), + "severity_lanes": _severity_lanes(snapshot.get("severity_lanes")), + "automation_loop_stages": _loop_stages(snapshot.get("automation_loop_stages")), + "alert_message_contract": _contract_fields(snapshot.get("alert_message_contract")), + "verification_stages": _verification_stages(snapshot.get("verification_stages")), + "workstreams": _workstreams(snapshot.get("workstreams")), + "required_operation_packet_fields": list(_REQUIRED_OPERATION_PACKET_FIELDS), + "expected_boundary_acks": dict(_EXPECTED_ACKS), + "blocked_actions": _strings(snapshot.get("blocked_actions")), + "no_false_green_rules": _rules(snapshot.get("no_false_green_rules")), + "cross_session_sync_checkpoints": _checkpoints( + snapshot.get("cross_session_sync_checkpoints") + ), + "boundary_markers": _boundary_markers(merged_summary), + "boundaries": { + "payload_persisted": False, + "secret_value_collection_allowed": False, + "host_write_authorized": False, + "wazuh_active_response_authorized": False, + "kali_active_scan_authorized": False, + "kali_execute_authorized": False, + "soar_action_authorized": False, + "auto_block_authorized": False, + "telegram_live_send_authorized": False, + "production_write_authorized": False, + "runtime_execution_authorized": False, + "runtime_gate_open": False, + "non_incident_guard_mode": "ai_controlled_apply_open_for_review_and_dry_run", + "break_glass_required_for_destructive_or_secret_or_live_response": True, + "not_authorization": True, + }, + } + + +def validate_iwooos_security_operation_packet( + operation_packet: dict[str, Any], + security_dir: Path | None = None, +) -> dict[str, Any]: + """Validate one redacted security operation packet without applying it.""" + contract = load_latest_iwooos_security_operating_system(security_dir) + snapshot = _load_snapshot(security_dir or _DEFAULT_SECURITY_DIR) + severity_values = { + item["severity"] + for item in _severity_lanes(snapshot.get("severity_lanes")) + if item.get("severity") + } + workstream_ids = { + item["workstream_id"] + for item in _workstreams(snapshot.get("workstreams")) + if item.get("workstream_id") + } + + findings: list[dict[str, Any]] = [] + if not isinstance(operation_packet, dict): + findings.append( + _finding( + "SOS-01", + "blocker", + "request_security_operation_packet_supplement", + "security operation packet must be a JSON object.", + [], + ) + ) + return _validation_result( + contract, "request_security_operation_packet_supplement", findings + ) + + sensitive_hits = _collect_sensitive_hits(operation_packet) + if sensitive_hits: + findings.append( + _finding( + "SOS-04", + "critical", + "quarantine_sensitive_payload", + "security operation 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(operation_packet) + if runtime_hits: + findings.append( + _finding( + "SOS-05", + "critical", + "reject_runtime_action_request", + "security operation packet requested live runtime execution; this validator only reviews evidence, dry-run, verifier, and writeback readiness.", + runtime_hits[:12], + ) + ) + return _validation_result(contract, "reject_runtime_action_request", findings) + + missing_fields = [ + field + for field in _REQUIRED_OPERATION_PACKET_FIELDS + if not _present(operation_packet.get(field)) + ] + if missing_fields: + findings.append( + _finding( + "SOS-01", + "blocker", + "request_security_operation_packet_supplement", + "security operation packet is missing required security loop fields.", + missing_fields, + ) + ) + + if operation_packet.get("operation_intent") != "validate_security_operation_loop_only": + findings.append( + _finding( + "SOS-02", + "blocker", + "request_operation_intent_fix", + "operation_intent must be validate_security_operation_loop_only.", + ["operation_intent"], + ) + ) + + if operation_packet.get("severity") not in severity_values: + findings.append( + _finding( + "SOS-03", + "blocker", + "request_severity_lane_fix", + "severity must match one committed IwoooS severity lane.", + ["severity"], + {"allowed_severities": sorted(severity_values)}, + ) + ) + + workstream_id = operation_packet.get("workstream_id") + if workstream_id is not None and workstream_id not in workstream_ids: + findings.append( + _finding( + "SOS-06", + "blocker", + "request_workstream_fix", + "workstream_id must match one committed IwoooS security workstream when supplied.", + ["workstream_id"], + ) + ) + + alias_issue = _validate_public_aliases(operation_packet.get("target_selector_aliases")) + if alias_issue: + findings.append( + _finding( + "SOS-07", + "blocker", + "request_target_selector_fix", + alias_issue, + ["target_selector_aliases"], + ) + ) + + bad_ack_fields = [ + field + for field, expected in _EXPECTED_ACKS.items() + if operation_packet.get(field) != expected + ] + if bad_ack_fields: + findings.append( + _finding( + "SOS-08", + "blocker", + "request_boundary_ack_fix", + "boundary acks must keep runtime execution, host writes, and secret collection closed until verifier-backed apply is separately authorized.", + bad_ack_fields, + ) + ) + + evidence_refs = operation_packet.get("redacted_evidence_refs") + if isinstance(evidence_refs, list) and len(evidence_refs) < 2: + findings.append( + _finding( + "SOS-09", + "blocker", + "request_evidence_ref_supplement", + "redacted_evidence_refs must include at least two independent public-safe evidence references.", + ["redacted_evidence_refs"], + ) + ) + + outcome = _first_blocking_lane(findings) or "accepted_for_security_operation_review_only" + if outcome == "accepted_for_security_operation_review_only": + findings.append( + _finding( + "SOS-10", + "info", + "security_operation_loop_ready", + "security operation packet passed no-persist review; next action is controlled check-mode / dry-run and verifier readback, not live runtime execution.", + [ + "source_of_truth_diff_ref", + "check_mode_plan_ref", + "dry_run_evidence_ref", + "post_apply_verifier_ref", + "km_playbook_writeback_ref", + ], + ) + ) + return _validation_result(contract, outcome, findings) + + +def _load_snapshot(directory: Path) -> dict[str, Any]: + path = directory / _SNAPSHOT_FILE + if not path.is_file(): + raise FileNotFoundError(f"{path}: IwoooS security operating system snapshot not found") + 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 _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 _strings(value: Any) -> list[str]: + if not isinstance(value, list): + return [] + return [item for item in value if isinstance(item, str)] + + +def _reference_frameworks(value: Any) -> list[dict[str, str]]: + if not isinstance(value, list): + return [] + items: list[dict[str, str]] = [] + for item in value: + if isinstance(item, dict): + items.append( + { + "framework_id": str(item.get("framework_id", "")), + "label": str(item.get("label", "")), + "source_url": str(item.get("source_url", "")), + } + ) + return items + + +def _operating_roles(value: Any) -> list[dict[str, Any]]: + if not isinstance(value, list): + return [] + items: list[dict[str, Any]] = [] + for item in value: + if isinstance(item, dict): + items.append( + { + "role_id": str(item.get("role_id", "")), + "label": str(item.get("label", "")), + "responsibility": str(item.get("responsibility", "")), + "runtime_gate_open": item.get("runtime_gate_open") is True, + } + ) + return items + + +def _severity_lanes(value: Any) -> list[dict[str, Any]]: + if not isinstance(value, list): + return [] + items: list[dict[str, Any]] = [] + for item in value: + if isinstance(item, dict): + items.append( + { + "severity": str(item.get("severity", "")), + "label": str(item.get("label", "")), + "triage_target": str(item.get("triage_target", "")), + "runtime_gate_open": item.get("runtime_gate_open") is True, + } + ) + return items + + +def _loop_stages(value: Any) -> list[dict[str, Any]]: + if not isinstance(value, list): + return [] + items: list[dict[str, Any]] = [] + for item in value: + if isinstance(item, dict): + items.append( + { + "stage_id": str(item.get("stage_id", "")), + "runtime_gate_open": item.get("runtime_gate_open") is True, + } + ) + return items + + +def _contract_fields(value: Any) -> list[dict[str, Any]]: + if not isinstance(value, list): + return [] + items: list[dict[str, Any]] = [] + for item in value: + if isinstance(item, dict): + items.append( + { + "field_id": str(item.get("field_id", "")), + "required": item.get("required") is True, + "raw_payload_allowed": item.get("raw_payload_allowed") is True, + } + ) + return items + + +def _verification_stages(value: Any) -> list[dict[str, Any]]: + if not isinstance(value, list): + return [] + items: list[dict[str, Any]] = [] + for item in value: + if isinstance(item, dict): + items.append( + { + "stage_id": str(item.get("stage_id", "")), + "accepted": item.get("accepted") is True, + "runtime_gate_open": item.get("runtime_gate_open") is True, + } + ) + return items + + +def _workstreams(value: Any) -> list[dict[str, str]]: + if not isinstance(value, list): + return [] + items: list[dict[str, str]] = [] + for item in value: + if isinstance(item, dict): + items.append( + { + "workstream_id": str(item.get("workstream_id", "")), + "lane_id": str(item.get("lane_id", "")), + "priority": str(item.get("priority", "")), + "title": str(item.get("title", "")), + "scope": str(item.get("scope", item.get("coverage", ""))), + } + ) + return items + + +def _rules(value: Any) -> list[dict[str, Any]]: + if not isinstance(value, list): + return [] + items: list[dict[str, Any]] = [] + for item in value: + if isinstance(item, dict): + items.append( + { + "rule_id": str(item.get("rule_id", "")), + "enforced": item.get("enforced") is True, + } + ) + return items + + +def _checkpoints(value: Any) -> list[dict[str, Any]]: + if not isinstance(value, list): + return [] + items: list[dict[str, Any]] = [] + for item in value: + if isinstance(item, dict): + items.append( + { + "checkpoint_id": str(item.get("checkpoint_id", "")), + "required": item.get("required") is True, + } + ) + return items + + +def _boundary_markers(summary: dict[str, int]) -> list[str]: + return [ + "iwooos_security_operating_system_api_visible=true", + "iwooos_security_operation_packet_validation_api_available=true", + f"iwooos_security_operating_system_reference_framework_count={summary['reference_framework_count']}", + f"iwooos_security_operating_system_operating_role_count={summary['operating_role_count']}", + f"iwooos_security_operating_system_severity_lane_count={summary['severity_lane_count']}", + f"iwooos_security_operating_system_workstream_count={summary['workstream_count']}", + f"iwooos_security_operating_system_p0_workstream_count={summary['p0_workstream_count']}", + f"iwooos_security_operating_system_alert_contract_field_count={summary['alert_contract_field_count']}", + f"iwooos_security_operating_system_automation_loop_stage_count={summary['automation_loop_stage_count']}", + f"iwooos_security_operating_system_verification_stage_count={summary['verification_stage_count']}", + f"iwooos_security_operating_system_operation_packet_required_field_count={summary['operation_packet_required_field_count']}", + f"iwooos_security_operating_system_evidence_weighted_percent={summary['evidence_weighted_security_operating_system_percent']}", + f"iwooos_security_operating_system_wazuh_registry_accepted_count={summary['wazuh_registry_accepted_count']}", + "iwooos_security_operating_system_operation_packet_validation_no_persist=true", + "iwooos_security_operating_system_runtime_gate_count=0", + "runtime_execution_authorized=false", + "host_write_authorized=false", + "secret_value_collection_allowed=false", + "not_authorization=true", + ] + + +def _require_boundaries(payload: dict[str, Any]) -> None: + summary = _summary(payload) + for key in ( + "runtime_gate_count", + "action_button_count", + "owner_response_received_count", + "owner_response_accepted_count", + "alert_receipt_accepted_count", + "incident_case_accepted_count", + "host_forensics_accepted_count", + "kali_scope_accepted_count", + ): + if _int(summary.get(key)) != 0: + raise ValueError(f"IwoooS security operating system summary.{key} must remain 0") + + expected_lengths = { + "reference_frameworks": _int(summary.get("reference_framework_count")), + "operating_roles": _int(summary.get("operating_role_count")), + "severity_lanes": _int(summary.get("severity_lane_count")), + "workstreams": _int(summary.get("workstream_count")), + "alert_message_contract": _int(summary.get("alert_contract_field_count")), + "automation_loop_stages": _int(summary.get("automation_loop_stage_count")), + "verification_stages": _int(summary.get("verification_stage_count")), + "no_false_green_rules": _int(summary.get("no_false_green_rule_count")), + "cross_session_sync_checkpoints": _int( + summary.get("cross_session_sync_checkpoint_count") + ), + "blocked_actions": _int(summary.get("blocked_action_count")), + } + for key, expected in expected_lengths.items(): + value = payload.get(key) + if not isinstance(value, list) or len(value) != expected: + raise ValueError( + f"IwoooS security operating system {key} must match summary count" + ) + + boundaries = payload.get("execution_boundaries") + if not isinstance(boundaries, dict): + raise ValueError("IwoooS security operating system execution_boundaries missing") + for key in _REQUIRED_FALSE_BOUNDARIES: + if boundaries.get(key) is not False: + raise ValueError( + f"IwoooS security operating system execution_boundaries.{key} must remain false" + ) + if boundaries.get("not_authorization") is not True: + raise ValueError("IwoooS security operating system not_authorization must remain true") + + +def _validation_result( + contract: dict[str, Any], + outcome_lane: str, + findings: list[dict[str, Any]], +) -> dict[str, Any]: + accepted = outcome_lane == "accepted_for_security_operation_review_only" + quarantined = outcome_lane == "quarantine_sensitive_payload" + rejected_runtime = outcome_lane == "reject_runtime_action_request" + supplement_required = not accepted and not quarantined and not rejected_runtime + return { + "schema_version": "iwooos_security_operation_packet_validation_result_v1", + "contract_schema_version": contract["schema_version"], + "status": outcome_lane, + "mode": "no_persist_security_operation_review_no_runtime_no_secret_collection", + "outcome_lane": outcome_lane, + "accepted_for_security_operation_review_only": accepted, + "quarantined": quarantined, + "runtime_action_rejected": rejected_runtime, + "summary": { + "security_operation_packet_received_count": 1, + "security_operation_packet_accepted_count": 1 if accepted else 0, + "security_operation_packet_supplement_required_count": 1 + if supplement_required + else 0, + "security_operation_packet_quarantined_count": 1 if quarantined else 0, + "security_operation_runtime_action_rejected_count": 1 + if rejected_runtime + else 0, + "runtime_gate_count": 0, + "host_write_authorized_count": 0, + "secret_value_collection_allowed_count": 0, + "wazuh_active_response_authorized_count": 0, + "kali_active_scan_authorized_count": 0, + "soar_action_authorized_count": 0, + "finding_count": len(findings), + }, + "validation_findings": findings, + "boundary_markers": [ + "iwooos_security_operation_packet_validation_received_count=1", + f"iwooos_security_operation_packet_validation_accepted_count={1 if accepted else 0}", + f"iwooos_security_operation_packet_validation_quarantined_count={1 if quarantined else 0}", + f"iwooos_security_operation_packet_validation_runtime_action_rejected_count={1 if rejected_runtime else 0}", + "iwooos_security_operation_packet_validation_no_persist=true", + "iwooos_security_operation_packet_validation_runtime_gate_count=0", + "runtime_execution_authorized=false", + "host_write_authorized=false", + "secret_value_collection_allowed=false", + "not_authorization=true", + ], + "boundaries": { + "payload_persisted": False, + "secret_value_collection_allowed": False, + "host_write_authorized": False, + "wazuh_active_response_authorized": False, + "kali_active_scan_authorized": False, + "kali_execute_authorized": False, + "soar_action_authorized": False, + "auto_block_authorized": False, + "telegram_live_send_authorized": False, + "production_write_authorized": False, + "runtime_execution_authorized": False, + "runtime_gate_open": False, + "not_authorization": True, + }, + "next_gate": ( + "controlled_check_mode_dry_run_then_post_apply_verifier_readback" + if accepted + else "security_operation_packet_fix_and_resubmit" + ), + } + + +def _finding( + check_id: str, + severity: str, + lane: str, + message: str, + field_paths: list[str], + extra: dict[str, Any] | None = None, +) -> dict[str, Any]: + payload: dict[str, Any] = { + "check_id": check_id, + "severity": severity, + "lane": lane, + "message": message, + "field_paths": field_paths, + } + if extra: + payload.update(extra) + return payload + + +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 _validate_public_aliases(value: Any) -> str | None: + aliases = value if isinstance(value, list) else [] + if not aliases or not all(isinstance(item, str) for item in aliases): + return "target_selector_aliases must be an array of public alias strings." + alias_set = set(aliases) + if len(aliases) != len(alias_set): + return "target_selector_aliases must not contain duplicates." + if any(not re.fullmatch(r"[a-z0-9][a-z0-9_-]{2,80}", item) for item in aliases): + return "target_selector_aliases must use public-safe alias ids only." + return None + + +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) + key_lower = key_text.lower() + for fragment in _FORBIDDEN_KEY_FRAGMENTS: + if fragment in key_lower: + hits.append({"category": f"key:{fragment}", "path": f"{path}.{key_text}"}) + hits.extend(_collect_sensitive_hits(item, f"{path}.{key_text}")) + elif isinstance(value, list): + for index, item in enumerate(value): + hits.extend(_collect_sensitive_hits(item, f"{path}[{index}]")) + elif isinstance(value, str): + for category, pattern in _SENSITIVE_TEXT_PATTERNS.items(): + if pattern.search(value): + hits.append({"category": category, "path": path}) + 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.lower() in _RUNTIME_ACTION_KEYS: + hits.append(f"{path}.{key_text}") + hits.extend(_collect_runtime_action_hits(item, f"{path}.{key_text}")) + elif isinstance(value, list): + for index, item in enumerate(value): + hits.extend(_collect_runtime_action_hits(item, f"{path}[{index}]")) + return hits + + +def _first_blocking_lane(findings: list[dict[str, Any]]) -> str | None: + for finding in findings: + if finding.get("severity") in {"critical", "blocker"}: + return str(finding.get("lane", "request_security_operation_packet_supplement")) + return None diff --git a/apps/api/tests/test_iwooos_security_operating_system.py b/apps/api/tests/test_iwooos_security_operating_system.py new file mode 100644 index 000000000..30ab94947 --- /dev/null +++ b/apps/api/tests/test_iwooos_security_operating_system.py @@ -0,0 +1,181 @@ +from __future__ import annotations + +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from src.api.v1.iwooos import router +from src.services.iwooos_security_operating_system import ( + load_latest_iwooos_security_operating_system, + validate_iwooos_security_operation_packet, +) + + +def _client() -> TestClient: + app = FastAPI() + app.include_router(router) + return TestClient(app) + + +def _valid_operation_packet() -> dict[str, object]: + return { + "operation_intent": "validate_security_operation_loop_only", + "event_title": "Wazuh registry parity drift review", + "severity": "SEV1", + "confidence": "high", + "asset_aliases": [ + "managed_core_node_a", + "managed_core_node_b", + ], + "asset_scope": "iwooos_wazuh_manager_registry_public_aliases", + "what_happened_plain_language": "manager registry parity requires review before any controlled apply", + "why_it_matters": "registry drift can hide unmanaged hosts and weaken intrusion detection", + "redacted_evidence_refs": [ + "evidence/iwooos/wazuh-registry-parity-redacted-v1", + "evidence/iwooos/post-enable-readback-redacted-v1", + ], + "ai_triage_lane": "wazuh_registry_truth", + "candidate_action": "prepare_controlled_check_mode_dry_run", + "owner_gate_and_verification": "commander_blanket_authorized_for_low_blast_radius_review_only", + "target_selector_aliases": [ + "managed_core_node_a", + "managed_core_node_b", + ], + "workstream_id": "P0-02", + "source_of_truth_diff_ref": "docs/security/iwooos-security-operating-system.snapshot.json#wazuh-registry-truth", + "check_mode_plan_ref": "playbooks/iwooos-security-operation-check-mode#redacted", + "dry_run_evidence_ref": "evidence/iwooos/security-operation-dry-run-redacted-v1", + "rollback_plan_ref": "playbooks/iwooos-security-operation-rollback#redacted", + "rollback_owner": "iwooos-security-reviewer", + "post_apply_verifier_ref": "verifiers/iwooos-security-operation-readback#public-safe", + "km_playbook_writeback_ref": "km/playbook-trust/iwooos-security-operation-loop-v1", + "audit_receipt_ref": "audit/iwooos-security-operation-packet-redacted-v1", + "runtime_boundary_ack": "runtime_gate_remains_closed_until_post_verifier_passes", + "host_write_boundary_ack": "no_host_write_performed_by_packet_validator", + "secret_boundary_ack": "no_secret_value_collected_or_submitted", + } + + +def test_iwooos_security_operating_system_readback_exposes_api_validator() -> None: + payload = load_latest_iwooos_security_operating_system() + + assert payload["schema_version"] == "iwooos_security_operating_system_readback_v1" + assert payload["source_schema_version"] == "iwooos_security_operating_system_v1" + assert payload["status"] == "iwooos_security_operating_system_ready_no_runtime_action" + assert payload["summary"]["reference_framework_count"] == 20 + assert payload["summary"]["operating_role_count"] == 10 + assert payload["summary"]["severity_lane_count"] == 5 + assert payload["summary"]["workstream_count"] == 24 + assert payload["summary"]["p0_workstream_count"] == 12 + assert payload["summary"]["automation_loop_stage_count"] == 8 + assert payload["summary"]["verification_stage_count"] == 12 + assert payload["summary"]["operation_packet_validator_available_count"] == 1 + assert payload["summary"]["operation_packet_required_field_count"] == 24 + assert payload["summary"]["wazuh_registry_accepted_count"] == 6 + assert payload["summary"]["runtime_gate_count"] == 0 + assert payload["operation_packet_validation_endpoint"] == ( + "/api/v1/iwooos/security-operating-system/validate-operation-packet" + ) + assert len(payload["required_operation_packet_fields"]) == 24 + assert payload["boundaries"]["payload_persisted"] is False + assert payload["boundaries"]["runtime_execution_authorized"] is False + assert payload["boundaries"]["host_write_authorized"] is False + assert payload["boundaries"]["secret_value_collection_allowed"] is False + assert payload["boundaries"]["not_authorization"] is True + assert any( + marker == "iwooos_security_operation_packet_validation_api_available=true" + for marker in payload["boundary_markers"] + ) + + +def test_iwooos_security_operating_system_api_is_public_safe() -> None: + response = _client().get("/api/v1/iwooos/security-operating-system") + + assert response.status_code == 200 + data = response.json() + assert data["schema_version"] == "iwooos_security_operating_system_readback_v1" + assert data["summary"]["operation_packet_validator_available_count"] == 1 + assert data["summary"]["runtime_gate_count"] == 0 + assert len(data["workstreams"]) == 24 + assert any(item["workstream_id"] == "P0-02" for item in data["workstreams"]) + assert "192.168.0." not in response.text + assert "工作視窗" not in response.text + assert "批准！繼續" not in response.text + assert "WAZUH_API_PASSWORD" not in response.text + + +def test_iwooos_security_operation_packet_validator_accepts_redacted_loop() -> None: + payload = validate_iwooos_security_operation_packet(_valid_operation_packet()) + + assert payload["schema_version"] == "iwooos_security_operation_packet_validation_result_v1" + assert payload["status"] == "accepted_for_security_operation_review_only" + assert payload["accepted_for_security_operation_review_only"] is True + assert payload["summary"]["security_operation_packet_received_count"] == 1 + assert payload["summary"]["security_operation_packet_accepted_count"] == 1 + assert payload["summary"]["runtime_gate_count"] == 0 + assert payload["summary"]["host_write_authorized_count"] == 0 + assert payload["summary"]["secret_value_collection_allowed_count"] == 0 + assert payload["boundaries"]["payload_persisted"] is False + assert payload["boundaries"]["runtime_execution_authorized"] is False + assert payload["next_gate"] == "controlled_check_mode_dry_run_then_post_apply_verifier_readback" + + +def test_iwooos_security_operation_packet_api_does_not_persist_or_open_gate() -> None: + client = _client() + before = client.get("/api/v1/iwooos/security-operating-system").json() + response = client.post( + "/api/v1/iwooos/security-operating-system/validate-operation-packet", + json=_valid_operation_packet(), + ) + after = client.get("/api/v1/iwooos/security-operating-system").json() + + assert response.status_code == 200 + result = response.json() + assert result["status"] == "accepted_for_security_operation_review_only" + assert result["summary"]["security_operation_packet_accepted_count"] == 1 + assert result["summary"]["runtime_gate_count"] == 0 + assert before["summary"] == after["summary"] + + +def test_iwooos_security_operation_packet_validator_requests_missing_fields() -> None: + packet = _valid_operation_packet() + packet.pop("post_apply_verifier_ref") + + payload = validate_iwooos_security_operation_packet(packet) + + assert payload["status"] == "request_security_operation_packet_supplement" + assert payload["summary"]["security_operation_packet_supplement_required_count"] == 1 + assert payload["summary"]["runtime_gate_count"] == 0 + assert any( + "post_apply_verifier_ref" in finding["field_paths"] + for finding in payload["validation_findings"] + ) + + +def test_iwooos_security_operation_packet_validator_quarantines_sensitive_payload() -> None: + packet = _valid_operation_packet() + packet["redacted_evidence_refs"] = [ + "bad ref includes 10.1.2.3 and Authorization: Bearer abcdefghijklmnop", + "evidence/iwooos/second-ref-redacted-v1", + ] + + payload = validate_iwooos_security_operation_packet(packet) + + assert payload["status"] == "quarantine_sensitive_payload" + assert payload["quarantined"] is True + assert payload["summary"]["security_operation_packet_quarantined_count"] == 1 + assert payload["summary"]["runtime_gate_count"] == 0 + assert "10.1.2.3" not in str(payload) + assert "Bearer abcdefghijklmnop" not in str(payload) + + +def test_iwooos_security_operation_packet_validator_rejects_runtime_action() -> None: + packet = _valid_operation_packet() + packet["wazuh_active_response"] = True + + payload = validate_iwooos_security_operation_packet(packet) + + assert payload["status"] == "reject_runtime_action_request" + assert payload["runtime_action_rejected"] is True + assert payload["summary"]["security_operation_runtime_action_rejected_count"] == 1 + assert payload["summary"]["wazuh_active_response_authorized_count"] == 0 + assert payload["summary"]["runtime_gate_count"] == 0