Postmortem: How a Lack of CKS-Certified Engineers Caused a Security Breach in Our Cluster

package main

import (
    "context"
    "flag"
    "fmt"
    "log"
    "os"
    "path/filepath"

    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/clientcmd"
    "k8s.io/client-go/util/homedir"
)

// ConfigScanner checks for security misconfigurations that led to our 2024 breach
type ConfigScanner struct {
    clientset *kubernetes.Clientset
    namespace string
    findings  []string
}

// NewConfigScanner initializes a K8s client and scanner instance
func NewConfigScanner(kubeconfig, namespace string) (*ConfigScanner, error) {
    if kubeconfig == "" {
        if home := homedir.HomeDir(); home != "" {
            kubeconfig = filepath.Join(home, ".kube", "config")
        } else {
            return nil, fmt.Errorf("no kubeconfig provided and home directory not found")
        }
    }

    config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
    if err != nil {
        return nil, fmt.Errorf("failed to build kubeconfig: %w", err)
    }

    clientset, err := kubernetes.NewForConfig(config)
    if err != nil {
        return nil, fmt.Errorf("failed to create kubernetes client: %w", err)
    }

    return &ConfigScanner{
        clientset: clientset,
        namespace: namespace,
        findings:  make([]string, 0),
    }, nil
}

// ScanAnonymousAccess checks for anonymous access to the K8s API server
func (s *ConfigScanner) ScanAnonymousAccess(ctx context.Context) {
    // Get cluster-wide RBAC cluster roles
    clusterRoles, err := s.clientset.RbacV1().ClusterRoles().List(ctx, metav1.ListOptions{})
    if err != nil {
        log.Printf("failed to list cluster roles: %v", err)
        return
    }

    for _, cr := range clusterRoles.Items {
        for _, rule := range cr.Rules {
            // Check for rules allowing anonymous users (system:anonymous) to perform actions
            for _, group := range rule.Groups {
                if group == "system:anonymous" {
                    s.findings = append(s.findings, fmt.Sprintf(
                        "CRITICAL: ClusterRole %s allows anonymous access via group %s with verbs %v",
                        cr.Name, group, rule.Verbs,
                    ))
                }
            }
        }
    }
}

// ScanPrivilegedPods checks for pods running with privileged: true
func (s *ConfigScanner) ScanPrivilegedPods(ctx context.Context) {
    pods, err := s.clientset.CoreV1().Pods(s.namespace).List(ctx, metav1.ListOptions{})
    if err != nil {
        log.Printf("failed to list pods in namespace %s: %v", s.namespace, err)
        return
    }

    for _, pod := range pods.Items {
        for _, container := range pod.Spec.Containers {
            if container.SecurityContext != nil && container.SecurityContext.Privileged != nil && *container.SecurityContext.Privileged {
                s.findings = append(s.findings, fmt.Sprintf(
                    "HIGH: Pod %s/%s has privileged container %s",
                    pod.Namespace, pod.Name, container.Name,
                ))
            }
        }
    }
}

func main() {
    var kubeconfig string
    var namespace string
    flag.StringVar(&kubeconfig, "kubeconfig", "", "Path to kubeconfig file")
    flag.StringVar(&namespace, "namespace", "default", "Namespace to scan")
    flag.Parse()

    ctx := context.Background()
    scanner, err := NewConfigScanner(kubeconfig, namespace)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Error initializing scanner: %v\n", err)
        os.Exit(1)
    }

    fmt.Println("Starting security scan...")
    scanner.ScanAnonymousAccess(ctx)
    scanner.ScanPrivilegedPods(ctx)

    if len(scanner.findings) == 0 {
        fmt.Println("No critical misconfigurations found.")
        return
    }

    fmt.Printf("\nFound %d misconfigurations:\n", len(scanner.findings))
    for _, f := range scanner.findings {
        fmt.Println(f)
    }
}

# CKS-compliant EKS cluster configuration (Terraform 1.7+)
# This configuration would have prevented the 2024 breach by enforcing Pod Security Admissions
# and restricting anonymous access by default

terraform {
  required_version = ">= 1.7.0"
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
    kubernetes = {
      source  = "hashicorp/kubernetes"
      version = "~> 2.23"
    }
  }
}

provider "aws" {
  region = var.aws_region
}

variable "aws_region" {
  type    = string
  default = "us-east-1"
}

variable "cluster_name" {
  type    = string
  default = "prod-secure-cluster"
}

# EKS cluster with CKS-recommended security controls
resource "aws_eks_cluster" "main" {
  name     = var.cluster_name
  role_arn = aws_iam_role.eks_cluster.arn
  version  = "1.29" # Kubernetes 1.29 has GA Pod Security Admissions

  vpc_config {
    subnet_ids              = aws_subnet.private[*].id
    endpoint_private_access = true
    endpoint_public_access  = false # Disable public API access to prevent external brute force
    security_group_ids      = [aws_security_group.eks_cluster.id]
  }

  # Enforce CKS-mandated encryption at rest
  encryption_config {
    resources = ["secrets"]
    provider {
      key_arn = aws_kms_key.eks_secrets.arn
    }
  }

  # Enable CloudWatch logging for all audit logs (required for postmortems)
  enabled_cluster_log_types = ["api", "audit", "authenticator", "controllerManager", "scheduler"]

  depends_on = [
    aws_iam_role_policy_attachment.eks_cluster_policy,
  ]

  tags = {
    Environment = "production"
    ManagedBy   = "terraform"
    CKSCompliant = "true"
  }
}

# KMS key for secret encryption
resource "aws_kms_key" "eks_secrets" {
  description             = "KMS key for EKS secret encryption"
  deletion_window_in_days = 10
  enable_key_rotation      = true # CKS requirement: rotate encryption keys annually

  tags = {
    Purpose = "eks-secret-encryption"
  }
}

# IAM role for EKS cluster
resource "aws_iam_role" "eks_cluster" {
  name = "${var.cluster_name}-cluster-role"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "eks.amazonaws.com"
        }
      }
    ]
  })
}

resource "aws_iam_role_policy_attachment" "eks_cluster_policy" {
  role       = aws_iam_role.eks_cluster.name
  policy_arn = "arn:aws:iam::aws:policy/AmazonEKSClusterPolicy"
}

# Private subnets for EKS (no public IPs for worker nodes)
resource "aws_subnet" "private" {
  count             = 3
  vpc_id            = aws_vpc.main.id
  cidr_block        = "10.0.${count.index + 1}.0/24"
  availability_zone = data.aws_availability_zones.available.names[count.index]

  tags = {
    "kubernetes.io/cluster/${var.cluster_name}" = "shared"
    Type                                          = "private"
  }
}

# Pod Security Admission configuration (replaces legacy PSP)
resource "kubernetes_manifest" "pod_security_admission" {
  manifest = {
    apiVersion = "policy/v1beta1"
    kind       = "PodSecurityPolicy"
    metadata = {
      name = "restricted-psa"
      annotations = {
        "pod-security.kubernetes.io/enforce" = "restricted"
        "pod-security.kubernetes.io/audit"   = "restricted"
        "pod-security.kubernetes.io/warn"    = "restricted"
      }
    }
    spec = {
      privileged                 = false
      allowPrivilegeEscalation   = false
      requiredDropCapabilities    = ["ALL"]
      volumes                    = ["configMap", "emptyDir", "projected", "secret", "downwardAPI", "persistentVolumeClaim"]
      hostNetwork                = false
      hostIPC                    = false
      hostPID                    = false
      runAsUser = {
        rule = "MustRunAsNonRoot"
      }
      seLinux = {
        rule = "RunAsAny"
      }
      fsGroup = {
        rule = "RunAsAny"
      }
      supplementalGroups = {
        rule = "RunAsAny"
      }
    }
  }

  depends_on = [aws_eks_cluster.main]
}

#!/usr/bin/env python3
"""
RBAC Auditor for CKS Compliance
Checks RBAC configurations against CNCF CKS v1.1 benchmarks
"""

import json
import sys
from kubernetes import client, config
from typing import List, Dict, Any

class RBACAuditor:
    def __init__(self, kubeconfig: str = None, namespace: str = "default"):
        """
        Initialize RBAC auditor with K8s client

        Args:
            kubeconfig: Path to kubeconfig file (defaults to ~/.kube/config)
            namespace: Namespace to audit (defaults to default)
        """
        try:
            if kubeconfig:
                config.load_kube_config(config_file=kubeconfig)
            else:
                config.load_kube_config()
        except Exception as e:
            raise RuntimeError(f"Failed to load kubeconfig: {str(e)}")

        self.api = client.RbacAuthorizationV1Api()
        self.namespace = namespace
        self.findings: List[Dict[str, Any]] = []

    def check_cluster_role_permissions(self) -> None:
        """
        Check for cluster roles with overly permissive permissions
        CKS Benchmark 3.1.1: Avoid cluster-admin role for non-admin users
        """
        try:
            cluster_roles = self.api.list_cluster_role()
        except Exception as e:
            print(f"ERROR: Failed to list cluster roles: {str(e)}", file=sys.stderr)
            return

        for cr in cluster_roles.items:
            # Check for cluster-admin role assigned to non-system users
            if cr.metadata.name == "cluster-admin":
                # Get role bindings for cluster-admin
                try:
                    bindings = self.api.list_cluster_role_binding()
                except Exception as e:
                    print(f"ERROR: Failed to list cluster role bindings: {str(e)}", file=sys.stderr)
                    continue

                for binding in bindings.items:
                    if binding.role_ref.name == "cluster-admin":
                        for subject in binding.subjects:
                            if subject.kind == "User" and not subject.name.startswith("system:"):
                                self.findings.append({
                                    "severity": "CRITICAL",
                                    "resource": "ClusterRoleBinding",
                                    "name": binding.metadata.name,
                                    "detail": f"cluster-admin role bound to user {subject.name}"
                                })

            # Check for wildcard verbs in cluster roles
            for rule in cr.rules:
                if "*" in rule.verbs:
                    self.findings.append({
                        "severity": "HIGH",
                        "resource": "ClusterRole",
                        "name": cr.metadata.name,
                        "detail": f"Wildcard verb (*) found in rule with resources {rule.resources}"
                    })

    def check_service_account_permissions(self) -> None:
        """
        Check service accounts with unnecessary cluster-wide access
        CKS Benchmark 3.2.3: Minimize service account access
        """
        try:
            service_accounts = self.api.list_namespaced_service_account(self.namespace)
        except Exception as e:
            print(f"ERROR: Failed to list service accounts in {self.namespace}: {str(e)}", file=sys.stderr)
            return

        for sa in service_accounts.items:
            # Check if service account is bound to any cluster roles
            try:
                bindings = self.api.list_cluster_role_binding()
            except Exception as e:
                print(f"ERROR: Failed to list cluster role bindings: {str(e)}", file=sys.stderr)
                continue

            for binding in bindings.items:
                for subject in binding.subjects:
                    if subject.kind == "ServiceAccount" and \
                       subject.name == sa.metadata.name and \
                       subject.namespace == sa.metadata.namespace:
                        self.findings.append({
                            "severity": "MEDIUM",
                            "resource": "ServiceAccount",
                            "name": f"{sa.metadata.namespace}/{sa.metadata.name}",
                            "detail": f"Service account bound to cluster role {binding.role_ref.name}"
                        })

    def generate_report(self) -> str:
        """Generate JSON report of findings"""
        return json.dumps({
            "namespace": self.namespace,
            "total_findings": len(self.findings),
            "findings": self.findings
        }, indent=2)

if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="CKS RBAC Auditor")
    parser.add_argument("--kubeconfig", help="Path to kubeconfig file")
    parser.add_argument("--namespace", default="default", help="Namespace to audit")
    args = parser.parse_args()

    try:
        auditor = RBACAuditor(kubeconfig=args.kubeconfig, namespace=args.namespace)
    except Exception as e:
        print(f"FATAL: Failed to initialize auditor: {str(e)}", file=sys.stderr)
        sys.exit(1)

    print("Auditing RBAC configurations...")
    auditor.check_cluster_role_permissions()
    auditor.check_service_account_permissions()

    report = auditor.generate_report()
    print(report)

    if auditor.findings:
        sys.exit(1)
    sys.exit(0)

# Sample CKS study lab: Scan for privileged pods
kubectl get pods --all-namespaces -o json | jq '.items[] | select(.spec.containers[].securityContext.privileged == true) | {namespace: .metadata.namespace, pod: .metadata.name, container: .spec.containers[].name}'

# Trivy CI step to scan container images
trivy image --exit-code 1 --severity HIGH,CRITICAL --no-progress myapp:latest

# Label namespace to enforce restricted PSA profile
kubectl label namespace prod pod-security.kubernetes.io/enforce=restricted --overwrite