Simple validation webhook with Python

import logging
import os
from flask import Flask, jsonify, request

app = Flask('webhook')
app.logger.addHandler(logging.StreamHandler())
app.logger.setLevel(logging.DEBUG)

#Health check
@app.route("/healthz", methods=['GET'])
def ping():
  return jsonify({'message': 'ok'})

REQUIRED_LABELS = ['ngaddons/ownerId', 'ngaddons/webexRoomId', 'ngaddons/appName']

@app.route('/validate', methods=['POST'])
def deployment_webhook():
  r = request.get_json()

  req = r.get('request', {})
  try:
    if not req:
      return send_response(False, '<no uid>', "Invalid request, no payload.request found")

    uid = req.get("uid", '')
    app.logger.debug(f"+ uid: {uid}")
    if not uid:
      return send_response(False, '<no uid>', "Invalid request, no payload.request.uid found")

    labels = req.get("object", {}).get("metadata", {}).get("labels")
  if 'ngaddons/bypass' in labels:
      return send_response(True, uid, "Request bypassed as 'ngaddons/bypass' is set")

    missing = [ l for l in REQUIRED_LABELS if l not in labels ]
    app.logger.debug(f"+ missing: {missing}")
    if missing:
      return send_response(False, uid, f"Missing labels: {missing}")

  except Exception as e:
    return send_response(False, uid, f"Webhook exception: {e}")

  #Send OK
  return send_response(True, uid, "Request has required labels")


#Function to respond back to the Admission Controller
def send_response(allowed, uid, message):
  return jsonify({
      "apiVersion": "admission.k8s.io/v1",
      "kind": "AdmissionReview",
      "response": {
        "allowed": allowed,
        "uid": uid,
        "status": {"message": message}
    }
  })


if __name__ == "__main__":
  ca_crt = '/etc/ssl/ca.crt'
  ca_key = '/etc/ssl/ca.key'
  app.run(ssl_context=(ca_crt, ca_key), port=5000, host='0.0.0.0', debug=True)

# Image: ashoka007/check-labels:0.1
FROM python:3.8-slim
WORKDIR /app
COPY requirements.txt /app
RUN pip install -r requirements.txt
COPY app.py /app
CMD python app.py

apiVersion: admissionregistration.k8s.io/v1
kind: ValidatingWebhookConfiguration
metadata:
  name: ngaddons-check-labels
  namespace: check-labels
webhooks:
  - name: ngaddons.check-labels.webhook
    failurePolicy: Fail
    sideEffects: None
    admissionReviewVersions: ["v1","v1beta1"]
    namespaceSelector:
      matchLabels:
        ngaddons/validation-webhooks: enabled
    rules:
      - apiGroups: ["apps", ""]
        resources:
          - "deployments"
          - "pods"
        apiVersions:
          - "*"
        operations:
          - CREATE
    clientConfig:
      service:
        name: ${WEBHOOK_SERVICE_NAME} # to be substituted
        namespace: ${WEBHOOK_NAMESPACE} # to be substituted
        path: /validate/
      caBundle: ${CA_BUNDLE} # to be substituted

# Configuration parameters are the key and DNS match is required
[ req ]
default_bits       = 2048
distinguished_name = req_distinguished_name
req_extensions     = req_ext
prompt             = no
[ req_distinguished_name ]
countryName                 = IN
stateOrProvinceName         = KAR
localityName                = BGL
organizationName            = ACME INC
commonName                  = check-labels 0.1
[ req_ext ]
subjectAltName = @alt_names
[alt_names]
DNS.1   = ${WEBHOOK_SERVICE_NAME}.${WEBHOOK_NAMESPACE}.svc

❯ openssl req -x509 -newkey rsa:4096 -nodes -out certs/ca.crt -keyout certs/ca.key -days 365 -config conf/ext.cnf -extensions req_ext

❯ kubectl create secret tls webhook-secret --cert=certs/ca.crt --key=certs/ca.key --namespace=${WEBHOOK_NAMESPACE}

❯ kubectl create namespace ${WEBHOOK_NAMESPACE}

apiVersion: apps/v1
kind: Deployment
metadata:
  name: webhook-dep
  namespace: ${WEBHOOK_NAMESPACE}
  labels:
    app: webhook
spec:
  replicas: 2
  selector:
    matchLabels:
      app: webhook
  template:
    metadata:
      labels:
        app: webhook
    spec:
      containers:
        - name: webhook-container
          image: ashoka007/check-labels:0.1
          volumeMounts:
            - mountPath: /etc/ssl
              name: webhook-certs
              readOnly: true
      volumes:
      - name: webhook-certs
        secret:
          secretName: webhook-secret

❯ kubectl expose deployment/webhook-dep --name=${WEBHOOK_SERVICE_NAME} --namespace=${WEBHOOK_NAMESPACE} --port=443 --target-port=5000

❯ kubectl create namespace demo-ns

# normal scenario
❯ kubectl run testpod --image=nginx -n demo-ns
pod/testpod created

# start enforcing validation
❯ kubectl label namespace demo-ns -l ngaddons/validation-webhooks=enabled

# validation fail
❯ kubectl run testpod2 --image=nginx -n demo-ns
Error from server: admission webhook "ngaddons.check-labels.webhook" denied the request: Missing labels: ['ngaddons/ownerCec', 'ngaddons/webexRoomId', 'ngaddons/appName']

# validation pass
❯ kubectl run testpod3 --image=nginx -n demo-ns -l=ngaddons/ownerId=ram -l=ngaddons/webexRoomId=rams-room-id -l=ngaddons/appName=rams-test-app
pod/testpod3 created

# validation bypass
❯ kubectl run testpod4 --image=nginx -n demo-ns -l=ngaddons/bypassed=1
pod/testpod4 created

❯ kubectl delete validatingwebhookconfigurations.admissionregistration.k8s.io check-labels