27a7190c2c
Some checks failed
CD Pipeline / tests (push) Waiting to run
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
Code Review / ai-code-review (push) Successful in 10s
CD Pipeline / build-and-deploy (push) Has been cancelled
CD Pipeline / post-deploy-checks (push) Has been cancelled
754 lines
29 KiB
Python
754 lines
29 KiB
Python
"""
|
|
IwoooS Wazuh live metadata owner gate readback.
|
|
|
|
This module only exposes committed gate metadata plus public-safe Wazuh route
|
|
aggregate status. It never reads secret values, never queries hosts, and never
|
|
authorizes Wazuh active response, scans, restarts, reloads, or host writes.
|
|
"""
|
|
|
|
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 = "wazuh-readonly-live-metadata-env-gate.snapshot.json"
|
|
_EXPECTED_SCHEMA = "iwooos_wazuh_readonly_live_metadata_env_gate_v1"
|
|
|
|
_REQUIRED_FALSE_BOUNDARIES = {
|
|
"argocd_sync_authorized",
|
|
"docker_restart_authorized",
|
|
"firewall_change_authorized",
|
|
"host_write_authorized",
|
|
"k8s_secret_patch_authorized",
|
|
"kali_active_scan_authorized",
|
|
"nginx_gateway_workaround_authorized",
|
|
"production_deploy_authorized",
|
|
"raw_wazuh_payload_storage_allowed",
|
|
"repo_write_authorized",
|
|
"runtime_execution_authorized",
|
|
"secret_value_collection_allowed",
|
|
"wazuh_active_response_authorized",
|
|
"wazuh_api_live_query_authorized",
|
|
}
|
|
|
|
_PLACEHOLDER_VALUES = {
|
|
"",
|
|
"pending",
|
|
"todo",
|
|
"tbd",
|
|
"n/a",
|
|
"na",
|
|
"owner_here",
|
|
"reviewer_here",
|
|
"redacted_ref_here",
|
|
"secret_source_metadata_ref_here",
|
|
"wazuh_manager_health_ref_here",
|
|
}
|
|
|
|
_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",
|
|
"password",
|
|
"private_key",
|
|
"raw_dashboard_request",
|
|
"raw_env",
|
|
"raw_log",
|
|
"raw_runtime_volume",
|
|
"raw_wazuh_payload",
|
|
"session",
|
|
"stored_api_password",
|
|
"token",
|
|
"unredacted_screenshot",
|
|
"wazuh_api_password",
|
|
}
|
|
|
|
_RUNTIME_ACTION_KEYS = {
|
|
"active_response_enable",
|
|
"agent_reenroll",
|
|
"agent_restart",
|
|
"argocd_sync",
|
|
"docker_restart",
|
|
"enable_live_metadata",
|
|
"enable_wazuh_live_metadata_without_owner_gate",
|
|
"execute_now",
|
|
"firewall_change",
|
|
"host_write",
|
|
"k8s_secret_patch",
|
|
"kali_active_scan",
|
|
"nginx_gateway_workaround",
|
|
"production_deploy_authorized",
|
|
"repo_write_authorized",
|
|
"runtime_execution_authorized",
|
|
"runtime_gate_open",
|
|
"wazuh_active_response",
|
|
"wazuh_active_response_authorized",
|
|
"wazuh_api_live_query",
|
|
"wazuh_api_live_query_authorized",
|
|
"wazuh_manager_restart",
|
|
}
|
|
|
|
|
|
def load_latest_iwooos_wazuh_live_metadata_gate(
|
|
security_dir: Path | None = None,
|
|
wazuh_live_status: dict[str, Any] | None = None,
|
|
wazuh_live_http_status: int = 0,
|
|
) -> dict[str, Any]:
|
|
"""Load the committed Wazuh live metadata owner gate as a public-safe payload."""
|
|
directory = security_dir or _DEFAULT_SECURITY_DIR
|
|
snapshot = _load_snapshot(directory)
|
|
_require_boundaries(snapshot)
|
|
|
|
summary = _summary(snapshot)
|
|
live_route = _live_route_summary(wazuh_live_status, wazuh_live_http_status)
|
|
merged_summary = {
|
|
"server_side_env_key_count": _int(summary.get("server_side_env_key_count")),
|
|
"required_owner_field_count": _int(summary.get("required_owner_field_count")),
|
|
"reviewer_check_count": _int(summary.get("reviewer_check_count")),
|
|
"outcome_lane_count": _int(summary.get("outcome_lane_count")),
|
|
"blocked_action_count": _int(summary.get("blocked_action_count")),
|
|
"production_route_readback_passed_count": _int(summary.get("production_route_readback_passed_count")),
|
|
"live_metadata_owner_response_received_count": _int(
|
|
summary.get("live_metadata_owner_response_received_count")
|
|
),
|
|
"live_metadata_owner_response_accepted_count": _int(
|
|
summary.get("live_metadata_owner_response_accepted_count")
|
|
),
|
|
"secret_source_metadata_accepted_count": _int(summary.get("secret_source_metadata_accepted_count")),
|
|
"wazuh_manager_health_ref_accepted_count": _int(
|
|
summary.get("wazuh_manager_health_ref_accepted_count")
|
|
),
|
|
"readonly_account_scope_accepted_count": _int(summary.get("readonly_account_scope_accepted_count")),
|
|
"post_enable_readback_passed_count": _int(summary.get("post_enable_readback_passed_count")),
|
|
"wazuh_api_live_query_authorized_count": _int(summary.get("wazuh_api_live_query_authorized_count")),
|
|
"wazuh_active_response_authorized_count": _int(summary.get("wazuh_active_response_authorized_count")),
|
|
"host_write_authorized_count": _int(summary.get("host_write_authorized_count")),
|
|
"runtime_gate_count": _int(summary.get("runtime_gate_count")),
|
|
"wazuh_live_route_http_status": live_route["http_status"],
|
|
"wazuh_live_route_degraded_count": live_route["degraded_count"],
|
|
"wazuh_live_readonly_api_enabled_count": live_route["readonly_api_enabled_count"],
|
|
"wazuh_live_agent_total": live_route["agent_total"],
|
|
"wazuh_live_metadata_available_count": live_route["metadata_available_count"],
|
|
"wazuh_live_status": live_route["status"],
|
|
}
|
|
|
|
return {
|
|
"schema_version": "iwooos_wazuh_live_metadata_gate_readback_v1",
|
|
"status": snapshot.get("status", "blocked_waiting_live_metadata_owner_response"),
|
|
"mode": "committed_snapshot_readback_with_public_safe_wazuh_route_metadata",
|
|
"source_refs": [f"docs/security/{_SNAPSHOT_FILE}", "GET /api/iwooos/wazuh"],
|
|
"live_metadata_owner_packet_validation_endpoint": (
|
|
"/api/v1/iwooos/wazuh-live-metadata-gate/validate-live-metadata-owner-packet"
|
|
),
|
|
"live_metadata_owner_packet_validation_mode": (
|
|
"no_persist_owner_metadata_review_no_wazuh_query_no_secret_collection"
|
|
),
|
|
"summary": merged_summary,
|
|
"items": _items(merged_summary),
|
|
"boundary_markers": _boundary_markers(merged_summary),
|
|
"boundaries": {
|
|
"runtime_execution_authorized": False,
|
|
"secret_value_collection_allowed": False,
|
|
"raw_wazuh_payload_storage_allowed": False,
|
|
"wazuh_api_live_query_authorized": False,
|
|
"wazuh_active_response_authorized": False,
|
|
"host_write_authorized": False,
|
|
"kali_active_scan_authorized": False,
|
|
"k8s_secret_patch_authorized": False,
|
|
"argocd_sync_authorized": False,
|
|
"docker_restart_authorized": False,
|
|
"nginx_gateway_workaround_authorized": False,
|
|
"firewall_change_authorized": False,
|
|
"not_authorization": True,
|
|
},
|
|
"no_false_green_rules": [
|
|
"正式路由 200 不是即時查詢授權",
|
|
"Wazuh 儀表板可見不是 manager registry 已驗收",
|
|
"機密來源只能交付中繼資料，不得交付明文、片段或雜湊",
|
|
"live metadata query、active response、host write 與 Kali active scan 是不同閘門",
|
|
"owner response accepted、post-enable readback 與 runtime gate 仍全部維持 0",
|
|
],
|
|
}
|
|
|
|
|
|
def validate_iwooos_wazuh_live_metadata_owner_packet(
|
|
owner_packet: dict[str, Any],
|
|
security_dir: Path | None = None,
|
|
wazuh_live_status: dict[str, Any] | None = None,
|
|
wazuh_live_http_status: int = 0,
|
|
) -> dict[str, Any]:
|
|
"""Validate one redacted live metadata owner packet without applying it."""
|
|
contract = load_latest_iwooos_wazuh_live_metadata_gate(
|
|
security_dir,
|
|
wazuh_live_status=wazuh_live_status,
|
|
wazuh_live_http_status=wazuh_live_http_status,
|
|
)
|
|
snapshot = _load_snapshot(security_dir or _DEFAULT_SECURITY_DIR)
|
|
required_fields = _strings(snapshot.get("required_owner_fields"))
|
|
|
|
findings: list[dict[str, Any]] = []
|
|
if not isinstance(owner_packet, dict):
|
|
findings.append(
|
|
_finding(
|
|
"LME-01",
|
|
"blocker",
|
|
"request_live_metadata_owner_packet_supplement",
|
|
"live metadata owner packet must be a JSON object.",
|
|
[],
|
|
)
|
|
)
|
|
return _validation_result(
|
|
contract, "request_live_metadata_owner_packet_supplement", findings
|
|
)
|
|
|
|
sensitive_hits = _collect_sensitive_hits(owner_packet)
|
|
if sensitive_hits:
|
|
findings.append(
|
|
_finding(
|
|
"LME-04",
|
|
"critical",
|
|
"quarantine_sensitive_payload",
|
|
"live metadata owner 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(owner_packet)
|
|
if runtime_hits:
|
|
findings.append(
|
|
_finding(
|
|
"LME-05",
|
|
"critical",
|
|
"reject_runtime_action_request",
|
|
"live metadata owner packet requested runtime execution; this validator only reviews redacted metadata evidence.",
|
|
runtime_hits[:12],
|
|
)
|
|
)
|
|
return _validation_result(contract, "reject_runtime_action_request", findings)
|
|
|
|
missing_fields = [
|
|
field
|
|
for field in required_fields
|
|
if not _present(owner_packet.get(field))
|
|
or _placeholder(owner_packet.get(field))
|
|
]
|
|
if missing_fields:
|
|
findings.append(
|
|
_finding(
|
|
"LME-01",
|
|
"blocker",
|
|
"request_live_metadata_owner_packet_supplement",
|
|
"live metadata owner packet is missing required metadata-only fields.",
|
|
missing_fields,
|
|
)
|
|
)
|
|
|
|
if owner_packet.get("no_secret_value_attestation") != "no_secret_value_collected":
|
|
findings.append(
|
|
_finding(
|
|
"LME-06",
|
|
"blocker",
|
|
"request_secret_boundary_ack_fix",
|
|
"no_secret_value_attestation must state no_secret_value_collected.",
|
|
["no_secret_value_attestation"],
|
|
)
|
|
)
|
|
|
|
if owner_packet.get("no_raw_payload_attestation") != "no_raw_wazuh_payload_stored":
|
|
findings.append(
|
|
_finding(
|
|
"LME-07",
|
|
"blocker",
|
|
"request_raw_payload_boundary_ack_fix",
|
|
"no_raw_payload_attestation must state no_raw_wazuh_payload_stored.",
|
|
["no_raw_payload_attestation"],
|
|
)
|
|
)
|
|
|
|
if (
|
|
owner_packet.get("active_response_separate_gate_ack")
|
|
!= "active_response_requires_separate_gate"
|
|
):
|
|
findings.append(
|
|
_finding(
|
|
"LME-08",
|
|
"blocker",
|
|
"request_active_response_boundary_ack_fix",
|
|
"active_response_separate_gate_ack must state active_response_requires_separate_gate.",
|
|
["active_response_separate_gate_ack"],
|
|
)
|
|
)
|
|
|
|
post_enable_command = str(owner_packet.get("post_enable_readback_command") or "")
|
|
if "wazuh-readonly-production-readback.py" not in post_enable_command:
|
|
findings.append(
|
|
_finding(
|
|
"LME-09",
|
|
"blocker",
|
|
"request_post_enable_readback_command_fix",
|
|
"post_enable_readback_command must reference the committed Wazuh readonly production readback script.",
|
|
["post_enable_readback_command"],
|
|
)
|
|
)
|
|
|
|
outcome = (
|
|
_first_blocking_lane(findings)
|
|
or "accepted_for_live_metadata_owner_review_only"
|
|
)
|
|
if outcome == "accepted_for_live_metadata_owner_review_only":
|
|
findings.append(
|
|
_finding(
|
|
"LME-10",
|
|
"info",
|
|
"live_metadata_owner_review_ready",
|
|
"live metadata owner packet passed no-persist metadata review; live Wazuh query and runtime gate remain closed.",
|
|
[
|
|
"secret_source_metadata_ref",
|
|
"wazuh_manager_health_ref",
|
|
"readonly_account_scope_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}: 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 _strings(value: Any) -> list[str]:
|
|
if not isinstance(value, list):
|
|
return []
|
|
return [item for item in value if isinstance(item, str)]
|
|
|
|
|
|
def _live_route_summary(payload: dict[str, Any] | None, http_status: int) -> dict[str, Any]:
|
|
if not isinstance(payload, dict):
|
|
return {
|
|
"status": "not_checked_by_snapshot_loader",
|
|
"http_status": 0,
|
|
"degraded_count": 1,
|
|
"readonly_api_enabled_count": 0,
|
|
"agent_total": 0,
|
|
"metadata_available_count": 0,
|
|
}
|
|
summary = _summary(payload)
|
|
status_text = str(payload.get("status") or "unknown")
|
|
metadata_available = status_text == "readonly_metadata_available"
|
|
return {
|
|
"status": status_text,
|
|
"http_status": http_status if isinstance(http_status, int) else 0,
|
|
"degraded_count": 0 if metadata_available else 1,
|
|
"readonly_api_enabled_count": _int(summary.get("readonly_api_enabled_count")),
|
|
"agent_total": _int(summary.get("agent_total")),
|
|
"metadata_available_count": 1 if metadata_available else 0,
|
|
}
|
|
|
|
|
|
def _items(summary: dict[str, Any]) -> list[dict[str, Any]]:
|
|
return [
|
|
_item(
|
|
"release_readback",
|
|
"ENV-1",
|
|
"route_readback_passed",
|
|
"steady" if summary["production_route_readback_passed_count"] == 1 else "warn",
|
|
{"route_readback": summary["production_route_readback_passed_count"]},
|
|
),
|
|
_item(
|
|
"server_env_owner",
|
|
"ENV-2",
|
|
(
|
|
"owner_metadata_accepted"
|
|
if summary["live_metadata_owner_response_accepted_count"]
|
|
else "waiting_owner_response"
|
|
),
|
|
(
|
|
"steady"
|
|
if summary["live_metadata_owner_response_accepted_count"]
|
|
else "locked"
|
|
),
|
|
{
|
|
"owner_received": summary["live_metadata_owner_response_received_count"],
|
|
"owner_accepted": summary["live_metadata_owner_response_accepted_count"],
|
|
},
|
|
),
|
|
_item(
|
|
"secret_metadata",
|
|
"ENV-3",
|
|
(
|
|
"secret_metadata_accepted"
|
|
if summary["secret_source_metadata_accepted_count"]
|
|
else "metadata_only_waiting_acceptance"
|
|
),
|
|
"steady" if summary["secret_source_metadata_accepted_count"] else "locked",
|
|
{"secret_metadata_accepted": summary["secret_source_metadata_accepted_count"]},
|
|
),
|
|
_item(
|
|
"manager_health",
|
|
"ENV-4",
|
|
(
|
|
"manager_health_ref_accepted"
|
|
if summary["wazuh_manager_health_ref_accepted_count"]
|
|
else "waiting_manager_health_ref"
|
|
),
|
|
"steady" if summary["wazuh_manager_health_ref_accepted_count"] else "warn",
|
|
{
|
|
"manager_health_accepted": summary["wazuh_manager_health_ref_accepted_count"],
|
|
"live_route_degraded": summary["wazuh_live_route_degraded_count"],
|
|
},
|
|
),
|
|
_item(
|
|
"readonly_scope",
|
|
"ENV-5",
|
|
(
|
|
"readonly_scope_ref_accepted"
|
|
if summary["readonly_account_scope_accepted_count"]
|
|
else "waiting_readonly_scope_ref"
|
|
),
|
|
"steady" if summary["readonly_account_scope_accepted_count"] else "warn",
|
|
{"readonly_scope_accepted": summary["readonly_account_scope_accepted_count"]},
|
|
),
|
|
_item(
|
|
"post_enable_readback",
|
|
"ENV-6",
|
|
"waiting_post_enable_readback",
|
|
"locked",
|
|
{
|
|
"post_enable_readback": summary["post_enable_readback_passed_count"],
|
|
"live_query_authorized": summary["wazuh_api_live_query_authorized_count"],
|
|
"runtime_gate": summary["runtime_gate_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, Any]) -> list[str]:
|
|
return [
|
|
f"正式路由讀回={summary['production_route_readback_passed_count']}",
|
|
f"負責人回覆接受={summary['live_metadata_owner_response_accepted_count']}",
|
|
f"機密來源中繼資料接受={summary['secret_source_metadata_accepted_count']}",
|
|
f"管理節點健康參照接受={summary['wazuh_manager_health_ref_accepted_count']}",
|
|
f"唯讀帳號範圍接受={summary['readonly_account_scope_accepted_count']}",
|
|
f"啟用後讀回={summary['post_enable_readback_passed_count']}",
|
|
f"Wazuh 即時查詢授權={summary['wazuh_api_live_query_authorized_count']}",
|
|
f"Wazuh 主動回應授權={summary['wazuh_active_response_authorized_count']}",
|
|
f"主機寫入授權={summary['host_write_authorized_count']}",
|
|
f"執行期閘門={summary['runtime_gate_count']}",
|
|
"機密明文收集=false",
|
|
"原始 Wazuh 載荷保存=false",
|
|
"K8s secret 手動修改=false",
|
|
"ArgoCD 手動同步=false",
|
|
"Docker 重啟=false",
|
|
"Nginx 或 gateway 繞路=false",
|
|
"防火牆變更=false",
|
|
"資安觀測節點主動掃描=false",
|
|
"不是執行授權=true",
|
|
]
|
|
|
|
|
|
def _validation_result(
|
|
contract: dict[str, Any],
|
|
outcome_lane: str,
|
|
findings: list[dict[str, Any]],
|
|
) -> dict[str, Any]:
|
|
accepted = outcome_lane == "accepted_for_live_metadata_owner_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_wazuh_live_metadata_owner_packet_validation_result_v1",
|
|
"contract_schema_version": contract["schema_version"],
|
|
"status": outcome_lane,
|
|
"mode": "no_persist_live_metadata_owner_packet_no_wazuh_query_no_secret_collection",
|
|
"outcome_lane": outcome_lane,
|
|
"accepted_for_live_metadata_owner_review_only": accepted,
|
|
"quarantined": quarantined,
|
|
"runtime_action_rejected": rejected_runtime,
|
|
"summary": {
|
|
"live_metadata_owner_response_received_count": 1,
|
|
"live_metadata_owner_response_accepted_count": 1 if accepted else 0,
|
|
"secret_source_metadata_accepted_count": 1 if accepted else 0,
|
|
"wazuh_manager_health_ref_accepted_count": 1 if accepted else 0,
|
|
"readonly_account_scope_accepted_count": 1 if accepted else 0,
|
|
"live_metadata_owner_response_supplement_required_count": 1
|
|
if supplement_required
|
|
else 0,
|
|
"live_metadata_owner_packet_quarantined_count": 1 if quarantined else 0,
|
|
"live_metadata_owner_runtime_action_rejected_count": 1
|
|
if rejected_runtime
|
|
else 0,
|
|
"post_enable_readback_passed_count": 0,
|
|
"wazuh_api_live_query_authorized_count": 0,
|
|
"wazuh_active_response_authorized_count": 0,
|
|
"host_write_authorized_count": 0,
|
|
"runtime_gate_count": 0,
|
|
"finding_count": len(findings),
|
|
},
|
|
"validation_findings": findings,
|
|
"boundary_markers": [
|
|
"wazuh_live_metadata_owner_packet_validation_received_count=1",
|
|
f"wazuh_live_metadata_owner_packet_validation_accepted_count={1 if accepted else 0}",
|
|
f"wazuh_live_metadata_owner_packet_validation_quarantined_count={1 if quarantined else 0}",
|
|
f"wazuh_live_metadata_owner_packet_validation_runtime_action_rejected_count={1 if rejected_runtime else 0}",
|
|
"wazuh_live_metadata_owner_packet_validation_no_persist=true",
|
|
"wazuh_live_metadata_owner_packet_validation_live_query_authorized_count=0",
|
|
"wazuh_live_metadata_owner_packet_validation_runtime_gate_count=0",
|
|
"secret_value_collection_allowed=false",
|
|
"raw_wazuh_payload_storage_allowed=false",
|
|
"wazuh_active_response_authorized=false",
|
|
"host_write_authorized=false",
|
|
"not_authorization=true",
|
|
],
|
|
"boundaries": {
|
|
"payload_persisted": False,
|
|
"runtime_execution_authorized": False,
|
|
"secret_value_collection_allowed": False,
|
|
"raw_wazuh_payload_storage_allowed": False,
|
|
"wazuh_api_live_query_authorized": False,
|
|
"wazuh_active_response_authorized": False,
|
|
"host_write_authorized": False,
|
|
"kali_active_scan_authorized": False,
|
|
"k8s_secret_patch_authorized": False,
|
|
"argocd_sync_authorized": False,
|
|
"docker_restart_authorized": False,
|
|
"nginx_gateway_workaround_authorized": False,
|
|
"firewall_change_authorized": False,
|
|
"runtime_gate_open": False,
|
|
"not_authorization": True,
|
|
},
|
|
"next_gate": "reviewer_validation_before_server_side_env_enable"
|
|
if accepted
|
|
else "live_metadata_owner_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 _placeholder(value: Any) -> bool:
|
|
if value is None:
|
|
return True
|
|
return str(value).strip().lower() in _PLACEHOLDER_VALUES
|
|
|
|
|
|
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(
|
|
{
|
|
"path": f"{path}.{key_text}",
|
|
"category": f"forbidden_key:{fragment}",
|
|
}
|
|
)
|
|
hits.extend(_collect_sensitive_hits(item, f"{path}.{key_text}"))
|
|
return hits
|
|
if isinstance(value, list):
|
|
for index, item in enumerate(value):
|
|
hits.extend(_collect_sensitive_hits(item, f"{path}[{index}]"))
|
|
return hits
|
|
if isinstance(value, str):
|
|
for category, pattern in _SENSITIVE_TEXT_PATTERNS.items():
|
|
if pattern.search(value):
|
|
hits.append({"path": path, "category": category})
|
|
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)
|
|
normalized_key = key_text.lower().replace("-", "_").replace(" ", "_")
|
|
if normalized_key in _RUNTIME_ACTION_KEYS and item not in (
|
|
False,
|
|
None,
|
|
"",
|
|
[],
|
|
{},
|
|
):
|
|
hits.append(f"{path}.{key_text}")
|
|
hits.extend(_collect_runtime_action_hits(item, f"{path}.{key_text}"))
|
|
return hits
|
|
if isinstance(value, list):
|
|
for index, item in enumerate(value):
|
|
hits.extend(_collect_runtime_action_hits(item, f"{path}[{index}]"))
|
|
return hits
|
|
if isinstance(value, str):
|
|
normalized = value.lower().replace("-", "_").replace(" ", "_")
|
|
if normalized in _RUNTIME_ACTION_KEYS:
|
|
hits.append(path)
|
|
return hits
|
|
|
|
|
|
def _first_blocking_lane(findings: list[dict[str, Any]]) -> str | None:
|
|
severity_order = {"critical": 0, "blocker": 1, "warn": 2, "info": 3}
|
|
blocking = [
|
|
finding
|
|
for finding in findings
|
|
if finding.get("severity") in {"critical", "blocker"}
|
|
]
|
|
if not blocking:
|
|
return None
|
|
blocking.sort(
|
|
key=lambda finding: severity_order.get(str(finding.get("severity")), 99)
|
|
)
|
|
return str(
|
|
blocking[0].get("lane") or "request_live_metadata_owner_packet_supplement"
|
|
)
|
|
|
|
|
|
def _require_boundaries(payload: dict[str, Any]) -> None:
|
|
summary = _summary(payload)
|
|
readiness_counts = {
|
|
"live_metadata_owner_response_received_count": _int(
|
|
summary.get("live_metadata_owner_response_received_count")
|
|
),
|
|
"live_metadata_owner_response_accepted_count": _int(
|
|
summary.get("live_metadata_owner_response_accepted_count")
|
|
),
|
|
"secret_source_metadata_accepted_count": _int(
|
|
summary.get("secret_source_metadata_accepted_count")
|
|
),
|
|
"wazuh_manager_health_ref_accepted_count": _int(
|
|
summary.get("wazuh_manager_health_ref_accepted_count")
|
|
),
|
|
"readonly_account_scope_accepted_count": _int(
|
|
summary.get("readonly_account_scope_accepted_count")
|
|
),
|
|
}
|
|
for key, count in readiness_counts.items():
|
|
if count not in {0, 1}:
|
|
raise ValueError(
|
|
f"Wazuh 即時中繼資料閘門 summary.{key} 必須維持 0 或 1"
|
|
)
|
|
accepted_count = readiness_counts["live_metadata_owner_response_accepted_count"]
|
|
if accepted_count > readiness_counts["live_metadata_owner_response_received_count"]:
|
|
raise ValueError("Wazuh 即時中繼資料閘門 accepted 不得大於 received")
|
|
for key in (
|
|
"secret_source_metadata_accepted_count",
|
|
"wazuh_manager_health_ref_accepted_count",
|
|
"readonly_account_scope_accepted_count",
|
|
):
|
|
if readiness_counts[key] > accepted_count:
|
|
raise ValueError(
|
|
f"Wazuh 即時中繼資料閘門 summary.{key} 不得超過 owner accepted"
|
|
)
|
|
|
|
for key in (
|
|
"post_enable_readback_passed_count",
|
|
"wazuh_api_live_query_authorized_count",
|
|
"wazuh_active_response_authorized_count",
|
|
"host_write_authorized_count",
|
|
"runtime_gate_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")
|