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

fix(wazuh): persist critical capability packets
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 0s
Details
CD Pipeline / cancel-stale-cd (push) Has been skipped
Details
CD Pipeline / tests (push) Successful in 2m28s
Details
CD Pipeline / build-and-deploy (push) Successful in 15m8s
Details
CD Pipeline / post-deploy-checks (push) Successful in 2m18s
Details

Browse Source
This commit is contained in:
ogt
2026-07-15 18:24:36 +08:00
parent e7f095b11c
commit 1387a30748
7 changed files with 785 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

343
apps/api/src/services/awooop_ansible_check_mode_service.py
View File

@@ -77,6 +77,15 @@ _EXECUTION_CAPABILITY_SAFETY_MARGIN_SECONDS = 5
_WORKING_MEMORY_PROJECTION_WINDOW_HOURS = 8 * 24
_WORKING_MEMORY_PROJECTION_CURSOR: dict[str, tuple[datetime, str]] = {}
_EXECUTION_CAPABILITY_FALLBACK_OPERATION_TYPE = "remediation_executed"
_WAZUH_ALERTMANAGER_INTEGRATION_CATALOG_ID = (
"ansible:wazuh-alertmanager-integration"
)
_WAZUH_PRIVILEGED_CONVERGENCE_CAPABILITY_MISSING = (
"wazuh_privileged_convergence_capability_missing"
)
_WAZUH_BREAK_GLASS_CAPABILITY_PACKET_OPERATION_TYPE = (
"ansible_break_glass_capability_work_item_queued"
)
_EXECUTION_CAPABILITY_OPERATION_TYPES = frozenset(
{
"ansible_executor_capability_issued",
@@ -6135,6 +6144,9 @@ async def backfill_missing_auto_repair_execution_receipts_once(
"retry_terminal_projection_lifecycle_written": 0,
"retry_terminal_projection_verified": 0,
"retry_terminal_projection_runtime_apply_executed": False,
"critical_break_glass_packet_scanned": 0,
"critical_break_glass_packet_written": 0,
"critical_break_glass_packet_error": None,
"skipped": 0,
"error": None,
}
@@ -6399,6 +6411,22 @@ async def backfill_missing_auto_repair_execution_receipts_once(
stats["telegram_receipt_acknowledged"] += 1
if closure.get("closed") is True:
stats["incident_closure_written"] += 1
break_glass_packet = (
await backfill_missing_wazuh_break_glass_capability_packets_once(
project_id=project_id,
window_hours=window_hours,
limit=max(1, min(limit, 5)),
)
)
stats["critical_break_glass_packet_scanned"] = int(
break_glass_packet.get("scanned") or 0
)
stats["critical_break_glass_packet_written"] = int(
break_glass_packet.get("written") or 0
)
stats["critical_break_glass_packet_error"] = (
str(break_glass_packet.get("error") or "")[:500] or None
)
except Exception as exc:
stats["error"] = f"{type(exc).__name__}: {exc}"[:500]
logger.warning("ansible_auto_repair_execution_receipt_backfill_failed", **stats)
@@ -8603,6 +8631,175 @@ async def _insert_skipped_candidate(
)
def _wazuh_privileged_convergence_capability_is_missing(
claim: AnsibleCheckModeClaim,
result: AnsibleRunResult,
) -> bool:
if (
claim.catalog_id != _WAZUH_ALERTMANAGER_INTEGRATION_CATALOG_ID
or result.returncode == 0
):
return False
return any(
_WAZUH_PRIVILEGED_CONVERGENCE_CAPABILITY_MISSING in value
for value in (result.stdout, result.stderr)
)
def _build_wazuh_break_glass_capability_packet(
*,
check_op_id: str,
source_candidate_op_id: str,
incident_id: str,
catalog_id: str,
source_input: Mapping[str, Any],
) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
automation_run_id = str(
source_input.get("automation_run_id") or source_candidate_op_id
).strip()[:180]
trace_id = str(
source_input.get("trace_id") or automation_run_id
).strip()[:180]
run_id = str(
source_input.get("run_id") or automation_run_id
).strip()[:180]
work_item_id = str(
source_input.get("work_item_id") or "P0-03-WAZUH-ALERT-INGRESS"
).strip()[:180]
source_recurrence = source_input.get("source_recurrence")
if not isinstance(source_recurrence, Mapping):
source_recurrence = {}
packet_uuid = uuid5(
NAMESPACE_URL,
f"awoooi:wazuh-break-glass:{automation_run_id}:{check_op_id}",
)
capability_packet_id = f"IWZ-BG-{packet_uuid.hex[:16].upper()}"
packet_input = {
"schema_version": "wazuh_privileged_convergence_capability_packet_v1",
"capability_packet_id": capability_packet_id,
"automation_run_id": automation_run_id,
"trace_id": trace_id,
"run_id": run_id,
"work_item_id": work_item_id,
"project_id": str(source_input.get("project_id") or "")[:180],
"run_namespace": str(
source_input.get("run_namespace") or ""
)[:180],
"proposal_source": str(
source_input.get("proposal_source") or ""
)[:180],
"source_receipt_ref": str(
source_input.get("source_receipt_ref") or ""
)[:180],
"source_recurrence": dict(source_recurrence),
"incident_id": str(incident_id or "").strip()[:180],
"source_candidate_op_id": str(source_candidate_op_id),
"source_check_mode_op_id": str(check_op_id),
"catalog_id": str(catalog_id),
"capability_type": "bounded_privileged_convergence",
"risk_level": "critical",
"policy_route": "critical_break_glass_queue",
"queue_owner": "ai_critical_capability_controller",
"same_run_correlation_required": True,
"execution_allowed": False,
"host_write_performed": False,
"secret_value_exposed": False,
"secret_value_logged": False,
"secret_value_read_back": False,
}
packet_output = {
"queue_status": "queued",
"failure_class": _WAZUH_PRIVILEGED_CONVERGENCE_CAPABILITY_MISSING,
"required_receipts": [
"capability_reference_presence",
"no_secret_privilege_preflight",
"bounded_check_mode_retry",
"independent_post_verifier",
"capability_revocation",
],
"next_controlled_action": (
"verify_bounded_privileged_capability_then_"
"requeue_same_run_check_mode"
),
"telegram_notification_emitted": False,
}
dry_run_result = {
"capability_packet_persisted": True,
"capability_granted": False,
"check_mode_replayed": False,
"apply_executed": False,
"host_write_performed": False,
"secret_value_exposed": False,
"secret_value_logged": False,
"secret_value_read_back": False,
}
return packet_input, packet_output, dry_run_result
async def _insert_wazuh_break_glass_capability_packet(
db: Any,
*,
check_op_id: str,
source_candidate_op_id: str,
incident_id: str,
catalog_id: str,
source_input: Mapping[str, Any],
) -> bool:
packet_input, packet_output, dry_run_result = (
_build_wazuh_break_glass_capability_packet(
check_op_id=check_op_id,
source_candidate_op_id=source_candidate_op_id,
incident_id=incident_id,
catalog_id=catalog_id,
source_input=source_input,
)
)
inserted = await db.execute(
text(f"""
INSERT INTO automation_operation_log (
operation_type, actor, status, incident_id,
input, output, dry_run_result, parent_op_id, tags
)
SELECT
'{_WAZUH_BREAK_GLASS_CAPABILITY_PACKET_OPERATION_TYPE}',
'ansible_check_mode_worker',
'pending',
:incident_db_id,
CAST(:input AS jsonb),
CAST(:output AS jsonb),
CAST(:dry_run_result AS jsonb),
CAST(:parent_op_id AS uuid),
ARRAY[
'ansible', 'wazuh', 'critical_break_glass',
'capability_packet', 'no_host_write', 'no_secret'
]::varchar[]
WHERE NOT EXISTS (
SELECT 1
FROM automation_operation_log existing
WHERE existing.operation_type =
'{_WAZUH_BREAK_GLASS_CAPABILITY_PACKET_OPERATION_TYPE}'
AND existing.input ->> 'capability_packet_id'
= :capability_packet_id
)
RETURNING op_id::text
"""),
{
"incident_db_id": _automation_operation_log_incident_id(
incident_id
),
"input": json.dumps(packet_input, ensure_ascii=False),
"output": json.dumps(packet_output, ensure_ascii=False),
"dry_run_result": json.dumps(
dry_run_result,
ensure_ascii=False,
),
"parent_op_id": check_op_id,
"capability_packet_id": packet_input["capability_packet_id"],
},
)
return inserted.scalar_one_or_none() is not None
async def finalize_check_mode_claim(
claim: AnsibleCheckModeClaim,
result: AnsibleRunResult,
@@ -8661,6 +8858,140 @@ async def finalize_check_mode_claim(
"op_id": claim.op_id,
},
)
if _wazuh_privileged_convergence_capability_is_missing(claim, result):
await _insert_wazuh_break_glass_capability_packet(
db,
check_op_id=claim.op_id,
source_candidate_op_id=claim.source_candidate_op_id,
incident_id=claim.incident_id,
catalog_id=claim.catalog_id,
source_input=claim.input_payload,
)
async def backfill_missing_wazuh_break_glass_capability_packets_once(
*,
project_id: str = "awoooi",
window_hours: int = 24,
limit: int = 5,
) -> dict[str, Any]:
"""Persist safe critical-capability packets for earlier failed Wazuh checks."""
stats: dict[str, Any] = {"scanned": 0, "written": 0, "error": None}
try:
async with get_db_context(project_id) as db:
await db.execute(text("SET LOCAL statement_timeout = '3000ms'"))
result = await db.execute(
text(f"""
SELECT
check_mode.op_id::text AS check_op_id,
candidate.op_id::text AS source_candidate_op_id,
coalesce(
check_mode.input ->> 'incident_id',
candidate.input ->> 'incident_id',
check_mode.incident_id::text
) AS incident_id,
coalesce(
candidate.input ->> 'automation_run_id',
candidate.op_id::text
) AS automation_run_id,
coalesce(
candidate.input ->> 'trace_id',
candidate.input ->> 'automation_run_id',
candidate.op_id::text
) AS trace_id,
coalesce(
candidate.input ->> 'run_id',
candidate.input ->> 'automation_run_id',
candidate.op_id::text
) AS run_id,
candidate.input ->> 'work_item_id' AS work_item_id,
candidate.input ->> 'project_id' AS project_id,
candidate.input ->> 'run_namespace' AS run_namespace,
candidate.input ->> 'proposal_source' AS proposal_source,
candidate.input ->> 'source_receipt_ref'
AS source_receipt_ref,
candidate.input -> 'source_recurrence'
AS source_recurrence
FROM automation_operation_log check_mode
JOIN automation_operation_log candidate
ON candidate.op_id = check_mode.parent_op_id
WHERE check_mode.operation_type =
'ansible_check_mode_executed'
AND check_mode.status = 'failed'
AND check_mode.created_at >= NOW() - (
:window_hours * INTERVAL '1 hour'
)
AND candidate.operation_type =
'ansible_candidate_matched'
AND candidate.input -> 'executor_candidates'
@> CAST(:catalog_match AS jsonb)
AND concat_ws(
' ',
check_mode.error,
check_mode.output ->> 'stdout_tail',
check_mode.output ->> 'stderr_tail',
check_mode.dry_run_result ->> 'stdout_tail',
check_mode.dry_run_result ->> 'stderr_tail'
) LIKE ('%' || :failure_marker || '%')
AND NOT EXISTS (
SELECT 1
FROM automation_operation_log packet
WHERE packet.operation_type =
'{_WAZUH_BREAK_GLASS_CAPABILITY_PACKET_OPERATION_TYPE}'
AND packet.parent_op_id = check_mode.op_id
)
ORDER BY check_mode.created_at DESC
LIMIT :limit
"""),
{
"window_hours": max(1, int(window_hours)),
"limit": max(1, min(int(limit), 20)),
"catalog_match": json.dumps(
[{
"catalog_id": (
_WAZUH_ALERTMANAGER_INTEGRATION_CATALOG_ID
)
}]
),
"failure_marker": (
_WAZUH_PRIVILEGED_CONVERGENCE_CAPABILITY_MISSING
),
},
)
rows = [dict(row) for row in result.mappings().all()]
stats["scanned"] = len(rows)
for row in rows:
source_input = {
"automation_run_id": row.get("automation_run_id"),
"trace_id": row.get("trace_id"),
"run_id": row.get("run_id"),
"work_item_id": row.get("work_item_id"),
"project_id": row.get("project_id"),
"run_namespace": row.get("run_namespace"),
"proposal_source": row.get("proposal_source"),
"source_receipt_ref": row.get("source_receipt_ref"),
"source_recurrence": row.get("source_recurrence"),
}
if await _insert_wazuh_break_glass_capability_packet(
db,
check_op_id=str(row.get("check_op_id") or ""),
source_candidate_op_id=str(
row.get("source_candidate_op_id") or ""
),
incident_id=str(row.get("incident_id") or ""),
catalog_id=_WAZUH_ALERTMANAGER_INTEGRATION_CATALOG_ID,
source_input=source_input,
):
stats["written"] += 1
except Exception as exc:
stats["error"] = f"{type(exc).__name__}: {exc}"[:500]
logger.warning(
"wazuh_break_glass_capability_packet_backfill_failed",
project_id=project_id,
**stats,
)
return stats
async def run_claimed_check_mode(
@@ -9144,6 +9475,18 @@ async def run_pending_check_modes_once(
receipt_stats.get("retry_terminal_projection_verified") or 0
),
"retry_terminal_projection_runtime_apply_executed": False,
"critical_break_glass_packet_scanned": int(
receipt_stats.get("critical_break_glass_packet_scanned") or 0
),
"critical_break_glass_packet_written": int(
receipt_stats.get("critical_break_glass_packet_written") or 0
),
"critical_break_glass_packet_error": (
str(
receipt_stats.get("critical_break_glass_packet_error") or ""
)[:500]
or None
),
"repair_receipt_backfill_error": (
str(receipt_stats.get("error") or "")[:500] or None
),

132
apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py
View File

@@ -130,6 +130,45 @@ _LIVE_RUNTIME_SQL = """
WHERE apply.operation_type = 'ansible_apply_executed'
ORDER BY apply.created_at DESC
LIMIT 1
), latest_break_glass_packet AS (
SELECT
packet.op_id::text AS break_glass_packet_receipt_id,
packet.status AS break_glass_packet_status,
packet.input ->> 'capability_packet_id' AS capability_packet_id,
packet.input ->> 'automation_run_id'
AS capability_automation_run_id,
packet.input ->> 'trace_id' AS capability_trace_id,
packet.input ->> 'run_id' AS capability_run_id,
packet.input ->> 'work_item_id' AS capability_work_item_id,
packet.input ->> 'project_id' AS capability_project_id,
packet.input ->> 'run_namespace' AS capability_run_namespace,
packet.input ->> 'proposal_source' AS capability_proposal_source,
packet.input -> 'source_recurrence'
AS capability_source_recurrence,
packet.input ->> 'capability_type' AS capability_type,
packet.input ->> 'risk_level' AS capability_risk_level,
packet.input ->> 'policy_route' AS capability_policy_route,
packet.output ->> 'queue_status' AS capability_queue_status,
coalesce(
(packet.input ->> 'execution_allowed')::boolean,
false
) AS capability_execution_allowed,
coalesce(
(packet.dry_run_result ->> 'capability_granted')::boolean,
false
) AS capability_granted,
coalesce(
(packet.dry_run_result ->> 'host_write_performed')::boolean,
false
) AS capability_packet_host_write_performed,
packet.created_at AS break_glass_packet_created_at
FROM automation_operation_log packet
JOIN latest_check check_mode
ON packet.parent_op_id::text = check_mode.check_op_id
WHERE packet.operation_type =
'ansible_break_glass_capability_work_item_queued'
ORDER BY packet.created_at DESC
LIMIT 1
), runtime_stages AS (
SELECT
coalesce(
@@ -239,6 +278,7 @@ _LIVE_RUNTIME_SQL = """
SELECT
candidate.*,
check_mode.*,
break_glass_packet.*,
apply.*,
runtime_stages.stage_ids,
runtime_stages.runtime_stage_identity_verified,
@@ -249,6 +289,7 @@ _LIVE_RUNTIME_SQL = """
telegram.*
FROM latest_candidate candidate
LEFT JOIN latest_check check_mode ON TRUE
LEFT JOIN latest_break_glass_packet break_glass_packet ON TRUE
LEFT JOIN latest_apply apply ON TRUE
LEFT JOIN runtime_stages ON TRUE
LEFT JOIN latest_verifier verifier ON TRUE
@@ -416,6 +457,43 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback(
)
check_failed = data.get("check_status") == "failed"
check_failure_class = _check_failure_class(data)
break_glass_packet_receipt_present = bool(
data.get("break_glass_packet_receipt_id")
)
capability_packet_id = str(data.get("capability_packet_id") or "")
break_glass_packet_identity_verified = bool(
break_glass_packet_receipt_present
and candidate_identity_verified
and check_identity_verified
and str(data.get("capability_automation_run_id") or "")
== candidate_op_id
and str(data.get("capability_trace_id") or "") == candidate_op_id
and str(data.get("capability_run_id") or "") == candidate_op_id
and str(data.get("capability_work_item_id") or "")
== expected_work_item_id
and str(data.get("capability_project_id") or "").strip()
and str(data.get("capability_run_namespace") or "")
== expected_run_namespace
and str(data.get("capability_proposal_source") or "")
== expected_proposal_source
and data.get("capability_source_recurrence")
== candidate_source_recurrence
)
break_glass_packet_present = all(
(
break_glass_packet_identity_verified,
data.get("break_glass_packet_status") == "pending",
capability_packet_id.startswith("IWZ-BG-"),
len(capability_packet_id) == 23,
data.get("capability_type") == "bounded_privileged_convergence",
data.get("capability_risk_level") == "critical",
data.get("capability_policy_route") == "critical_break_glass_queue",
data.get("capability_queue_status") == "queued",
data.get("capability_execution_allowed") is False,
data.get("capability_granted") is False,
data.get("capability_packet_host_write_performed") is False,
)
)
apply_terminal = data.get("apply_status") in {"success", "failed"}
apply_passed = apply_terminal and str(
data.get("apply_returncode") or ""
@@ -495,6 +573,16 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback(
active_blockers.append("wazuh_controlled_check_mode_failed")
if check_failure_class:
active_blockers.append(check_failure_class)
if check_failure_class == _PRIVILEGED_CONVERGENCE_CAPABILITY_MISSING:
active_blockers.append(
"critical_break_glass_capability_packet_pending"
if break_glass_packet_present
else (
"critical_break_glass_capability_packet_identity_mismatch"
if break_glass_packet_receipt_present
else "critical_break_glass_capability_packet_missing"
)
)
if apply_failed:
active_blockers.append("wazuh_controlled_apply_failed")
if apply_passed and not verifier_passed:
@@ -536,6 +624,9 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback(
"dispatch_candidate_count": 1 if candidate_present else 0,
"check_mode_passed_count": 1 if check_passed else 0,
"check_mode_failed_count": 1 if check_failed else 0,
"critical_break_glass_capability_packet_count": (
1 if break_glass_packet_present else 0
),
"runtime_execution_performed_count": 1 if apply_terminal else 0,
"bounded_posture_execution_passed_count": (
1 if apply_passed and not ingress_mode else 0
@@ -580,12 +671,36 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback(
],
"missing_stage_ids": missing_stage_ids,
"check_failure_class": check_failure_class,
"critical_break_glass_capability_packet": (
{
"receipt_id": str(
data.get("break_glass_packet_receipt_id") or ""
),
"capability_packet_id": str(
capability_packet_id
),
"status": "pending",
"queue_status": "queued",
"capability_type": "bounded_privileged_convergence",
"risk_level": "critical",
"policy_route": "critical_break_glass_queue",
"execution_allowed": False,
"capability_granted": False,
"host_write_performed": False,
"secret_value_exposed": False,
"secret_value_logged": False,
"secret_value_read_back": False,
}
if break_glass_packet_present
else None
),
"next_safe_action": _next_action(
ingress_mode=ingress_mode,
candidate_present=candidate_present,
check_passed=check_passed,
check_failed=check_failed,
check_failure_class=check_failure_class,
break_glass_packet_present=break_glass_packet_present,
apply_terminal=apply_terminal,
apply_passed=apply_passed,
verifier_passed=verifier_passed,
@@ -609,6 +724,17 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback(
"command_output_returned": False,
"inventory_identity_returned": False,
"secret_value_collection_allowed": False,
"critical_break_glass_required": (
check_failure_class
== _PRIVILEGED_CONVERGENCE_CAPABILITY_MISSING
),
"critical_break_glass_capability_packet_persisted": (
break_glass_packet_present
),
"critical_break_glass_capability_packet_identity_verified": (
break_glass_packet_identity_verified
),
"critical_break_glass_capability_granted": False,
"github_api_used": False,
},
"source_refs": [
@@ -658,6 +784,7 @@ def _next_action(
check_passed: bool,
check_failed: bool,
check_failure_class: str | None,
break_glass_packet_present: bool,
apply_terminal: bool,
apply_passed: bool,
verifier_passed: bool,
@@ -684,6 +811,11 @@ def _next_action(
check_failure_class
== _PRIVILEGED_CONVERGENCE_CAPABILITY_MISSING
):
if break_glass_packet_present:
return (
"critical_break_glass_capability_packet_queued_waiting_"
"bounded_capability_verifier_then_retry"
)
return (
"queue_critical_break_glass_privileged_convergence_capability_"
"then_bounded_retry"