Admission Webhook 开发

1. Admission Webhooks 是什么

Webhook 指的是中途拦截请求做定制化处理，包含两种类型：

2. Webhook 的执行时机

Admission Webhook 本质是 API Server 的一个 webhook 调用。总结出两种 webhook 的触发时机：

1. 总体来说，Admission Webhooks 机制是在 API Server 接收到请求、执行授权检查后和将请求持久化到 etcd 之前触发。
2. 详细地看，Mutating Webhook 在 Validating Webhook 之前触发，因此可以先对资源进行修改，然后再进行验证。

3. Admission Webhook 的运行流程

Pod 创建流程中 Webhook 执行时机：

4. Mutating Webhook 返回格式

Mutating Webhook 的返回格式是一个 AdmissionReview 对象，与请求使用相同的类型，只是填充了 response 部分。

完整结构

JSON{
  "kind": "AdmissionReview",
  "apiVersion": "admission.k8s.io/v1",
  "response": {
    "uid": "请求中的 uid，必须原样返回",
    "allowed": true,
    "patchType": "JSONPatch",
    "patch": "Base64 编码的 JSON Patch",
    "status": {
      "code": 200,
      "message": "可选的状态信息"
    }
  }
}

各字段说明

三种典型响应

① 放行并修改（Mutate）

JSON{
  "kind": "AdmissionReview",
  "apiVersion": "admission.k8s.io/v1",
  "response": {
    "uid": "abc-123",
    "allowed": true,
    "patchType": "JSONPatch",
    "patch": "W3sib3AiOiJhZGQiLCJwYXRoIjoiL21ldGFkYXRhL2xhYmVscy9lbnZpcm9ubWVudCIsInZhbHVlIjoicHJvZHVjdGlvbiJ9XQ=="
  }
}

② 直接拒绝

JSON{
  "kind": "AdmissionReview",
  "apiVersion": "admission.k8s.io/v1",
  "response": {
    "uid": "abc-123",
    "allowed": false,
    "status": {
      "code": 403,
      "message": "Pod 不允许使用 latest 标签"
    }
  }
}

③ 直接放行（不修改）

JSON{
  "kind": "AdmissionReview",
  "apiVersion": "admission.k8s.io/v1",
  "response": {
    "uid": "abc-123",
    "allowed": true
  }
}

JSON Patch (RFC 6902)[
  {"op": "add", "path": "/metadata/labels/environment", "value": "production"}
]

5. 代码示例

webhook.py

Pythonfrom flask import Flask, request, jsonify
import json
import ssl
import base64

app = Flask(__name__)


def create_patch(metadata):
    """
    创建 JSON Patch 以添加 'mutate' 注释。
    如果 metadata.annotations 不存在，则首先创建该路径。
    """
    if 'labels' in metadata:
        dic = metadata['labels']
    else:
        dic = {}

    patch = [
        {'op': 'add', 'path': '/metadata/labels', 'value': dic},
        {'op': 'add', 'path': '/metadata/labels/environment', 'value': 'production'}
    ]

    patch_json = json.dumps(patch)
    patch_base64 = base64.b64encode(patch_json.encode('utf-8')).decode('utf-8')
    return patch_base64


@app.route('/mutate', methods=['POST'])
def mutate():
    """
    处理 Mutating Webhook 的请求，对 Pod 对象应用 JSON Patch。
    """
    admission_review = request.get_json()

    if 'request' not in admission_review or 'object' not in admission_review['request']:
        return jsonify({
            'kind': 'AdmissionReview',
            'apiVersion': 'admission.k8s.io/v1',
            'response': {
                'allowed': False,
                'status': {'message': 'Invalid AdmissionReview format'}
            }
        })

    req = admission_review['request']
    metadata = req['object']['metadata']
    patch_json = create_patch(metadata)

    admission_response = {
        'kind': 'AdmissionReview',
        'apiVersion': 'admission.k8s.io/v1',
        'response': {
            'uid': req['uid'],
            'allowed': True,
            'patchType': 'JSONPatch',
            'patch': patch_json
        }
    }

    return jsonify(admission_response)


if __name__ == '__main__':
    context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    context.load_cert_chain('/certs/tls.crt', '/certs/tls.key')
    app.run(host='0.0.0.0', port=443, ssl_context=context)

6. Validating Webhook 代码示例

和 Mutating Webhook 不同，Validating Webhook 只做校验、不修改对象，响应中不需要 patch / patchType，只需要返回 allowed: true/false。

下面实现一个常见的安全策略：禁止 Pod 使用 :latest 镜像标签 + 禁止特权容器 + 禁止 hostPath 挂载。

validate_webhook.py

Pythonfrom flask import Flask, request, jsonify
import ssl

app = Flask(__name__)

# ========== 校验规则 ==========

ALLOWED_REGISTRIES = []  # 留空=不限制镜像仓库；填入 ["company.registry.io"] 则只允许该仓库

FORBIDDEN_TAGS = [":latest", ":LATEST", ""]   # 禁止的镜像标签

FORBIDDEN_HOST_PATHS = ["/", "/etc", "/var", "/proc", "/sys", "/dev"]  # 禁止的 hostPath 前缀


def validate_pod(pod):
    """
    对 Pod 对象执行校验，返回 (allowed, reason)。
    allowed=True 表示通过，allowed=False 表示拒绝，reason 为拒绝原因。
    """
    containers = pod.get('spec', {}).get('containers', [])
    init_containers = pod.get('spec', {}).get('initContainers', [])
    all_containers = containers + init_containers

    # ---- 1. 检查镜像标签 ----
    for container in all_containers:
        image = container.get('image', '')

        # 检查是否使用了禁止的标签
        for tag in FORBIDDEN_TAGS:
            if tag and image.endswith(tag):
                return False, f"容器 '{container['name']}' 使用了禁止的镜像标签: {image}（不允许 :latest 或无标签）"

        # 检查镜像仓库白名单（如果配置了）
        if ALLOWED_REGISTRIES:
            allowed = any(image.startswith(registry) for registry in ALLOWED_REGISTRIES)
            if not allowed:
                return False, f"容器 '{container['name']}' 的镜像 '{image}' 不在允许的仓库列表中"

    # ---- 2. 检查特权容器 ----
    for container in all_containers:
        security_context = container.get('securityContext', {})
        if security_context.get('privileged', False):
            return False, f"容器 '{container['name']}' 使用了特权模式（privileged: true），这是被禁止的"

        if security_context.get('runAsUser', -1) == 0:
            return False, f"容器 '{container['name']}' 以 root 用户运行（runAsUser: 0），这是被禁止的"

    # ---- 3. 检查 hostPath 挂载 ----
    volumes = pod.get('spec', {}).get('volumes', [])
    for volume in volumes:
        host_path = volume.get('hostPath', {})
        if host_path:
            path = host_path.get('path', '')
            for forbidden in FORBIDDEN_HOST_PATHS:
                if path == forbidden or path.startswith(forbidden + '/'):
                    return False, f"Pod 使用了危险的 hostPath 挂载: {path}（禁止挂载宿主机敏感目录）"

    # ---- 4. 检查 hostNetwork / hostPID / hostIPC ----
    pod_spec = pod.get('spec', {})
    if pod_spec.get('hostNetwork', False):
        return False, "Pod 使用了 hostNetwork: true，这是被禁止的"
    if pod_spec.get('hostPID', False):
        return False, "Pod 使用了 hostPID: true，这是被禁止的"
    if pod_spec.get('hostIPC', False):
        return False, "Pod 使用了 hostIPC: true，这是被禁止的"

    return True, ""


# ========== HTTP Handler ==========

@app.route('/validate', methods=['POST'])
def validate():
    """
    处理 Validating Webhook 的请求，对 Pod 对象执行安全策略校验。
    """
    admission_review = request.get_json()

    # 验证请求格式
    if 'request' not in admission_review or 'object' not in admission_review['request']:
        return jsonify({
            'kind': 'AdmissionReview',
            'apiVersion': 'admission.k8s.io/v1',
            'response': {
                'allowed': False,
                'status': {'message': 'Invalid AdmissionReview format'}
            }
        })

    req = admission_review['request']
    pod = req['object']

    # 执行校验
    allowed, reason = validate_pod(pod)

    if allowed:
        # 放行：不需要 patch，只返回 allowed: true
        admission_response = {
            'kind': 'AdmissionReview',
            'apiVersion': 'admission.k8s.io/v1',
            'response': {
                'uid': req['uid'],
                'allowed': True
            }
        }
    else:
        # 拒绝：返回 allowed: false + 拒绝原因
        admission_response = {
            'kind': 'AdmissionReview',
            'apiVersion': 'admission.k8s.io/v1',
            'response': {
                'uid': req['uid'],
                'allowed': False,
                'status': {
                    'code': 403,
                    'message': reason
                }
            }
        }

    print(f"[VALIDATE] allowed={allowed}, reason={reason or 'N/A'}")
    return jsonify(admission_response)


if __name__ == '__main__':
    context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    context.load_cert_chain('/certs/tls.crt', '/certs/tls.key')
    app.run(host='0.0.0.0', port=443, ssl_context=context)

Mutating vs Validating 代码对比

校验规则说明

ValidatingWebhookConfiguration

和 MutatingWebhookConfiguration 结构几乎一样，只是 kind 不同、path 指向 /validate：

YAMLapiVersion: admissionregistration.k8s.io/v1
kind: ValidatingWebhookConfiguration
metadata:
  name: example-validating-webhook
webhooks:
  - name: validate.example.webhook.com
    clientConfig:
      service:
        name: webhook-service
        namespace: default
        path: "/validate"                 # 指向 /validate 而非 /mutate
      caBundle: "..."                     # 同一个 CA 证书
    rules:
      - operations: ["CREATE", "UPDATE"]  # 创建和更新都校验
        apiGroups: [""]
        apiVersions: ["v1"]
        resources: ["pods"]
    admissionReviewVersions: ["v1"]
    sideEffects: None
    failurePolicy: Fail                   # webhook 不可用时拒绝请求（安全优先）
    timeoutSeconds: 5

两种 Webhook 部署在一起

实际项目中，Mutating 和 Validating 通常部署在同一个 Service 里，共用证书和 Pod，只是路径不同：

Python — 合并版本from flask import Flask, request, jsonify
import json, ssl, base64

app = Flask(__name__)

# ... create_patch() 和 validate_pod() 的代码同上 ...

@app.route('/mutate', methods=['POST'])
def mutate():
    # ... Mutating 逻辑 ...

@app.route('/validate', methods=['POST'])
def validate():
    # ... Validating 逻辑 ...

if __name__ == '__main__':
    context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    context.load_cert_chain('/certs/tls.crt', '/certs/tls.key')
    app.run(host='0.0.0.0', port=443, ssl_context=context)

对应的 K8s 资源关系：

7. 制作镜像

Dockerfile

DockerfileFROM python:3.9-slim
WORKDIR /app
COPY webhook.py .
RUN pip install Flask -i https://pypi.tuna.tsinghua.edu.cn/simple
CMD ["python", "webhook.py"]

构建 & 推送

Shelldocker tag webhook:latest uhub.service.ucloud.cn/mute-webhook/mute:latest
docker push uhub.service.ucloud.cn/mute-webhook/mute:latest

8. 生成证书

Shell — gen-webhook-cert.sh#!/bin/bash
set -e
SERVICE=webhook-service
NAMESPACE=default
SECRET=webhook-certs
DIR=certs

rm -rf ${DIR}
mkdir -p ${DIR} && cd ${DIR}

echo "===== 1. 生成 CA ====="
openssl genrsa -out ca.key 2048
openssl req -x509 -new -nodes -key ca.key \
  -subj "/CN=${SERVICE}.${NAMESPACE}.svc" \
  -days 36500 -out ca.crt

echo "===== 2. 生成 Webhook 服务证书 ====="
cat > server.cnf << EOF
[req]
default_bits = 2048
prompt = no
default_md = sha256
req_extensions = v3_req
distinguished_name = dn
[dn]
CN = ${SERVICE}.${NAMESPACE}.svc
[v3_req]
basicConstraints = CA:FALSE
keyUsage = critical, digitalSignature, keyEncipherment
extendedKeyUsage = serverAuth
subjectAltName = @alt_names
[alt_names]
DNS.1 = ${SERVICE}
DNS.2 = ${SERVICE}.${NAMESPACE}
DNS.3 = ${SERVICE}.${NAMESPACE}.svc
DNS.4 = ${SERVICE}.${NAMESPACE}.svc.cluster.local
EOF

openssl genrsa -out webhook.key 2048
openssl req -new -key webhook.key -out webhook.csr -config server.cnf
openssl x509 -req -in webhook.csr \
  -CA ca.crt -CAkey ca.key -CAcreateserial \
  -out webhook.crt -days 36500 \
  -extensions v3_req -extfile server.cnf

echo "===== 3. 创建 K8s Secret ====="
kubectl create secret tls ${SECRET} \
  --cert=webhook.crt \
  --key=webhook.key \
  --namespace=${NAMESPACE} \
  --dry-run=client -o yaml | kubectl apply -f -

echo ""
echo "===== 4. caBundle（用于 WebhookConfiguration）====="
cat ca.crt | base64 | tr -d '\n'
echo ""
echo "===== 完成！ ====="
ls -la ca.crt webhook.crt webhook.key

9. MutatingWebhookConfiguration

它就是整个机制的"关卡规则"。光有 webhook 程序运行着是不够的，K8s API Server 根本不知道它的存在，也不知道什么时候该把请求转发给它。MutatingWebhookConfiguration 就是告诉 API Server 的那张"通知单"。

YAMLapiVersion: admissionregistration.k8s.io/v1
kind: MutatingWebhookConfiguration
metadata:
  name: example-mutating-webhook
webhooks:
  - name: example.webhook.com          # webhook 的唯一标识名
    clientConfig:                       # 告诉 API Server 怎么找到你的 webhook
      service:
        name: webhook-service           # Service 名
        namespace: default              # 命名空间
        path: "/mutate"                 # 请求路径
      caBundle: "..."                   # CA 证书的 Base64
    rules:                              # 什么情况下触发这个 webhook
      - operations: ["CREATE"]          # 创建资源时触发
        apiGroups: [""]                 # 核心 API 组
        apiVersions: ["v1"]             # API 版本
        resources: ["pods"]             # 只对 Pod 生效
    admissionReviewVersions: ["v1"]     # 支持的 AdmissionReview 版本
    sideEffects: None                   # 声明 webhook 没有副作用

关键点逐个解释

整体流程