diff --git a/apps/api/src/jobs/asset_scanner_job.py b/apps/api/src/jobs/asset_scanner_job.py index 5e3b63fda..65b448a5d 100644 --- a/apps/api/src/jobs/asset_scanner_job.py +++ b/apps/api/src/jobs/asset_scanner_job.py @@ -29,9 +29,13 @@ from __future__ import annotations import asyncio import hashlib +import ipaddress import json as _json import re +import socket +import ssl import time as _time +from datetime import UTC, datetime from pathlib import Path from typing import Any @@ -47,6 +51,14 @@ _FIRST_DELAY_SEC = 60 # 啟動後等 60s 再首掃 (其他 service init) _KUBECTL_TIMEOUT_SEC = 30 _LOOP_BACKOFF_SEC = 300 # 異常時 backoff 5 分鐘 _PROJECT_ID = "awoooi" +_PUBLIC_TLS_PROBE_TIMEOUT_SEC = 5.0 +_PUBLIC_TLS_PROBE_CONCURRENCY = 4 +_PUBLIC_TLS_WARNING_DAYS = 30 +_PUBLIC_TLS_CRITICAL_DAYS = 7 + + +class _PublicTlsAddressPolicyError(RuntimeError): + pass _OWNER_SIGNAL_KEYS = ( "wooo.work/owner-team", @@ -1052,6 +1064,317 @@ def _normalize_public_domain(value: Any) -> str | None: return domain +def _classify_public_tls_expiry( + expires_epoch: float, + *, + now_epoch: float | None = None, +) -> tuple[str, int, str]: + current_epoch = _time.time() if now_epoch is None else now_epoch + days_remaining = int((expires_epoch - current_epoch) // 86400) + if days_remaining < 0: + return "critical", days_remaining, "tls_certificate_expired" + if days_remaining < _PUBLIC_TLS_CRITICAL_DAYS: + return "critical", days_remaining, "tls_certificate_expiry_critical" + if days_remaining < _PUBLIC_TLS_WARNING_DAYS: + return "warning", days_remaining, "tls_certificate_expiry_warning" + return "healthy", days_remaining, "tls_certificate_valid" + + +def _public_tls_probe_result( + *, + status: str, + evidenced: bool, + healthy: bool, + chain_verified: bool, + hostname_verified: bool, + reason_code: str, + expires_at: str | None = None, + days_remaining: int | None = None, +) -> dict[str, Any]: + return { + "live_tls_probe_executed": True, + "live_tls_probe_status": status, + "live_tls_probe_evidenced": evidenced, + "live_tls_healthy": healthy, + "tls_chain_verified": chain_verified, + "tls_hostname_verified": hostname_verified, + "tls_expires_at": expires_at, + "tls_days_remaining": days_remaining, + "tls_probe_reason_code": reason_code, + "certificate_material_read": False, + "private_key_material_read": False, + "secret_value_read": False, + } + + +def _validated_public_tls_addresses(addresses: list[str]) -> list[str]: + validated: list[str] = [] + seen: set[str] = set() + for raw_address in addresses: + try: + address = ipaddress.ip_address(raw_address) + except ValueError as exc: + raise _PublicTlsAddressPolicyError( + "tls_endpoint_address_invalid" + ) from exc + if not address.is_global: + raise _PublicTlsAddressPolicyError("tls_endpoint_address_not_public") + normalized = str(address) + if normalized not in seen: + seen.add(normalized) + validated.append(normalized) + if not validated: + raise _PublicTlsAddressPolicyError("tls_endpoint_address_missing") + return validated + + +async def _resolve_public_tls_addresses(domain: str) -> list[str]: + records = await asyncio.wait_for( + asyncio.get_running_loop().getaddrinfo( + domain, + 443, + type=socket.SOCK_STREAM, + proto=socket.IPPROTO_TCP, + ), + timeout=_PUBLIC_TLS_PROBE_TIMEOUT_SEC, + ) + return _validated_public_tls_addresses( + [str(record[4][0]) for record in records] + ) + + +async def _probe_public_tls_endpoint(domain: str) -> dict[str, Any]: + """Verify one public TLS endpoint without reading certificate/key files.""" + writer: asyncio.StreamWriter | None = None + try: + addresses = await _resolve_public_tls_addresses(domain) + context = ssl.create_default_context() + + async def connect() -> tuple[asyncio.StreamReader, asyncio.StreamWriter]: + last_network_error: OSError | None = None + for address in addresses: + try: + return await asyncio.open_connection( + address, + 443, + ssl=context, + server_hostname=domain, + ) + except ssl.SSLError: + raise + except OSError as exc: + last_network_error = exc + if last_network_error is not None: + raise last_network_error + raise OSError("tls_endpoint_connection_missing") + + _, writer = await asyncio.wait_for( + connect(), + timeout=_PUBLIC_TLS_PROBE_TIMEOUT_SEC, + ) + ssl_object = writer.get_extra_info("ssl_object") + if ssl_object is None: + raise ValueError("tls_socket_metadata_missing") + certificate = ssl_object.getpeercert() + if not isinstance(certificate, dict) or not certificate.get("notAfter"): + raise ValueError("tls_expiry_metadata_missing") + + expires_epoch = ssl.cert_time_to_seconds(str(certificate["notAfter"])) + status, days_remaining, reason_code = _classify_public_tls_expiry( + expires_epoch + ) + return _public_tls_probe_result( + status=status, + evidenced=True, + healthy=status == "healthy", + chain_verified=True, + hostname_verified=True, + reason_code=reason_code, + expires_at=datetime.fromtimestamp(expires_epoch, UTC) + .replace(microsecond=0) + .isoformat(), + days_remaining=days_remaining, + ) + except ssl.SSLCertVerificationError: + return _public_tls_probe_result( + status="critical", + evidenced=True, + healthy=False, + chain_verified=False, + hostname_verified=False, + reason_code="tls_certificate_verification_failed", + ) + except _PublicTlsAddressPolicyError: + return _public_tls_probe_result( + status="unavailable", + evidenced=False, + healthy=False, + chain_verified=False, + hostname_verified=False, + reason_code="tls_endpoint_address_policy_rejected", + ) + except (TimeoutError, asyncio.TimeoutError, OSError): + return _public_tls_probe_result( + status="unavailable", + evidenced=False, + healthy=False, + chain_verified=False, + hostname_verified=False, + reason_code="tls_endpoint_unavailable", + ) + except (KeyError, TypeError, ValueError): + return _public_tls_probe_result( + status="critical", + evidenced=True, + healthy=False, + chain_verified=False, + hostname_verified=False, + reason_code="tls_certificate_metadata_invalid", + ) + except Exception: + return _public_tls_probe_result( + status="unavailable", + evidenced=False, + healthy=False, + chain_verified=False, + hostname_verified=False, + reason_code="tls_probe_internal_error", + ) + finally: + if writer is not None: + writer.close() + + +async def _probe_public_tls_endpoints( + domains: list[str], +) -> dict[str, dict[str, Any]]: + semaphore = asyncio.Semaphore(_PUBLIC_TLS_PROBE_CONCURRENCY) + + async def bounded(domain: str) -> tuple[str, dict[str, Any]]: + async with semaphore: + return domain, await _probe_public_tls_endpoint(domain) + + rows = await asyncio.gather(*(bounded(domain) for domain in sorted(set(domains)))) + return dict(rows) + + +def _apply_domain_tls_probe_results( + assets: list[dict[str, Any]], + relationships: list[dict[str, str]], + probe_results: dict[str, dict[str, Any]], +) -> tuple[list[dict[str, Any]], list[dict[str, str]]]: + """Join bounded endpoint results back to domain and certificate assets.""" + assets_by_key = {str(asset["asset_key"]): asset for asset in assets} + result_by_domain_key: dict[str, dict[str, Any]] = {} + + for asset in assets: + if asset.get("asset_type") != "website" or not str( + asset.get("asset_key") or "" + ).startswith("repo/nginx-domain/"): + continue + domain = _normalize_public_domain(asset.get("host")) + if domain is None: + continue + result = probe_results.get(domain) + if result is None: + result = _public_tls_probe_result( + status="unavailable", + evidenced=False, + healthy=False, + chain_verified=False, + hostname_verified=False, + reason_code="tls_probe_result_missing", + ) + asset["metadata"] = {**(asset.get("metadata") or {}), **result} + result_by_domain_key[str(asset["asset_key"])] = result + + certificate_domains: dict[str, list[dict[str, Any]]] = {} + for relationship in relationships: + if relationship.get("relationship_type") != "authenticates_via": + continue + certificate_key = str(relationship.get("to_key") or "") + domain_result = result_by_domain_key.get( + str(relationship.get("from_key") or "") + ) + if domain_result is not None and certificate_key in assets_by_key: + certificate_domains.setdefault(certificate_key, []).append(domain_result) + + for certificate_key, endpoint_results in certificate_domains.items(): + certificate = assets_by_key[certificate_key] + endpoint_count = len(endpoint_results) + evidenced_count = sum( + result.get("live_tls_probe_evidenced") is True + for result in endpoint_results + ) + healthy_count = sum( + result.get("live_tls_healthy") is True for result in endpoint_results + ) + warning_count = sum( + result.get("live_tls_probe_status") == "warning" + for result in endpoint_results + ) + critical_count = sum( + result.get("live_tls_probe_status") == "critical" + for result in endpoint_results + ) + unavailable_count = endpoint_count - evidenced_count + days_remaining = [ + int(result["tls_days_remaining"]) + for result in endpoint_results + if isinstance(result.get("tls_days_remaining"), int) + ] + all_evidenced = endpoint_count > 0 and evidenced_count == endpoint_count + all_healthy = all_evidenced and healthy_count == endpoint_count + + if critical_count > 0: + aggregate_status = "critical" + reason_code = "tls_certificate_endpoint_critical" + elif warning_count > 0: + aggregate_status = "warning" + reason_code = "tls_certificate_endpoint_warning" + elif unavailable_count > 0: + aggregate_status = "unavailable" + reason_code = "tls_certificate_endpoint_unavailable" + elif all_healthy: + aggregate_status = "healthy" + reason_code = "tls_certificate_endpoints_healthy" + else: + aggregate_status = "critical" + reason_code = "tls_certificate_endpoint_state_invalid" + + certificate["metadata"] = { + **(certificate.get("metadata") or {}), + "live_tls_probe_executed": True, + "live_tls_probe_status": aggregate_status, + "live_tls_probe_evidenced": all_evidenced, + "live_tls_healthy": all_healthy, + "tls_chain_verified": all( + result.get("tls_chain_verified") is True + for result in endpoint_results + ), + "tls_hostname_verified": all( + result.get("tls_hostname_verified") is True + for result in endpoint_results + ), + "tls_endpoint_count": endpoint_count, + "tls_evidenced_endpoint_count": evidenced_count, + "tls_healthy_endpoint_count": healthy_count, + "tls_warning_endpoint_count": warning_count, + "tls_critical_endpoint_count": critical_count, + "tls_unavailable_endpoint_count": unavailable_count, + "tls_min_days_remaining": ( + min(days_remaining) if days_remaining else None + ), + "tls_probe_reason_code": reason_code, + "certificate_material_read": False, + "private_key_material_read": False, + "raw_certificate_path_stored": False, + "secret_value_read": False, + } + + return assets, relationships + + def _build_domain_tls_inventory_assets( payload: dict[str, Any], ) -> tuple[list[dict[str, Any]], list[dict[str, str]]]: @@ -1204,7 +1527,14 @@ async def _collect_domain_tls_inventory_assets() -> ( tuple[list[dict[str, Any]], list[dict[str, str]]] ): payload = await asyncio.to_thread(_load_domain_tls_inventory_payload) - return _build_domain_tls_inventory_assets(payload) + assets, relationships = _build_domain_tls_inventory_assets(payload) + domains = [ + str(asset["host"]) + for asset in assets + if asset.get("asset_type") == "website" and asset.get("host") + ] + probe_results = await _probe_public_tls_endpoints(domains) + return _apply_domain_tls_probe_results(assets, relationships, probe_results) async def _collect_runtime_assets() -> ( 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 297bd96de..863efab7a 100644 --- a/apps/api/src/services/iwooos_security_asset_control_plane.py +++ b/apps/api/src/services/iwooos_security_asset_control_plane.py @@ -370,6 +370,8 @@ def _build_security_program_domains( package_digest_unpinned_count: int, certificate_live_verified_percent: int, certificate_live_unverified_count: int, + certificate_live_healthy_percent: int, + certificate_live_unhealthy_count: int, ) -> list[dict[str, Any]]: scope_by_id = {row["scope_id"]: row for row in asset_scopes} function_by_id = {row["function_id"]: row for row in control_functions} @@ -399,11 +401,12 @@ def _build_security_program_domains( supply_chain_gap_code = "runtime_image_digest_pin_missing" else: supply_chain_gap_code = "sbom_signature_dependency_evidence_missing" - network_tls_gap_code = ( - "certificate_live_tls_probe_missing" - if certificate_live_unverified_count > 0 - else "network_dns_tls_runtime_evidence_missing" - ) + if certificate_live_unverified_count > 0: + network_tls_gap_code = "certificate_live_tls_probe_missing" + elif certificate_live_unhealthy_count > 0: + network_tls_gap_code = "certificate_live_tls_unhealthy" + else: + network_tls_gap_code = "network_dns_tls_runtime_evidence_missing" domain_values = ( ( @@ -461,6 +464,7 @@ def _build_security_program_domains( 100 if by_type.get("network", 0) > 0 else 0, 100 if by_type.get("certificate", 0) > 0 else 0, certificate_live_verified_percent, + certificate_live_healthy_percent, compliance_percent("ssl_cert_valid"), ), by_type.get("network", 0) @@ -687,6 +691,10 @@ def build_iwooos_security_asset_control_plane( int(_value(row, "unverified_certificate_count", 0) or 0) for row in inventory_source ) + certificate_live_unhealthy_count = sum( + int(_value(row, "unhealthy_certificate_count", 0) or 0) + for row in inventory_source + ) present_types = { asset_type for asset_type in _ASSET_TYPES if by_type.get(asset_type, 0) > 0 } @@ -907,6 +915,16 @@ def build_iwooos_security_asset_control_plane( by_type.get("certificate", 0), ), certificate_live_unverified_count=certificate_live_unverified_count, + certificate_live_healthy_percent=_percent( + max( + 0, + by_type.get("certificate", 0) + - certificate_live_unverified_count + - certificate_live_unhealthy_count, + ), + by_type.get("certificate", 0), + ), + certificate_live_unhealthy_count=certificate_live_unhealthy_count, ) work_items: list[dict[str, Any]] = [] @@ -962,17 +980,21 @@ def build_iwooos_security_asset_control_plane( ), ) ) - if certificate_live_unverified_count > 0: + certificate_live_gap_count = ( + certificate_live_unverified_count + certificate_live_unhealthy_count + ) + if certificate_live_gap_count > 0: work_items.append( _work_item( "AIA-P0-006-02B", "P0", "certificate_live_verification", - "驗證 public TLS 憑證", - certificate_live_unverified_count, + "驗證並修復 public TLS 憑證", + certificate_live_gap_count, ( "執行 bounded TLS handshake metadata probe，驗證 hostname、" - "chain 與 expiry；不得讀取 private key 或 certificate volume。" + "chain 與 expiry；對 warning/critical 建立受控修復，" + "不得讀取 private key 或 certificate volume。" ), ) ) @@ -1132,6 +1154,7 @@ def build_iwooos_security_asset_control_plane( "certificate_live_unverified_count": ( certificate_live_unverified_count ), + "certificate_live_unhealthy_count": certificate_live_unhealthy_count, "automation_coverage_green_percent": _percent( coverage_green, coverage_total ), @@ -1221,10 +1244,23 @@ def build_iwooos_security_asset_control_plane( by_type.get("certificate", 0) - certificate_live_unverified_count, ), + "live_tls_evidenced_count": max( + 0, + by_type.get("certificate", 0) + - certificate_live_unverified_count, + ), + "live_tls_healthy_count": max( + 0, + by_type.get("certificate", 0) + - certificate_live_unverified_count + - certificate_live_unhealthy_count, + ), "live_tls_unverified_count": certificate_live_unverified_count, + "live_tls_unhealthy_count": certificate_live_unhealthy_count, "live_tls_health_proven": ( by_type.get("certificate", 0) > 0 and certificate_live_unverified_count == 0 + and certificate_live_unhealthy_count == 0 ), "raw_certificate_path_returned": False, "certificate_material_read": False, @@ -1272,6 +1308,7 @@ def build_unavailable_iwooos_security_asset_control_plane( "package_digest_unpinned_count": 0, "forbidden_github_supply_chain_asset_count": 0, "certificate_live_unverified_count": 0, + "certificate_live_unhealthy_count": 0, "automation_coverage_green_percent": 0, "automation_coverage_unknown_count": 0, "compliance_violation_count": 0, @@ -1438,7 +1475,10 @@ def build_unavailable_iwooos_security_asset_control_plane( "certificate_inventory": { "certificate_asset_count": 0, "live_tls_verified_count": 0, + "live_tls_evidenced_count": 0, + "live_tls_healthy_count": 0, "live_tls_unverified_count": 0, + "live_tls_unhealthy_count": 0, "live_tls_health_proven": False, "raw_certificate_path_returned": False, "certificate_material_read": False, @@ -1672,7 +1712,18 @@ class IwoooSSecurityAssetControlPlaneService: metadata->>'live_tls_probe_evidenced', 'false' ) <> 'true' - ) AS unverified_certificate_count + ) AS unverified_certificate_count, + count(*) FILTER ( + WHERE asset_type = 'certificate' + AND COALESCE( + metadata->>'live_tls_probe_evidenced', + 'false' + ) = 'true' + AND COALESCE( + metadata->>'live_tls_healthy', + 'false' + ) <> 'true' + ) AS unhealthy_certificate_count FROM asset_inventory WHERE lifecycle_state = 'active' GROUP BY asset_type diff --git a/apps/api/tests/test_asset_scanner_job.py b/apps/api/tests/test_asset_scanner_job.py index 2aa5ffa7a..ab5c03554 100644 --- a/apps/api/tests/test_asset_scanner_job.py +++ b/apps/api/tests/test_asset_scanner_job.py @@ -596,7 +596,9 @@ def test_domain_tls_inventory_projects_fingerprints_without_raw_paths() -> None: assert len(assets) == 3 assert {asset["asset_type"] for asset in assets} == {"website", "certificate"} - certificate = next(asset for asset in assets if asset["asset_type"] == "certificate") + certificate = next( + asset for asset in assets if asset["asset_type"] == "certificate" + ) assert certificate["metadata"]["raw_certificate_path_stored"] is False assert certificate["metadata"]["certificate_material_read"] is False assert certificate["metadata"]["private_key_material_read"] is False @@ -607,6 +609,170 @@ def test_domain_tls_inventory_projects_fingerprints_without_raw_paths() -> None: assert "privkey.pem" not in public_text +def test_public_tls_expiry_classification_is_conservative() -> None: + now_epoch = 1_700_000_000.0 + + assert asset_scanner_job._classify_public_tls_expiry( + now_epoch + (45 * 86400), now_epoch=now_epoch + ) == ("healthy", 45, "tls_certificate_valid") + assert asset_scanner_job._classify_public_tls_expiry( + now_epoch + (20 * 86400), now_epoch=now_epoch + ) == ("warning", 20, "tls_certificate_expiry_warning") + assert asset_scanner_job._classify_public_tls_expiry( + now_epoch + (3 * 86400), now_epoch=now_epoch + ) == ("critical", 3, "tls_certificate_expiry_critical") + assert asset_scanner_job._classify_public_tls_expiry( + now_epoch - 1, now_epoch=now_epoch + ) == ("critical", -1, "tls_certificate_expired") + + +def test_public_tls_address_policy_allows_only_global_addresses() -> None: + assert asset_scanner_job._validated_public_tls_addresses( + ["1.1.1.1", "2606:4700:4700::1111", "1.1.1.1"] + ) == ["1.1.1.1", "2606:4700:4700::1111"] + + for rejected in ( + "127.0.0.1", + "10.0.0.1", + "169.254.169.254", + "192.168.0.110", + "::1", + "fc00::1", + "not-an-address", + ): + try: + asset_scanner_job._validated_public_tls_addresses([rejected]) + except asset_scanner_job._PublicTlsAddressPolicyError: + pass + else: + raise AssertionError(f"non-public TLS address accepted: {rejected}") + + +def test_domain_tls_probe_aggregates_shared_certificate_without_false_green() -> ( + None +): + payload = { + "schema_version": "domain_tls_certbot_inventory_v1", + "generated_at": "2026-07-14T00:00:00+08:00", + "mode": "repo_only_from_nginx_source_of_truth", + "execution_boundaries": { + "secret_value_collected": False, + "live_tls_probe_executed": False, + "certbot_renew_executed": False, + "nginx_reload_executed": False, + "host_write_executed": False, + }, + "summary": { + "managed_domain_count": 2, + "unique_certificate_path_count": 1, + }, + "managed_domains": [ + { + "domain": domain, + "certificate_paths": ["/redacted/shared/fullchain.pem"], + } + for domain in ("app.example.test", "api.example.test") + ], + } + assets, relationships = asset_scanner_job._build_domain_tls_inventory_assets( + payload + ) + probe_results = { + "app.example.test": asset_scanner_job._public_tls_probe_result( + status="healthy", + evidenced=True, + healthy=True, + chain_verified=True, + hostname_verified=True, + reason_code="tls_certificate_valid", + expires_at="2026-09-01T00:00:00+00:00", + days_remaining=49, + ), + "api.example.test": asset_scanner_job._public_tls_probe_result( + status="warning", + evidenced=True, + healthy=False, + chain_verified=True, + hostname_verified=True, + reason_code="tls_certificate_expiry_warning", + expires_at="2026-08-01T00:00:00+00:00", + days_remaining=18, + ), + } + + assets, _ = asset_scanner_job._apply_domain_tls_probe_results( + assets, relationships, probe_results + ) + certificate = next( + asset for asset in assets if asset["asset_type"] == "certificate" + ) + metadata = certificate["metadata"] + + assert metadata["live_tls_probe_evidenced"] is True + assert metadata["live_tls_healthy"] is False + assert metadata["live_tls_probe_status"] == "warning" + assert metadata["tls_endpoint_count"] == 2 + assert metadata["tls_evidenced_endpoint_count"] == 2 + assert metadata["tls_healthy_endpoint_count"] == 1 + assert metadata["tls_warning_endpoint_count"] == 1 + assert metadata["tls_unavailable_endpoint_count"] == 0 + assert metadata["tls_min_days_remaining"] == 18 + assert metadata["certificate_material_read"] is False + assert metadata["private_key_material_read"] is False + + +def test_domain_tls_probe_keeps_shared_certificate_unverified_when_one_is_missing() -> ( + None +): + payload = { + "schema_version": "domain_tls_certbot_inventory_v1", + "generated_at": "2026-07-14T00:00:00+08:00", + "mode": "repo_only_from_nginx_source_of_truth", + "execution_boundaries": { + "secret_value_collected": False, + "live_tls_probe_executed": False, + "certbot_renew_executed": False, + "nginx_reload_executed": False, + "host_write_executed": False, + }, + "summary": { + "managed_domain_count": 2, + "unique_certificate_path_count": 1, + }, + "managed_domains": [ + { + "domain": domain, + "certificate_paths": ["/redacted/shared/fullchain.pem"], + } + for domain in ("app.example.test", "api.example.test") + ], + } + assets, relationships = asset_scanner_job._build_domain_tls_inventory_assets( + payload + ) + assets, _ = asset_scanner_job._apply_domain_tls_probe_results( + assets, + relationships, + { + "app.example.test": asset_scanner_job._public_tls_probe_result( + status="healthy", + evidenced=True, + healthy=True, + chain_verified=True, + hostname_verified=True, + reason_code="tls_certificate_valid", + days_remaining=49, + ) + }, + ) + certificate = next(asset for asset in assets if asset["asset_type"] == "certificate") + + assert certificate["metadata"]["live_tls_probe_evidenced"] is False + assert certificate["metadata"]["live_tls_healthy"] is False + assert certificate["metadata"]["tls_evidenced_endpoint_count"] == 1 + assert certificate["metadata"]["tls_unavailable_endpoint_count"] == 1 + + def test_domain_tls_inventory_fails_closed_when_secret_boundary_is_not_false() -> ( None ): 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 3eaf6482b..347a4a652 100644 --- a/apps/api/tests/test_iwooos_security_asset_control_plane.py +++ b/apps/api/tests/test_iwooos_security_asset_control_plane.py @@ -330,7 +330,10 @@ def test_certificate_assets_remain_open_until_live_tls_probe_is_evidenced() -> N assert payload["certificate_inventory"] == { "certificate_asset_count": 10, "live_tls_verified_count": 0, + "live_tls_evidenced_count": 0, + "live_tls_healthy_count": 0, "live_tls_unverified_count": 10, + "live_tls_unhealthy_count": 0, "live_tls_health_proven": False, "raw_certificate_path_returned": False, "certificate_material_read": False, @@ -351,6 +354,95 @@ def test_certificate_assets_remain_open_until_live_tls_probe_is_evidenced() -> N assert domains["network_dns_tls"]["status"] != "controlled" +def test_unhealthy_live_tls_evidence_remains_a_p0_gap() -> 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="full", + latest_success_ended_at=now - timedelta(minutes=1), + latest_total_assets=3, + ), + inventory_rows=[ + _row( + asset_type="certificate", + cnt=3, + unowned_count=3, + stale_count=0, + unverified_certificate_count=0, + unhealthy_certificate_count=1, + ), + ], + coverage_rows=[], + relationship_row=_row(relationship_count=0, orphan_asset_count=3), + compliance_rows=[], + 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"]["certificate_live_unverified_count"] == 0 + assert payload["summary"]["certificate_live_unhealthy_count"] == 1 + assert payload["certificate_inventory"]["live_tls_evidenced_count"] == 3 + assert payload["certificate_inventory"]["live_tls_healthy_count"] == 2 + assert payload["certificate_inventory"]["live_tls_unhealthy_count"] == 1 + assert payload["certificate_inventory"]["live_tls_health_proven"] is False + work_item = next( + item + for item in payload["work_items"] + if item["work_item_id"] == "AIA-P0-006-02B" + ) + assert work_item["gap_count"] == 1 + domains = { + domain["domain_id"]: domain for domain in payload["security_program_domains"] + } + assert domains["network_dns_tls"]["gap_code"] == ( + "certificate_live_tls_unhealthy" + ) + + +def test_fully_evidenced_healthy_tls_inventory_proves_health() -> 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="full", + latest_success_ended_at=now - timedelta(minutes=1), + latest_total_assets=10, + ), + inventory_rows=[ + _row( + asset_type="certificate", + cnt=10, + unowned_count=10, + stale_count=0, + unverified_certificate_count=0, + unhealthy_certificate_count=0, + ), + ], + coverage_rows=[], + relationship_row=_row(relationship_count=10, orphan_asset_count=0), + compliance_rows=[], + automation_rows=[], + ai_trace_row=_row(trace_count=0, accepted_trace_count=0), + siem_row=_row(), + audit_row=_row(), + generated_at=now, + ) + + assert payload["certificate_inventory"]["live_tls_evidenced_count"] == 10 + assert payload["certificate_inventory"]["live_tls_healthy_count"] == 10 + assert payload["certificate_inventory"]["live_tls_unverified_count"] == 0 + assert payload["certificate_inventory"]["live_tls_unhealthy_count"] == 0 + assert payload["certificate_inventory"]["live_tls_health_proven"] is True + assert all( + item["work_item_id"] != "AIA-P0-006-02B" + for item in payload["work_items"] + ) + + def test_partial_source_failure_is_visible_without_erasing_core_inventory() -> None: source_health = [ {"source_id": "asset_discovery", "status": "ready", "reason_code": None}, @@ -547,6 +639,7 @@ def test_service_bounds_source_concurrency_for_database_budget(monkeypatch) -> N assert "forbidden_source_domain_detected" in inventory_query assert "digest_pinned" in inventory_query assert "live_tls_probe_evidenced" in inventory_query + assert "live_tls_healthy" in inventory_query def test_public_api_returns_live_aggregate(monkeypatch) -> None: diff --git a/apps/web/src/components/iwooos/security-asset-control-plane-cockpit.tsx b/apps/web/src/components/iwooos/security-asset-control-plane-cockpit.tsx index 1c94b1432..d75d99fb4 100644 --- a/apps/web/src/components/iwooos/security-asset-control-plane-cockpit.tsx +++ b/apps/web/src/components/iwooos/security-asset-control-plane-cockpit.tsx @@ -6,6 +6,7 @@ import { Boxes, CheckCircle2, Database, + LockKeyhole, RefreshCw, Radar, ShieldAlert, @@ -48,6 +49,11 @@ type Copy = { digestPinned: string; forbiddenSources: string; imagePolicy: string; + publicTls: string; + certificates: string; + tlsEvidence: string; + tlsHealthy: string; + tlsPolicy: string; compliant: string; repair: string; notEvidenced: string; @@ -88,6 +94,11 @@ const COPY: Record<"zh" | "en", Copy> = { digestPinned: "Digest 鎖定", forbiddenSources: "禁用來源", imagePolicy: "映像政策", + publicTls: "Public TLS", + certificates: "憑證資產", + tlsEvidence: "Live 驗證", + tlsHealthy: "健康憑證", + tlsPolicy: "TLS 政策", compliant: "符合", repair: "需修復", notEvidenced: "未盤點", @@ -126,6 +137,11 @@ const COPY: Record<"zh" | "en", Copy> = { digestPinned: "Digest pinned", forbiddenSources: "Forbidden sources", imagePolicy: "Image policy", + publicTls: "Public TLS", + certificates: "Certificates", + tlsEvidence: "Live evidence", + tlsHealthy: "Healthy certs", + tlsPolicy: "TLS policy", compliant: "PASS", repair: "REPAIR", notEvidenced: "NO DATA", @@ -247,6 +263,69 @@ function TruthCell({ ); } +function CertificateTruthStrip({ + copy, + payload, +}: { + copy: Copy; + payload: SecurityAssetControlPlanePayload | null; +}) { + const certificate = payload?.certificate_inventory; + const assetCount = certificate?.certificate_asset_count ?? 0; + const evidencedCount = + certificate?.live_tls_evidenced_count ?? certificate?.live_tls_verified_count; + const healthyCount = certificate?.live_tls_healthy_count; + const healthProven = certificate?.live_tls_health_proven === true; + + return ( +
+
+
+
+ 0 ? "healthy" : "critical"} + /> + + + +
+
+ ); +} + function StageStrip({ title, stages, @@ -358,6 +437,8 @@ export function SecurityAssetControlPlaneCockpit({ + +
{ }, }; expect(isSecurityAssetControlPlanePayload(unsafeSupplyChain)).toBe(false); + + const unsafeCertificateInventory = { + ...safe, + certificate_inventory: { + ...safe.certificate_inventory, + private_key_material_read: true, + }, + }; + expect(isSecurityAssetControlPlanePayload(unsafeCertificateInventory)).toBe(false); }); it("does not show healthy before same-run closure is proven", () => { @@ -194,6 +217,20 @@ describe("security asset control plane projection", () => { expect(securityControlPlaneTone(value)).toBe("critical"); }); + it("keeps missing or unhealthy live TLS evidence critical", () => { + const value = payload(); + value.ai_automation.same_run_closed_loop_proven = true; + expect(securityControlPlaneTone(value)).toBe("healthy"); + + value.certificate_inventory!.live_tls_healthy_count = 9; + value.certificate_inventory!.live_tls_unhealthy_count = 1; + value.certificate_inventory!.live_tls_health_proven = false; + expect(securityControlPlaneTone(value)).toBe("critical"); + + value.certificate_inventory = undefined; + expect(securityControlPlaneTone(value)).toBe("critical"); + }); + it("keeps P0 work ahead of P1 regardless of API order", () => { const items = topSecurityControlPlaneWorkItems(payload(), 2); expect(items.map((item) => item.work_item_id)).toEqual([ diff --git a/apps/web/src/lib/security-asset-control-plane.ts b/apps/web/src/lib/security-asset-control-plane.ts index 3630688e7..e7860c478 100644 --- a/apps/web/src/lib/security-asset-control-plane.ts +++ b/apps/web/src/lib/security-asset-control-plane.ts @@ -16,6 +16,8 @@ export type SecurityControlPlaneSummary = { relationship_count: number; package_digest_unpinned_count?: number; forbidden_github_supply_chain_asset_count?: number; + certificate_live_unverified_count?: number; + certificate_live_unhealthy_count?: number; automation_coverage_green_percent: number; automation_coverage_unknown_count: number; compliance_violation_count: number; @@ -84,6 +86,19 @@ export type SecurityControlPlaneSupplyChain = { github_api_used: false; }; +export type SecurityControlPlaneCertificateInventory = { + certificate_asset_count: number; + live_tls_verified_count: number; + live_tls_evidenced_count?: number; + live_tls_healthy_count?: number; + live_tls_unverified_count: number; + live_tls_unhealthy_count?: number; + live_tls_health_proven: boolean; + raw_certificate_path_returned: false; + certificate_material_read: false; + private_key_material_read: false; +}; + export type SecurityAssetControlPlanePayload = { schema_version: "iwooos_security_asset_control_plane_v1"; generated_at: string; @@ -156,6 +171,7 @@ export type SecurityAssetControlPlanePayload = { immutable_external_audit_store_evidenced: boolean; }; supply_chain?: SecurityControlPlaneSupplyChain; + certificate_inventory?: SecurityControlPlaneCertificateInventory; source_health: Array<{ source_id: string; status: "ready" | "unavailable"; @@ -212,6 +228,26 @@ export function isSecurityAssetControlPlanePayload( return false; } } + if (value.certificate_inventory !== undefined) { + if (!isRecord(value.certificate_inventory)) return false; + if ( + typeof value.certificate_inventory.certificate_asset_count !== "number" || + typeof value.certificate_inventory.live_tls_verified_count !== "number" || + (value.certificate_inventory.live_tls_evidenced_count !== undefined && + typeof value.certificate_inventory.live_tls_evidenced_count !== "number") || + (value.certificate_inventory.live_tls_healthy_count !== undefined && + typeof value.certificate_inventory.live_tls_healthy_count !== "number") || + typeof value.certificate_inventory.live_tls_unverified_count !== "number" || + (value.certificate_inventory.live_tls_unhealthy_count !== undefined && + typeof value.certificate_inventory.live_tls_unhealthy_count !== "number") || + typeof value.certificate_inventory.live_tls_health_proven !== "boolean" || + value.certificate_inventory.raw_certificate_path_returned !== false || + value.certificate_inventory.certificate_material_read !== false || + value.certificate_inventory.private_key_material_read !== false + ) { + return false; + } + } return ( typeof value.summary.managed_asset_count === "number" && typeof value.summary.source_count === "number" && @@ -245,6 +281,15 @@ export function securityControlPlaneTone( ) { return "critical"; } + if ( + !payload.certificate_inventory || + payload.certificate_inventory.certificate_asset_count === 0 || + payload.certificate_inventory.live_tls_unverified_count > 0 || + (payload.certificate_inventory.live_tls_unhealthy_count ?? 0) > 0 || + !payload.certificate_inventory.live_tls_health_proven + ) { + return "critical"; + } if ( payload.summary.compliance_violation_count > 0 || payload.summary.nist_controlled_percent < 50 ||