From d4bc3cb4535a20435881f2b599140fc2c1d8558e Mon Sep 17 00:00:00 2001
From: ogt <ogt@awoooi.local>
Date: Tue, 14 Jul 2026 14:24:46 +0800
Subject: [PATCH] feat(security): project runtime service asset types

---
 apps/api/src/jobs/asset_scanner_job.py   | 279 +++++++++++++++++++++++
 apps/api/tests/test_asset_scanner_job.py | 202 ++++++++++++++++
 2 files changed, 481 insertions(+)

diff --git a/apps/api/src/jobs/asset_scanner_job.py b/apps/api/src/jobs/asset_scanner_job.py
index b0eb1626c..3441b04a4 100644
--- a/apps/api/src/jobs/asset_scanner_job.py
+++ b/apps/api/src/jobs/asset_scanner_job.py
@@ -416,12 +416,113 @@ def _build_workload_asset(item: dict[str, Any], kind: str) -> dict[str, Any]:
     }
 
 
+def _build_workload_component_assets(
+    item: dict[str, Any],
+    kind: str,
+) -> tuple[list[dict[str, Any]], list[dict[str, str]]]:
+    """Project workload labels into product and observability component assets."""
+    meta = item.get("metadata", {}) or {}
+    labels = meta.get("labels", {}) or {}
+    namespace = meta.get("namespace") or "default"
+    name = meta.get("name") or "unknown"
+    signals = " ".join(
+        str(value)
+        for value in (
+            name,
+            labels.get("app"),
+            labels.get("component"),
+            labels.get("system"),
+            labels.get("app.kubernetes.io/name"),
+            labels.get("app.kubernetes.io/component"),
+        )
+        if value
+    ).lower()
+    tokens = {token for token in re.split(r"[^a-z0-9]+", signals) if token}
+
+    component_types: list[str] = []
+    if tokens & {"web", "frontend", "ui", "portal"}:
+        component_types.append("frontend")
+    if tokens & {"api", "backend", "server", "worker", "processor", "broker"}:
+        component_types.append("backend")
+    if tokens & {"grafana", "dashboard", "kibana"}:
+        component_types.append("dashboard")
+    if tokens & {
+        "log",
+        "logs",
+        "logging",
+        "promtail",
+        "fluentbit",
+        "fluent",
+        "vector",
+        "otel",
+        "opentelemetry",
+    }:
+        component_types.append("log_stream")
+
+    owner = _normalize_owner_team(labels.get("system")) or _owner_from_namespace(
+        namespace
+    )
+    workload_key = f"k8s/{kind.lower()}/{namespace}/{name}"
+    assets: list[dict[str, Any]] = []
+    relationships: list[dict[str, str]] = []
+    for component_type in component_types:
+        component_owner = owner
+        if component_owner is None and component_type in {"dashboard", "log_stream"}:
+            component_owner = "platform_security"
+        component_key = (
+            f"k8s/component/{component_type}/{kind.lower()}/{namespace}/{name}"
+        )
+        tags = [
+            "source:k8s_workload_labels",
+            f"component_type:{component_type}",
+        ]
+        if component_owner:
+            tags.append(f"owner:{component_owner}")
+        assets.append(
+            {
+                "asset_key": component_key,
+                "asset_type": component_type,
+                "host": None,
+                "namespace": namespace,
+                "name": name,
+                "metadata": {
+                    "kind": f"{kind}ComponentProjection",
+                    "workload_asset_key": workload_key,
+                    "label_signals": {
+                        key: str(labels[key])
+                        for key in (
+                            "app",
+                            "component",
+                            "system",
+                            "environment",
+                            "app.kubernetes.io/name",
+                            "app.kubernetes.io/component",
+                        )
+                        if labels.get(key)
+                    },
+                    "container_command_or_env_read": False,
+                    "secret_value_read": False,
+                },
+                "tags": tags,
+            }
+        )
+        relationships.append(
+            {
+                "from_key": component_key,
+                "to_key": workload_key,
+                "relationship_type": "depends_on",
+            }
+        )
+    return assets, relationships
+
+
 def _build_service_asset(item: dict[str, Any]) -> dict[str, Any]:
     """Service → asset_inventory row (asset_type='k8s_resource')."""
     meta = item.get("metadata", {}) or {}
     ns = meta.get("namespace") or "default"
     name = meta.get("name") or "unknown"
     spec = item.get("spec", {}) or {}
+    labels = meta.get("labels", {}) or {}
 
     return {
         "asset_key": f"k8s/service/{ns}/{name}",
@@ -435,11 +536,57 @@ def _build_service_asset(item: dict[str, Any]) -> dict[str, Any]:
             "cluster_ip": spec.get("clusterIP"),
             "selector": spec.get("selector", {}),
             "ports": spec.get("ports", []),
+            "labels": labels,
         },
         "tags": [f"svc_type:{spec.get('type', '')}"],
     }
 
 
+def _build_service_network_asset(item: dict[str, Any]) -> dict[str, Any]:
+    """Build the logical network identity exposed by one Kubernetes Service."""
+    meta = item.get("metadata", {}) or {}
+    spec = item.get("spec", {}) or {}
+    namespace = meta.get("namespace") or "default"
+    name = meta.get("name") or "unknown"
+    labels = meta.get("labels", {}) or {}
+    owner = _normalize_owner_team(labels.get("system")) or _owner_from_namespace(
+        namespace
+    )
+    tags = ["source:k8s_service", "network_kind:service"]
+    if owner:
+        tags.append(f"owner:{owner}")
+
+    return {
+        "asset_key": f"k8s/network/service/{namespace}/{name}",
+        "asset_type": "network",
+        "host": None,
+        "namespace": namespace,
+        "name": name,
+        "metadata": {
+            "kind": "KubernetesServiceNetwork",
+            "service_type": spec.get("type"),
+            "cluster_ip": spec.get("clusterIP"),
+            "cluster_ips": spec.get("clusterIPs", []),
+            "ip_families": spec.get("ipFamilies", []),
+            "external_name": spec.get("externalName"),
+            "ports": [
+                {
+                    "name": port.get("name"),
+                    "protocol": port.get("protocol"),
+                    "port": port.get("port"),
+                    "target_port": port.get("targetPort"),
+                    "node_port": port.get("nodePort"),
+                }
+                for port in spec.get("ports", []) or []
+            ],
+            "labels": labels,
+            "packet_payload_read": False,
+            "secret_value_read": False,
+        },
+        "tags": tags,
+    }
+
+
 def _build_node_asset(item: dict[str, Any]) -> dict[str, Any]:
     """Node → asset_inventory row (asset_type='host')."""
     meta = item.get("metadata", {}) or {}
@@ -572,6 +719,98 @@ def _build_secret_reference_asset(
     }
 
 
+def _is_service_registry_configmap(item: dict[str, Any]) -> bool:
+    metadata = item.get("metadata", {}) or {}
+    return (
+        metadata.get("namespace") == "awoooi-prod"
+        and metadata.get("name") == "service-registry"
+    )
+
+
+def _collect_service_registry_assets(
+    item: dict[str, Any],
+) -> tuple[list[dict[str, Any]], list[dict[str, str]]]:
+    """Read allowlisted service identity fields from the runtime registry."""
+    import yaml
+
+    if not _is_service_registry_configmap(item):
+        return [], []
+    raw_registry = str(
+        ((item.get("data") or {}).get("service-registry.yaml") or "")
+    )
+    if not raw_registry or len(raw_registry) > 1_000_000:
+        raise RuntimeError("service_registry_config_invalid")
+    payload = yaml.safe_load(raw_registry)
+    services = payload.get("services") if isinstance(payload, dict) else None
+    if not isinstance(services, list):
+        raise RuntimeError("service_registry_services_invalid")
+
+    assets: list[dict[str, Any]] = []
+    relationships: list[dict[str, str]] = []
+    for row in services:
+        if not isinstance(row, dict):
+            continue
+        name = _normalize_owner_team(row.get("name"))
+        if not name:
+            continue
+        host = str(row.get("host") or "").strip()
+        display_name = str(row.get("display_name") or "")
+        reason = str(row.get("reason") or "")
+        classification_text = f"{name} {display_name} {reason}".lower()
+        asset_types: list[str] = []
+        if name == "grafana" or "dashboard" in classification_text:
+            asset_types.append("dashboard")
+        if (
+            "redis" in classification_text
+            or "cache" in classification_text
+            or "快取" in classification_text
+        ):
+            asset_types.append("cache")
+        if any(
+            marker in classification_text
+            for marker in ("message queue", "task queue", "queue", "佇列", "stream")
+        ):
+            asset_types.append("message_queue")
+
+        owner = "awoooi" if name.startswith("awoooi_") else "platform_security"
+        for asset_type in asset_types:
+            asset_key = f"service-registry/{asset_type}/{name}"
+            assets.append(
+                {
+                    "asset_key": asset_key,
+                    "asset_type": asset_type,
+                    "host": host if _is_valid_ipv4(host) else None,
+                    "namespace": "awoooi-prod",
+                    "name": name,
+                    "metadata": {
+                        "kind": "ServiceRegistryProjection",
+                        "display_name": display_name,
+                        "stateful_level": str(row.get("stateful_level") or "unknown"),
+                        "alert_only": row.get("alert_only") is True,
+                        "requires_pre_backup": row.get("requires_pre_backup") is True,
+                        "container_count": len(row.get("containers") or []),
+                        "source_configmap": "awoooi-prod/service-registry",
+                        "allowlisted_fields_only": True,
+                        "raw_config_persisted": False,
+                        "secret_value_read": False,
+                    },
+                    "tags": [
+                        "source:k8s_service_registry",
+                        f"owner:{owner}",
+                    ],
+                }
+            )
+            if _is_valid_ipv4(host):
+                relationships.append(
+                    {
+                        "from_key": asset_key,
+                        "to_key": f"host/{host}",
+                        "relationship_type": "depends_on",
+                    }
+                )
+    return assets, relationships
+
+
 def _collect_ingress_assets(
     payload: dict[str, Any],
 ) -> tuple[list[dict[str, Any]], list[dict[str, str]]]:
@@ -838,7 +1077,21 @@ async def _collect_runtime_assets() -> (
             payload = await _fetch_kubectl_json(resource)
             for item in payload.get("items", []) or []:
                 assets.append(_build_workload_asset(item, kind))
+                component_assets, component_relationships = (
+                    _build_workload_component_assets(item, kind)
+                )
+                assets.extend(component_assets)
+                relationships.extend(component_relationships)
             reconciled_prefixes.add(f"k8s/{kind.lower()}/")
+            for component_type in (
+                "frontend",
+                "backend",
+                "dashboard",
+                "log_stream",
+            ):
+                reconciled_prefixes.add(
+                    f"k8s/component/{component_type}/{kind.lower()}/"
+                )
         except Exception as e:
             logger.warning(f"collect_{resource}_failed", error=str(e))
             collector_errors.append(f"k8s_{resource}")
@@ -848,7 +1101,16 @@ async def _collect_runtime_assets() -> (
         payload = await _fetch_kubectl_json("services")
         for item in payload.get("items", []) or []:
             svc = _build_service_asset(item)
+            network = _build_service_network_asset(item)
             assets.append(svc)
+            assets.append(network)
+            relationships.append(
+                {
+                    "from_key": svc["asset_key"],
+                    "to_key": network["asset_key"],
+                    "relationship_type": "depends_on",
+                }
+            )
 
             # 為該 Service 找出匹配的 Pod
             selector = (item.get("spec", {}) or {}).get("selector") or {}
@@ -871,6 +1133,7 @@ async def _collect_runtime_assets() -> (
                         }
                     )
         reconciled_prefixes.add("k8s/service/")
+        reconciled_prefixes.add("k8s/network/service/")
     except Exception as e:
         logger.warning("collect_services_failed", error=str(e))
         collector_errors.append("k8s_services")
@@ -878,9 +1141,25 @@ async def _collect_runtime_assets() -> (
     # 5. ConfigMaps
     try:
         payload = await _fetch_kubectl_json("configmaps")
+        service_registry_seen = False
         for item in payload.get("items", []) or []:
             assets.append(_build_configmap_asset(item))
+            if _is_service_registry_configmap(item):
+                service_registry_seen = True
+                registry_assets, registry_relationships = (
+                    _collect_service_registry_assets(item)
+                )
+                assets.extend(registry_assets)
+                relationships.extend(registry_relationships)
         reconciled_prefixes.add("k8s/configmap/")
+        if service_registry_seen:
+            reconciled_prefixes.update(
+                (
+                    "service-registry/dashboard/",
+                    "service-registry/cache/",
+                    "service-registry/message_queue/",
+                )
+            )
     except Exception as e:
         logger.warning("collect_configmaps_failed", error=str(e))
         collector_errors.append("k8s_configmaps")
diff --git a/apps/api/tests/test_asset_scanner_job.py b/apps/api/tests/test_asset_scanner_job.py
index 87783250d..f1b5683e9 100644
--- a/apps/api/tests/test_asset_scanner_job.py
+++ b/apps/api/tests/test_asset_scanner_job.py
@@ -554,6 +554,106 @@ def test_k8s_metadata_assets_never_read_volume_secret_or_command_data() -> None:
     assert secret["metadata"]["secret_value_read"] is False
 
 
+def test_workload_network_and_service_registry_projections_are_bounded() -> None:
+    web_assets, web_relationships = (
+        asset_scanner_job._build_workload_component_assets(
+            {
+                "metadata": {
+                    "namespace": "awoooi-prod",
+                    "name": "awoooi-web",
+                    "labels": {
+                        "app": "awoooi-web",
+                        "system": "awoooi",
+                    },
+                }
+            },
+            "Deployment",
+        )
+    )
+    log_assets, _ = asset_scanner_job._build_workload_component_assets(
+        {
+            "metadata": {
+                "namespace": "observability",
+                "name": "otel-collector",
+                "labels": {
+                    "app.kubernetes.io/component": "log-collector",
+                },
+            }
+        },
+        "DaemonSet",
+    )
+    network = asset_scanner_job._build_service_network_asset(
+        {
+            "metadata": {
+                "namespace": "awoooi-prod",
+                "name": "awoooi-api",
+                "labels": {"system": "awoooi"},
+            },
+            "spec": {
+                "type": "ClusterIP",
+                "clusterIP": "10.43.0.10",
+                "ports": [
+                    {
+                        "name": "http",
+                        "protocol": "TCP",
+                        "port": 80,
+                        "targetPort": 8000,
+                    }
+                ],
+            },
+        }
+    )
+    registry_assets, registry_relationships = (
+        asset_scanner_job._collect_service_registry_assets(
+            {
+                "metadata": {
+                    "namespace": "awoooi-prod",
+                    "name": "service-registry",
+                },
+                "data": {
+                    "service-registry.yaml": """
+services:
+  - name: grafana
+    display_name: Grafana
+    host: 192.168.0.110
+    stateful_level: STANDARD_HITL
+    containers: [grafana]
+  - name: redis
+    display_name: Redis cache
+    host: 192.168.0.188
+    stateful_level: CRITICAL_HITL
+    containers: [redis]
+  - name: sentry-redis
+    display_name: Sentry task queue
+    host: 192.168.0.110
+    stateful_level: CRITICAL_HITL
+    containers: [sentry-redis]
+""",
+                },
+            }
+        )
+    )
+
+    assert [asset["asset_type"] for asset in web_assets] == ["frontend"]
+    assert web_relationships[0]["to_key"] == (
+        "k8s/deployment/awoooi-prod/awoooi-web"
+    )
+    assert [asset["asset_type"] for asset in log_assets] == ["log_stream"]
+    assert network["asset_type"] == "network"
+    assert network["metadata"]["packet_payload_read"] is False
+    assert {asset["asset_type"] for asset in registry_assets} == {
+        "dashboard",
+        "cache",
+        "message_queue",
+    }
+    assert all(
+        asset["metadata"]["raw_config_persisted"] is False
+        and asset["metadata"]["secret_value_read"] is False
+        for asset in registry_assets
+    )
+    assert len(registry_relationships) == len(registry_assets)
+
+
 def test_runtime_collection_integrates_new_asset_types_and_reconciliation_prefixes(
     monkeypatch,
 ) -> None:
@@ -584,6 +684,94 @@ def test_runtime_collection_integrates_new_asset_types_and_reconciliation_prefix
                     }
                 ]
             }
+        if resource == "deployments":
+            return {
+                "items": [
+                    {
+                        "metadata": {
+                            "namespace": "awoooi-prod",
+                            "name": "awoooi-web",
+                            "labels": {
+                                "app": "awoooi-web",
+                                "system": "awoooi",
+                            },
+                        },
+                        "spec": {},
+                        "status": {},
+                    },
+                    {
+                        "metadata": {
+                            "namespace": "awoooi-prod",
+                            "name": "awoooi-api",
+                            "labels": {
+                                "app": "awoooi-api",
+                                "system": "awoooi",
+                            },
+                        },
+                        "spec": {},
+                        "status": {},
+                    },
+                ]
+            }
+        if resource == "daemonsets":
+            return {
+                "items": [
+                    {
+                        "metadata": {
+                            "namespace": "observability",
+                            "name": "otel-collector",
+                            "labels": {
+                                "app.kubernetes.io/component": "log-collector"
+                            },
+                        },
+                        "spec": {},
+                        "status": {},
+                    }
+                ]
+            }
+        if resource == "services":
+            return {
+                "items": [
+                    {
+                        "metadata": {
+                            "namespace": "awoooi-prod",
+                            "name": "awoooi-api",
+                            "labels": {"system": "awoooi"},
+                        },
+                        "spec": {
+                            "type": "ClusterIP",
+                            "clusterIP": "10.43.0.10",
+                            "ports": [{"port": 80, "targetPort": 8000}],
+                        },
+                    }
+                ]
+            }
+        if resource == "configmaps":
+            return {
+                "items": [
+                    {
+                        "metadata": {
+                            "namespace": "awoooi-prod",
+                            "name": "service-registry",
+                        },
+                        "data": {
+                            "service-registry.yaml": """
+services:
+  - name: grafana
+    display_name: Grafana
+    host: 192.168.0.110
+    stateful_level: STANDARD_HITL
+    containers: [grafana]
+  - name: sentry-redis
+    display_name: Sentry task queue
+    host: 192.168.0.110
+    stateful_level: CRITICAL_HITL
+    containers: [sentry-redis]
+""",
+                        },
+                    }
+                ]
+            }
         if resource == "ingresses":
             return {
                 "items": [
@@ -663,6 +851,13 @@ def test_runtime_collection_integrates_new_asset_types_and_reconciliation_prefix
         "api_endpoint",
         "certificate",
         "scheduled_job",
+        "network",
+        "frontend",
+        "backend",
+        "log_stream",
+        "dashboard",
+        "cache",
+        "message_queue",
     } <= {asset["asset_type"] for asset in assets}
     assert {
         "k8s/secret-ref/",
@@ -671,6 +866,13 @@ def test_runtime_collection_integrates_new_asset_types_and_reconciliation_prefix
         "k8s/ingress-route/",
         "k8s/certificate-ref/",
         "k8s/cronjob/",
+        "k8s/network/service/",
+        "k8s/component/frontend/deployment/",
+        "k8s/component/backend/deployment/",
+        "k8s/component/log_stream/daemonset/",
+        "service-registry/dashboard/",
+        "service-registry/cache/",
+        "service-registry/message_queue/",
     } <= set(prefixes)
     assert any(
         relationship["to_key"] == "k8s/secret-ref/prod/database-ref"