awoooi/scripts/security/wazuh-readonly-production-readback.py

#!/usr/bin/env python3
"""
IwoooS Wazuh 只讀 API production readback 驗收。

本工具只做 HTTPS GET，不收 secret、不連 Wazuh manager、不做 active
response、不碰主機。預設模式用於部署後驗收：production API 不可是 404。
若 release 尚未部署，可加 --allow-predeploy-404 記錄目前仍未上線。
"""

from __future__ import annotations

import argparse
import json
import re
import sys
from dataclasses import dataclass
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen


DEFAULT_URL = "https://awoooi.wooo.work/api/iwooos/wazuh"
EXPECTED_SCHEMA = "iwooos_wazuh_readonly_status_v1"
ALLOWED_STATUSES = {
    "disabled_waiting_iwooos_wazuh_owner_gate",
    "misconfigured_missing_server_side_wazuh_env",
    "wazuh_auth_token_missing",
    "wazuh_readonly_metadata_unavailable",
    "wazuh_agent_registry_empty",
    "wazuh_agent_registry_below_expected",
    "readonly_metadata_available",
}
FORBIDDEN_RESPONSE_PATTERNS = [
    ("private_ipv4", re.compile(r"\b(?:10|127|172\.(?:1[6-9]|2\d|3[01])|192\.168)\.\d{1,3}\.\d{1,3}\b")),
    ("known_secret_shape", re.compile(r"Wooo-[0-9]{6,}")),
    ("token_like_field", re.compile(r'"(?:token|password|secret|private_key|runner_token)"\s*:', re.IGNORECASE)),
    (
        "raw_payload_marker",
        re.compile(r"raw[_ -]?(?:log|event|alert|body|request|response)", re.IGNORECASE),
    ),
    ("legacy_fake_soc_copy", re.compile(r"IWOOOS SOC Dashboard|Threat Blocked|Recent Automated Responses", re.IGNORECASE)),
]


@dataclass(frozen=True)
class HttpResult:
    status_code: int
    body: str
    content_type: str


def fetch_url(url: str, timeout_seconds: float) -> HttpResult:
    request = Request(url, headers={"Accept": "application/json", "User-Agent": "iwooos-wazuh-readback/1.0"})
    try:
        with urlopen(request, timeout=timeout_seconds) as response:
            body = response.read().decode("utf-8", errors="replace")
            return HttpResult(
                status_code=response.status,
                body=body,
                content_type=response.headers.get("content-type", ""),
            )
    except HTTPError as error:
        body = error.read().decode("utf-8", errors="replace")
        return HttpResult(
            status_code=error.code,
            body=body,
            content_type=error.headers.get("content-type", ""),
        )
    except URLError as error:
        raise SystemExit(f"BLOCKED production readback network_error={error}") from error


def load_json_body(body: str) -> dict[str, Any]:
    try:
        payload = json.loads(body)
    except json.JSONDecodeError as error:
        raise SystemExit(f"BLOCKED production readback non_json_response: {error}") from error
    if not isinstance(payload, dict):
        raise SystemExit("BLOCKED production readback response_not_object")
    return payload


def require_false(boundaries: dict[str, Any], key: str) -> None:
    if boundaries.get(key) is not False:
        raise SystemExit(f"BLOCKED production readback boundaries.{key}: expected false")


def require_zero(summary: dict[str, Any], key: str) -> None:
    if summary.get(key) != 0:
        raise SystemExit(f"BLOCKED production readback summary.{key}: expected 0")


def validate_payload(result: HttpResult, *, allow_predeploy_404: bool) -> dict[str, Any]:
    if result.status_code == 404 and allow_predeploy_404:
        payload = load_json_body(result.body)
        if payload.get("detail") != "Not Found":
            raise SystemExit("BLOCKED predeploy readback 404 body is not FastAPI Not Found")
        return {
            "schema_version": "iwooos_wazuh_production_readback_v1",
            "status": "predeploy_404_observed",
            "http_status": result.status_code,
            "runtime_gate_count": 0,
        }

    if result.status_code == 404:
        raise SystemExit("BLOCKED production readback returned 404; Wazuh FastAPI compatibility route is not deployed")
    if result.status_code not in {200, 502, 503}:
        raise SystemExit(f"BLOCKED production readback unexpected_http_status={result.status_code}")

    for pattern_id, pattern in FORBIDDEN_RESPONSE_PATTERNS:
        if pattern.search(result.body):
            raise SystemExit(f"BLOCKED production readback response leaked forbidden pattern {pattern_id}")

    payload = load_json_body(result.body)
    if payload.get("schema_version") != EXPECTED_SCHEMA:
        raise SystemExit(f"BLOCKED production readback schema_version={payload.get('schema_version')!r}")
    if payload.get("status") not in ALLOWED_STATUSES:
        raise SystemExit(f"BLOCKED production readback status={payload.get('status')!r}")
    if payload.get("mode") != "metadata_only_no_active_response_no_raw_payload":
        raise SystemExit(f"BLOCKED production readback mode={payload.get('mode')!r}")

    summary = payload.get("summary")
    if not isinstance(summary, dict):
        raise SystemExit("BLOCKED production readback summary missing")
    boundaries = payload.get("boundaries")
    if not isinstance(boundaries, dict):
        raise SystemExit("BLOCKED production readback boundaries missing")

    for key in [
        "wazuh_manager_query_accepted_count",
        "wazuh_event_accepted_count",
        "host_forensics_accepted_count",
        "active_response_authorized_count",
        "host_write_authorized_count",
        "runtime_gate_count",
    ]:
        require_zero(summary, key)

    for key in [
        "active_response_authorized",
        "host_write_authorized",
        "secret_value_collection_allowed",
        "raw_wazuh_payload_storage_allowed",
        "agent_identity_public_display_allowed",
        "internal_ip_public_display_allowed",
    ]:
        require_false(boundaries, key)

    if boundaries.get("not_authorization") is not True:
        raise SystemExit("BLOCKED production readback boundaries.not_authorization: expected true")

    return {
        "schema_version": "iwooos_wazuh_production_readback_v1",
        "status": "production_readback_passed",
        "http_status": result.status_code,
        "api_status": payload["status"],
        "configured": bool(payload.get("configured")),
        "runtime_gate_count": summary["runtime_gate_count"],
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="IwoooS Wazuh 只讀 API production readback 驗收")
    parser.add_argument("--url", default=DEFAULT_URL, help="production Wazuh read-only API URL")
    parser.add_argument("--timeout-seconds", type=float, default=10.0)
    parser.add_argument(
        "--allow-predeploy-404",
        action="store_true",
        help="僅供尚未部署時記錄 404 現況；正式部署驗收不得使用",
    )
    parser.add_argument("--json", action="store_true", help="輸出 JSON 摘要")
    args = parser.parse_args()

    result = fetch_url(args.url, args.timeout_seconds)
    report = validate_payload(result, allow_predeploy_404=args.allow_predeploy_404)
    if args.json:
        print(json.dumps(report, ensure_ascii=False, sort_keys=True))
    else:
        print(
            "WAZUH_READONLY_PRODUCTION_READBACK_OK "
            f"status={report['status']} "
            f"http={report['http_status']} "
            f"runtime_gate={report['runtime_gate_count']}"
        )
    return 0


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