fix(iwooos): 接上 Wazuh 只讀 API 邊界

This commit is contained in:
ogt
2026-06-24 21:19:23 +08:00
parent 6b9a09a01a
commit 6a83ae48a1
10 changed files with 1275 additions and 0 deletions

View File

@@ -0,0 +1,165 @@
"""
IwoooS 安全治理 API。
Wazuh 接線採用只讀 metadata 模式:預設關閉、不保存 raw payload、
不公開 agent 原名 / 內網 IP、不啟用 active response。
"""
from __future__ import annotations
import os
from base64 import b64encode
from typing import Any
from urllib.parse import urljoin, urlparse
import httpx
from fastapi import APIRouter
from fastapi.responses import JSONResponse
router = APIRouter(tags=["IwoooS Security"])
REQUEST_TIMEOUT_SECONDS = 5.0
def _wazuh_env() -> dict[str, str]:
return {
"enabled": os.getenv("IWOOOS_WAZUH_READONLY_ENABLED", "").strip().lower(),
"base_url": os.getenv("WAZUH_API_BASE_URL", "").strip(),
"username": os.getenv("WAZUH_API_USERNAME", "").strip(),
"password": os.getenv("WAZUH_API_PASSWORD", "").strip(),
}
def _https_url(value: str) -> str | None:
parsed = urlparse(value)
if parsed.scheme != "https" or not parsed.netloc:
return None
return value.rstrip("/") + "/"
def _boundary_response(status_text: str, http_status: int = 200) -> JSONResponse:
return JSONResponse(
status_code=http_status,
content={
"schema_version": "iwooos_wazuh_readonly_status_v1",
"status": status_text,
"mode": "metadata_only_no_active_response_no_raw_payload",
"configured": False,
"summary": {
"wazuh_platform_reported_count": 1,
"readonly_api_enabled_count": 0,
"wazuh_manager_query_accepted_count": 0,
"wazuh_event_accepted_count": 0,
"host_forensics_accepted_count": 0,
"active_response_authorized_count": 0,
"host_write_authorized_count": 0,
"runtime_gate_count": 0,
},
"boundaries": _boundaries(),
},
)
def _boundaries() -> dict[str, bool]:
return {
"active_response_authorized": False,
"host_write_authorized": False,
"secret_value_collection_allowed": False,
"raw_wazuh_payload_storage_allowed": False,
"agent_identity_public_display_allowed": False,
"internal_ip_public_display_allowed": False,
"not_authorization": True,
}
def _redacted_agent(agent: dict[str, Any], index: int) -> dict[str, Any]:
os_info = agent.get("os") if isinstance(agent.get("os"), dict) else {}
return {
"alias": f"agent-{index + 1:02d}",
"status": agent.get("status", "unknown"),
"os": os_info.get("platform") or os_info.get("name") or "unknown",
"last_seen_present": bool(agent.get("lastKeepAlive")),
}
async def _fetch_json(client: httpx.AsyncClient, url: str, headers: dict[str, str]) -> dict[str, Any]:
response = await client.get(url, headers=headers)
response.raise_for_status()
payload = response.json()
return payload if isinstance(payload, dict) else {}
async def _wazuh_readonly_status() -> JSONResponse:
env = _wazuh_env()
if env["enabled"] != "true":
return _boundary_response("disabled_waiting_iwooos_wazuh_owner_gate")
base_url = _https_url(env["base_url"])
if not base_url or not env["username"] or not env["password"]:
return _boundary_response("misconfigured_missing_server_side_wazuh_env", 503)
try:
auth_header = b64encode(f"{env['username']}:{env['password']}".encode("utf-8")).decode("ascii")
async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT_SECONDS) as client:
auth = await _fetch_json(
client,
urljoin(base_url, "security/user/authenticate"),
{"Authorization": f"Basic {auth_header}"},
)
token = (auth.get("data") or {}).get("token")
if not token:
return _boundary_response("wazuh_auth_token_missing", 502)
bearer_headers = {"Authorization": f"Bearer {token}"}
status_payload = await _fetch_json(
client,
urljoin(base_url, "agents/summary/status"),
bearer_headers,
)
agents_payload = await _fetch_json(
client,
urljoin(base_url, "agents?limit=100&select=id,status,os.name,os.platform,lastKeepAlive"),
bearer_headers,
)
except (httpx.HTTPError, ValueError):
return _boundary_response("wazuh_readonly_metadata_unavailable", 502)
connection = ((status_payload.get("data") or {}).get("connection") or {})
affected_items = ((agents_payload.get("data") or {}).get("affected_items") or [])
if not isinstance(affected_items, list):
affected_items = []
return JSONResponse(
content={
"schema_version": "iwooos_wazuh_readonly_status_v1",
"status": "readonly_metadata_available",
"mode": "metadata_only_no_active_response_no_raw_payload",
"configured": True,
"summary": {
"wazuh_platform_reported_count": 1,
"readonly_api_enabled_count": 1,
"agent_total": connection.get("total", len(affected_items)),
"agent_active": connection.get("active", 0),
"agent_disconnected": connection.get("disconnected", 0),
"agent_pending": connection.get("pending", 0),
"wazuh_manager_query_accepted_count": 0,
"wazuh_event_accepted_count": 0,
"host_forensics_accepted_count": 0,
"active_response_authorized_count": 0,
"host_write_authorized_count": 0,
"runtime_gate_count": 0,
},
"agents": [_redacted_agent(agent, index) for index, agent in enumerate(affected_items[:20])],
"boundaries": _boundaries(),
},
)
@router.get("/api/iwooos/wazuh")
async def get_iwooos_wazuh_readonly_status_compat() -> JSONResponse:
return await _wazuh_readonly_status()
@router.get("/api/v1/iwooos/wazuh")
async def get_iwooos_wazuh_readonly_status_v1() -> JSONResponse:
return await _wazuh_readonly_status()

View File

@@ -60,6 +60,7 @@ from src.api.v1 import (
# Import API routers
from src.api.v1 import health as health_v1
from src.api.v1 import incidents as incidents_v1 # Phase 6.4: Decision Proposal
from src.api.v1 import iwooos as iwooos_v1 # IwoooS security governance API
from src.api.v1 import knowledge as knowledge_v1 # KB Phase 1: Knowledge Base
from src.api.v1 import learning as learning_v1 # Phase D-G P0: Learning API
from src.api.v1 import metrics as metrics_v1 # Phase 7: Gold Metrics (真實血脈)
@@ -1035,6 +1036,7 @@ async def global_exception_handler(_request: Request, exc: Exception) -> JSONRes
# =============================================================================
# New v1 API routes
app.include_router(iwooos_v1.router, tags=["IwoooS Security"])
app.include_router(health_v1.router, prefix="/api/v1", tags=["Health"])
app.include_router(csrf_v1.router, prefix="/api/v1", tags=["Security"]) # Phase 20
app.include_router(dashboard_v1.router, prefix="/api/v1", tags=["Dashboard"])

View File

@@ -0,0 +1,127 @@
from __future__ import annotations
import httpx
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from src.api.v1.iwooos import router
def _client() -> TestClient:
app = FastAPI()
app.include_router(router)
return TestClient(app)
def test_iwooos_wazuh_compat_route_returns_disabled_boundary_by_default(monkeypatch: pytest.MonkeyPatch):
monkeypatch.delenv("IWOOOS_WAZUH_READONLY_ENABLED", raising=False)
monkeypatch.delenv("WAZUH_API_BASE_URL", raising=False)
monkeypatch.delenv("WAZUH_API_USERNAME", raising=False)
monkeypatch.delenv("WAZUH_API_PASSWORD", raising=False)
response = _client().get("/api/iwooos/wazuh")
assert response.status_code == 200
data = response.json()
assert data["schema_version"] == "iwooos_wazuh_readonly_status_v1"
assert data["status"] == "disabled_waiting_iwooos_wazuh_owner_gate"
assert data["configured"] is False
assert data["summary"]["runtime_gate_count"] == 0
assert data["boundaries"]["active_response_authorized"] is False
assert data["boundaries"]["host_write_authorized"] is False
assert data["boundaries"]["raw_wazuh_payload_storage_allowed"] is False
assert data["boundaries"]["internal_ip_public_display_allowed"] is False
def test_iwooos_wazuh_v1_route_rejects_missing_server_side_env(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setenv("IWOOOS_WAZUH_READONLY_ENABLED", "true")
monkeypatch.setenv("WAZUH_API_BASE_URL", "")
monkeypatch.setenv("WAZUH_API_USERNAME", "")
monkeypatch.setenv("WAZUH_API_PASSWORD", "")
response = _client().get("/api/v1/iwooos/wazuh")
assert response.status_code == 503
data = response.json()
assert data["status"] == "misconfigured_missing_server_side_wazuh_env"
assert data["configured"] is False
assert data["summary"]["runtime_gate_count"] == 0
def test_iwooos_wazuh_rejects_non_https_base_url(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setenv("IWOOOS_WAZUH_READONLY_ENABLED", "true")
monkeypatch.setenv("WAZUH_API_BASE_URL", "http://wazuh.example.test:55000")
monkeypatch.setenv("WAZUH_API_USERNAME", "readonly")
monkeypatch.setenv("WAZUH_API_PASSWORD", "placeholder")
response = _client().get("/api/iwooos/wazuh")
assert response.status_code == 503
data = response.json()
assert data["status"] == "misconfigured_missing_server_side_wazuh_env"
assert data["boundaries"]["secret_value_collection_allowed"] is False
def test_iwooos_wazuh_live_response_is_metadata_only(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setenv("IWOOOS_WAZUH_READONLY_ENABLED", "true")
monkeypatch.setenv("WAZUH_API_BASE_URL", "https://wazuh.example.test:55000")
monkeypatch.setenv("WAZUH_API_USERNAME", "readonly")
monkeypatch.setenv("WAZUH_API_PASSWORD", "placeholder")
def handler(request: httpx.Request) -> httpx.Response:
if request.url.path == "/security/user/authenticate":
return httpx.Response(200, json={"data": {"token": "token-value"}})
if request.url.path == "/agents/summary/status":
return httpx.Response(
200,
json={"data": {"connection": {"total": 2, "active": 1, "disconnected": 1, "pending": 0}}},
)
if request.url.path == "/agents":
return httpx.Response(
200,
json={
"data": {
"affected_items": [
{
"id": "001",
"name": "host-110-private-name",
"ip": "192.168.0.110",
"status": "active",
"os": {"platform": "linux"},
"lastKeepAlive": "2026-06-24T13:00:00Z",
}
]
}
},
)
return httpx.Response(404)
transport = httpx.MockTransport(handler)
original_async_client = httpx.AsyncClient
def client_factory(*args, **kwargs):
kwargs["transport"] = transport
return original_async_client(*args, **kwargs)
monkeypatch.setattr(httpx, "AsyncClient", client_factory)
response = _client().get("/api/iwooos/wazuh")
assert response.status_code == 200
data = response.json()
assert data["status"] == "readonly_metadata_available"
assert data["configured"] is True
assert data["summary"]["agent_total"] == 2
assert data["summary"]["runtime_gate_count"] == 0
assert data["agents"] == [
{
"alias": "agent-01",
"status": "active",
"os": "linux",
"last_seen_present": True,
}
]
assert "host-110-private-name" not in response.text
assert "192.168.0.110" not in response.text
assert "token-value" not in response.text

View File

@@ -164,6 +164,43 @@
**邊界**:本輪只有 read-only live checks、docs-only 更新與安全 handoff artifact 同步;沒有 Wazuh / SOC UI/API 修改,沒有 SSH 寫主機,沒有 Docker / Nginx / firewall / K8s / ArgoCD runtime 寫入,沒有 active scan沒有讀 secret沒有使用或保存聊天中的密碼。
## 2026-06-24IwoooS Wazuh public API 404 根因與只讀相容路由收斂
**背景**:使用者追問 Wazuh 為什麼仍不能訪問。只讀 production check 顯示 `https://awoooi.wooo.work/api/iwooos/wazuh``404 {"detail":"Not Found"}`,而 `https://awoooi.wooo.work/zh-TW/iwooos``200`。判定根因不是 IwoooS 前台不存在,而是 production `/api` 目前落到 FastAPI 後端;既有 Wazuh Next.js API route 沒有被 public gateway 暴露FastAPI 端也尚未提供相容 route。
**完成**
- 新增 FastAPI 端 `GET /api/iwooos/wazuh``GET /api/v1/iwooos/wazuh`,回傳與前端 Next.js route 一致的 `iwooos_wazuh_readonly_status_v1`
- route 預設回 `disabled_waiting_iwooos_wazuh_owner_gate`,不再需要用 404 表達未啟用狀態。
- live Wazuh 查詢仍需 `IWOOOS_WAZUH_READONLY_ENABLED=true``WAZUH_API_BASE_URL``WAZUH_API_USERNAME``WAZUH_API_PASSWORD` 全部由 server-side env 提供;未設定或非 HTTPS 時回 `misconfigured_missing_server_side_wazuh_env`
- live response 僅回 metadataagent alias、狀態、OS 類別、last_seen_present 與 aggregate counts不回 Wazuh raw payload、agent 原名、內網 IP、token 或 secret。
- 新增 `wazuh-readonly-route-boundary-guard.py`,同時掃 Next.js route、FastAPI route 與 IwoooS 前台,阻擋硬編 Wazuh 內網 URL / port、帳密、`NODE_TLS_REJECT_UNAUTHORIZED`、假 SOC dashboard、假 CVE、raw payload 或 legacy dashboard component 回流。
- `security-mirror-progress-guard.py` 已直接呼叫此 guard讓 Wazuh 接線邊界進入既有 IwoooS security mirror gate。
- 新增 `wazuh-readonly-production-readback.py`,供 release 後驗證 production `/api/iwooos/wazuh` 不再 404且 schema、status、0 / false 邊界與防洩漏條件都正確predeploy 404 只能用 `--allow-predeploy-404` 記錄現況,不可當正式驗收。
- 新增 `wazuh-readonly-release-gate.py``wazuh-readonly-release-gate.snapshot.json`,固定 source-side 已完成、Gitea push / production deploy / production readback 尚未完成,並由 `security-mirror-progress-guard.py` 驗證。
**驗證**
- `pytest apps/api/tests/test_iwooos_wazuh_api.py``4 passed`
- `python3 scripts/security/wazuh-readonly-route-boundary-guard.py --root .``WAZUH_READONLY_ROUTE_BOUNDARY_GUARD_OK route=2 public_ui_files=1 forbidden=0 runtime_gate=0`
- `python3 scripts/security/wazuh-readonly-release-gate.py --root .``WAZUH_READONLY_RELEASE_GATE_OK source=1 push=0 deploy=0 readback=0 runtime_gate=0`
- `python3 scripts/security/security-mirror-progress-guard.py --root .``SECURITY_MIRROR_PROGRESS_GUARD_OK`
- `python3 -m py_compile apps/api/src/api/v1/iwooos.py scripts/security/wazuh-readonly-route-boundary-guard.py scripts/security/security-mirror-progress-guard.py` 通過。
- `python3 scripts/security/wazuh-readonly-production-readback.py --allow-predeploy-404 --json` 可記錄尚未部署現況;正式部署後需不加 allow flag且不得回 404。
- `git diff --check` 通過。
**完成度 / 狀態**
- Wazuh public API 404 source-side 修補:`100%`
- Wazuh route boundary source guard`100%`
- Production readback 驗收腳本:`100%`
- Wazuh release gate snapshot / guard`100%`
- Production deploy / readback`0%`,尚未推送與部署。
- Wazuh server-side env enable`0%`,尚未由 secrets / env gate 啟用。
- Wazuh event refs、host forensic refs、containment decision、recovery proof accepted全部 `0%`
- active response、host write、Kali active scan、firewall / Nginx / Docker / K8s runtime action全部 `0 / false`
**邊界**:本輪只做 source-side API 相容路由、測試與 guard沒有 SSH、沒有查 live Wazuh API、沒有讀或保存 secret、沒有改 Nginx / firewall / Docker / K8s、沒有 active scan、沒有 Wazuh active response、沒有 Telegram 實發、沒有 production deploy。
**Release handoff 補充**:受控 workspace 的 Gitea HTTPS push 因非互動式 credential 缺失失敗;本輪未複製或使用舊 workspace 內嵌明文 token。已新增 `docs/security/IWOOOS-WAZUH-READONLY-API-RELEASE-HANDOFF.md`,供具備正式 Gitea / release 權限的 lane 合併 `codex/iwooos-wazuh-boundary-guard-20260624` 分支 HEAD 或同等 patch並以 production `/api/iwooos/wazuh` readback 驗證不再 404。
## 2026-06-2421:04 recovery readback 與 MOMO V10.651 雙機基準收斂
**背景**:前一輪 MOMO workspace readback 指到 `V10.646`,但 21:04 live health 已回 `V10.651`。因此本輪重新比對 Gitea `wooo/ewoooc` `main`、正式站 `/health`、Mac Mini / MacBook Pro Codex workspace 與 full-stack cold-start避免「網站可用」和「版本 / 資料最新」互相混淆。

View File

@@ -0,0 +1,144 @@
# IwoooS Wazuh 只讀 API Release Handoff
> 狀態source-side 修補完成,等待具備正式 Gitea push / release 權限的 lane 合併與部署。
> 本文件不包含 secret、token、內網 Wazuh URL、raw log、raw Wazuh payload 或工作視窗逐字稿。
## 根因判定
`https://awoooi.wooo.work/api/iwooos/wazuh` production 目前回 `404 {"detail":"Not Found"}`,同時 `https://awoooi.wooo.work/zh-TW/iwooos``200`
判定根因:
- production `/api` 目前落到 FastAPI 後端。
- 既有 `apps/web/src/app/api/iwooos/wazuh/route.ts` 是 Next.js route沒有被 public gateway 暴露到這條 production path。
- FastAPI 後端原本沒有 `/api/iwooos/wazuh` 相容 route因此回 404。
- 這個 404 不能被解讀成 Wazuh manager 一定故障,也不能被解讀成 Wazuh 已未安裝。
## Source-Side 修補
本地 commit
- `codex/iwooos-wazuh-boundary-guard-20260624` 分支 HEAD`fix(iwooos): 接上 Wazuh 只讀 API 邊界`
變更範圍:
- `apps/api/src/api/v1/iwooos.py`
- `apps/api/src/main.py`
- `apps/api/tests/test_iwooos_wazuh_api.py`
- `scripts/security/wazuh-readonly-route-boundary-guard.py`
- `scripts/security/wazuh-readonly-production-readback.py`
- `scripts/security/wazuh-readonly-release-gate.py`
- `scripts/security/security-mirror-progress-guard.py`
- `docs/security/wazuh-readonly-release-gate.snapshot.json`
- `docs/LOGBOOK.md`
完成內容:
- 新增 FastAPI `GET /api/iwooos/wazuh`
- 新增 FastAPI `GET /api/v1/iwooos/wazuh`
- 預設回 `disabled_waiting_iwooos_wazuh_owner_gate`,避免 production 繼續用 404 表示未啟用。
- live Wazuh 查詢仍需 `IWOOOS_WAZUH_READONLY_ENABLED=true` 與 server-side env`WAZUH_API_BASE_URL``WAZUH_API_USERNAME``WAZUH_API_PASSWORD`
- 強制 Wazuh base URL 使用 HTTPS。
- 回傳資料只允許 metadataagent alias、status、OS 類別、last_seen_present 與 aggregate counts。
- 不回傳 raw Wazuh payload、agent 原名、內網 IP、token、password 或 secret。
- 新增 source guard阻擋硬編 Wazuh 內網 URL / port、帳密、關 TLS、假 SOC dashboard、假 CVE、raw payload 與 legacy dashboard component 回流。
- 新增 production readback 腳本,部署後可直接驗證 public API 不再 404、schema / status / boundary 正確,且沒有 raw payload、內網 IP、agent 原名或 secret 洩漏。
- 新增 release gate snapshot 與 guard固定 source-side 已完成、Gitea push / production deploy / production readback 尚未完成,避免後續把 predeploy 404 誤判成通過。
## 已完成驗證
已在 `/Users/ogt/codex-workspaces/awoooi-dev` 執行:
```bash
pytest apps/api/tests/test_iwooos_wazuh_api.py
python3 scripts/security/wazuh-readonly-route-boundary-guard.py --root .
python3 scripts/security/wazuh-readonly-release-gate.py --root .
python3 scripts/security/security-mirror-progress-guard.py --root .
python3 scripts/ops/doc-secrets-sanity-check.py docs apps/api/src/api/v1/iwooos.py apps/web/src/app/api/iwooos/wazuh/route.ts scripts/security/wazuh-readonly-route-boundary-guard.py
python3 -m py_compile apps/api/src/api/v1/iwooos.py scripts/security/wazuh-readonly-route-boundary-guard.py scripts/security/security-mirror-progress-guard.py
git diff --check
```
驗證結果:
- `pytest apps/api/tests/test_iwooos_wazuh_api.py``4 passed`
- `wazuh-readonly-route-boundary-guard``route=2 public_ui_files=1 forbidden=0 runtime_gate=0`
- `wazuh-readonly-release-gate``source=1 push=0 deploy=0 readback=0 runtime_gate=0`
- `security-mirror-progress-guard``SECURITY_MIRROR_PROGRESS_GUARD_OK`
- `doc-secrets-sanity-check``DOC_SECRET_SANITY_OK scanned_files=967`
- `py_compile`:通過。
- `git diff --check`:通過。
尚未部署前的 production 現況記錄:
```bash
python3 scripts/security/wazuh-readonly-production-readback.py --allow-predeploy-404 --json
```
預期只可回 `status=predeploy_404_observed`。正式部署驗收不得加 `--allow-predeploy-404`
## Release 前 Gate
合併 / 部署前需確認:
- 使用具備正式權限的 Gitea lane 合併 `codex/iwooos-wazuh-boundary-guard-20260624` 分支 HEAD 或同等 patch不得 force push。
- 不得複製舊 workspace 的內嵌明文 Gitea token。
- 不得把 Wazuh URL、帳密、token、cookie、private key、runner token 或 webhook secret 寫入 repo。
- 不得為了讓 API 變 200 而直接改 Nginx、Docker、K8s、firewall、Wazuh manager、Wazuh rule、Wazuh decoder 或 Wazuh active response。
- 若要啟用 live metadata query必須由正式 secrets / env 注入 `IWOOOS_WAZUH_READONLY_ENABLED=true` 與 Wazuh server-side env且要先有 owner gate。
## Production Readback 預期
部署後、尚未啟用 Wazuh env 時:
```bash
curl -sS https://awoooi.wooo.work/api/iwooos/wazuh
```
預期:
- HTTP `200`
- `schema_version=iwooos_wazuh_readonly_status_v1`
- `status=disabled_waiting_iwooos_wazuh_owner_gate`
- `configured=false`
- `runtime_gate_count=0`
- `active_response_authorized=false`
- `host_write_authorized=false`
- 不含內網 IP、agent 原名、token、password、raw payload。
正式驗收命令:
```bash
python3 scripts/security/wazuh-readonly-production-readback.py --json
```
正式驗收不接受 404若仍回 404代表 FastAPI 相容 route 尚未部署或 gateway 尚未接到新 API。
若 owner gate 與 server-side env 已正式啟用:
- 成功時可回 `readonly_metadata_available`
- Wazuh 不可達時可回 `wazuh_readonly_metadata_unavailable`
- 任何情況都不得回 raw payload、agent 原名、內網 IP、secret。
- 任何情況都不得因 route 可用而自動打開 active response、host write、Kali active scan 或 SOAR action。
## 完成度
| 項目 | 完成度 | 狀態 |
|---|---:|---|
| Wazuh public API 404 source-side 修補 | `100%` | 已完成本地分支 HEAD |
| Wazuh route boundary source guard | `100%` | 已納入 `security-mirror-progress-guard` |
| Production readback 驗收腳本 | `100%` | 已完成;正式部署後不得接受 404 |
| Wazuh release gate snapshot / guard | `100%` | 已完成;固定 push/deploy/readback 仍 blocked |
| Gitea push | `0%` | 受控 workspace HTTPS credential 缺失 |
| Production deploy / readback | `0%` | 等待 release lane |
| Wazuh server-side env enable | `0%` | 等待 owner gate 與 secrets 注入 |
| Wazuh event refs / host forensic refs accepted | `0%` | 尚未收到合格證據 |
| Wazuh active response / host write / Kali active scan | `0%` | 必須維持 false |
## 下一步優先序
1. 解決受控 workspace Gitea HTTPS push 認證,或由正式 release lane 合併 `codex/iwooos-wazuh-boundary-guard-20260624` 分支 HEAD。
2. 部署後先驗證 `/api/iwooos/wazuh` 不再 404且預設 disabled 邊界正確。
3. 另開 owner gate 決定是否啟用 server-side Wazuh read-only metadata query。
4. 收件 Wazuh manager health ref、agent status ref、event refs、host forensic refs 與 containment / recovery proof。
5. 仍禁止 active response、host write、firewall / Nginx / Docker / K8s runtime action、Kali active scan、secret 明文收集。

View File

@@ -0,0 +1,96 @@
{
"execution_boundaries": {
"agent_identity_public_display_allowed": false,
"force_push_allowed": false,
"host_read_authorized": false,
"host_write_authorized": false,
"internal_ip_public_display_allowed": false,
"kali_active_scan_authorized": false,
"not_authorization": true,
"production_deploy_authorized": false,
"raw_wazuh_payload_storage_allowed": false,
"runtime_execution_authorized": false,
"secret_value_collection_allowed": false,
"wazuh_active_response_authorized": false,
"wazuh_api_live_query_authorized": false
},
"generated_at": "2026-06-24T21:36:00+08:00",
"missing_required_source_paths": [],
"mode": "repo_release_gate_no_runtime_no_secret_collection",
"operator_interpretation": [
"此 gate 通過不代表 production 已部署,只代表 source-side Wazuh read-only API 與 guard 可交接。",
"正式 release 前不得用 predeploy 404 當成功,也不得為了修 404 直接改 Nginx、Docker、K8s、firewall 或 Wazuh secret。",
"live Wazuh metadata query 必須另走 owner gate 與 server-side envactive response、host write、Kali active scan 仍為 0 / false。"
],
"release_gates": [
{
"gate_id": "source_side_fastapi_route",
"required_evidence": "FastAPI /api/iwooos/wazuh 與 /api/v1/iwooos/wazuh source path present",
"runtime_authorized": false,
"status": "passed"
},
{
"gate_id": "source_boundary_guard",
"required_evidence": "wazuh-readonly-route-boundary-guard.py 通過",
"runtime_authorized": false,
"status": "passed"
},
{
"gate_id": "production_readback_script",
"required_evidence": "wazuh-readonly-production-readback.py 可在 release 後不接受 404",
"runtime_authorized": false,
"status": "passed"
},
{
"gate_id": "gitea_branch_push",
"required_evidence": "具備正式權限的 lane 推送或合併 codex/iwooos-wazuh-boundary-guard-20260624",
"runtime_authorized": false,
"status": "blocked_credential_required"
},
{
"gate_id": "production_deploy",
"required_evidence": "Gitea CD / deploy marker 指向已合併 Wazuh fix 的 commit",
"runtime_authorized": false,
"status": "blocked_waiting_release_lane"
},
{
"gate_id": "production_readback",
"required_evidence": "python3 scripts/security/wazuh-readonly-production-readback.py --json 通過且不回 404",
"runtime_authorized": false,
"status": "blocked_waiting_deploy"
},
{
"gate_id": "wazuh_live_metadata_env",
"required_evidence": "server-side env 與 owner gate不得硬編 secret",
"runtime_authorized": false,
"status": "blocked_owner_gate_required"
}
],
"required_source_paths": [
"apps/api/src/api/v1/iwooos.py",
"apps/api/tests/test_iwooos_wazuh_api.py",
"apps/web/src/app/api/iwooos/wazuh/route.ts",
"docs/security/IWOOOS-WAZUH-READONLY-API-RELEASE-HANDOFF.md",
"scripts/security/wazuh-readonly-production-readback.py",
"scripts/security/wazuh-readonly-route-boundary-guard.py"
],
"schema_version": "iwooos_wazuh_readonly_release_gate_v1",
"status": "blocked_waiting_gitea_push_and_production_deploy",
"summary": {
"active_response_authorized_count": 0,
"gitea_push_complete_count": 0,
"host_forensics_ref_accepted_count": 0,
"host_write_authorized_count": 0,
"missing_required_source_path_count": 0,
"predeploy_404_observed_count": 1,
"production_deploy_complete_count": 0,
"production_readback_passed_count": 0,
"production_readback_script_complete_count": 1,
"release_handoff_complete_count": 1,
"route_boundary_guard_complete_count": 1,
"runtime_gate_count": 0,
"source_side_fix_complete_count": 1,
"wazuh_event_ref_accepted_count": 0,
"wazuh_server_side_env_enabled_count": 0
}
}

View File

@@ -87,6 +87,14 @@ def validate(root: Path) -> None:
str(root / "scripts" / "security" / "public-frontend-env-guard.py")
)
public_frontend_env_guard["validate"](root)
wazuh_readonly_route_boundary_guard = runpy.run_path(
str(root / "scripts" / "security" / "wazuh-readonly-route-boundary-guard.py")
)
wazuh_readonly_route_boundary_guard["validate"](root)
wazuh_readonly_release_gate = runpy.run_path(
str(root / "scripts" / "security" / "wazuh-readonly-release-gate.py")
)
wazuh_readonly_release_gate["validate"](root)
telegram_alert_readability_guard = runpy.run_path(
str(root / "scripts" / "security" / "telegram-alert-readability-guard.py")
)

View File

@@ -0,0 +1,184 @@
#!/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",
"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[_ -]?(?:wazuh|payload|log)", 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())

View File

@@ -0,0 +1,186 @@
#!/usr/bin/env python3
"""
IwoooS Wazuh 只讀 API release gate。
本工具只檢查 repo 內 source、snapshot 與 gate 狀態,不連 production、
不查 Wazuh、不讀 secret、不做 deploy。目的在於固定「source-side 已完成」
與「Gitea push / production deploy / production readback 尚未完成」的界線。
"""
from __future__ import annotations
import argparse
import json
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/wazuh-readonly-release-gate.snapshot.json")
REQUIRED_SOURCE_PATHS = [
"apps/api/src/api/v1/iwooos.py",
"apps/api/tests/test_iwooos_wazuh_api.py",
"apps/web/src/app/api/iwooos/wazuh/route.ts",
"docs/security/IWOOOS-WAZUH-READONLY-API-RELEASE-HANDOFF.md",
"scripts/security/wazuh-readonly-production-readback.py",
"scripts/security/wazuh-readonly-route-boundary-guard.py",
]
def now_iso() -> str:
return datetime.now(TAIPEI).replace(microsecond=0).isoformat()
def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]:
missing_paths = [path for path in REQUIRED_SOURCE_PATHS if not (root / path).exists()]
source_ready = not missing_paths
return {
"schema_version": "iwooos_wazuh_readonly_release_gate_v1",
"generated_at": generated_at or now_iso(),
"status": "blocked_waiting_gitea_push_and_production_deploy",
"mode": "repo_release_gate_no_runtime_no_secret_collection",
"required_source_paths": REQUIRED_SOURCE_PATHS,
"summary": {
"source_side_fix_complete_count": 1 if source_ready else 0,
"route_boundary_guard_complete_count": 1 if (root / "scripts/security/wazuh-readonly-route-boundary-guard.py").exists() else 0,
"production_readback_script_complete_count": 1 if (root / "scripts/security/wazuh-readonly-production-readback.py").exists() else 0,
"release_handoff_complete_count": 1 if (root / "docs/security/IWOOOS-WAZUH-READONLY-API-RELEASE-HANDOFF.md").exists() else 0,
"missing_required_source_path_count": len(missing_paths),
"gitea_push_complete_count": 0,
"production_deploy_complete_count": 0,
"production_readback_passed_count": 0,
"predeploy_404_observed_count": 1,
"wazuh_server_side_env_enabled_count": 0,
"wazuh_event_ref_accepted_count": 0,
"host_forensics_ref_accepted_count": 0,
"active_response_authorized_count": 0,
"host_write_authorized_count": 0,
"runtime_gate_count": 0,
},
"release_gates": [
{
"gate_id": "source_side_fastapi_route",
"status": "passed",
"required_evidence": "FastAPI /api/iwooos/wazuh 與 /api/v1/iwooos/wazuh source path present",
"runtime_authorized": False,
},
{
"gate_id": "source_boundary_guard",
"status": "passed",
"required_evidence": "wazuh-readonly-route-boundary-guard.py 通過",
"runtime_authorized": False,
},
{
"gate_id": "production_readback_script",
"status": "passed",
"required_evidence": "wazuh-readonly-production-readback.py 可在 release 後不接受 404",
"runtime_authorized": False,
},
{
"gate_id": "gitea_branch_push",
"status": "blocked_credential_required",
"required_evidence": "具備正式權限的 lane 推送或合併 codex/iwooos-wazuh-boundary-guard-20260624",
"runtime_authorized": False,
},
{
"gate_id": "production_deploy",
"status": "blocked_waiting_release_lane",
"required_evidence": "Gitea CD / deploy marker 指向已合併 Wazuh fix 的 commit",
"runtime_authorized": False,
},
{
"gate_id": "production_readback",
"status": "blocked_waiting_deploy",
"required_evidence": "python3 scripts/security/wazuh-readonly-production-readback.py --json 通過且不回 404",
"runtime_authorized": False,
},
{
"gate_id": "wazuh_live_metadata_env",
"status": "blocked_owner_gate_required",
"required_evidence": "server-side env 與 owner gate不得硬編 secret",
"runtime_authorized": False,
},
],
"execution_boundaries": {
"runtime_execution_authorized": False,
"production_deploy_authorized": False,
"wazuh_api_live_query_authorized": False,
"wazuh_active_response_authorized": False,
"host_read_authorized": False,
"host_write_authorized": False,
"kali_active_scan_authorized": False,
"secret_value_collection_allowed": False,
"raw_wazuh_payload_storage_allowed": False,
"internal_ip_public_display_allowed": False,
"agent_identity_public_display_allowed": False,
"force_push_allowed": False,
"not_authorization": True,
},
"missing_required_source_paths": missing_paths,
"operator_interpretation": [
"此 gate 通過不代表 production 已部署,只代表 source-side Wazuh read-only API 與 guard 可交接。",
"正式 release 前不得用 predeploy 404 當成功,也不得為了修 404 直接改 Nginx、Docker、K8s、firewall 或 Wazuh secret。",
"live Wazuh metadata query 必須另走 owner gate 與 server-side envactive response、host write、Kali active scan 仍為 0 / false。",
],
}
def load_json(path: Path) -> dict[str, Any]:
return json.loads(path.read_text(encoding="utf-8"))
def validate(root: Path) -> None:
report = build_report(root)
snapshot_path = root / SNAPSHOT_PATH
if not snapshot_path.exists():
raise SystemExit(f"BLOCKED Wazuh release gate snapshot missing: {SNAPSHOT_PATH.as_posix()}")
snapshot = load_json(snapshot_path)
expected_summary = report["summary"]
for key, expected in expected_summary.items():
actual = snapshot.get("summary", {}).get(key)
if actual != expected:
raise SystemExit(f"BLOCKED Wazuh release gate summary.{key}: expected {expected!r}, got {actual!r}")
if snapshot.get("schema_version") != "iwooos_wazuh_readonly_release_gate_v1":
raise SystemExit("BLOCKED Wazuh release gate schema_version mismatch")
if snapshot.get("status") != "blocked_waiting_gitea_push_and_production_deploy":
raise SystemExit("BLOCKED Wazuh release gate status mismatch")
for key, value in snapshot.get("execution_boundaries", {}).items():
if key == "not_authorization":
if value is not True:
raise SystemExit("BLOCKED Wazuh release gate not_authorization must be true")
elif value is not False:
raise SystemExit(f"BLOCKED Wazuh release gate execution_boundaries.{key}: expected false")
def main() -> int:
parser = argparse.ArgumentParser(description="IwoooS Wazuh 只讀 API release gate")
parser.add_argument("--root", default=".", help="repository root")
parser.add_argument("--output", help="寫出 JSON 報告")
parser.add_argument("--generated-at", help="固定報告時間,供 committed snapshot 使用")
args = parser.parse_args()
root = Path(args.root).resolve()
report = build_report(root, args.generated_at)
if args.output:
output = Path(args.output)
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
validate(root)
summary = report["summary"]
print(
"WAZUH_READONLY_RELEASE_GATE_OK "
f"source={summary['source_side_fix_complete_count']} "
f"push={summary['gitea_push_complete_count']} "
f"deploy={summary['production_deploy_complete_count']} "
f"readback={summary['production_readback_passed_count']} "
f"runtime_gate={summary['runtime_gate_count']}"
)
return 0
if __name__ == "__main__":
sys.exit(main())

View File

@@ -0,0 +1,326 @@
#!/usr/bin/env python3
"""
檢查 IwoooS Wazuh 只讀 API 接線邊界。
本 guard 只掃描 repo 內 source不連線 Wazuh、不讀主機、不收 secret、
不啟用 active response也不做任何 production runtime 動作。
"""
from __future__ import annotations
import argparse
import json
import re
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
TAIPEI = timezone(timedelta(hours=8))
NEXT_ROUTE_PATH = Path("apps/web/src/app/api/iwooos/wazuh/route.ts")
BACKEND_ROUTE_PATH = Path("apps/api/src/api/v1/iwooos.py")
PUBLIC_PAGE_PATH = Path("apps/web/src/app/[locale]/iwooos/page.tsx")
PUBLIC_COMPONENT_ROOT = Path("apps/web/src/components/iwooos")
@dataclass(frozen=True)
class ForbiddenPattern:
pattern_id: str
pattern: re.Pattern[str]
scope: str
ROUTE_REQUIRED_TOKENS = [
"IWOOOS_WAZUH_READONLY_ENABLED",
"WAZUH_API_BASE_URL",
"WAZUH_API_USERNAME",
"WAZUH_API_PASSWORD",
"process.env",
"requireHttpsBaseUrl",
"metadata_only_no_active_response_no_raw_payload",
"active_response_authorized: false",
"host_write_authorized: false",
"runtime_gate_count: 0",
"secret_value_collection_allowed: false",
"raw_wazuh_payload_storage_allowed: false",
"agent_identity_public_display_allowed: false",
"internal_ip_public_display_allowed: false",
"not_authorization: true",
"redactedAgent",
"alias: `agent-",
]
BACKEND_REQUIRED_TOKENS = [
"/api/iwooos/wazuh",
"/api/v1/iwooos/wazuh",
"IWOOOS_WAZUH_READONLY_ENABLED",
"WAZUH_API_BASE_URL",
"WAZUH_API_USERNAME",
"WAZUH_API_PASSWORD",
"metadata_only_no_active_response_no_raw_payload",
"active_response_authorized_count",
"host_write_authorized_count",
"runtime_gate_count",
"raw_wazuh_payload_storage_allowed",
"internal_ip_public_display_allowed",
"_redacted_agent",
]
FORBIDDEN_PATTERNS = [
ForbiddenPattern(
"hardcoded_wazuh_private_url",
re.compile(r"https?://(?:10|127|172\.(?:1[6-9]|2\d|3[01])|192\.168)\.[^\s'\"`]+", re.IGNORECASE),
"route_and_public_ui",
),
ForbiddenPattern(
"wazuh_default_api_port_literal",
re.compile(r"(?<![A-Za-z0-9_])55000(?![A-Za-z0-9_])"),
"route_and_public_ui",
),
ForbiddenPattern(
"node_tls_reject_unauthorized_disabled",
re.compile(r"NODE_TLS_REJECT_UNAUTHORIZED|rejectUnauthorized\s*:\s*false", re.IGNORECASE),
"route_and_public_ui",
),
ForbiddenPattern(
"hardcoded_wazuh_password_assignment",
re.compile(r"WAZUH_API_PASSWORD\s*=\s*['\"]|password\s*:\s*['\"]|passwd\s*:\s*['\"]", re.IGNORECASE),
"route",
),
ForbiddenPattern(
"hardcoded_wazuh_username_assignment",
re.compile(r"WAZUH_API_USERNAME\s*=\s*['\"]|username\s*:\s*['\"]|user\s*:\s*['\"]", re.IGNORECASE),
"route",
),
ForbiddenPattern(
"known_secret_shape",
re.compile(r"Wooo-[0-9]{6,}"),
"route_and_public_ui",
),
ForbiddenPattern(
"fake_soc_dashboard_copy",
re.compile(
r"IWOOOS SOC Dashboard|Threat Blocked|Wazuh Agents Status|Protected Nodes|Recent Automated Responses",
re.IGNORECASE,
),
"public_ui",
),
ForbiddenPattern(
"fake_cve_or_alert_fixture",
re.compile(r"CVE-2025-55182|recentAlerts\s*:\s*\[|vulnerabilities\s*:\s*\[", re.IGNORECASE),
"public_ui",
),
ForbiddenPattern(
"legacy_wazuh_dashboard_component",
re.compile(r"FifaWazuhMonitor"),
"public_ui",
),
ForbiddenPattern(
"raw_wazuh_payload_public_copy",
re.compile(r"rawWazuhPayload|raw_wazuh_payload\s*[:=]\s*[\[{]|raw Wazuh payload\s*[:=]", re.IGNORECASE),
"public_ui",
),
]
def now_iso() -> str:
return datetime.now(TAIPEI).replace(microsecond=0).isoformat()
def read_text(path: Path) -> str:
return path.read_text(encoding="utf-8")
def collect_public_ui_files(root: Path) -> list[Path]:
files: list[Path] = []
page = root / PUBLIC_PAGE_PATH
if page.exists():
files.append(PUBLIC_PAGE_PATH)
component_root = root / PUBLIC_COMPONENT_ROOT
if component_root.exists():
files.extend(
path.relative_to(root)
for path in sorted(component_root.rglob("*"))
if path.is_file() and path.suffix in {".ts", ".tsx", ".js", ".jsx"}
)
return files
def source_lines(text: str) -> list[tuple[int, str]]:
return list(enumerate(text.splitlines(), start=1))
def pattern_applies(pattern: ForbiddenPattern, source_kind: str) -> bool:
if pattern.scope == "route_and_public_ui":
return True
if pattern.scope == "route":
return source_kind == "route"
if pattern.scope == "public_ui":
return source_kind == "public_ui"
return False
def collect_forbidden_matches(root: Path) -> list[dict[str, Any]]:
targets: list[tuple[str, Path]] = [("route", NEXT_ROUTE_PATH), ("route", BACKEND_ROUTE_PATH)]
targets.extend(("public_ui", path) for path in collect_public_ui_files(root))
matches: list[dict[str, Any]] = []
for source_kind, relative_path in targets:
path = root / relative_path
if not path.exists():
continue
for line_number, line in source_lines(read_text(path)):
for forbidden in FORBIDDEN_PATTERNS:
if pattern_applies(forbidden, source_kind) and forbidden.pattern.search(line):
matches.append(
{
"path": relative_path.as_posix(),
"line": line_number,
"pattern_id": forbidden.pattern_id,
"source_kind": source_kind,
}
)
return matches
def collect_missing_required_tokens(route_text: str, required_tokens: list[str]) -> list[str]:
return [token for token in required_tokens if token not in route_text]
def build_report(root: Path, generated_at: str | None = None) -> dict[str, Any]:
next_route = root / NEXT_ROUTE_PATH
backend_route = root / BACKEND_ROUTE_PATH
next_route_present = next_route.exists()
backend_route_present = backend_route.exists()
next_route_text = read_text(next_route) if next_route_present else ""
backend_route_text = read_text(backend_route) if backend_route_present else ""
public_ui_files = collect_public_ui_files(root)
missing_required_tokens = {
NEXT_ROUTE_PATH.as_posix(): collect_missing_required_tokens(next_route_text, ROUTE_REQUIRED_TOKENS),
BACKEND_ROUTE_PATH.as_posix(): collect_missing_required_tokens(backend_route_text, BACKEND_REQUIRED_TOKENS),
}
missing_required_token_count = sum(len(tokens) for tokens in missing_required_tokens.values())
forbidden_matches = collect_forbidden_matches(root)
return {
"schema_version": "wazuh_readonly_route_boundary_guard_v1",
"generated_at": generated_at or now_iso(),
"status": (
"pass"
if next_route_present and backend_route_present and not missing_required_token_count and not forbidden_matches
else "blocked"
),
"mode": "repo_source_scan_no_runtime_no_secret_collection",
"guarded_route_paths": [NEXT_ROUTE_PATH.as_posix(), BACKEND_ROUTE_PATH.as_posix()],
"guarded_public_ui_paths": [path.as_posix() for path in public_ui_files],
"required_route_tokens": {
NEXT_ROUTE_PATH.as_posix(): ROUTE_REQUIRED_TOKENS,
BACKEND_ROUTE_PATH.as_posix(): BACKEND_REQUIRED_TOKENS,
},
"forbidden_pattern_ids": [pattern.pattern_id for pattern in FORBIDDEN_PATTERNS],
"summary": {
"route_present_count": int(next_route_present) + int(backend_route_present),
"next_route_present_count": 1 if next_route_present else 0,
"backend_route_present_count": 1 if backend_route_present else 0,
"public_ui_file_count": len(public_ui_files),
"required_token_count": len(ROUTE_REQUIRED_TOKENS) + len(BACKEND_REQUIRED_TOKENS),
"missing_required_token_count": missing_required_token_count,
"forbidden_pattern_count": len(FORBIDDEN_PATTERNS),
"forbidden_match_count": len(forbidden_matches),
"readonly_api_default_closed_count": sum(
"IWOOOS_WAZUH_READONLY_ENABLED" in text for text in [next_route_text, backend_route_text]
),
"server_side_env_required_count": sum(
token in text
for text in [next_route_text, backend_route_text]
for token in ["WAZUH_API_BASE_URL", "WAZUH_API_USERNAME", "WAZUH_API_PASSWORD"]
),
"tls_disable_match_count": sum(
1
for item in forbidden_matches
if item["pattern_id"] == "node_tls_reject_unauthorized_disabled"
),
"hardcoded_private_url_match_count": sum(
1 for item in forbidden_matches if item["pattern_id"] == "hardcoded_wazuh_private_url"
),
"fake_soc_fixture_match_count": sum(
1
for item in forbidden_matches
if item["pattern_id"] in {"fake_soc_dashboard_copy", "fake_cve_or_alert_fixture"}
),
"active_response_authorized_count": 0,
"host_write_authorized_count": 0,
"runtime_gate_count": 0,
"action_button_count": 0,
},
"execution_boundaries": {
"runtime_execution_authorized": False,
"wazuh_api_live_query_authorized": False,
"wazuh_active_response_authorized": False,
"host_read_authorized": False,
"host_write_authorized": False,
"secret_value_collection_allowed": False,
"raw_wazuh_payload_storage_allowed": False,
"agent_identity_public_display_allowed": False,
"internal_ip_public_display_allowed": False,
"frontend_public_raw_alert_display_allowed": False,
"action_buttons_allowed": False,
"not_authorization": True,
},
"missing_required_tokens": missing_required_tokens,
"forbidden_matches": forbidden_matches,
"operator_interpretation": [
"Wazuh API code path 必須預設關閉,只有 server-side env 與 owner gate 允許只讀 metadata 查詢。",
"不得硬編 Wazuh 內網 URL、使用者、密碼或關閉 TLS 驗證。",
"前台不得顯示假 SOC dashboard、假 CVE、假 automated response、raw payload、agent 原名或內網 IP。",
"此 guard 通過只代表 source 邊界合格,不代表 Wazuh live query、active response、host containment 或 runtime remediation 已授權。",
],
}
def validate(root: Path) -> None:
report = build_report(root)
errors: list[str] = []
if report["summary"]["next_route_present_count"] != 1:
errors.append(f"{NEXT_ROUTE_PATH.as_posix()}: Wazuh Next.js 只讀 route 不存在")
if report["summary"]["backend_route_present_count"] != 1:
errors.append(f"{BACKEND_ROUTE_PATH.as_posix()}: Wazuh FastAPI 相容 route 不存在")
for path, tokens in report["missing_required_tokens"].items():
for token in tokens:
errors.append(f"{path}: 缺少必要只讀邊界 token {token!r}")
for item in report["forbidden_matches"]:
errors.append(
f"{item['path']}:{item['line']}: 命中 Wazuh 邊界 forbidden pattern {item['pattern_id']}"
)
if errors:
raise SystemExit("BLOCKED Wazuh readonly route boundary guard:\n" + "\n".join(f"- {error}" for error in errors))
def main() -> None:
parser = argparse.ArgumentParser(description="檢查 IwoooS Wazuh 只讀 API 接線邊界")
parser.add_argument("--root", default=".", help="repository root")
parser.add_argument("--output", help="寫出 JSON 報告")
parser.add_argument("--generated-at", help="固定報告時間,供 committed snapshot 使用")
args = parser.parse_args()
root = Path(args.root).resolve()
report = build_report(root, args.generated_at)
if args.output:
output = Path(args.output)
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
validate(root)
summary = report["summary"]
print(
"WAZUH_READONLY_ROUTE_BOUNDARY_GUARD_OK "
f"route={summary['route_present_count']} "
f"public_ui_files={summary['public_ui_file_count']} "
f"forbidden={summary['forbidden_match_count']} "
f"runtime_gate={summary['runtime_gate_count']}"
)
if __name__ == "__main__":
main()