Postmortem: How a Kubernetes 1.32 RBAC Misconfiguration Led to a Data Breach in Q1 2026

// rbac_check.go
// Demonstration of the RBAC permission check that was missing from the CI pipeline
// in the breached cluster. This code would have detected the misconfigured RoleBinding
// before deployment.
package main

import (
    "context"
    "flag"
    "fmt"
    "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"
    "k8s.io/apiserver/pkg/authorization/authorizer"
    "k8s.io/apiserver/pkg/authorization/authorizerfactory"
    "k8s.io/apiserver/pkg/authorization/union"
    rbacv1 "k8s.io/api/rbac/v1"
)

// checkServiceAccountRBAC validates that a ServiceAccount only has access to allowed resources
// Returns a list of non-compliant permissions, or an error if the check fails.
func checkServiceAccountRBAC(ctx context.Context, clientset *kubernetes.Clientset, saNamespace, saName string, allowedResources []string) ([]string, error) {
    // Fetch all RoleBindings in the ServiceAccount's namespace
    roleBindings, err := clientset.RbacV1().RoleBindings(saNamespace).List(ctx, metav1.ListOptions{})
    if err != nil {
        return nil, fmt.Errorf("failed to list RoleBindings in namespace %s: %w", saNamespace, err)
    }

    var nonCompliant []string

    // Iterate over all RoleBindings to find ones referencing the target ServiceAccount
    for _, rb := range roleBindings.Items {
        // Check if the RoleBinding references our ServiceAccount
        saRef := rb.Subjects
        for _, subj := range saRef {
            if subj.Kind == "ServiceAccount" && subj.Name == saName && subj.Namespace == saNamespace {
                // Fetch the associated Role
                role, err := clientset.RbacV1().Roles(saNamespace).Get(ctx, rb.RoleRef.Name, metav1.GetOptions{})
                if err != nil {
                    return nil, fmt.Errorf("failed to get Role %s in namespace %s: %w", rb.RoleRef.Name, saNamespace, err)
                }

                // Check each rule in the Role against allowed resources
                for _, rule := range role.Rules {
                    // Check if the rule grants access to resources not in allowedResources
                    for _, res := range rule.Resources {
                        isAllowed := false
                        for _, allowed := range allowedResources {
                            if res == allowed {
                                isAllowed = true
                                break
                            }
                        }
                        if !isAllowed {
                            nonCompliant = append(nonCompliant, fmt.Sprintf(
                                "Role %s (via RoleBinding %s) grants %v on %s, which is not allowed for SA %s/%s",
                                role.Name, rb.Name, rule.Verbs, res, saNamespace, saName,
                            ))
                        }
                    }
                }
            }
        }
    }

    return nonCompliant, nil
}

func main() {
    // Parse kubeconfig flag
    var kubeconfig *string
    if home := homedir.HomeDir(); home != "" {
        kubeconfig = flag.String("kubeconfig", filepath.Join(home, ".kube", "config"), "(optional) absolute path to the kubeconfig file")
    } else {
        kubeconfig = flag.String("kubeconfig", "", "absolute path to the kubeconfig file")
    }
    saNamespace := flag.String("sa-namespace", "default", "Namespace of the ServiceAccount to check")
    saName := flag.String("sa-name", "app-sa", "Name of the ServiceAccount to check")
    flag.Parse()

    // Validate required flags
    if *saNamespace == "" || *saName == "" {
        fmt.Fprintf(os.Stderr, "Error: sa-namespace and sa-name are required\n")
        flag.Usage()
        os.Exit(1)
    }

    // Build config from kubeconfig
    config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Error building kubeconfig: %v\n", err)
        os.Exit(1)
    }

    // Create clientset
    clientset, err := kubernetes.NewForConfig(config)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Error creating Kubernetes clientset: %v\n", err)
        os.Exit(1)
    }

    // Define allowed resources for the ServiceAccount (configmaps only, as per policy)
    allowedResources := []string{"configmaps"}

    // Run RBAC check
    ctx := context.Background()
    nonCompliant, err := checkServiceAccountRBAC(ctx, clientset, *saNamespace, *saName, allowedResources)
    if err != nil {
        fmt.Fprintf(os.Stderr, "Error checking RBAC: %v\n", err)
        os.Exit(1)
    }

    // Report results
    if len(nonCompliant) == 0 {
        fmt.Printf("✅ ServiceAccount %s/%s has no non-compliant RBAC permissions\n", *saNamespace, *saName)
        os.Exit(0)
    }

    fmt.Printf("❌ ServiceAccount %s/%s has %d non-compliant RBAC permissions:\n", *saNamespace, *saName, len(nonCompliant))
    for _, msg := range nonCompliant {
        fmt.Printf("  - %s\n", msg)
    }
    os.Exit(1)
}

// rbac_admission_webhook.go
// Kubernetes 1.32 ValidatingAdmissionWebhook implementation that blocks RoleBindings
// granting unauthorized access to secrets. This webhook was not enabled in the breached cluster.
package main

import (
    "context"
    "encoding/json"
    "flag"
    "fmt"
    "io"
    "net/http"
    "os"
    "strings"

    admissionv1 "k8s.io/api/admission/v1"
    rbacv1 "k8s.io/api/rbac/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/runtime"
    "k8s.io/apimachinery/pkg/util/validation/field"
)

const (
    // Restricted resources that require explicit approval
    restrictedResources = "secrets"
    // Allowed verbs for restricted resources (empty = no access)
    allowedVerbsForRestricted = ""
)

// admitFunc handles AdmissionReview requests and returns AdmissionResponse
type admitFunc func(admissionv1.AdmissionReview) *admissionv1.AdmissionResponse

// serveAdmission handles HTTP requests to the admission webhook
func serveAdmission(w http.ResponseWriter, r *http.Request, admit admitFunc) {
    body, err := io.ReadAll(r.Body)
    if err != nil {
        http.Error(w, "Failed to read request body", http.StatusBadRequest)
        return
    }
    defer r.Body.Close()

    // Parse AdmissionReview request
    var admissionReview admissionv1.AdmissionReview
    if err := json.Unmarshal(body, &admissionReview); err != nil {
        http.Error(w, "Failed to unmarshal AdmissionReview", http.StatusBadRequest)
        return
    }

    // Validate request has required fields
    if admissionReview.Request == nil {
        http.Error(w, "AdmissionReview request is nil", http.StatusBadRequest)
        return
    }

    // Process admission request
    response := admit(admissionReview)

    // Set response type
    response.UID = admissionReview.Request.UID
    admissionReview.Response = response
    admissionReview.Kind = "AdmissionReview"
    admissionReview.APIVersion = "admission.k8s.io/v1"

    // Write response
    resp, err := json.Marshal(admissionReview)
    if err != nil {
        http.Error(w, "Failed to marshal AdmissionReview response", http.StatusInternalServerError)
        return
    }

    w.Header().Set("Content-Type", "application/json")
    w.Write(resp)
}

// validateRoleBinding checks if a RoleBinding grants unauthorized access to restricted resources
func validateRoleBinding(roleBinding *rbacv1.RoleBinding) (*admissionv1.AdmissionResponse, error) {
    // Skip if the RoleBinding references a ClusterRole (handled by ClusterRoleBinding webhook)
    if roleBinding.RoleRef.Kind != "Role" {
        return &admissionv1.AdmissionResponse{Allowed: true}, nil
    }

    // Fetch the associated Role (in a real webhook, this would call the Kubernetes API)
    // For this example, we simulate fetching the Role from the request if it's a create/update
    // In practice, you'd use a cached client or API call here
    var role rbacv1.Role
    if roleBinding.RoleRef.APIVersion == "rbac.authorization.k8s.io/v1" {
        // In a real implementation, you'd fetch the Role by name from the cluster
        // This is a simplified check that assumes the Role is included in the request (not standard)
        // For brevity, we skip Role fetching and check the RoleBinding's subjects (simulated)
    }

    // Check if the RoleBinding grants access to restricted resources
    // Simulated check: if the RoleBinding is for the "app-sa" ServiceAccount, check its Role
    // In the breach, the Role had rules granting get on secrets
    for _, subj := range roleBinding.Subjects {
        if subj.Kind == "ServiceAccount" && subj.Name == "app-sa" {
            // Simulated Role check: in reality, fetch the Role and check rules
            // For this example, we assume the Role has a rule granting get on secrets
            // which is non-compliant
            return &admissionv1.AdmissionResponse{
                Allowed: false,
                Result: &metav1.Status{
                    Message: "RoleBinding grants unauthorized access to secrets for ServiceAccount app-sa",
                    Reason:  metav1.StatusReasonForbidden,
                    Code:    http.StatusForbidden,
                },
            }, nil
        }
    }

    return &admissionv1.AdmissionResponse{Allowed: true}, nil
}

func main() {
    port := flag.Int("port", 8443, "Port to listen on")
    tlsCert := flag.String("tls-cert", "/etc/webhook/cert.pem", "Path to TLS certificate")
    tlsKey := flag.String("tls-key", "/etc/webhook/key.pem", "Path to TLS private key")
    flag.Parse()

    // Validate flags
    if *port <= 0 || *port > 65535 {
        fmt.Fprintf(os.Stderr, "Invalid port: %d\n", *port)
        os.Exit(1)
    }

    // Define admission handler
    admit := func(review admissionv1.AdmissionReview) *admissionv1.AdmissionResponse {
        req := review.Request
        // Only handle RoleBinding resources
        if req.Resource.Resource != "rolebindings" {
            return &admissionv1.AdmissionResponse{Allowed: true}
        }

        // Deserialize the RoleBinding from the request
        var roleBinding rbacv1.RoleBinding
        if err := json.Unmarshal(req.Object.Raw, &roleBinding); err != nil {
            return &admissionv1.AdmissionResponse{
                Allowed: false,
                Result: &metav1.Status{
                    Message: fmt.Sprintf("Failed to unmarshal RoleBinding: %v", err),
                    Code:    http.StatusBadRequest,
                },
            }
        }

        // Validate the RoleBinding
        resp, err := validateRoleBinding(&roleBinding)
        if err != nil {
            return &admissionv1.AdmissionResponse{
                Allowed: false,
                Result: &metav1.Status{
                    Message: fmt.Sprintf("Validation error: %v", err),
                    Code:    http.StatusInternalServerError,
                },
            }
        }
        return resp
    }

    // Register HTTP handler
    http.HandleFunc("/validate-rolebinding", func(w http.ResponseWriter, r *http.Request) {
        serveAdmission(w, r, admit)
    })

    // Start TLS server
    fmt.Printf("Starting admission webhook on port %d...\n", *port)
    if err := http.ListenAndServeTLS(fmt.Sprintf(":%d", *port), *tlsCert, *tlsKey, nil); err != nil {
        fmt.Fprintf(os.Stderr, "Failed to start server: %v\n", err)
        os.Exit(1)
    }
}

// rbac_benchmark.go
// Benchmarks RBAC misconfiguration detection time across Kubernetes 1.31 and 1.32
// using the new kubectl auth reconcile and legacy check methods.
package main

import (
    "context"
    "fmt"
    "math/rand"
    "sync"
    "time"

    rbacv1 "k8s.io/api/rbac/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes/fake"
)

// generateMockRoleBindings creates n mock RoleBindings, 10% of which are misconfigured
func generateMockRoleBindings(n int, namespace string) []rbacv1.RoleBinding {
    bindings := make([]rbacv1.RoleBinding, 0, n)
    for i := 0; i < n; i++ {
        isMisconfigured := rand.Float64() < 0.1 // 10% misconfigured
        rules := []rbacv1.PolicyRule{
            {
                Verbs:     []string{"get", "list"},
                Resources: []string{"configmaps"},
            },
        }
        if isMisconfigured {
            // Add unauthorized secrets access
            rules = append(rules, rbacv1.PolicyRule{
                Verbs:     []string{"get"},
                Resources: []string{"secrets"},
            })
        }

        // Create a mock Role
        role := rbacv1.Role{
            ObjectMeta: metav1.ObjectMeta{
                Name:      fmt.Sprintf("role-%d", i),
                Namespace: namespace,
            },
            Rules: rules,
        }

        // Create RoleBinding referencing the Role
        binding := rbacv1.RoleBinding{
            ObjectMeta: metav1.ObjectMeta{
                Name:      fmt.Sprintf("binding-%d", i),
                Namespace: namespace,
            },
            RoleRef: rbacv1.RoleRef{
                APIGroup: "rbac.authorization.k8s.io",
                Kind:     "Role",
                Name:     role.Name,
            },
            Subjects: []rbacv1.Subject{
                {
                    Kind:      "ServiceAccount",
                    Name:      "app-sa",
                    Namespace: namespace,
                },
            },
        }
        bindings = append(bindings, binding)
    }
    return bindings
}

// legacyCheckRBAC simulates the legacy (pre-1.32) RBAC check method
func legacyCheckRBAC(bindings []rbacv1.RoleBinding) int {
    misconfigured := 0
    for _, binding := range bindings {
        // Simulate fetching the Role (legacy method does this per binding)
        time.Sleep(10 * time.Microsecond) // Simulate API latency
        for _, rule := range binding.RoleRef.Name {
            // Simplified check: if binding has secrets in rules, count as misconfigured
            // In real legacy check, this would fetch the Role and check rules
            _ = rule // Simulate work
        }
        // Check if binding allows secrets (simplified)
        if binding.RoleRef.Name != "" {
            misconfigured++
        }
    }
    return misconfigured
}

// newCheckRBAC simulates the Kubernetes 1.32 optimized RBAC check with batch fetching
func newCheckRBAC(bindings []rbacv1.RoleBinding) int {
    misconfigured := 0
    // Batch fetch all Roles (simulated in 1.32)
    time.Sleep(50 * time.Microsecond) // Simulate batch API call
    for _, binding := range bindings {
        // Simplified check: no per-binding API call
        if binding.RoleRef.Name != "" {
            misconfigured++
        }
    }
    return misconfigured
}

func main() {
    rand.Seed(time.Now().UnixNano())
    namespace := "default"
    bindingCounts := []int{1000, 5000, 10000, 50000}

    fmt.Println("RBAC Misconfiguration Check Benchmark (K8s 1.31 vs 1.32)")
    fmt.Println("=========================================================")
    fmt.Printf("%-10s | %-15s | %-15s | %-10s\n", "Bindings", "Legacy (1.31)", "New (1.32)", "Speedup")
    fmt.Println("---------------------------------------------------------")

    for _, count := range bindingCounts {
        bindings := generateMockRoleBindings(count, namespace)

        // Benchmark legacy check
        start := time.Now()
        legacyMisconfig := legacyCheckRBAC(bindings)
        legacyDuration := time.Since(start)

        // Benchmark new check
        start = time.Now()
        newMisconfig := newCheckRBAC(bindings)
        newDuration := time.Since(start)

        // Calculate speedup
        speedup := float64(legacyDuration) / float64(newDuration)

        fmt.Printf("%-10d | %-15s | %-15s | %-10.2fx\n",
            count,
            legacyDuration.Round(time.Microsecond).String(),
            newDuration.Round(time.Microsecond).String(),
            speedup,
        )

        // Validate results match
        if legacyMisconfig != newMisconfig {
            fmt.Printf("Warning: Result mismatch for %d bindings: legacy=%d new=%d\n", count, legacyMisconfig, newMisconfig)
        }
    }

    // Concurrent check benchmark (1.32 feature)
    fmt.Println("\nConcurrent Check Benchmark (K8s 1.32 only)")
    fmt.Println("===========================================")
    bindings := generateMockRoleBindings(10000, namespace)
    workerCounts := []int{1, 2, 4, 8, 16}

    for _, workers := range workerCounts {
        start := time.Now()
        var wg sync.WaitGroup
        misconfigured := 0
        mutex := sync.Mutex{}

        // Split bindings into worker chunks
        chunkSize := len(bindings) / workers
        for i := 0; i < workers; i++ {
            wg.Add(1)
            go func(i int) {
                defer wg.Done()
                startIdx := i * chunkSize
                endIdx := startIdx + chunkSize
                if i == workers-1 {
                    endIdx = len(bindings)
                }
                localMisconfig := 0
                for _, binding := range bindings[startIdx:endIdx] {
                    // Simulate check
                    time.Sleep(1 * time.Microsecond)
                    if binding.RoleRef.Name != "" {
                        localMisconfig++
                    }
                }
                mutex.Lock()
                misconfigured += localMisconfig
                mutex.Unlock()
            }(i)
        }
        wg.Wait()
        duration := time.Since(start)
        fmt.Printf("Workers: %-2d | Duration: %-15s | Misconfigured: %d\n", workers, duration.Round(time.Microsecond).String(), misconfigured)
    }
}

# GitHub Actions step for RBAC CI check
- name: Run kubectl auth reconcile
  run: |
    kubectl auth reconcile -f rbac/ --dry-run=server -o patch > rbac-patch.yaml
    if [ -s rbac-patch.yaml ]; then
      echo "❌ RBAC misconfigurations detected:"
      cat rbac-patch.yaml
      exit 1
    else
      echo "✅ RBAC configuration matches desired state"
    fi

# Kubernetes audit policy for RBAC resources
apiVersion: audit.k8s.io/v1
kind: Policy
rules:
  - level: Metadata
    resources:
      - group: rbac.authorization.k8s.io
        resources: ["roles", "rolebindings", "clusterroles", "clusterrolebindings"]
    verbs: ["create", "update", "delete", "patch"]

# Gatekeeper policy to block RoleBindings granting secrets access
apiVersion: gatekeeper.sh/v1beta1
kind: ConstraintTemplate
metadata:
  name: k8sblocksecretaccess
spec:
  crd:
    spec:
      names:
        kind: K8sBlockSecretAccess
  targets:
    - target: admission.k8s.io
      rego: |
        package k8sblocksecretaccess

        violation[{"msg": msg}] {
          input.review.object.kind == "RoleBinding"
          roleRef := input.review.object.roleRef
          roleRef.kind == "Role"
          # Fetch the Role (simulated, in practice use data.inventory)
          # For brevity, check if the RoleBinding references a Role that allows secrets
          input.review.object.metadata.labels["allow-secrets"] != "true"
          msg := sprintf("RoleBinding %v grants unauthorized secrets access", [input.review.object.metadata.name])
        }