282 lines
10 KiB
Python
282 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
# ruff: noqa: E402, I001
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
from datetime import UTC, datetime, timedelta
|
|
from types import SimpleNamespace
|
|
|
|
os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test")
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from src.api.v1.iwooos import router
|
|
from src.services.iwooos_security_asset_control_plane import (
|
|
IwoooSSecurityAssetControlPlaneService,
|
|
build_iwooos_security_asset_control_plane,
|
|
build_unavailable_iwooos_security_asset_control_plane,
|
|
)
|
|
|
|
|
|
def _row(**values):
|
|
return SimpleNamespace(**values)
|
|
|
|
|
|
def _ready_payload() -> dict:
|
|
now = datetime(2026, 7, 11, 4, 0, tzinfo=UTC)
|
|
discovery = _row(
|
|
latest_status="success",
|
|
latest_scan_depth="full",
|
|
latest_started_at=now - timedelta(minutes=31),
|
|
latest_ended_at=now - timedelta(minutes=30),
|
|
latest_success_ended_at=now - timedelta(minutes=30),
|
|
latest_total_assets=24,
|
|
latest_new_assets=2,
|
|
latest_modified_assets=1,
|
|
latest_disappeared_assets=0,
|
|
)
|
|
inventory = [
|
|
_row(asset_type="host", cnt=3, unowned_count=1, stale_count=0),
|
|
_row(asset_type="k8s_workload", cnt=5, unowned_count=0, stale_count=1),
|
|
_row(asset_type="website", cnt=2, unowned_count=0, stale_count=0),
|
|
_row(asset_type="api_endpoint", cnt=4, unowned_count=0, stale_count=0),
|
|
_row(asset_type="database", cnt=2, unowned_count=0, stale_count=0),
|
|
_row(asset_type="log_stream", cnt=4, unowned_count=0, stale_count=0),
|
|
_row(asset_type="monitoring_target", cnt=3, unowned_count=0, stale_count=0),
|
|
_row(asset_type="ai_agent", cnt=1, unowned_count=0, stale_count=0),
|
|
]
|
|
coverage = [
|
|
_row(dimension=dimension, status="green", cnt=20)
|
|
for dimension in (
|
|
"auto_monitoring",
|
|
"auto_alerting",
|
|
"auto_rule_creation",
|
|
"auto_rule_matching",
|
|
"auto_playbook",
|
|
"auto_remediation",
|
|
"auto_km_creation",
|
|
)
|
|
]
|
|
compliance = [
|
|
_row(dimension=dimension, status="compliant", cnt=20)
|
|
for dimension in (
|
|
"ssl_cert_valid",
|
|
"cve_scan",
|
|
"secret_rotated",
|
|
"backup_tested",
|
|
"audit_log_enabled",
|
|
"access_reviewed",
|
|
"encryption_at_rest",
|
|
)
|
|
]
|
|
automation = [
|
|
_row(operation_type="asset_discovered", status="success", cnt=1),
|
|
_row(operation_type="rule_matched", status="success", cnt=3),
|
|
_row(operation_type="rule_created", status="success", cnt=2),
|
|
_row(operation_type="ansible_check_mode_executed", status="dry_run", cnt=2),
|
|
_row(operation_type="ansible_apply_executed", status="success", cnt=1),
|
|
_row(operation_type="remediation_verified", status="success", cnt=1),
|
|
_row(operation_type="km_created", status="success", cnt=1),
|
|
_row(operation_type="ansible_apply_executed", status="failed", cnt=1),
|
|
]
|
|
return build_iwooos_security_asset_control_plane(
|
|
discovery_row=discovery,
|
|
inventory_rows=inventory,
|
|
coverage_rows=coverage,
|
|
relationship_row=_row(relationship_count=21, orphan_asset_count=3),
|
|
compliance_rows=compliance,
|
|
automation_rows=automation,
|
|
ai_trace_row=_row(trace_count=4, accepted_trace_count=3),
|
|
siem_row=_row(
|
|
rule_total=12,
|
|
approved_rule_count=10,
|
|
fired_rule_count=8,
|
|
noisy_rule_count=1,
|
|
ai_generated_rule_count=3,
|
|
incident_total_24h=5,
|
|
open_incident_count_24h=2,
|
|
resolved_incident_count_24h=3,
|
|
decision_chain_count_24h=5,
|
|
normalized_event_count_24h=18,
|
|
mttr_minutes_24h=14.25,
|
|
),
|
|
audit_row=_row(
|
|
k8s_audit_count_24h=2,
|
|
k8s_audit_failure_count_24h=0,
|
|
mcp_audit_count_24h=11,
|
|
mcp_audit_failure_count_24h=1,
|
|
asset_change_count_24h=4,
|
|
asset_change_ai_analysis_count_24h=3,
|
|
),
|
|
generated_at=now,
|
|
)
|
|
|
|
|
|
def test_builds_live_asset_siem_audit_and_ai_control_plane_without_false_closure() -> (
|
|
None
|
|
):
|
|
payload = _ready_payload()
|
|
|
|
assert payload["schema_version"] == "iwooos_security_asset_control_plane_v1"
|
|
assert payload["status"] == "ready"
|
|
assert payload["source_status"] == "live_database_aggregate"
|
|
assert payload["summary"]["managed_asset_count"] == 24
|
|
assert payload["summary"]["present_asset_type_count"] == 8
|
|
assert payload["summary"]["siem_stage_coverage_percent"] == 100
|
|
assert payload["summary"]["ai_loop_stage_coverage_percent"] == 100
|
|
assert payload["summary"]["verified_remediation_receipt_count_24h"] == 1
|
|
assert payload["discovery"]["fresh"] is True
|
|
assert payload["ai_automation"]["same_run_closed_loop_proven"] is False
|
|
assert payload["ai_automation"]["same_run_closed_loop_count"] == 0
|
|
assert len(payload["nist_csf_functions"]) == 6
|
|
assert payload["summary"]["security_program_domain_count"] == 16
|
|
assert len(payload["security_program_domains"]) == 16
|
|
domain_by_id = {
|
|
domain["domain_id"]: domain for domain in payload["security_program_domains"]
|
|
}
|
|
assert domain_by_id["endpoint_xdr_wazuh"]["status"] == "not_evidenced"
|
|
assert domain_by_id["siem_soc_threat_intel"]["controlled_percent"] == 50
|
|
assert len(payload["siem"]["pipeline"]) == 8
|
|
assert payload["security_audit"]["mcp_tool_audit_count_24h"] == 11
|
|
assert any(
|
|
item["work_item_id"] == "AIA-P0-006-02" for item in payload["work_items"]
|
|
)
|
|
assert any(
|
|
item["work_item_id"] == "AIA-P1-001-01" for item in payload["work_items"]
|
|
)
|
|
|
|
public_text = json.dumps(payload, ensure_ascii=False)
|
|
assert "192.168." not in public_text
|
|
assert "postgresql://" not in public_text
|
|
assert "target_resource" not in public_text
|
|
assert '"event_payload":' not in public_text
|
|
assert 'secret_value_collection_allowed": true' not in public_text.lower()
|
|
|
|
|
|
def test_unknown_coverage_and_compliance_create_p0_work_items() -> None:
|
|
now = datetime(2026, 7, 11, 4, 0, tzinfo=UTC)
|
|
payload = build_iwooos_security_asset_control_plane(
|
|
discovery_row=_row(
|
|
latest_status="success",
|
|
latest_scan_depth="shallow",
|
|
latest_started_at=now - timedelta(minutes=10),
|
|
latest_ended_at=now - timedelta(minutes=9),
|
|
latest_success_ended_at=now - timedelta(minutes=9),
|
|
latest_total_assets=1,
|
|
),
|
|
inventory_rows=[
|
|
_row(asset_type="host", cnt=1, unowned_count=1, stale_count=0),
|
|
],
|
|
coverage_rows=[
|
|
_row(dimension="auto_monitoring", status="unknown", cnt=1),
|
|
],
|
|
relationship_row=_row(relationship_count=0, orphan_asset_count=1),
|
|
compliance_rows=[
|
|
_row(dimension="cve_scan", status="unknown", cnt=1),
|
|
],
|
|
automation_rows=[],
|
|
ai_trace_row=_row(trace_count=0, accepted_trace_count=0),
|
|
siem_row=_row(),
|
|
audit_row=_row(),
|
|
generated_at=now,
|
|
)
|
|
|
|
assert payload["summary"]["automation_coverage_green_percent"] == 0
|
|
assert payload["summary"]["automation_coverage_unknown_count"] == 1
|
|
assert payload["summary"]["compliance_unknown_count"] == 1
|
|
assert payload["summary"]["siem_observed_stage_count"] == 0
|
|
assert payload["summary"]["ai_loop_observed_stage_count"] == 0
|
|
work_item_ids = {item["work_item_id"] for item in payload["work_items"]}
|
|
assert {
|
|
"AIA-P0-006-02",
|
|
"AIA-P0-006-03",
|
|
"AIA-P0-006-04",
|
|
"AIA-P0-006-05",
|
|
"AIA-P0-006-06",
|
|
"AIA-P0-007-01",
|
|
"AIA-P0-008-01",
|
|
} <= work_item_ids
|
|
|
|
|
|
def test_unavailable_payload_is_fixed_degraded_contract_without_raw_error() -> None:
|
|
payload = build_unavailable_iwooos_security_asset_control_plane(
|
|
"database_query_failed"
|
|
)
|
|
|
|
assert payload["status"] == "degraded"
|
|
assert payload["source_status"] == "live_database_unavailable"
|
|
assert payload["reason_code"] == "database_query_failed"
|
|
assert payload["summary"]["managed_asset_count"] == 0
|
|
assert payload["discovery"]["fresh"] is False
|
|
assert payload["boundaries"]["raw_asset_identity_returned"] is False
|
|
assert payload["boundaries"]["raw_event_payload_returned"] is False
|
|
|
|
|
|
def test_service_redacts_database_exception_details(monkeypatch) -> None:
|
|
service = IwoooSSecurityAssetControlPlaneService()
|
|
|
|
async def fail_load():
|
|
raise RuntimeError("postgresql://operator:raw-secret@private-host/runtime")
|
|
|
|
monkeypatch.setattr(service, "_load_snapshot", fail_load)
|
|
payload = asyncio.run(service.get_snapshot())
|
|
text = json.dumps(payload)
|
|
|
|
assert payload["reason_code"] == "database_query_failed"
|
|
assert "raw-secret" not in text
|
|
assert "private-host" not in text
|
|
|
|
|
|
def test_service_uses_canonical_awoooi_project_scope(monkeypatch) -> None:
|
|
service = IwoooSSecurityAssetControlPlaneService()
|
|
requested_projects: list[str] = []
|
|
|
|
class FailingContext:
|
|
async def __aenter__(self):
|
|
raise RuntimeError("stop after project scope capture")
|
|
|
|
async def __aexit__(self, exc_type, exc, traceback):
|
|
return False
|
|
|
|
def fake_db_context(project_id: str):
|
|
requested_projects.append(project_id)
|
|
return FailingContext()
|
|
|
|
monkeypatch.setattr(
|
|
"src.services.iwooos_security_asset_control_plane.get_db_context",
|
|
fake_db_context,
|
|
)
|
|
payload = asyncio.run(service.get_snapshot())
|
|
|
|
assert requested_projects == ["awoooi"]
|
|
assert payload["status"] == "degraded"
|
|
assert payload["reason_code"] == "database_query_failed"
|
|
|
|
|
|
def test_public_api_returns_live_aggregate(monkeypatch) -> None:
|
|
payload = _ready_payload()
|
|
|
|
class FakeService:
|
|
async def get_snapshot(self):
|
|
return payload
|
|
|
|
monkeypatch.setattr(
|
|
"src.api.v1.iwooos.get_iwooos_security_asset_control_plane_service",
|
|
lambda: FakeService(),
|
|
)
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
response = TestClient(app).get("/api/v1/iwooos/security-asset-control-plane")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["schema_version"] == "iwooos_security_asset_control_plane_v1"
|
|
assert data["summary"]["managed_asset_count"] == 24
|
|
assert data["summary"]["siem_stage_coverage_percent"] == 100
|
|
assert data["ai_automation"]["same_run_closed_loop_proven"] is False
|
|
assert "192.168." not in response.text
|
|
assert "raw-secret" not in response.text
|