Mutating webhook for kubectl annotations for any resource

main.go

package main

import (
	"context"
	"crypto/tls"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"net/http"
	"os"
	"os/signal"
	"syscall"

	admissionv1beta1 "k8s.io/api/admission/v1beta1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/serializer"
	"k8s.io/klog/v2"
)

var (
	port     int
	certFile string
	keyFile  string
	urlPath  string
)

var (
	runtimeScheme = runtime.NewScheme()
	codecs        = serializer.NewCodecFactory(runtimeScheme)
	deserializer  = codecs.UniversalDeserializer()
)

func main() {
	flag.IntVar(&port, "port", 443, "Webhook server port.")
	flag.StringVar(&certFile, "cert-file", "/etc/webhook/certs/cert.pem", "File containing the x509 Certificate for HTTPS.")
	flag.StringVar(&keyFile, "key-file", "/etc/webhook/certs/key.pem", "File containing the x509 private key to --tlsCertFile.")
	flag.StringVar(&urlPath, "url-path", "/mutate", "URL path to serve requests on")
	flag.Parse()

	pair, err := tls.LoadX509KeyPair(certFile, keyFile)
	if err != nil {
		klog.Errorf("Failed to load key pair: %v", err)
	}

	server := &http.Server{
		Addr:      fmt.Sprintf(":%d", port),
		TLSConfig: &tls.Config{Certificates: []tls.Certificate{pair}},
	}

	mux := http.NewServeMux()
	mux.HandleFunc(urlPath, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		var body []byte
		if r.Body != nil {
			if data, err := io.ReadAll(r.Body); err == nil {
				body = data
			}
		}
		if len(body) == 0 {
			klog.Errorf("empty body")
			http.Error(w, "empty body", http.StatusBadRequest)
			return
		}

		contentType := r.Header.Get("Content-Type")
		if contentType != "application/json" {
			klog.Errorf("Content-Type=%s, expect application/json", contentType)
			http.Error(w, "invalid Content-Type, expect `application/json`", http.StatusUnsupportedMediaType)
			return
		}

		var response *admissionv1beta1.AdmissionResponse
		request := admissionv1beta1.AdmissionReview{}
		if _, _, err := deserializer.Decode(body, nil, &request); err != nil {
			klog.Errorf("Can't decode body: %v", err)
			response = &admissionv1beta1.AdmissionResponse{
				Result: &metav1.Status{
					Message: err.Error(),
				},
			}
		} else {
			fmt.Println(r.URL.Path)
			response = mutate(&request)
		}

		review := admissionv1beta1.AdmissionReview{}
		if response != nil {
			review.Response = response
			if request.Request != nil {
				review.Response.UID = request.Request.UID
			}
		}

		resp, err := json.Marshal(review)
		if err != nil {
			klog.Errorf("Can't encode response: %v", err)
			http.Error(w, fmt.Sprintf("could not encode response: %v", err), http.StatusInternalServerError)
			return
		}
		klog.Infof("Ready to write reponse ...")
		if _, err := w.Write(resp); err != nil {
			klog.Errorf("Can't write response: %v", err)
			http.Error(w, fmt.Sprintf("could not write response: %v", err), http.StatusInternalServerError)
		}
	}))
	server.Handler = mux

	go func() {
		if err := server.ListenAndServeTLS("", ""); err != nil {
			klog.Errorf("Failed to listen and serve webhook server: %v", err)
		}
	}()

	klog.Info("Server started")

	signalChan := make(chan os.Signal, 1)
	signal.Notify(signalChan, syscall.SIGINT, syscall.SIGTERM)
	<-signalChan

	klog.Infof("Got OS shutdown signal, shutting down webhook server gracefully...")
	if err := server.Shutdown(context.Background()); err != nil {
		klog.Errorf("Server shutdown failed: %v", err)
	}
}

func mutate(request *admissionv1beta1.AdmissionReview) *admissionv1beta1.AdmissionResponse {
	data := make(map[string]interface{})
	if err := json.Unmarshal(request.Request.Object.Raw, &data); err != nil {
		return &admissionv1beta1.AdmissionResponse{Result: &metav1.Status{Message: err.Error()}}
	}
	obj := unstructured.Unstructured{Object: data}
	annotations := obj.GetAnnotations()
	annotations["kubectl.kubenetes.io/last-applied-configuration"] = ""

	jsonPatch := []map[string]interface{}{
		{
			"op":    "replace",
			"path":  "/metadata/annotations",
			"value": annotations,
		},
	}
	patchBytes, err := json.Marshal(jsonPatch)
	if err != nil {
		return &admissionv1beta1.AdmissionResponse{Result: &metav1.Status{Message: err.Error()}}
	}

	return &admissionv1beta1.AdmissionResponse{
		Allowed: true,
		Patch:   patchBytes,
		PatchType: func() *admissionv1beta1.PatchType {
			pt := admissionv1beta1.PatchTypeJSONPatch
			return &pt
		}(),
	}
}