455 lines
16 KiB
Python
455 lines
16 KiB
Python
"""Kubernetes provider plugin for infrastructure discovery.
|
|
|
|
Uses the official kubernetes-client library to discover deployments, services,
|
|
ingresses, config maps, persistent volumes, and namespaces from a Kubernetes
|
|
cluster. Detects CPU architecture from node labels.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Callable
|
|
|
|
from kubernetes import client, config
|
|
|
|
from iac_reverse.models import (
|
|
CpuArchitecture,
|
|
DiscoveredResource,
|
|
PlatformCategory,
|
|
ProviderType,
|
|
ScanProgress,
|
|
ScanResult,
|
|
)
|
|
from iac_reverse.plugin_base import ProviderPlugin
|
|
from iac_reverse.scanner.scanner import AuthenticationError
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Mapping from kubernetes.io/arch label values to CpuArchitecture enum
|
|
_ARCH_LABEL_MAP: dict[str, CpuArchitecture] = {
|
|
"amd64": CpuArchitecture.AMD64,
|
|
"arm": CpuArchitecture.ARM,
|
|
"arm64": CpuArchitecture.AARCH64,
|
|
"aarch64": CpuArchitecture.AARCH64,
|
|
}
|
|
|
|
_SUPPORTED_RESOURCE_TYPES = [
|
|
"kubernetes_deployment",
|
|
"kubernetes_service",
|
|
"kubernetes_ingress",
|
|
"kubernetes_config_map",
|
|
"kubernetes_persistent_volume",
|
|
"kubernetes_namespace",
|
|
]
|
|
|
|
|
|
class KubernetesPlugin(ProviderPlugin):
|
|
"""Kubernetes provider plugin using the official kubernetes-client.
|
|
|
|
Authenticates via kubeconfig file and discovers cluster resources
|
|
including deployments, services, ingresses, config maps, persistent
|
|
volumes, and namespaces.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._api_client: client.ApiClient | None = None
|
|
self._core_v1: client.CoreV1Api | None = None
|
|
self._apps_v1: client.AppsV1Api | None = None
|
|
self._networking_v1: client.NetworkingV1Api | None = None
|
|
|
|
def authenticate(self, credentials: dict[str, str]) -> None:
|
|
"""Load kubeconfig and initialize Kubernetes API clients.
|
|
|
|
Args:
|
|
credentials: Dict with keys:
|
|
- kubeconfig_path: Path to the kubeconfig file (required)
|
|
- context: Kubernetes context name (optional)
|
|
|
|
Raises:
|
|
AuthenticationError: If kubeconfig cannot be loaded.
|
|
"""
|
|
kubeconfig_path = credentials.get("kubeconfig_path")
|
|
if not kubeconfig_path:
|
|
raise AuthenticationError(
|
|
provider_name="kubernetes",
|
|
reason="kubeconfig_path is required in credentials",
|
|
)
|
|
|
|
context = credentials.get("context") or None
|
|
|
|
try:
|
|
config.load_kube_config(
|
|
config_file=kubeconfig_path,
|
|
context=context,
|
|
)
|
|
except Exception as exc:
|
|
raise AuthenticationError(
|
|
provider_name="kubernetes",
|
|
reason=f"Failed to load kubeconfig from '{kubeconfig_path}': {exc}",
|
|
) from exc
|
|
|
|
self._api_client = client.ApiClient()
|
|
self._core_v1 = client.CoreV1Api(self._api_client)
|
|
self._apps_v1 = client.AppsV1Api(self._api_client)
|
|
self._networking_v1 = client.NetworkingV1Api(self._api_client)
|
|
|
|
def get_platform_category(self) -> PlatformCategory:
|
|
"""Return CONTAINER_ORCHESTRATION platform category."""
|
|
return PlatformCategory.CONTAINER_ORCHESTRATION
|
|
|
|
def list_endpoints(self) -> list[str]:
|
|
"""Return node addresses as endpoints.
|
|
|
|
Returns:
|
|
List of node internal IP addresses or hostnames.
|
|
"""
|
|
if self._core_v1 is None:
|
|
return []
|
|
|
|
try:
|
|
nodes = self._core_v1.list_node()
|
|
endpoints: list[str] = []
|
|
for node in nodes.items:
|
|
if node.status and node.status.addresses:
|
|
for addr in node.status.addresses:
|
|
if addr.type == "InternalIP":
|
|
endpoints.append(addr.address)
|
|
break
|
|
else:
|
|
# Fallback to first address
|
|
endpoints.append(node.status.addresses[0].address)
|
|
return endpoints
|
|
except Exception as exc:
|
|
logger.warning("Failed to list node endpoints: %s", exc)
|
|
return []
|
|
|
|
def list_supported_resource_types(self) -> list[str]:
|
|
"""Return all Kubernetes resource types this plugin can discover."""
|
|
return list(_SUPPORTED_RESOURCE_TYPES)
|
|
|
|
def detect_architecture(self, endpoint: str) -> CpuArchitecture:
|
|
"""Detect CPU architecture from node labels.
|
|
|
|
Queries node labels for 'kubernetes.io/arch' to determine the
|
|
CPU architecture. Falls back to AMD64 if the label is not found.
|
|
|
|
Args:
|
|
endpoint: Node IP address or hostname to query.
|
|
|
|
Returns:
|
|
CpuArchitecture enum value for the node.
|
|
"""
|
|
if self._core_v1 is None:
|
|
return CpuArchitecture.AMD64
|
|
|
|
try:
|
|
nodes = self._core_v1.list_node()
|
|
for node in nodes.items:
|
|
# Match node by address
|
|
if node.status and node.status.addresses:
|
|
node_addresses = [
|
|
addr.address for addr in node.status.addresses
|
|
]
|
|
if endpoint in node_addresses:
|
|
labels = node.metadata.labels or {}
|
|
arch_label = labels.get(
|
|
"kubernetes.io/arch",
|
|
labels.get("beta.kubernetes.io/arch", "amd64"),
|
|
)
|
|
return _ARCH_LABEL_MAP.get(
|
|
arch_label, CpuArchitecture.AMD64
|
|
)
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"Failed to detect architecture for endpoint '%s': %s",
|
|
endpoint,
|
|
exc,
|
|
)
|
|
|
|
return CpuArchitecture.AMD64
|
|
|
|
def discover_resources(
|
|
self,
|
|
endpoints: list[str],
|
|
resource_types: list[str],
|
|
progress_callback: Callable[[ScanProgress], None],
|
|
) -> ScanResult:
|
|
"""Discover Kubernetes resources across all namespaces.
|
|
|
|
Args:
|
|
endpoints: List of node addresses (used for architecture detection).
|
|
resource_types: List of resource type strings to discover.
|
|
progress_callback: Callable for progress updates.
|
|
|
|
Returns:
|
|
ScanResult with all discovered resources.
|
|
"""
|
|
resources: list[DiscoveredResource] = []
|
|
warnings: list[str] = []
|
|
errors: list[str] = []
|
|
|
|
# Determine architecture from first endpoint
|
|
architecture = CpuArchitecture.AMD64
|
|
if endpoints:
|
|
architecture = self.detect_architecture(endpoints[0])
|
|
|
|
endpoint_str = endpoints[0] if endpoints else "cluster"
|
|
total_types = len(resource_types)
|
|
|
|
for idx, resource_type in enumerate(resource_types):
|
|
try:
|
|
discovered = self._discover_resource_type(
|
|
resource_type, architecture, endpoint_str
|
|
)
|
|
resources.extend(discovered)
|
|
except Exception as exc:
|
|
error_msg = (
|
|
f"Error discovering {resource_type}: {exc}"
|
|
)
|
|
errors.append(error_msg)
|
|
logger.error(error_msg)
|
|
|
|
progress_callback(
|
|
ScanProgress(
|
|
current_resource_type=resource_type,
|
|
resources_discovered=len(resources),
|
|
resource_types_completed=idx + 1,
|
|
total_resource_types=total_types,
|
|
)
|
|
)
|
|
|
|
return ScanResult(
|
|
resources=resources,
|
|
warnings=warnings,
|
|
errors=errors,
|
|
scan_timestamp="",
|
|
profile_hash="",
|
|
)
|
|
|
|
def _discover_resource_type(
|
|
self,
|
|
resource_type: str,
|
|
architecture: CpuArchitecture,
|
|
endpoint: str,
|
|
) -> list[DiscoveredResource]:
|
|
"""Discover resources of a specific type.
|
|
|
|
Args:
|
|
resource_type: The resource type string to discover.
|
|
architecture: Detected CPU architecture.
|
|
endpoint: Endpoint string for the resource.
|
|
|
|
Returns:
|
|
List of DiscoveredResource objects.
|
|
"""
|
|
dispatch = {
|
|
"kubernetes_deployment": self._discover_deployments,
|
|
"kubernetes_service": self._discover_services,
|
|
"kubernetes_ingress": self._discover_ingresses,
|
|
"kubernetes_config_map": self._discover_config_maps,
|
|
"kubernetes_persistent_volume": self._discover_persistent_volumes,
|
|
"kubernetes_namespace": self._discover_namespaces,
|
|
}
|
|
|
|
handler = dispatch.get(resource_type)
|
|
if handler is None:
|
|
return []
|
|
|
|
return handler(architecture, endpoint)
|
|
|
|
def _discover_deployments(
|
|
self, architecture: CpuArchitecture, endpoint: str
|
|
) -> list[DiscoveredResource]:
|
|
"""Discover all deployments across namespaces."""
|
|
results: list[DiscoveredResource] = []
|
|
deployments = self._apps_v1.list_deployment_for_all_namespaces()
|
|
|
|
for dep in deployments.items:
|
|
name = dep.metadata.name
|
|
namespace = dep.metadata.namespace
|
|
results.append(
|
|
DiscoveredResource(
|
|
resource_type="kubernetes_deployment",
|
|
unique_id=f"{namespace}/{name}",
|
|
name=name,
|
|
provider=ProviderType.KUBERNETES,
|
|
platform_category=PlatformCategory.CONTAINER_ORCHESTRATION,
|
|
architecture=architecture,
|
|
endpoint=endpoint,
|
|
attributes={
|
|
"namespace": namespace,
|
|
"replicas": dep.spec.replicas if dep.spec else None,
|
|
"labels": dict(dep.metadata.labels or {}),
|
|
},
|
|
raw_references=[
|
|
f"kubernetes_namespace:{namespace}",
|
|
],
|
|
)
|
|
)
|
|
|
|
return results
|
|
|
|
def _discover_services(
|
|
self, architecture: CpuArchitecture, endpoint: str
|
|
) -> list[DiscoveredResource]:
|
|
"""Discover all services across namespaces."""
|
|
results: list[DiscoveredResource] = []
|
|
services = self._core_v1.list_service_for_all_namespaces()
|
|
|
|
for svc in services.items:
|
|
name = svc.metadata.name
|
|
namespace = svc.metadata.namespace
|
|
results.append(
|
|
DiscoveredResource(
|
|
resource_type="kubernetes_service",
|
|
unique_id=f"{namespace}/{name}",
|
|
name=name,
|
|
provider=ProviderType.KUBERNETES,
|
|
platform_category=PlatformCategory.CONTAINER_ORCHESTRATION,
|
|
architecture=architecture,
|
|
endpoint=endpoint,
|
|
attributes={
|
|
"namespace": namespace,
|
|
"type": svc.spec.type if svc.spec else None,
|
|
"cluster_ip": svc.spec.cluster_ip if svc.spec else None,
|
|
"labels": dict(svc.metadata.labels or {}),
|
|
},
|
|
raw_references=[
|
|
f"kubernetes_namespace:{namespace}",
|
|
],
|
|
)
|
|
)
|
|
|
|
return results
|
|
|
|
def _discover_ingresses(
|
|
self, architecture: CpuArchitecture, endpoint: str
|
|
) -> list[DiscoveredResource]:
|
|
"""Discover all ingresses across namespaces."""
|
|
results: list[DiscoveredResource] = []
|
|
ingresses = self._networking_v1.list_ingress_for_all_namespaces()
|
|
|
|
for ing in ingresses.items:
|
|
name = ing.metadata.name
|
|
namespace = ing.metadata.namespace
|
|
results.append(
|
|
DiscoveredResource(
|
|
resource_type="kubernetes_ingress",
|
|
unique_id=f"{namespace}/{name}",
|
|
name=name,
|
|
provider=ProviderType.KUBERNETES,
|
|
platform_category=PlatformCategory.CONTAINER_ORCHESTRATION,
|
|
architecture=architecture,
|
|
endpoint=endpoint,
|
|
attributes={
|
|
"namespace": namespace,
|
|
"labels": dict(ing.metadata.labels or {}),
|
|
},
|
|
raw_references=[
|
|
f"kubernetes_namespace:{namespace}",
|
|
],
|
|
)
|
|
)
|
|
|
|
return results
|
|
|
|
def _discover_config_maps(
|
|
self, architecture: CpuArchitecture, endpoint: str
|
|
) -> list[DiscoveredResource]:
|
|
"""Discover all config maps across namespaces."""
|
|
results: list[DiscoveredResource] = []
|
|
config_maps = self._core_v1.list_config_map_for_all_namespaces()
|
|
|
|
for cm in config_maps.items:
|
|
name = cm.metadata.name
|
|
namespace = cm.metadata.namespace
|
|
results.append(
|
|
DiscoveredResource(
|
|
resource_type="kubernetes_config_map",
|
|
unique_id=f"{namespace}/{name}",
|
|
name=name,
|
|
provider=ProviderType.KUBERNETES,
|
|
platform_category=PlatformCategory.CONTAINER_ORCHESTRATION,
|
|
architecture=architecture,
|
|
endpoint=endpoint,
|
|
attributes={
|
|
"namespace": namespace,
|
|
"data_keys": list((cm.data or {}).keys()),
|
|
"labels": dict(cm.metadata.labels or {}),
|
|
},
|
|
raw_references=[
|
|
f"kubernetes_namespace:{namespace}",
|
|
],
|
|
)
|
|
)
|
|
|
|
return results
|
|
|
|
def _discover_persistent_volumes(
|
|
self, architecture: CpuArchitecture, endpoint: str
|
|
) -> list[DiscoveredResource]:
|
|
"""Discover all persistent volumes (cluster-scoped)."""
|
|
results: list[DiscoveredResource] = []
|
|
pvs = self._core_v1.list_persistent_volume()
|
|
|
|
for pv in pvs.items:
|
|
name = pv.metadata.name
|
|
results.append(
|
|
DiscoveredResource(
|
|
resource_type="kubernetes_persistent_volume",
|
|
unique_id=name,
|
|
name=name,
|
|
provider=ProviderType.KUBERNETES,
|
|
platform_category=PlatformCategory.CONTAINER_ORCHESTRATION,
|
|
architecture=architecture,
|
|
endpoint=endpoint,
|
|
attributes={
|
|
"capacity": (
|
|
dict(pv.spec.capacity)
|
|
if pv.spec and pv.spec.capacity
|
|
else {}
|
|
),
|
|
"access_modes": (
|
|
list(pv.spec.access_modes)
|
|
if pv.spec and pv.spec.access_modes
|
|
else []
|
|
),
|
|
"storage_class": (
|
|
pv.spec.storage_class_name if pv.spec else None
|
|
),
|
|
"labels": dict(pv.metadata.labels or {}),
|
|
},
|
|
raw_references=[],
|
|
)
|
|
)
|
|
|
|
return results
|
|
|
|
def _discover_namespaces(
|
|
self, architecture: CpuArchitecture, endpoint: str
|
|
) -> list[DiscoveredResource]:
|
|
"""Discover all namespaces."""
|
|
results: list[DiscoveredResource] = []
|
|
namespaces = self._core_v1.list_namespace()
|
|
|
|
for ns in namespaces.items:
|
|
name = ns.metadata.name
|
|
results.append(
|
|
DiscoveredResource(
|
|
resource_type="kubernetes_namespace",
|
|
unique_id=name,
|
|
name=name,
|
|
provider=ProviderType.KUBERNETES,
|
|
platform_category=PlatformCategory.CONTAINER_ORCHESTRATION,
|
|
architecture=architecture,
|
|
endpoint=endpoint,
|
|
attributes={
|
|
"status": (
|
|
ns.status.phase if ns.status else None
|
|
),
|
|
"labels": dict(ns.metadata.labels or {}),
|
|
},
|
|
raw_references=[],
|
|
)
|
|
)
|
|
|
|
return results
|