Kubernetes Security for ML Workloads

# Enumerate GPU operator components in a cluster
import subprocess
import json
 
def enumerate_gpu_operator(namespace="gpu-operator"):
    """Identify GPU operator attack surface in a Kubernetes cluster."""
    components = {}
 
    # Vind GPU-operator-pods en hun privilegeniveaus
    pods = json.loads(subprocess.check_output([
        "kubectl", "get", "pods", "-n", namespace,
        "-o", "json"
    ]))
 
    for pod in pods["items"]:
        pod_name = pod["metadata"]["name"]
        containers = pod["spec"].get("containers", [])
 
        for container in containers:
            security_context = container.get("securityContext", {})
            pod_security = pod["spec"].get("securityContext", {})
 
            components[pod_name] = {
                "privileged": security_context.get("privileged", False),
                "host_pid": pod["spec"].get("hostPID", False),
                "host_network": pod["spec"].get("hostNetwork", False),
                "volume_mounts": [
                    vm["mountPath"] for vm in container.get("volumeMounts", [])
                ],
                "capabilities": security_context.get("capabilities", {}),
            }
 
    return components

# Default device plugin socket locations
# /var/lib/kubelet/device-plugins/nvidia.sock
# /var/lib/kubelet/device-plugins/kubelet.sock
 
# An attacker with node access can:
# 1. Register a malicious device plugin that intercepts GPU allocation
# 2. Monitor device plugin traffic to observe which pods request GPUs
# 3. Manipulate device allocation to redirect GPU access

# Probe KubeFlow notebook server for Kubernetes access
import requests
import os
 
def assess_notebook_k8s_access():
    """Assess Kubernetes access from a KubeFlow notebook pod."""
    findings = []
 
    # Controleer op een serviceaccount-token
    token_path = "/var/run/secrets/kubernetes.io/serviceaccount/token"
    if os.path.exists(token_path):
        with open(token_path) as f:
            token = f.read()
        findings.append({
            "finding": "Service account token accessible",
            "token_preview": token[:50] + "...",
        })
 
    # Probeer de pods in de namespace op te sommen
    namespace_path = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"
    namespace = open(namespace_path).read() if os.path.exists(namespace_path) else "default"
 
    k8s_host = os.environ.get("KUBERNETES_SERVICE_HOST", "kubernetes.default.svc")
    k8s_port = os.environ.get("KUBERNETES_SERVICE_PORT", "443")
 
    try:
        resp = requests.get(
            f"https://{k8s_host}:{k8s_port}/api/v1/namespaces/{namespace}/pods",
            headers={"Authorization": f"Bearer {token}"},
            verify="/var/run/secrets/kubernetes.io/serviceaccount/ca.crt",
            timeout=5,
        )
        if resp.status_code == 200:
            pod_names = [p["metadata"]["name"] for p in resp.json()["items"]]
            findings.append({
                "finding": "Can list pods in namespace",
                "severity": "HIGH",
                "pods": pod_names,
            })
    except Exception as e:
        findings.append({"finding": "API access failed", "error": str(e)})
 
    # Controleer op GPU-toegang vanuit de notebook
    try:
        import torch
        if torch.cuda.is_available():
            findings.append({
                "finding": "GPU access from notebook",
                "gpu_count": torch.cuda.device_count(),
                "gpu_name": torch.cuda.get_device_name(0),
            })
    except ImportError:
        pass
 
    return findings

# Malicious pipeline component that exploits Kubernetes access
# This demonstrates the attack pattern - not for use in unauthorized testing
apiVersion: argoproj.io/v1alpha1
kind: Workflow
spec:
  templates:
    - name: malicious-step
      container:
        image: python:3.11
        command: ["python", "-c"]
        args:
          - |
            # Pipeline pods often have broad namespace access
            # for reading artifacts, secrets, and config maps
            import subprocess
            # Enumerate accessible secrets
            result = subprocess.run(
                ["kubectl", "get", "secrets", "-A", "-o", "name"],
                capture_output=True, text=True
            )
            print(result.stdout)

# Enumerate model serving endpoints across namespaces
def enumerate_inference_services():
    """Find all inference services in a Kubernetes cluster."""
    import subprocess
    import json
 
    services = {}
 
    # KServe InferenceService-resources
    try:
        kserve = json.loads(subprocess.check_output([
            "kubectl", "get", "inferenceservices", "--all-namespaces", "-o", "json"
        ]))
        for item in kserve["items"]:
            name = item["metadata"]["name"]
            ns = item["metadata"]["namespace"]
            url = item.get("status", {}).get("url", "unknown")
            services[f"{ns}/{name}"] = {
                "type": "kserve",
                "url": url,
                "ready": item.get("status", {}).get("conditions", []),
            }
    except subprocess.CalledProcessError:
        pass
 
    # Seldon-deployments
    try:
        seldon = json.loads(subprocess.check_output([
            "kubectl", "get", "seldondeployments", "--all-namespaces", "-o", "json"
        ]))
        for item in seldon["items"]:
            name = item["metadata"]["name"]
            ns = item["metadata"]["namespace"]
            services[f"{ns}/{name}"] = {
                "type": "seldon",
                "replicas": item["spec"].get("replicas", 1),
                "predictors": [
                    p["name"] for p in item["spec"].get("predictors", [])
                ],
            }
    except subprocess.CalledProcessError:
        pass
 
    return services

# Assess distributed training network exposure
def assess_distributed_training_security(namespace="training"):
    """Check for insecure distributed training configurations."""
    import subprocess
    import json
 
    findings = []
 
    # Vind trainingspods met open communicatiepoorten
    pods = json.loads(subprocess.check_output([
        "kubectl", "get", "pods", "-n", namespace,
        "-l", "training-job",
        "-o", "json"
    ]))
 
    for pod in pods["items"]:
        containers = pod["spec"].get("containers", [])
        for container in containers:
            ports = container.get("ports", [])
            env_vars = {
                e["name"]: e.get("value", "")
                for e in container.get("env", [])
            }
 
            # Controleer op NCCL- en Gloo-communicatiepoorten
            nccl_port = env_vars.get("NCCL_SOCKET_IFNAME")
            master_addr = env_vars.get("MASTER_ADDR")
            master_port = env_vars.get("MASTER_PORT")
 
            if master_addr and master_port:
                findings.append({
                    "pod": pod["metadata"]["name"],
                    "master_addr": master_addr,
                    "master_port": master_port,
                    "severity": "MEDIUM",
                    "note": "Distributed training master endpoint exposed",
                })
 
            # Controleer op shared-memory-mounts (vereist voor NCCL)
            volumes = pod["spec"].get("volumes", [])
            for vol in volumes:
                if vol.get("emptyDir", {}).get("medium") == "Memory":
                    findings.append({
                        "pod": pod["metadata"]["name"],
                        "finding": "Shared memory (dshm) mount detected",
                        "note": "Required for NCCL but may allow cross-container data access",
                    })
 
    return findings

# Common overly permissive RBAC for ML platform service accounts
# This pattern appears frequently in KubeFlow and similar platforms
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: ml-platform-admin
rules:
  # Broad pod management for training jobs
  - apiGroups: [""]
    resources: ["pods", "pods/exec", "pods/log"]
    verbs: ["*"]
  # Secret access for model registry credentials
  - apiGroups: [""]
    resources: ["secrets"]
    verbs: ["get", "list", "watch"]
  # PVC management for datasets and models
  - apiGroups: [""]
    resources: ["persistentvolumeclaims"]
    verbs: ["*"]
  # CRD access for ML operators
  - apiGroups: ["kubeflow.org"]
    resources: ["*"]
    verbs: ["*"]

def audit_ml_rbac(namespace="kubeflow"):
    """Audit RBAC permissions for ML platform service accounts."""
    import subprocess
    import json
 
    dangerous_permissions = []
 
    # Haal alle serviceaccounts in de ML-namespace op
    sa_list = json.loads(subprocess.check_output([
        "kubectl", "get", "serviceaccounts", "-n", namespace, "-o", "json"
    ]))
 
    for sa in sa_list["items"]:
        sa_name = sa["metadata"]["name"]
 
        # Controleer wat deze SA kan doen met auth can-i
        for resource in ["secrets", "pods/exec", "configmaps", "nodes"]:
            for verb in ["get", "list", "create", "delete"]:
                result = subprocess.run(
                    ["kubectl", "auth", "can-i", verb, resource,
                     "--as", f"system:serviceaccount:{namespace}:{sa_name}",
                     "-n", namespace],
                    capture_output=True, text=True
                )
                if "yes" in result.stdout:
                    dangerous_permissions.append({
                        "service_account": f"{namespace}:{sa_name}",
                        "resource": resource,
                        "verb": verb,
                    })
 
    return dangerous_permissions

# Commonly misconfigured shared storage for ML workloads
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: model-storage
spec:
  accessModes:
    - ReadWriteMany  # Multiple pods can write simultaneously
  resources:
    requests:
      storage: 500Gi
  storageClassName: nfs-client  # NFS often lacks access controls