From ea857399517abed2824e536a2dda7e77463512c3 Mon Sep 17 00:00:00 2001 From: ogt Date: Sat, 11 Jul 2026 19:59:18 +0800 Subject: [PATCH] fix(iwooos): isolate control plane source reads --- .../iwooos_security_asset_control_plane.py | 170 ++++++++++++++---- ...est_iwooos_security_asset_control_plane.py | 58 +++++- 2 files changed, 191 insertions(+), 37 deletions(-) diff --git a/apps/api/src/services/iwooos_security_asset_control_plane.py b/apps/api/src/services/iwooos_security_asset_control_plane.py index 79644b691..c7f61ff2b 100644 --- a/apps/api/src/services/iwooos_security_asset_control_plane.py +++ b/apps/api/src/services/iwooos_security_asset_control_plane.py @@ -1069,6 +1069,9 @@ def build_unavailable_iwooos_security_asset_control_plane( "source_status": "live_database_unavailable", "reason_code": reason_code, "summary": { + "source_count": 9, + "ready_source_count": 0, + "unavailable_source_count": 9, "managed_asset_count": 0, "expected_asset_type_count": len(_ASSET_TYPES), "present_asset_type_count": 0, @@ -1212,6 +1215,24 @@ def build_unavailable_iwooos_security_asset_control_plane( "accepted_ai_trace_count_24h": 0, "immutable_external_audit_store_evidenced": False, }, + "source_health": [ + { + "source_id": source_id, + "status": "unavailable", + "reason_code": reason_code, + } + for source_id in ( + "asset_discovery", + "asset_inventory", + "automation_coverage", + "asset_relationship", + "security_compliance", + "automation_operations", + "ai_decision_trace", + "siem_runtime", + "audit_runtime", + ) + ], "work_items": [ _work_item( "AIA-P0-006-01", @@ -1250,6 +1271,44 @@ def build_unavailable_iwooos_security_asset_control_plane( } +async def _read_public_source( + db: Any, + *, + source_id: str, + statement: Any, + result_mode: str, + default: Any, +) -> tuple[Any, dict[str, Any]]: + """Isolate one aggregate query so an optional source cannot abort the readback.""" + try: + async with db.begin_nested(): + result = await db.execute(statement) + if result_mode == "all": + value = result.fetchall() + elif result_mode == "one": + value = result.one() + elif result_mode == "one_or_none": + value = result.one_or_none() + else: + raise ValueError("unsupported_result_mode") + return value, { + "source_id": source_id, + "status": "ready", + "reason_code": None, + } + except Exception as exc: # noqa: BLE001 - isolate and redact source failures + logger.warning( + "iwooos_security_asset_control_plane_source_unavailable", + source_id=source_id, + error_type=type(exc).__name__, + ) + return default, { + "source_id": source_id, + "status": "unavailable", + "reason_code": f"{source_id}_query_failed", + } + + class IwoooSSecurityAssetControlPlaneService: """Read aggregate production security state with a bounded DB timeout.""" @@ -1274,8 +1333,12 @@ class IwoooSSecurityAssetControlPlaneService: async def _load_snapshot(self) -> dict[str, Any]: async with get_db_context("awoooi") as db: - discovery_result = await db.execute( - _sql( + source_health: list[dict[str, Any]] = [] + + discovery_row, health = await _read_public_source( + db, + source_id="asset_discovery", + statement=_sql( """ SELECT latest.status AS latest_status, @@ -1303,12 +1366,16 @@ class IwoooSSecurityAssetControlPlaneService: LIMIT 1 ) successful ON true """ - ) + ), + result_mode="one_or_none", + default=None, ) - discovery_row = discovery_result.one_or_none() + source_health.append(health) - inventory_result = await db.execute( - _sql( + inventory_rows, health = await _read_public_source( + db, + source_id="asset_inventory", + statement=_sql( """ SELECT asset_type, @@ -1324,12 +1391,16 @@ class IwoooSSecurityAssetControlPlaneService: GROUP BY asset_type ORDER BY asset_type """ - ) + ), + result_mode="all", + default=[], ) - inventory_rows = inventory_result.fetchall() + source_health.append(health) - coverage_result = await db.execute( - _sql( + coverage_rows, health = await _read_public_source( + db, + source_id="automation_coverage", + statement=_sql( """ SELECT dimension, coverage_status AS status, count(*) AS cnt FROM asset_coverage_snapshot @@ -1343,12 +1414,16 @@ class IwoooSSecurityAssetControlPlaneService: GROUP BY dimension, coverage_status ORDER BY dimension, coverage_status """ - ) + ), + result_mode="all", + default=[], ) - coverage_rows = coverage_result.fetchall() + source_health.append(health) - relationship_result = await db.execute( - _sql( + relationship_row, health = await _read_public_source( + db, + source_id="asset_relationship", + statement=_sql( """ SELECT (SELECT count(*) FROM asset_relationship WHERE is_active) AS relationship_count, @@ -1363,12 +1438,16 @@ class IwoooSSecurityAssetControlPlaneService: FROM asset_inventory asset WHERE asset.lifecycle_state = 'active' """ - ) + ), + result_mode="one", + default={}, ) - relationship_row = relationship_result.one() + source_health.append(health) - compliance_result = await db.execute( - _sql( + compliance_rows, health = await _read_public_source( + db, + source_id="security_compliance", + statement=_sql( """ WITH latest AS ( SELECT DISTINCT ON (asset_id, dimension) @@ -1381,12 +1460,16 @@ class IwoooSSecurityAssetControlPlaneService: GROUP BY dimension, status ORDER BY dimension, status """ - ) + ), + result_mode="all", + default=[], ) - compliance_rows = compliance_result.fetchall() + source_health.append(health) - automation_result = await db.execute( - _sql( + automation_rows, health = await _read_public_source( + db, + source_id="automation_operations", + statement=_sql( """ SELECT operation_type, status, count(*) AS cnt FROM automation_operation_log @@ -1394,12 +1477,16 @@ class IwoooSSecurityAssetControlPlaneService: GROUP BY operation_type, status ORDER BY operation_type, status """ - ) + ), + result_mode="all", + default=[], ) - automation_rows = automation_result.fetchall() + source_health.append(health) - ai_trace_result = await db.execute( - _sql( + ai_trace_row, health = await _read_public_source( + db, + source_id="ai_decision_trace", + statement=_sql( """ SELECT count(*) AS trace_count, @@ -1408,12 +1495,16 @@ class IwoooSSecurityAssetControlPlaneService: JOIN automation_operation_log operation ON operation.op_id = trace.op_id WHERE operation.created_at >= NOW() - INTERVAL '24 hours' """ - ) + ), + result_mode="one", + default={}, ) - ai_trace_row = ai_trace_result.one() + source_health.append(health) - siem_result = await db.execute( - _sql( + siem_row, health = await _read_public_source( + db, + source_id="siem_runtime", + statement=_sql( """ SELECT (SELECT count(*) FROM alert_rule_catalog) AS rule_total, @@ -1454,12 +1545,16 @@ class IwoooSSecurityAssetControlPlaneService: AND resolved_at >= NOW() - INTERVAL '24 hours') AS mttr_minutes_24h """ - ) + ), + result_mode="one", + default={}, ) - siem_row = siem_result.one() + source_health.append(health) - audit_result = await db.execute( - _sql( + audit_row, health = await _read_public_source( + db, + source_id="audit_runtime", + statement=_sql( """ SELECT (SELECT count(*) FROM audit_logs @@ -1486,9 +1581,11 @@ class IwoooSSecurityAssetControlPlaneService: AND ai_analysis IS NOT NULL) AS asset_change_ai_analysis_count_24h """ - ) + ), + result_mode="one", + default={}, ) - audit_row = audit_result.one() + source_health.append(health) return build_iwooos_security_asset_control_plane( discovery_row=discovery_row, @@ -1500,6 +1597,7 @@ class IwoooSSecurityAssetControlPlaneService: ai_trace_row=ai_trace_row, siem_row=siem_row, audit_row=audit_row, + source_health=source_health, ) diff --git a/apps/api/tests/test_iwooos_security_asset_control_plane.py b/apps/api/tests/test_iwooos_security_asset_control_plane.py index f6db4f084..85d153d29 100644 --- a/apps/api/tests/test_iwooos_security_asset_control_plane.py +++ b/apps/api/tests/test_iwooos_security_asset_control_plane.py @@ -16,6 +16,7 @@ from fastapi.testclient import TestClient from src.api.v1.iwooos import router from src.services.iwooos_security_asset_control_plane import ( IwoooSSecurityAssetControlPlaneService, + _read_public_source, build_iwooos_security_asset_control_plane, build_unavailable_iwooos_security_asset_control_plane, ) @@ -25,7 +26,7 @@ def _row(**values): return SimpleNamespace(**values) -def _ready_payload() -> dict: +def _ready_payload(source_health: list[dict] | None = None) -> dict: now = datetime(2026, 7, 11, 4, 0, tzinfo=UTC) discovery = _row( latest_status="success", @@ -111,6 +112,7 @@ def _ready_payload() -> dict: asset_change_count_24h=4, asset_change_ai_analysis_count_24h=3, ), + source_health=source_health, generated_at=now, ) @@ -201,6 +203,60 @@ def test_unknown_coverage_and_compliance_create_p0_work_items() -> None: } <= work_item_ids +def test_partial_source_failure_is_visible_without_erasing_core_inventory() -> None: + source_health = [ + {"source_id": "asset_discovery", "status": "ready", "reason_code": None}, + {"source_id": "asset_inventory", "status": "ready", "reason_code": None}, + { + "source_id": "audit_runtime", + "status": "unavailable", + "reason_code": "audit_runtime_query_failed", + }, + ] + payload = _ready_payload(source_health) + + assert payload["status"] == "degraded" + assert payload["source_status"] == "live_database_partial" + assert payload["summary"]["managed_asset_count"] == 24 + assert payload["summary"]["ready_source_count"] == 2 + assert payload["summary"]["unavailable_source_count"] == 1 + assert payload["source_health"] == source_health + assert payload["work_items"][0]["work_item_id"] == "AIA-P0-006-00" + + +def test_read_public_source_returns_redacted_reason_code_on_query_failure() -> None: + class NestedContext: + async def __aenter__(self): + return None + + async def __aexit__(self, exc_type, exc, traceback): + return False + + class FailingDb: + def begin_nested(self): + return NestedContext() + + async def execute(self, statement): + raise RuntimeError("private table detail") + + value, health = asyncio.run( + _read_public_source( + FailingDb(), + source_id="audit_runtime", + statement="SELECT 1", + result_mode="one", + default={}, + ) + ) + + assert value == {} + assert health == { + "source_id": "audit_runtime", + "status": "unavailable", + "reason_code": "audit_runtime_query_failed", + } + + def test_unavailable_payload_is_fixed_degraded_contract_without_raw_error() -> None: payload = build_unavailable_iwooos_security_asset_control_plane( "database_query_failed"