Webhooks

Admission webhooks are HTTP callbacks that receive admission requests and can mutate or validate Kubernetes API objects. They’re the most flexible way to implement custom admission control logic without modifying the Kubernetes API server.

Types of Webhooks

Mutating Webhooks

Mutating webhooks can modify objects before they’re persisted. They run first and can:

Add default values
Inject sidecar containers
Set default labels or annotations
Modify resource specifications

Validating Webhooks

Validating webhooks can only accept or reject requests. They run after mutating webhooks and can:

Enforce security policies
Validate business rules
Check compliance requirements
Reject invalid configurations

Creating a Validating Webhook

1. Create the Webhook Server

Here’s a simple validating webhook in Python using Flask:

from flask import Flask, request, jsonify
import base64
import json

app = Flask(__name__)

@app.route('/validate', methods=['POST'])
def validate():
    admission_review = request.json
    pod = admission_review['request']['object']
    
    # Check if pod runs as root
    containers = pod.get('spec', {}).get('containers', [])
    for container in containers:
        security_context = container.get('securityContext', {})
        if security_context.get('runAsUser') == 0:
            return jsonify({
                'apiVersion': 'admission.k8s.io/v1',
                'kind': 'AdmissionReview',
                'response': {
                    'uid': admission_review['request']['uid'],
                    'allowed': False,
                    'status': {
                        'message': 'Pods cannot run as root user (UID 0)'
                    }
                }
            })
    
    # Allow the request
    return jsonify({
        'apiVersion': 'admission.k8s.io/v1',
        'kind': 'AdmissionReview',
        'response': {
            'uid': admission_review['request']['uid'],
            'allowed': True
        }
    })

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=443, ssl_context='adhoc')

2. Deploy the Webhook Server

apiVersion: apps/v1
kind: Deployment
metadata:
  name: validating-webhook
  namespace: default
spec:
  replicas: 2
  selector:
    matchLabels:
      app: validating-webhook
  template:
    metadata:
      labels:
        app: validating-webhook
    spec:
      containers:
      - name: webhook
        image: my-registry/validating-webhook:latest
        ports:
        - containerPort: 443
---
apiVersion: v1
kind: Service
metadata:
  name: validating-webhook
  namespace: default
spec:
  selector:
    app: validating-webhook
  ports:
  - port: 443
    targetPort: 443

3. Create TLS Certificate

Webhooks require HTTPS. Generate a certificate:

# Create certificate signing request
cat <<EOF | kubectl apply -f -
apiVersion: certificates.k8s.io/v1
kind: CertificateSigningRequest
metadata:
  name: validating-webhook.default
spec:
  request: $(cat validating-webhook.csr | base64 | tr -d '\n')
  signerName: kubernetes.io/legacy-unknown
  usages:
  - digital signature
  - key encipherment
  - server auth
EOF

# Approve the CSR
kubectl certificate approve validating-webhook.default

# Get the certificate
kubectl get csr validating-webhook.default -o jsonpath='{.status.certificate}' | base64 -d > validating-webhook.crt

4. Create ValidatingWebhookConfiguration

apiVersion: admissionregistration.k8s.io/v1
kind: ValidatingWebhookConfiguration
metadata:
  name: pod-security-validator
webhooks:
- name: pod-security-validator.default.svc
  clientConfig:
    service:
      name: validating-webhook
      namespace: default
      path: "/validate"
    caBundle: <base64-encoded-ca-cert>
  rules:
  - apiGroups: [""]
    apiVersions: ["v1"]
    operations: ["CREATE", "UPDATE"]
    resources: ["pods"]
  admissionReviewVersions: ["v1", "v1beta1"]
  sideEffects: None
  failurePolicy: Fail

Creating a Mutating Webhook

Example: Inject Sidecar Container

from flask import Flask, request, jsonify
import base64
import json

app = Flask(__name__)

@app.route('/mutate', methods=['POST'])
def mutate():
    admission_review = request.json
    pod = admission_review['request']['object']
    
    # Add sidecar container
    sidecar = {
        'name': 'log-collector',
        'image': 'fluent/fluent-bit:latest',
        'volumeMounts': [{
            'name': 'varlog',
            'mountPath': '/var/log'
        }]
    }
    
    if 'containers' not in pod['spec']:
        pod['spec']['containers'] = []
    
    pod['spec']['containers'].append(sidecar)
    
    # Create patch
    patch = [{
        'op': 'add',
        'path': '/spec/containers/-',
        'value': sidecar
    }]
    
    patch_base64 = base64.b64encode(json.dumps(patch).encode()).decode()
    
    return jsonify({
        'apiVersion': 'admission.k8s.io/v1',
        'kind': 'AdmissionReview',
        'response': {
            'uid': admission_review['request']['uid'],
            'allowed': True,
            'patchType': 'JSONPatch',
            'patch': patch_base64
        }
    })

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=443, ssl_context='adhoc')

MutatingWebhookConfiguration

apiVersion: admissionregistration.k8s.io/v1
kind: MutatingWebhookConfiguration
metadata:
  name: sidecar-injector
webhooks:
- name: sidecar-injector.default.svc
  clientConfig:
    service:
      name: mutating-webhook
      namespace: default
      path: "/mutate"
    caBundle: <base64-encoded-ca-cert>
  rules:
  - apiGroups: [""]
    apiVersions: ["v1"]
    operations: ["CREATE"]
    resources: ["pods"]
  admissionReviewVersions: ["v1", "v1beta1"]
  sideEffects: None
  failurePolicy: Ignore

Webhook Configuration Options

Rules

Define which resources trigger the webhook:

rules:
- apiGroups: ["apps"]           # API group
  apiVersions: ["v1"]            # API version
  operations: ["CREATE", "UPDATE"]  # Operations to intercept
  resources: ["deployments"]     # Resource type
  scope: "Namespaced"            # Namespaced or Cluster-scoped

Failure Policy

What happens if the webhook fails:

Fail - Reject the request (use for security-critical webhooks)
Ignore - Allow the request (use for non-critical mutations)

failurePolicy: Fail  # or Ignore

Side Effects

Declare if the webhook has side effects (external API calls, etc.):

sideEffects: None  # or Unknown, NoneOnDryRun

Timeout

How long to wait for webhook response:

timeoutSeconds: 10

Object Selectors

Only intercept specific objects:

objectSelector:
  matchLabels:
    webhook: enabled

Namespace Selectors

Only intercept resources in specific namespaces:

namespaceSelector:
  matchLabels:
    admission: enabled

Best Practices

Make webhooks fast - Target < 100ms response time
Use failure policy wisely - Fail for security, Ignore for convenience
Handle timeouts gracefully - Set appropriate timeouts
Idempotent mutations - Mutations should be safe to apply multiple times
Test thoroughly - Webhooks affect all matching requests
Monitor performance - Track latency and error rates
Use dry-run - Test webhooks with --dry-run=server
Document side effects - Clearly declare if webhooks have side effects

Common Patterns

Security Validation

Reject pods that don’t meet security requirements:

def validate_security(pod):
    violations = []
    
    # Check for privileged containers
    for container in pod['spec']['containers']:
        if container.get('securityContext', {}).get('privileged'):
            violations.append('Privileged containers not allowed')
    
    # Check for host network
    if pod['spec'].get('hostNetwork'):
        violations.append('Host network not allowed')
    
    return violations

Default Injection

Add default values to all pods:

def inject_defaults(pod):
    # Add default labels
    if 'labels' not in pod['metadata']:
        pod['metadata']['labels'] = {}
    pod['metadata']['labels']['managed-by'] = 'webhook'
    
    # Set default service account
    if 'serviceAccountName' not in pod['spec']:
        pod['spec']['serviceAccountName'] = 'default'

Resource Defaults

Set default resource requests and limits:

def inject_resources(pod):
    for container in pod['spec']['containers']:
        if 'resources' not in container:
            container['resources'] = {
                'requests': {'cpu': '100m', 'memory': '128Mi'},
                'limits': {'cpu': '500m', 'memory': '512Mi'}
            }

Troubleshooting

Check webhook status:

kubectl get validatingwebhookconfiguration
kubectl get mutatingwebhookconfiguration

Test webhook manually:

# Create a test pod
kubectl run test-pod --image=nginx --dry-run=server -o yaml | kubectl apply -f -

# Check webhook logs
kubectl logs -n default deployment/validating-webhook

Verify webhook is called:

Check API server logs for webhook calls:

kubectl logs -n kube-system kube-apiserver-<node> | grep webhook

Common Issues

Certificate problems - Ensure CA bundle is correct
Network issues - Verify service is reachable
Timeout errors - Increase timeout or optimize webhook
JSON patch errors - Validate patch format for mutating webhooks