wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 4 Packages Projects Releases Wiki Activity

feat(security): project runtime service asset types

Browse Source
This commit is contained in:
ogt
2026-07-14 14:24:46 +08:00
parent f963452b3e
commit d4bc3cb453
2 changed files with 481 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

279
apps/api/src/jobs/asset_scanner_job.py
View File

@@ -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")

202
apps/api/tests/test_asset_scanner_job.py
View File

@@ -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"