From fe95d61a34b5e21a73b19adf4121fa662d788132 Mon Sep 17 00:00:00 2001 From: ogt Date: Tue, 14 Jul 2026 10:05:53 +0800 Subject: [PATCH] fix(security): restore control plane maintenance ownership --- apps/api/src/core/config.py | 8 ++++++ apps/api/src/workers/signal_worker.py | 28 +++++++++++++++++++ ..._signal_worker_ansible_executor_binding.py | 19 ++++++++++++- k8s/awoooi-prod/06-deployment-api.yaml | 2 ++ k8s/awoooi-prod/08-deployment-worker.yaml | 2 ++ 5 files changed, 58 insertions(+), 1 deletion(-) diff --git a/apps/api/src/core/config.py b/apps/api/src/core/config.py index aced8db6a..bb8edbeed 100644 --- a/apps/api/src/core/config.py +++ b/apps/api/src/core/config.py @@ -751,6 +751,14 @@ class Settings(BaseSettings): "then enqueue them for the existing check-mode worker." ), ) + ENABLE_SECURITY_CONTROL_PLANE_MAINTENANCE_WORKER: bool = Field( + default=False, + description=( + "True=run externally read-only discovery plus bounded rule catalog, " + "compliance, coverage, and asset drift control-plane writebacks in the " + "dedicated signal worker." + ), + ) AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_INTERVAL_SECONDS: int = Field( default=600, ge=60, diff --git a/apps/api/src/workers/signal_worker.py b/apps/api/src/workers/signal_worker.py index bee848313..1ef309923 100644 --- a/apps/api/src/workers/signal_worker.py +++ b/apps/api/src/workers/signal_worker.py @@ -621,6 +621,7 @@ async def _main() -> None: worker = await init_signal_worker() ansible_candidate_backfill_task: asyncio.Task[Any] | None = None backup_restore_legacy_backfill_task: asyncio.Task[Any] | None = None + security_maintenance_tasks: list[asyncio.Task[Any]] = [] if settings.ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER: from src.jobs.awooop_ansible_candidate_backfill_job import ( run_awooop_ansible_candidate_backfill_loop, @@ -662,6 +663,29 @@ async def _main() -> None: endpoint_class=relay_preflight["endpoint_class"], telegram_dispatch_performed=False, ) + if settings.ENABLE_SECURITY_CONTROL_PLANE_MAINTENANCE_WORKER: + from src.jobs.asset_change_tracker_job import run_asset_change_tracker_loop + from src.jobs.asset_scanner_job import run_asset_scanner_loop + from src.jobs.compliance_scanner_job import run_compliance_scanner_loop + from src.jobs.coverage_evaluator_job import run_coverage_evaluator_loop + from src.jobs.rule_catalog_sync_job import run_rule_catalog_sync_loop + + security_maintenance_loops = ( + ("run_asset_scanner_loop", run_asset_scanner_loop), + ("run_rule_catalog_sync_loop", run_rule_catalog_sync_loop), + ("run_compliance_scanner_loop", run_compliance_scanner_loop), + ("run_coverage_evaluator_loop", run_coverage_evaluator_loop), + ("run_asset_change_tracker_loop", run_asset_change_tracker_loop), + ) + security_maintenance_tasks = [ + asyncio.create_task(loop(), name=task_name) + for task_name, loop in security_maintenance_loops + ] + logger.info( + "signal_worker_security_control_plane_maintenance_started", + task_names=[task.get_name() for task in security_maintenance_tasks], + task_count=len(security_maintenance_tasks), + ) # Setup graceful shutdown shutdown_event = asyncio.Event() @@ -699,6 +723,10 @@ async def _main() -> None: await backup_restore_legacy_backfill_task except asyncio.CancelledError: pass + for task in security_maintenance_tasks: + task.cancel() + if security_maintenance_tasks: + await asyncio.gather(*security_maintenance_tasks, return_exceptions=True) await worker.stop() await close_worker_redis_pool() # 關閉 Worker 專屬連線 await close_redis_pool() diff --git a/apps/api/tests/test_signal_worker_ansible_executor_binding.py b/apps/api/tests/test_signal_worker_ansible_executor_binding.py index 37b6a5620..4356a0108 100644 --- a/apps/api/tests/test_signal_worker_ansible_executor_binding.py +++ b/apps/api/tests/test_signal_worker_ansible_executor_binding.py @@ -8,13 +8,22 @@ import yaml from src.workers import ansible_executor_broker, signal_worker -def test_signal_worker_only_produces_ansible_candidates() -> None: +def test_signal_worker_produces_candidates_and_owns_security_maintenance() -> None: source = inspect.getsource(signal_worker._main) assert "run_awooop_ansible_candidate_backfill_loop" in source assert "signal_worker_ansible_candidate_backfill_loop_started" in source assert "ansible_candidate_backfill_task.cancel()" in source assert "await ansible_candidate_backfill_task" in source + assert "ENABLE_SECURITY_CONTROL_PLANE_MAINTENANCE_WORKER" in source + assert "run_asset_scanner_loop" in source + assert "run_rule_catalog_sync_loop" in source + assert "run_compliance_scanner_loop" in source + assert "run_coverage_evaluator_loop" in source + assert "run_asset_change_tracker_loop" in source + assert "signal_worker_security_control_plane_maintenance_started" in source + assert "task.cancel()" in source + assert "await asyncio.gather(*security_maintenance_tasks" in source assert "run_awooop_ansible_check_mode_loop" not in source assert "signal_worker_ansible_check_mode_loop_started" not in source @@ -37,6 +46,7 @@ def test_worker_manifest_has_no_execution_transport_or_write_loop() -> None: assert 'command: ["python", "-m", "src.workers.signal_worker"]' in manifest assert "serviceAccountName: awoooi-worker" in manifest assert "ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER" in manifest + assert "ENABLE_SECURITY_CONTROL_PLANE_MAINTENANCE_WORKER" in manifest assert "AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_STARTUP_SLEEP_SECONDS" in manifest assert "ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER" in manifest assert 'value: "false"' in manifest @@ -45,6 +55,12 @@ def test_worker_manifest_has_no_execution_transport_or_write_loop() -> None: assert "mountPath: /run/secrets/ssh_mcp_key" not in manifest assert "mountPath: /etc/ssh-mcp/known_hosts" not in manifest assert "secretName: ssh-mcp-key" not in manifest + deployment = yaml.safe_load(manifest) + worker_env = { + item["name"]: item.get("value") + for item in deployment["spec"]["template"]["spec"]["containers"][0]["env"] + } + assert worker_env["ENABLE_SECURITY_CONTROL_PLANE_MAINTENANCE_WORKER"] == "true" def test_execution_broker_is_only_workload_with_ssh_transport() -> None: @@ -95,6 +111,7 @@ def test_api_manifest_has_read_only_identity_and_no_ssh_executor_mounts() -> Non assert pod_spec["serviceAccountName"] == "awoooi-api" assert env["SSH_MCP_ENABLED"] == "false" assert env["DATABASE_NULL_POOL"] == "true" + assert env["AWOOOI_API_BACKGROUND_TASKS_ENABLED"] == "false" assert env["ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER"] == "false" assert env["ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER"] == "false" assert env["ENABLE_IWOOOS_WAZUH_MANAGER_POSTURE_EXECUTOR"] == "true" diff --git a/k8s/awoooi-prod/06-deployment-api.yaml b/k8s/awoooi-prod/06-deployment-api.yaml index e3ddd1f24..70f1b4055 100644 --- a/k8s/awoooi-prod/06-deployment-api.yaml +++ b/k8s/awoooi-prod/06-deployment-api.yaml @@ -174,6 +174,8 @@ spec: # 2026-07-11 Codex: release each API connection after use while # workload-specific DB identities are being split and verified. value: "true" + - name: AWOOOI_API_BACKGROUND_TASKS_ENABLED + value: "false" - name: IWOOOS_WAZUH_READONLY_ENABLED # 2026-06-30 Codex: controlled GitOps enablement after owner # metadata, manager registry acceptance, dry-run, rollback, and diff --git a/k8s/awoooi-prod/08-deployment-worker.yaml b/k8s/awoooi-prod/08-deployment-worker.yaml index 0ffda7f61..104dcd018 100644 --- a/k8s/awoooi-prod/08-deployment-worker.yaml +++ b/k8s/awoooi-prod/08-deployment-worker.yaml @@ -176,6 +176,8 @@ spec: value: "968bd794d16f53192484a381477aa7c091ed9648" - name: ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER value: "true" + - name: ENABLE_SECURITY_CONTROL_PLANE_MAINTENANCE_WORKER + value: "true" - name: AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_INTERVAL_SECONDS value: "600" - name: AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_BATCH_LIMIT