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

Merge remote-tracking branch 'gitea/main' into codex/github-private-backup-readback-20260627

Browse Source
This commit is contained in:
Your Name
2026-06-27 11:44:58 +08:00
parent 91c32f7b5f fccd8874fc
commit 9b120fcbbd
9 changed files with 635 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -26,6 +26,9 @@ from src.services.iwooos_wazuh_readonly_status import (
from src.services.iwooos_wazuh_live_metadata_gate import (
load_latest_iwooos_wazuh_live_metadata_gate,
)
from src.services.iwooos_wazuh_owner_evidence_preflight import (
load_latest_iwooos_wazuh_owner_evidence_preflight,
)
from src.services.public_redaction import redact_public_lan_topology
@@ -79,6 +82,34 @@ async def get_iwooos_wazuh_live_metadata_gate() -> dict[str, Any]:
) from exc
@router.get(
"/api/v1/iwooos/wazuh-owner-evidence-preflight",
response_model=dict[str, Any],
summary="取得 Wazuh 負責人證據收件預檢讀回",
description=(
"讀取已提交的 Wazuh 代理清單負責人證據收件預檢，回傳公開安全的欄位數、"
"審查檢查、分流、拒收內容計數與 0 / false 邊界。此端點不查 Wazuh、"
"不讀主機、不保存原始載荷、不收機密明文、不啟用主動回應、不改 Nginx / "
"Docker / K8s / firewall。"
),
)
async def get_iwooos_wazuh_owner_evidence_preflight() -> dict[str, Any]:
"""回傳 Wazuh manager registry 負責人證據收件預檢只讀狀態。"""
try:
payload = await asyncio.to_thread(load_latest_iwooos_wazuh_owner_evidence_preflight)
return redact_public_lan_topology(payload)
except FileNotFoundError as exc:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=str(exc),
) from exc
except (json.JSONDecodeError, ValueError) as exc:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"IwoooS Wazuh 負責人證據預檢無效：{exc}",
) from exc
@router.get(
"/api/v1/iwooos/runtime-security-readback",
response_model=dict[str, Any],

63
apps/api/src/services/iwooos_runtime_security_readback.py
View File

@@ -21,6 +21,7 @@ _SNAPSHOT_FILES = {
"wazuh_coverage": "wazuh-managed-host-coverage-gate.snapshot.json",
"wazuh_runtime": "wazuh-agent-visibility-runtime-gate.snapshot.json",
"wazuh_live_metadata_gate": "wazuh-readonly-live-metadata-env-gate.snapshot.json",
"wazuh_owner_evidence_preflight": "wazuh-agent-visibility-owner-evidence-preflight.snapshot.json",
"kali_status": "kali-integration-status.snapshot.json",
"soc_control": "soc-siem-kali-wazuh-integration-control.snapshot.json",
"alert_readability": "telegram-alert-readability-guard.snapshot.json",
@@ -33,6 +34,7 @@ _EXPECTED_SCHEMAS = {
"wazuh_coverage": "wazuh_managed_host_coverage_gate_v1",
"wazuh_runtime": "wazuh_agent_visibility_runtime_gate_v1",
"wazuh_live_metadata_gate": "iwooos_wazuh_readonly_live_metadata_env_gate_v1",
"wazuh_owner_evidence_preflight": "wazuh_agent_visibility_owner_evidence_preflight_v1",
"kali_status": "kali_integration_status_v1",
"soc_control": "soc_siem_kali_wazuh_integration_control_v1",
"alert_readability": "telegram_alert_readability_guard_v1",
@@ -73,6 +75,7 @@ def load_latest_iwooos_runtime_security_readback(
owner_gap_summary = _summary(snapshots["owner_gap"])
wazuh_summary = _summary(snapshots["wazuh_coverage"])
live_metadata_gate_summary = _summary(snapshots["wazuh_live_metadata_gate"])
owner_evidence_preflight_summary = _summary(snapshots["wazuh_owner_evidence_preflight"])
soc_summary = _summary(snapshots["soc_control"])
alert_summary = _summary(snapshots["alert_readability"])
dispatch_summary = _summary(snapshots["owner_dispatch"])
@@ -95,7 +98,7 @@ def load_latest_iwooos_runtime_security_readback(
"source_refs": source_refs,
"summary": {
"source_snapshot_count": len(source_refs),
"p0_lane_count": 8,
"p0_lane_count": 9,
"control_plane_visibility_percent": _average_percent(
soc_summary.get("coverage_percent_after_soc_integration_control"),
intrusion_summary.get("coverage_percent_after_prevention_control"),
@@ -139,6 +142,36 @@ def load_latest_iwooos_runtime_security_readback(
"wazuh_live_metadata_gate_live_query_authorized_count": _int(
live_metadata_gate_summary.get("wazuh_api_live_query_authorized_count")
),
"wazuh_owner_evidence_required_field_count": _int(
owner_evidence_preflight_summary.get("required_field_count")
),
"wazuh_owner_evidence_reviewer_check_count": _int(
owner_evidence_preflight_summary.get("reviewer_check_count")
),
"wazuh_owner_evidence_outcome_lane_count": _int(
owner_evidence_preflight_summary.get("outcome_lane_count")
),
"wazuh_owner_evidence_forbidden_payload_count": _int(
owner_evidence_preflight_summary.get("forbidden_payload_count")
),
"wazuh_owner_evidence_expected_alias_count": _int(
owner_evidence_preflight_summary.get("expected_scope_alias_count")
),
"wazuh_owner_evidence_registry_export_received_count": _int(
owner_evidence_preflight_summary.get("registry_export_received_count")
),
"wazuh_owner_evidence_registry_export_accepted_count": _int(
owner_evidence_preflight_summary.get("registry_export_accepted_count")
),
"wazuh_owner_evidence_received_count": _int(
owner_evidence_preflight_summary.get("owner_evidence_received_count")
),
"wazuh_owner_evidence_accepted_count": _int(
owner_evidence_preflight_summary.get("owner_evidence_accepted_count")
),
"wazuh_owner_evidence_runtime_gate_count": _int(
owner_evidence_preflight_summary.get("runtime_gate_count")
),
"kali_active_scan_authorized_count": _int(soc_summary.get("kali_active_scan_authorized_count")),
"kali_execute_authorized_count": _int(soc_summary.get("kali_execute_authorized_count")),
"kali_finding_envelope_accepted_count": _int(soc_summary.get("kali_finding_envelope_accepted_count")),
@@ -206,6 +239,30 @@ def load_latest_iwooos_runtime_security_readback(
},
["docs/security/wazuh-readonly-live-metadata-env-gate.snapshot.json"],
),
_lane(
"wazuh_owner_evidence_preflight",
snapshots["wazuh_owner_evidence_preflight"].get(
"status",
"owner_evidence_preflight_ready_no_runtime_action",
),
0,
"locked",
"補齊 manager registry 脫敏封包、逐主機矩陣、Dashboard API 狀態與負責人決策",
{
"required_fields": owner_evidence_preflight_summary.get("required_field_count", 0),
"reviewer_checks": owner_evidence_preflight_summary.get("reviewer_check_count", 0),
"outcome_lanes": owner_evidence_preflight_summary.get("outcome_lane_count", 0),
"forbidden_payloads": owner_evidence_preflight_summary.get("forbidden_payload_count", 0),
"owner_received": owner_evidence_preflight_summary.get("owner_evidence_received_count", 0),
"owner_accepted": owner_evidence_preflight_summary.get("owner_evidence_accepted_count", 0),
"registry_export_accepted": owner_evidence_preflight_summary.get(
"registry_export_accepted_count",
0,
),
"runtime_gate": owner_evidence_preflight_summary.get("runtime_gate_count", 0),
},
["docs/security/wazuh-agent-visibility-owner-evidence-preflight.snapshot.json"],
),
_lane(
"wazuh_dashboard_api",
"degraded_api_connection_not_green",
@@ -303,6 +360,7 @@ def load_latest_iwooos_runtime_security_readback(
"告警格式合約不代表通知已實發或已取得 receipt",
"Wazuh 正式只讀路由 disabled 或退化時仍是 P0 紅燈",
"Wazuh 即時中繼資料必須先通過負責人、機密中繼資料、唯讀範圍與啟用後讀回",
"Wazuh 負責人證據預檢 ready 不代表已收件、已接受或可啟用 active response",
],
}
@@ -421,6 +479,9 @@ def _require_runtime_boundaries(snapshots: dict[str, dict[str, Any]]) -> None:
"secret_value_collection_allowed_count",
"wazuh_api_live_query_authorized_count",
"wazuh_active_response_authorized_count",
"active_response_authorized_count",
"registry_export_accepted_count",
"owner_evidence_accepted_count",
"post_enable_readback_passed_count",
):
if key in summary and _int(summary.get(key)) != 0:

264
apps/api/src/services/iwooos_wazuh_owner_evidence_preflight.py Normal file
View File

@@ -0,0 +1,264 @@
"""
IwoooS Wazuh owner evidence preflight readback.
This module only exposes committed, public-safe preflight metadata. It never
queries Wazuh, never reads secret values, and never authorizes active response,
host writes, scans, restarts, reloads, or gateway changes.
"""
from __future__ import annotations
import json
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 = "wazuh-agent-visibility-owner-evidence-preflight.snapshot.json"
_EXPECTED_SCHEMA = "wazuh_agent_visibility_owner_evidence_preflight_v1"
_REQUIRED_FALSE_BOUNDARIES = {
"agent_identity_public_display_allowed",
"host_write_authorized",
"internal_ip_public_display_allowed",
"raw_wazuh_payload_storage_allowed",
"runtime_execution_authorized",
"secret_value_collection_allowed",
"wazuh_active_response_authorized",
"wazuh_api_live_query_authorized",
}
def load_latest_iwooos_wazuh_owner_evidence_preflight(
security_dir: Path | None = None,
) -> dict[str, Any]:
"""Load the committed Wazuh owner evidence preflight as a public-safe payload."""
directory = security_dir or _DEFAULT_SECURITY_DIR
snapshot = _load_snapshot(directory)
_require_boundaries(snapshot)
summary = _summary(snapshot)
contract = snapshot.get("registry_export_contract")
contract = contract if isinstance(contract, dict) else {}
merged_summary = {
"required_field_count": _int(summary.get("required_field_count")),
"reviewer_check_count": _int(summary.get("reviewer_check_count")),
"outcome_lane_count": _int(summary.get("outcome_lane_count")),
"forbidden_payload_count": _int(summary.get("forbidden_payload_count")),
"expected_scope_alias_count": _int(summary.get("expected_scope_alias_count")),
"per_host_required_field_count": _int(summary.get("per_host_required_field_count")),
"allowed_collection_method_count": _len(contract.get("allowed_collection_methods")),
"registry_export_received_count": _int(summary.get("registry_export_received_count")),
"registry_export_accepted_count": _int(summary.get("registry_export_accepted_count")),
"owner_evidence_received_count": _int(summary.get("owner_evidence_received_count")),
"owner_evidence_accepted_count": _int(summary.get("owner_evidence_accepted_count")),
"owner_evidence_rejected_count": _int(summary.get("owner_evidence_rejected_count")),
"owner_evidence_quarantined_count": _int(summary.get("owner_evidence_quarantined_count")),
"runtime_gate_count": _int(summary.get("runtime_gate_count")),
"wazuh_api_live_query_authorized_count": 0,
"active_response_authorized_count": _int(summary.get("active_response_authorized_count")),
"host_write_authorized_count": _int(summary.get("host_write_authorized_count")),
"secret_value_collection_allowed_count": _int(summary.get("secret_value_collection_allowed_count")),
}
return {
"schema_version": "iwooos_wazuh_owner_evidence_preflight_readback_v1",
"status": snapshot.get("status", "owner_evidence_preflight_ready_no_runtime_action"),
"mode": "committed_snapshot_readback_redacted_metadata_only",
"source_refs": [
f"docs/security/{_SNAPSHOT_FILE}",
"scripts/security/wazuh-agent-visibility-owner-evidence-preflight.py",
],
"summary": merged_summary,
"items": _items(merged_summary),
"boundary_markers": _boundary_markers(merged_summary),
"boundaries": {
"wazuh_api_live_query_authorized": False,
"wazuh_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,
"runtime_execution_authorized": False,
"not_authorization": True,
},
"no_false_green_rules": [
"負責人證據預檢 ready 不代表已收件或已接受",
"Wazuh 儀表板可見不是 manager registry counts 已驗收",
"Dashboard index pattern 三綠勾不可替代 API connection、API version 或 manager registry",
"agent service active、TCP 連線或舊截圖不可替代逐主機 registry matrix",
"收件封包若夾帶原始紀錄、內網識別、agent 原名或機密，必須隔離，不得渲染到前台",
"active response、host write、firewall、Nginx、Docker、K8s 或 secret 變更一律不是這個預檢授權",
],
}
def _load_snapshot(directory: Path) -> dict[str, Any]:
path = directory / _SNAPSHOT_FILE
if not path.is_file():
raise FileNotFoundError(f"{path}: Wazuh 負責人證據預檢快照不存在")
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 _len(value: Any) -> int:
return len(value) if isinstance(value, list) else 0
def _items(summary: dict[str, int]) -> list[dict[str, Any]]:
return [
_item(
"scope_aliases",
"EV-0",
"scope_aliases_ready",
"warn",
{
"expected_scope_aliases": summary["expected_scope_alias_count"],
"allowed_collection_methods": summary["allowed_collection_method_count"],
},
),
_item(
"registry_counts",
"EV-1",
"waiting_redacted_counts",
"warn",
{
"registry_export_received": summary["registry_export_received_count"],
"registry_export_accepted": summary["registry_export_accepted_count"],
},
),
_item(
"per_host_matrix",
"EV-2",
"waiting_per_host_matrix",
"warn",
{"per_host_required_fields": summary["per_host_required_field_count"]},
),
_item(
"time_window",
"EV-3",
"waiting_time_window",
"warn",
{"owner_received": summary["owner_evidence_received_count"]},
),
_item(
"health_refs",
"EV-4",
"waiting_health_refs",
"warn",
{"reviewer_checks": summary["reviewer_check_count"]},
),
_item(
"redaction",
"EV-5",
"reject_sensitive_payloads",
"locked",
{
"forbidden_payloads": summary["forbidden_payload_count"],
"quarantined": summary["owner_evidence_quarantined_count"],
},
),
_item(
"owner_decision",
"EV-6",
"waiting_owner_decision",
"warn",
{
"owner_received": summary["owner_evidence_received_count"],
"owner_accepted": summary["owner_evidence_accepted_count"],
},
),
_item(
"runtime_boundary",
"EV-7",
"runtime_closed",
"locked",
{
"runtime_gate": summary["runtime_gate_count"],
"live_query": summary["wazuh_api_live_query_authorized_count"],
"active_response": summary["active_response_authorized_count"],
"host_write": summary["host_write_authorized_count"],
},
),
]
def _item(
item_id: str,
check: str,
state_key: str,
tone: str,
metrics: dict[str, int],
) -> dict[str, Any]:
return {
"item_id": item_id,
"check": check,
"state_key": state_key,
"tone": tone,
"metrics": metrics,
}
def _boundary_markers(summary: dict[str, int]) -> list[str]:
return [
f"必要欄位={summary['required_field_count']}",
f"審查檢查={summary['reviewer_check_count']}",
f"結果分流={summary['outcome_lane_count']}",
f"拒收敏感類型={summary['forbidden_payload_count']}",
f"公開節點別名={summary['expected_scope_alias_count']}",
f"逐主機矩陣欄位={summary['per_host_required_field_count']}",
f"允許收集方式={summary['allowed_collection_method_count']}",
f"registry export 已收件={summary['registry_export_received_count']}",
f"registry export 已接受={summary['registry_export_accepted_count']}",
f"負責人證據已收件={summary['owner_evidence_received_count']}",
f"負責人證據已接受={summary['owner_evidence_accepted_count']}",
f"負責人證據已隔離={summary['owner_evidence_quarantined_count']}",
f"執行期閘門={summary['runtime_gate_count']}",
f"Wazuh 即時查詢={summary['wazuh_api_live_query_authorized_count']}",
f"Wazuh 主動回應={summary['active_response_authorized_count']}",
f"主機寫入={summary['host_write_authorized_count']}",
f"機密明文收集={summary['secret_value_collection_allowed_count']}",
"原始 Wazuh 載荷保存=false",
"agent 身分前台顯示=false",
"內網識別前台顯示=false",
"不是執行授權=true",
]
def _require_boundaries(payload: dict[str, Any]) -> None:
summary = _summary(payload)
for key in (
"registry_export_accepted_count",
"owner_evidence_accepted_count",
"runtime_gate_count",
"active_response_authorized_count",
"host_write_authorized_count",
"secret_value_collection_allowed_count",
):
if _int(summary.get(key)) != 0:
raise ValueError(f"Wazuh 負責人證據預檢 summary.{key} 必須維持 0")
boundaries = payload.get("execution_boundaries")
if not isinstance(boundaries, dict):
raise ValueError("Wazuh 負責人證據預檢 execution_boundaries 缺失")
for key in _REQUIRED_FALSE_BOUNDARIES:
if boundaries.get(key) is not False:
raise ValueError(f"Wazuh 負責人證據預檢 execution_boundaries.{key} 必須維持 false")
if boundaries.get("not_authorization") is not True:
raise ValueError("Wazuh 負責人證據預檢 not_authorization 必須維持 true")