awoooi/scripts/security/iwooos-security-operating-system.py

#!/usr/bin/env python3
"""IwoooS 資安作戰系統產生器。

本工具只產生 repo 內 snapshot，將主流資安框架、SOC / CSIRT / DevSecOps
分工、即時危害分流、告警訊息合約、Wazuh / Kali / Nginx / host / CI/CD
驗證節點與 AI 自動化閉環，收斂成一份可重跑的作戰系統。

它不連線主機、不呼叫 Wazuh / Kali、不 SSH、不讀 secret、不送 Telegram、
不 reload Nginx / Alertmanager、不改 firewall / K8s / workflow，也不啟用
active response、active scan、SOAR 或 auto block。
"""

from __future__ import annotations

import argparse
import json
import re
import subprocess
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any


TAIPEI = timezone(timedelta(hours=8))
SNAPSHOT_PATH = Path("docs/security/iwooos-security-operating-system.snapshot.json")
SCHEMA_VERSION = "iwooos_security_operating_system_v1"

REFERENCE_FRAMEWORKS = [
    ("nist_csf_2_0", "NIST CSF 2.0", "https://www.nist.gov/cyberframework"),
    ("nist_sp_800_61_r3", "NIST SP 800-61 Rev. 3", "https://csrc.nist.gov/pubs/sp/800/61/r3/final"),
    ("cis_controls_v8_1", "CIS Controls v8.1", "https://www.cisecurity.org/controls/v8-1"),
    ("cisa_zero_trust", "CISA Zero Trust Maturity Model", "https://www.cisa.gov/resources-tools/resources/zero-trust-maturity-model"),
    ("cisa_kev", "CISA Known Exploited Vulnerabilities", "https://www.cisa.gov/known-exploited-vulnerabilities-catalog"),
    ("first_epss", "FIRST EPSS", "https://www.first.org/epss/"),
    ("mitre_attack", "MITRE ATT&CK Enterprise", "https://attack.mitre.org/matrices/enterprise/"),
    ("mitre_d3fend", "MITRE D3FEND", "https://d3fend.mitre.org/"),
    ("owasp_asvs", "OWASP ASVS", "https://owasp.org/www-project-application-security-verification-standard/"),
    ("owasp_samm", "OWASP SAMM", "https://owaspsamm.org/"),
    ("wazuh_xdr_siem", "Wazuh XDR / SIEM", "https://documentation.wazuh.com/current/index.html"),
    ("wazuh_active_response", "Wazuh Active Response", "https://documentation.wazuh.com/current/user-manual/capabilities/active-response/index.html"),
    ("prometheus_alertmanager", "Prometheus Alertmanager", "https://prometheus.io/docs/alerting/latest/alertmanager/"),
    ("opentelemetry", "OpenTelemetry", "https://opentelemetry.io/docs/what-is-opentelemetry/"),
    ("ocsf", "Open Cybersecurity Schema Framework", "https://ocsf.io/"),
    ("sigma", "Sigma detection rules", "https://sigmahq.io/sigma/"),
    ("slsa", "SLSA", "https://slsa.dev/"),
    ("spdx_cyclonedx", "SPDX / CycloneDX", "https://spdx.dev/"),
    ("sigstore_cosign", "Sigstore / Cosign", "https://docs.sigstore.dev/cosign/signing/signing_with_containers/"),
    ("nist_ai_rmf", "NIST AI RMF", "https://www.nist.gov/itl/ai-risk-management-framework"),
]

OPERATING_ROLES = [
    ("security_program_owner", "資安作戰負責人", "維護控制面、優先序、完成度與停止線。"),
    ("soc_reviewer", "SOC 審查人", "審查告警、SIEM、Wazuh、Kali 與 no-false-green evidence。"),
    ("incident_commander", "事故指揮", "統一 severity、scope、containment 候選與跨專案同步。"),
    ("platform_owner", "平台負責人", "負責 host、Docker、systemd、Nginx、K8s、ArgoCD 與 public gateway 影響判讀。"),
    ("service_owner", "服務負責人", "負責產品、API、網站、admin、webhook 與 AI provider route 的驗證。"),
    ("evidence_custodian", "證據保管人", "維護脫敏 refs、chain of custody、retention 與 raw absence attestation。"),
    ("change_manager", "變更管理人", "確認維護窗口、rollback owner、postcheck、operator notification 與 freeze。"),
    ("supply_chain_owner", "供應鏈負責人", "負責 workflow、runner、Harbor、SBOM、SLSA、Cosign、KEV / package SLA。"),
    ("ai_security_reviewer", "AI 安全審查人", "審核 AI agent tool 權限、prompt redaction、過度代理與成本邊界。"),
    ("executive_risk_owner", "風險負責人", "接受風險、例外期限、資源優先序與治理報告。"),
]

SEVERITY_LANES = [
    ("SEV0", "已確認入侵或 active exploitation", "15 分鐘內形成 case / freeze / containment 候選；不得無 owner 直接執行。"),
    ("SEV1", "公開入口高風險、KEV、credential exposure、Wazuh agent 消失", "30 分鐘內形成 owner packet、證據缺口與維護窗口草案。"),
    ("SEV2", "Nginx / firewall / runner / workflow / runtime drift", "4 小時內完成 diff、owner、rollback 與 postcheck 計畫。"),
    ("SEV3", "告警噪音、coverage gap、dashboard degradation", "1 個工作日內進入 backlog 與 no-false-green 修正。"),
    ("SEV4", "治理、文件、成熟度與低風險 hardening", "納入週期報告與例外期限，不得混成緊急事件。"),
]

WORKSTREAMS = [
    ("P0-01", "asset_exposure_graph", "P0", "資產 / 暴露面總圖", "host、domain、route、service、port、package、repo、runner、secret metadata、backup、AI agent"),
    ("P0-02", "wazuh_registry_truth", "P0", "Wazuh manager registry truth", "agent total、active、disconnected、last seen、expected minimum、dashboard / API mismatch"),
    ("P0-03", "host_intrusion_forensics", "P0", "主機入侵與鑑識", "auth、sudo、process、network、FIM、persistence、package、service、Docker event"),
    ("P0-04", "gateway_config_control", "P0", "Nginx / Gateway config-control", "source-to-live diff、rendered diff、nginx test ref、route smoke、rollback"),
    ("P0-05", "network_access_baseline", "P0", "SSH / firewall / WireGuard / NodePort baseline", "before / after、actor、impact、operator notification、restoration evidence"),
    ("P0-06", "secret_identity_hygiene", "P0", "身分與 secret metadata", "SSH、sudo、deploy key、runner token name、webhook secret name、OIDC、break-glass"),
    ("P0-07", "alert_readability_receipt", "P0", "告警可讀性與 receipt", "Telegram / Alertmanager / Wazuh alert card、dedupe、noise budget、receipt"),
    ("P0-08", "incident_case_gate", "P0", "Incident case gate", "case id、timeline、owner、decision、containment、recovery、postcheck、lesson learned"),
    ("P0-09", "kev_exposure_patch_priority", "P0", "KEV / exposure / package SLA", "CISA KEV、EPSS、public exposure、asset criticality、maintenance window"),
    ("P0-10", "backup_restore_forensic_retention", "P0", "備份 / 還原 / 鑑識保存", "restore drill、offsite、escrow、chain of custody、retention、rollback proof"),
    ("P0-11", "runner_workflow_supply_chain", "P0", "Runner / workflow / supply-chain", "Gitea、workflow、runner、deploy key、Harbor、SBOM、Cosign、SLSA"),
    ("P0-12", "ai_agent_permission_gate", "P0", "AI Agent 權限閘", "tool allowlist、redaction、cost、privacy、approval、excessive agency"),
    ("P1-01", "kali_evidence_envelope", "P1", "Kali 112 evidence envelope", "health、tool version、scope、normalized finding、active scan approval packet"),
    ("P1-02", "detection_as_code", "P1", "Detection-as-code", "ATT&CK、D3FEND、Sigma、測試資料、false-positive budget、rule owner"),
    ("P1-03", "ndr_passive_sensor", "P1", "NDR passive sensor", "Suricata、Zeek、DNS / TLS / HTTP / flow logs；不開 IPS"),
    ("P1-04", "k8s_docker_hardening", "P1", "K8s / Docker hardening", "CIS / NSA-CISA 對照、Pod Security、RBAC、NetworkPolicy、audit log"),
    ("P1-05", "appsec_api_asvs", "P1", "AppSec / API ASVS", "auth、authorization、session、rate limit、CORS、security headers、webhook abuse case"),
    ("P1-06", "sbom_slsa_cosign", "P1", "SBOM / SLSA / Cosign", "SPDX、CycloneDX、VEX、provenance、artifact signing、verify"),
    ("P1-07", "soar_dry_run_case_enrichment", "P1", "SOAR dry-run / case enrichment", "TheHive / Cortex 類 case draft、enrichment、blast radius、rollback"),
    ("P1-08", "grc_exception_register", "P1", "GRC / exception register", "risk register、accepted risk、expiry、audit evidence、control owner"),
    ("P2-01", "ueba_behavior_baseline", "P2", "UEBA / 行為基線", "使用者、service account、runner、AI agent、host process、egress baseline"),
    ("P2-02", "purple_team_validation", "P2", "Purple-team / tabletop", "ATT&CK emulation、BAS / canary、偵測回歸；需授權 scope"),
    ("P2-03", "mdr_247_process", "P2", "MDR / 24x7 流程", "on-call、升級、SLA、交接、值班報表、演練"),
    ("P2-04", "exposure_management_graph", "P2", "Exposure management graph", "外部攻擊面、弱點、身份、雲端、repo、AI agent、資料流"),
]

ALERT_CONTRACT_FIELDS = [
    "event_title",
    "severity_and_confidence",
    "asset_alias_and_scope",
    "what_happened_plain_language",
    "why_it_matters",
    "redacted_evidence_refs",
    "ai_triage_lane",
    "next_candidate_action",
    "owner_gate_and_verification",
]

AUTOMATION_LOOP_STAGES = [
    "sensor_evidence",
    "normalizer_redaction",
    "ai_triage_lane",
    "candidate_generation",
    "owner_gate",
    "execution_boundary",
    "verifier_readback",
    "learning_writeback",
]

VERIFICATION_STAGES = [
    "source_guard",
    "snapshot_schema",
    "redaction_guard",
    "owner_packet_preflight",
    "wazuh_registry_readback",
    "kali_scope_readback",
    "alert_receipt_readback",
    "route_desktop_mobile_smoke",
    "postcheck_metrics",
    "cross_session_sync",
    "logbook_update",
    "no_false_green_review",
]

NO_FALSE_GREEN_RULES = [
    "route_200_is_not_security_clearance",
    "dashboard_up_is_not_agent_registry",
    "agent_active_is_not_intrusion_closed",
    "alert_quiet_is_not_alert_chain_healthy",
    "backup_fresh_is_not_restore_drill",
    "cd_success_is_not_runtime_authorization",
    "ui_visible_is_not_owner_acceptance",
    "awooop_approval_is_not_security_approval",
    "external_agent_claim_is_not_forensic_proof",
    "transport_connection_is_not_registry_acceptance",
    "source_snapshot_is_not_live_truth",
    "general_continue_is_not_maintenance_window",
]

CROSS_SESSION_SYNC_CHECKPOINTS = [
    "fetch_gitea_main_before_work",
    "share_commit_and_run_ids",
    "share_production_readback",
    "declare_runtime_boundaries",
    "freeze_same_host_or_same_gateway_edits",
    "record_owner_gate_state",
    "update_logbook_after_stage",
]

BLOCKED_ACTIONS = [
    "ssh_write",
    "host_live_secret_read",
    "wazuh_active_response_enable",
    "kali_active_scan",
    "kali_execute",
    "nginx_reload",
    "firewall_change",
    "docker_restart",
    "systemd_restart",
    "argocd_sync",
    "kubectl_apply",
    "workflow_modification",
    "secret_rotation",
    "telegram_live_send",
    "soar_action",
    "auto_block",
    "production_write",
    "force_push",
]

FORBIDDEN_TEXT_PATTERNS = [
    re.compile(r"\b(?:10|127|172\.(?:1[6-9]|2\d|3[01])|192\.168)\.\d{1,3}\.\d{1,3}\b"),
    re.compile(r"Authorization\s*:", re.IGNORECASE),
    re.compile(r"Bearer\s+[A-Za-z0-9._-]{10,}", re.IGNORECASE),
    re.compile(r"Basic\s+[A-Za-z0-9+/=]{10,}", re.IGNORECASE),
    re.compile(r"password\s*[:=]\s*['\"][^'\"]+['\"]", re.IGNORECASE),
    re.compile(r"token\s*[:=]\s*['\"][^'\"]+['\"]", re.IGNORECASE),
    re.compile(r"cookie\s*[:=]\s*['\"][^'\"]+['\"]", re.IGNORECASE),
    re.compile(r"<codex_delegation>", re.IGNORECASE),
]


def git_short_sha(root: Path) -> str:
    try:
        result = subprocess.run(
            ["git", "rev-parse", "--short", "HEAD"],
            cwd=root,
            check=True,
            capture_output=True,
            text=True,
        )
        return result.stdout.strip()
    except Exception:
        return "unknown"


def build_snapshot(root: Path, generated_at: str) -> dict[str, Any]:
    p0 = [item for item in WORKSTREAMS if item[2] == "P0"]
    p1 = [item for item in WORKSTREAMS if item[2] == "P1"]
    p2 = [item for item in WORKSTREAMS if item[2] == "P2"]
    return {
        "schema_version": SCHEMA_VERSION,
        "generated_at": generated_at,
        "git_commit": git_short_sha(root),
        "status": "iwooos_security_operating_system_ready_no_runtime_action",
        "mode": "repo_snapshot_guard_frontstage_only",
        "reference_frameworks": [
            {"framework_id": framework_id, "label": label, "source_url": source_url}
            for framework_id, label, source_url in REFERENCE_FRAMEWORKS
        ],
        "operating_roles": [
            {"role_id": role_id, "label": label, "responsibility": responsibility, "runtime_gate_open": False}
            for role_id, label, responsibility in OPERATING_ROLES
        ],
        "severity_lanes": [
            {"severity": severity, "label": label, "triage_target": target, "runtime_gate_open": False}
            for severity, label, target in SEVERITY_LANES
        ],
        "workstreams": [
            {
                "workstream_id": workstream_id,
                "lane_id": lane_id,
                "priority": priority,
                "title": title,
                "scope": scope,
                "owner_packet_required": True,
                "runtime_gate_open": False,
            }
            for workstream_id, lane_id, priority, title, scope in WORKSTREAMS
        ],
        "alert_message_contract": [
            {"field_id": field_id, "required": True, "raw_payload_allowed": False}
            for field_id in ALERT_CONTRACT_FIELDS
        ],
        "automation_loop_stages": [
            {"stage_id": stage_id, "runtime_gate_open": False} for stage_id in AUTOMATION_LOOP_STAGES
        ],
        "verification_stages": [
            {"stage_id": stage_id, "accepted": False} for stage_id in VERIFICATION_STAGES
        ],
        "no_false_green_rules": [
            {"rule_id": rule_id, "enforced": True} for rule_id in NO_FALSE_GREEN_RULES
        ],
        "cross_session_sync_checkpoints": [
            {"checkpoint_id": checkpoint_id, "required": True} for checkpoint_id in CROSS_SESSION_SYNC_CHECKPOINTS
        ],
        "blocked_actions": BLOCKED_ACTIONS,
        "summary": {
            "reference_framework_count": len(REFERENCE_FRAMEWORKS),
            "operating_role_count": len(OPERATING_ROLES),
            "severity_lane_count": len(SEVERITY_LANES),
            "workstream_count": len(WORKSTREAMS),
            "p0_workstream_count": len(p0),
            "p1_workstream_count": len(p1),
            "p2_workstream_count": len(p2),
            "alert_contract_field_count": len(ALERT_CONTRACT_FIELDS),
            "automation_loop_stage_count": len(AUTOMATION_LOOP_STAGES),
            "verification_stage_count": len(VERIFICATION_STAGES),
            "no_false_green_rule_count": len(NO_FALSE_GREEN_RULES),
            "cross_session_sync_checkpoint_count": len(CROSS_SESSION_SYNC_CHECKPOINTS),
            "blocked_action_count": len(BLOCKED_ACTIONS),
            "source_control_artifact_percent": 100,
            "evidence_weighted_security_operating_system_percent": 56,
            "soc_siem_framework_percent": 92,
            "wazuh_manager_registry_acceptance_percent": 0,
            "runtime_response_percent": 0,
            "owner_response_received_count": 0,
            "owner_response_accepted_count": 0,
            "wazuh_registry_accepted_count": 0,
            "kali_scope_accepted_count": 0,
            "alert_receipt_accepted_count": 0,
            "incident_case_accepted_count": 0,
            "host_forensics_accepted_count": 0,
            "runtime_gate_count": 0,
            "action_button_count": 0,
        },
        "execution_boundaries": {
            "runtime_execution_authorized": False,
            "host_write_authorized": False,
            "wazuh_active_response_authorized": False,
            "kali_active_scan_authorized": False,
            "kali_execute_authorized": False,
            "nginx_reload_authorized": False,
            "firewall_change_authorized": False,
            "secret_value_collection_allowed": False,
            "telegram_live_send_authorized": False,
            "soar_action_authorized": False,
            "auto_block_authorized": False,
            "production_write_authorized": False,
            "not_authorization": True,
        },
    }


def load_json(path: Path) -> dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8"))


def assert_equal(label: str, actual: Any, expected: Any) -> None:
    if actual != expected:
        raise SystemExit(f"BLOCKED {label}: expected {expected!r}, got {actual!r}")


def assert_false(label: str, actual: Any) -> None:
    assert_equal(label, actual, False)


def collect_string_values(value: Any) -> list[str]:
    if isinstance(value, str):
        return [value]
    if isinstance(value, list):
        values: list[str] = []
        for item in value:
            values.extend(collect_string_values(item))
        return values
    if isinstance(value, dict):
        values = []
        for item in value.values():
            values.extend(collect_string_values(item))
        return values
    return []


def validate_no_forbidden_text(snapshot: dict[str, Any]) -> None:
    for text in collect_string_values(snapshot):
        for pattern in FORBIDDEN_TEXT_PATTERNS:
            if pattern.search(text):
                raise SystemExit("BLOCKED iwooos_security_operating_system: snapshot contains forbidden sensitive text")


def validate(root: Path) -> None:
    snapshot = load_json(root / SNAPSHOT_PATH)
    assert_equal("schema_version", snapshot.get("schema_version"), SCHEMA_VERSION)
    assert_equal("status", snapshot.get("status"), "iwooos_security_operating_system_ready_no_runtime_action")
    assert_equal("mode", snapshot.get("mode"), "repo_snapshot_guard_frontstage_only")
    summary = snapshot.get("summary", {})
    expected_counts = {
        "reference_framework_count": 20,
        "operating_role_count": 10,
        "severity_lane_count": 5,
        "workstream_count": 24,
        "p0_workstream_count": 12,
        "p1_workstream_count": 8,
        "p2_workstream_count": 4,
        "alert_contract_field_count": 9,
        "automation_loop_stage_count": 8,
        "verification_stage_count": 12,
        "no_false_green_rule_count": 12,
        "cross_session_sync_checkpoint_count": 7,
        "blocked_action_count": 18,
        "source_control_artifact_percent": 100,
        "evidence_weighted_security_operating_system_percent": 56,
        "soc_siem_framework_percent": 92,
        "wazuh_manager_registry_acceptance_percent": 0,
        "runtime_response_percent": 0,
        "owner_response_received_count": 0,
        "owner_response_accepted_count": 0,
        "wazuh_registry_accepted_count": 0,
        "kali_scope_accepted_count": 0,
        "alert_receipt_accepted_count": 0,
        "incident_case_accepted_count": 0,
        "host_forensics_accepted_count": 0,
        "runtime_gate_count": 0,
        "action_button_count": 0,
    }
    for key, expected in expected_counts.items():
        assert_equal(f"summary.{key}", summary.get(key), expected)
    boundaries = snapshot.get("execution_boundaries", {})
    for key, value in boundaries.items():
        if key == "not_authorization":
            assert_equal(f"execution_boundaries.{key}", value, True)
        else:
            assert_false(f"execution_boundaries.{key}", value)
    validate_no_forbidden_text(snapshot)


def main() -> int:
    parser = argparse.ArgumentParser(description="IwoooS 資安作戰系統")
    parser.add_argument("--root", default=".", help="repo root")
    parser.add_argument("--output", help="寫出 JSON snapshot")
    parser.add_argument("--generated-at", help="固定報告時間，供 committed snapshot 使用")
    parser.add_argument("--json", action="store_true")
    args = parser.parse_args()

    root = Path(args.root).resolve()
    generated_at = args.generated_at or datetime.now(TAIPEI).isoformat(timespec="seconds")
    if args.output:
        output = Path(args.output)
        if not output.is_absolute():
            output = root / output
        snapshot = build_snapshot(root, generated_at)
        output.parent.mkdir(parents=True, exist_ok=True)
        output.write_text(json.dumps(snapshot, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")

    validate(root)
    snapshot = load_json(root / SNAPSHOT_PATH)
    if args.json:
        print(json.dumps(snapshot, ensure_ascii=False, sort_keys=True))
        return 0
    summary = snapshot["summary"]
    print(
        "IWOOOS_SECURITY_OPERATING_SYSTEM_OK "
        f"frameworks={summary['reference_framework_count']} "
        f"workstreams={summary['workstream_count']} "
        f"p0={summary['p0_workstream_count']} "
        f"alert_contract={summary['alert_contract_field_count']} "
        f"evidence_percent={summary['evidence_weighted_security_operating_system_percent']} "
        f"runtime_gate={summary['runtime_gate_count']}"
    )
    return 0


if __name__ == "__main__":
    sys.exit(main())