knete

A lightweight Python framework for building Kubernetes controllers and admission webhooks.

Features

• Controllers — watch any Kubernetes resource and react to create/update/delete events
• Validating webhooks — reject requests that violate policy
• Mutating webhooks — patch objects on the way in
• Composite mutating webhooks — fan-out multiple mutators to a single endpoint, accumulating all patches in one round-trip
• Decorator-style webhooks — register plain functions instead of subclasses
• Manifest generator — python -m knete <file.py> generates all k8s manifests (Deployment, Service, RBAC, WebhookConfigurations) including cert-manager TLS support

Installation

pip install knete

Quick start

Controller

from knete import Controller, Manager

class PodController(Controller):
    class Meta:
        resource  = "v1/pods"
        namespace = "default"

    def on_create(self, name, namespace, labels, spec):
        print(f"Pod created: {namespace}/{name}")

    def on_delete(self, name):
        print(f"Pod deleted: {name}")

Manager().register(PodController).start()

Validating webhook

from knete import ValidatingWebhook, AdmissionResponse, WebhookServer

class NoLatestTagWebhook(ValidatingWebhook):
    class Meta:
        resource   = "apps/v1/deployments"
        operations = ["CREATE", "UPDATE"]
        name       = "no-latest-tag"

    def validate(self, name, spec) -> AdmissionResponse:
        for c in spec.get("template", {}).get("spec", {}).get("containers", []):
            if c["image"].endswith(":latest") or ":" not in c["image"]:
                return AdmissionResponse.deny(f"image {c['image']!r} uses :latest tag")
        return AdmissionResponse.allow()

server = WebhookServer(cert_file="tls.crt", key_file="tls.key")
server.register(NoLatestTagWebhook)
server.start()

Mutating webhook

from knete import MutatingWebhook, AdmissionResponse

class InjectLabelWebhook(MutatingWebhook):
    class Meta:
        resource   = "v1/pods"
        operations = ["CREATE"]
        name       = "inject-label"

    def mutate(self, labels) -> AdmissionResponse:
        if "managed-by" in labels:
            return AdmissionResponse.allow()
        return AdmissionResponse.patch([
            {"op": "add", "path": "/metadata/labels/managed-by", "value": "knete"},
        ])

Multiple validators

Register each validating webhook separately — Kubernetes calls them in parallel, so total latency is max(validators) rather than sum(validators):

server.register(NoLatestTagWebhook)     # POST /validate/no-latest-tag
server.register(RequiredLabelsWebhook)  # POST /validate/required-labels

Decorator style

Skip the class entirely and register plain functions directly on the server:

server = WebhookServer(cert_file="tls.crt", key_file="tls.key")

@server.validate("apps/v1/deployments", operations=["CREATE", "UPDATE"], name="no-latest-tag")
def no_latest_tag(name, spec) -> AdmissionResponse:
    for c in spec.get("template", {}).get("spec", {}).get("containers", []):
        if c["image"].endswith(":latest") or ":" not in c["image"]:
            return AdmissionResponse.deny(f"image {c['image']!r} uses :latest tag")

@server.mutate("v1/pods", operations=["CREATE"], name="inject-label")
def inject_label(labels) -> AdmissionResponse:
    if "managed-by" not in labels:
        return AdmissionResponse.patch([
            {"op": "add", "path": "/metadata/labels/managed-by", "value": "knete"},
        ])

server.start()

Both styles can be mixed freely. name defaults to the function name if omitted.

Manifest generator

Generate all Kubernetes manifests for your operator or webhook server:

# Operator
python -m knete examples/my_operator.py --name my-operator --image my-operator:v1

# Webhook with cert-manager TLS (generates a self-signed ClusterIssuer)
python -m knete examples/my_webhook.py --name my-webhook --image my-webhook:v1 --cert-manager

# Webhook using an existing ClusterIssuer
python -m knete examples/my_webhook.py --name my-webhook --image my-webhook:v1 --cert-manager my-issuer

This writes k8s/ manifests and a Containerfile ready to build. Apply with:

kubectl apply -f k8s/

Examples

Requirements

• Python 3.10+
• cert-manager (optional, for webhook TLS)