From 964beb2564002028423903a8daa20fb4e6570556 Mon Sep 17 00:00:00 2001 From: ogt Date: Sat, 11 Jul 2026 21:01:51 +0800 Subject: [PATCH] fix(security): restore compliance scanner readback --- apps/api/src/jobs/compliance_scanner_job.py | 26 ++++--- .../iwooos_security_asset_control_plane.py | 9 ++- apps/api/tests/test_compliance_scanner_job.py | 72 +++++++++++++++++++ ...est_iwooos_security_asset_control_plane.py | 9 +++ 4 files changed, 105 insertions(+), 11 deletions(-) create mode 100644 apps/api/tests/test_compliance_scanner_job.py diff --git a/apps/api/src/jobs/compliance_scanner_job.py b/apps/api/src/jobs/compliance_scanner_job.py index 8453f3315..5023254bf 100644 --- a/apps/api/src/jobs/compliance_scanner_job.py +++ b/apps/api/src/jobs/compliance_scanner_job.py @@ -29,7 +29,7 @@ from __future__ import annotations import asyncio import json as _json import time as _time -from datetime import datetime, timedelta, timezone +from datetime import UTC, datetime, timedelta, timezone from typing import Any import structlog @@ -42,6 +42,7 @@ logger = structlog.get_logger(__name__) _FIRST_DELAY_SEC = 180 _LOOP_BACKOFF_SEC = 1800 _DAILY_TRIGGER_HOUR_TAIPEI = 3 +_PROJECT_ID = "awoooi" # 7 維 compliance (ADR-090 schema CHECK) _DIMENSIONS = ( @@ -143,10 +144,11 @@ async def scan_once(triggered_by: str = "cron") -> dict[str, int]: async def _fetch_active_assets() -> list[dict[str, Any]]: """從 asset_inventory 撈所有 active asset.""" from sqlalchemy import text as _sql + from src.db.base import get_db_context try: - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: result = await db.execute( _sql(""" SELECT asset_id, asset_key, asset_type, metadata @@ -196,6 +198,7 @@ async def _write_compliance_for_asset(asset: dict[str, Any]) -> tuple[int, int, Returns: (snapshots_written, violations_count, warnings_count) """ from sqlalchemy import text as _sql + from src.db.base import get_db_context snapshots = 0 @@ -206,7 +209,7 @@ async def _write_compliance_for_asset(asset: dict[str, Any]) -> tuple[int, int, dimension_results = await asyncio.to_thread(_evaluate_all_dimensions, asset) try: - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: for dim, (status, detail) in dimension_results.items(): await db.execute( _sql(""" @@ -297,10 +300,9 @@ def _check_ssl_cert(asset: dict[str, Any]) -> tuple[str, dict]: if not https_target: return ("unknown", {"reason": "no https scrape_url / instance"}) - import ssl import socket + import ssl from urllib.parse import urlparse - from datetime import datetime try: parsed = urlparse(https_target) @@ -331,8 +333,8 @@ def _check_ssl_cert(asset: dict[str, Any]) -> tuple[str, dict]: def _check_ssl_cert_via_verified_socket(hostname: str, port: int) -> tuple[str, dict]: """用 verified socket 拿 dict form cert, 取 notAfter 判斷剩餘天數.""" - import ssl import socket + import ssl from datetime import datetime try: @@ -383,7 +385,7 @@ def _check_secret_rotation(asset: dict[str, Any]) -> tuple[str, dict]: except (ValueError, TypeError): return ("unknown", {"reason": f"unparseable timestamp: {created_ts[:50]}"}) - now_utc = datetime.now(timezone.utc) + now_utc = datetime.now(UTC) age_days = (now_utc - created).days if age_days > _SECRET_ROTATION_WARNING_DAYS: @@ -437,6 +439,7 @@ async def _llm_analyze_compliance_posture( """ try: import json as _j + from src.services.llm_json_parser import parse_llm_json_response from src.services.openclaw import get_openclaw @@ -470,8 +473,12 @@ async def _send_telegram_posture( """推 Telegram 合規摘要 + 互動按鈕 (P0 修).""" try: import html + from src.core.config import settings - from src.services.ai_advisory_helpers import build_ai_advisory_keyboard, is_snoozed + from src.services.ai_advisory_helpers import ( + build_ai_advisory_keyboard, + is_snoozed, + ) from src.services.telegram_gateway import get_telegram_gateway target_chat_id = settings.SRE_GROUP_CHAT_ID @@ -543,11 +550,12 @@ async def _send_telegram_posture( async def _log_aol(stats: dict[str, int], duration_ms: int, triggered_by: str, error: str | None) -> None: try: from sqlalchemy import text as _sql + from src.db.base import get_db_context aol_status = "failed" if error else "success" - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: await db.execute( _sql(""" INSERT INTO automation_operation_log ( 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 cfd3773ea..004c69f7c 100644 --- a/apps/api/src/services/iwooos_security_asset_control_plane.py +++ b/apps/api/src/services/iwooos_security_asset_control_plane.py @@ -1486,10 +1486,15 @@ class IwoooSSecurityAssetControlPlaneService: source_id="security_compliance", statement=_sql( """ - WITH latest AS ( + WITH recent AS MATERIALIZED ( + SELECT snapshot_id, asset_id, dimension, status, detected_at + FROM asset_compliance_snapshot + ORDER BY snapshot_id DESC + LIMIT 100000 + ), latest AS ( SELECT DISTINCT ON (asset_id, dimension) asset_id, dimension, status - FROM asset_compliance_snapshot + FROM recent ORDER BY asset_id, dimension, detected_at DESC ) SELECT dimension, status, count(*) AS cnt diff --git a/apps/api/tests/test_compliance_scanner_job.py b/apps/api/tests/test_compliance_scanner_job.py new file mode 100644 index 000000000..5c1458e4b --- /dev/null +++ b/apps/api/tests/test_compliance_scanner_job.py @@ -0,0 +1,72 @@ +from __future__ import annotations + +import asyncio + +from src.jobs import compliance_scanner_job + + +class _AssetRow: + asset_id = 1 + asset_key = "host/test" + asset_type = "host" + metadata = {} + + +class _Result: + def fetchall(self): + return [_AssetRow()] + + +class _Database: + async def execute(self, statement, parameters=None): + return _Result() + + +class _Context: + async def __aenter__(self): + return _Database() + + async def __aexit__(self, exc_type, exc, traceback): + return False + + +def test_background_compliance_scanner_scopes_database_operations(monkeypatch) -> None: + requested_projects: list[str | None] = [] + + def fake_db_context(project_id: str | None = None): + requested_projects.append(project_id) + return _Context() + + monkeypatch.setattr("src.db.base.get_db_context", fake_db_context) + monkeypatch.setattr( + compliance_scanner_job, + "_evaluate_all_dimensions", + lambda asset: {"audit_log_enabled": ("compliant", {"source": "test"})}, + ) + + async def exercise_compliance_operations() -> None: + assets = await compliance_scanner_job._fetch_active_assets() + assert assets == [ + { + "asset_id": 1, + "asset_key": "host/test", + "asset_type": "host", + "metadata": {}, + } + ] + + written, violations, warnings = ( + await compliance_scanner_job._write_compliance_for_asset(assets[0]) + ) + assert (written, violations, warnings) == (1, 0, 0) + + await compliance_scanner_job._log_aol( + stats={"assets_scanned": 1, "snapshots_written": 1}, + duration_ms=10, + triggered_by="cron", + error=None, + ) + + asyncio.run(exercise_compliance_operations()) + + assert requested_projects == ["awoooi"] * 3 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 e684eb9b0..8f5f4fb5b 100644 --- a/apps/api/tests/test_iwooos_security_asset_control_plane.py +++ b/apps/api/tests/test_iwooos_security_asset_control_plane.py @@ -345,6 +345,7 @@ def test_service_uses_canonical_awoooi_project_scope(monkeypatch) -> None: def test_service_bounds_source_concurrency_for_database_budget(monkeypatch) -> None: service = IwoooSSecurityAssetControlPlaneService() concurrency = {"active": 0, "maximum": 0} + statements: list[str] = [] class EmptyResult: def fetchall(self): @@ -358,6 +359,7 @@ def test_service_bounds_source_concurrency_for_database_budget(monkeypatch) -> N class SlowDb: async def execute(self, statement): + statements.append(str(statement)) concurrency["active"] += 1 concurrency["maximum"] = max(concurrency["maximum"], concurrency["active"]) await asyncio.sleep(0.02) @@ -381,6 +383,13 @@ def test_service_bounds_source_concurrency_for_database_budget(monkeypatch) -> N assert payload["summary"]["source_count"] == 9 assert payload["summary"]["ready_source_count"] == 9 assert payload["summary"]["unavailable_source_count"] == 0 + compliance_query = next( + statement + for statement in statements + if "asset_compliance_snapshot" in statement + ) + assert "ORDER BY snapshot_id DESC" in compliance_query + assert "LIMIT 100000" in compliance_query def test_public_api_returns_live_aggregate(monkeypatch) -> None: