WIP: My first Operator and CRDs

What and why

In my current home server setup I have Keycloak to handle SSO, and as far as I know there isn’t a clean way to configure realms and create new clients without using the WebUI or manually calling the REST API.

This is a nice opportunity to get some experience implementing an operator and some CRDs.

Getting started

Since I don’t have any prior experience writing CRDs, I will ask qwen-coder for assistance.

Starting with CRDs

Asking qwen “Can you help me create a custom Kubernetes CRD for managing keycloak realms and OIDC clients?” spits out custom resource definitions and manifests to test them out.

With a bit of editing — mostly names and schema — we have a working first draft of two CRDs, one for Realms and one for Clients:

apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: realms.keycloak.organiccode.net
spec:
  group: keycloak.organiccode.net
  versions:
    - name: v1alpha1
      served: true
      storage: true
      schema:
        openAPIV3Schema:
          type: object
          properties:
            spec:
              type: object
              properties:
                realmName: { type: string }
                displayName: { type: string }
                enabled: { type: boolean }
  scope: Cluster
  names:
    plural: realms
    singular: realm
    kind: KeycloakRealm
    shortNames: [kr]

The operator

As I want to make the operator in Python — because I know Python, Python is a straight-forward PoC language, and it has a client library for integrating with Keycloak — qwen proposes to use Kopf (Kubernetes Operators in Python Framework).

import kopf
from kubernetes import client, config
import requests
import os
from urllib.parse import urljoin

config.load_incluster_config()

KEYCLOAK_URL = os.getenv("KEYCLOAK_URL", "http://keycloak:8080")
KEYCLOAK_ADMIN_USER = os.getenv("KEYCLOAK_ADMIN_USER", "admin")
KEYCLOAK_ADMIN_PASSWORD = os.getenv("KEYCLOAK_ADMIN_PASSWORD", "admin")


def get_keycloak_token():
    url = urljoin(KEYCLOAK_URL, "/auth/realms/master/protocol/openid-connect/token")
    data = {
        "grant_type": "password",
        "client_id": "admin-cli",
        "username": KEYCLOAK_ADMIN_USER,
        "password": KEYCLOAK_ADMIN_PASSWORD,
    }
    response = requests.post(url, data=data)
    response.raise_for_status()
    return response.json()["access_token"]


@kopf.on.create("keycloak.organiccode.net", "v1alpha1", "realms")
def create_realm(spec, name, namespace, **kwargs):
    token = get_keycloak_token()
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
    data = {
        "realm": spec["realmName"],
        "displayName": spec.get("displayName", spec["realmName"]),
        "enabled": spec.get("enabled", True),
    }
    url = urljoin(KEYCLOAK_URL, "/auth/admin/realms")
    response = requests.post(url, headers=headers, json=data)
    response.raise_for_status()
    return {"message": f"Realm {name} created successfully"}

Question is — how do we test this in a simple and fast manner? Worst case we spin up a minikube cluster and deploy Keycloak. Let’s push on and worry about testing later. :|

Bundling it up with Helm

helm create keycloak-realm-operator gets us the default Helm project structure. After adding/removing some files and cleaning up template variable names, we get something like this:

keycloak-realm-operator/
├── Chart.yaml
├── templates
│   ├── crds
│   │   ├── client.yaml
│   │   └── realm.yaml
│   ├── deployment.yaml
│   ├── _helpers.tpl
│   ├── roles.yaml
│   └── serviceaccount.yaml
└── values.yaml

Taking it for a test drive

docker build -t keycloak-realm-operator:latest .
minikube image load keycloak-realm-operator:latest
helm install keycloak-realm-operator keycloak-realm-operator

We now have a running pod. Looking at the pod logs, there are some exceptions due to missing permissions for the service account, but nothing too hard to hammer out. After sorting out the permissions, refactoring out the Keycloak API calls into their own module, and debugging some small inconsistencies between the client and the API, we have a working proof of concept.

Continuing

Now that we have a working shell, we can start to make improvements.

Creating secrets from kopf

When we create a new client in Keycloak, we want to inform the actual client of the client secret. Let’s create a wrapper for the Kubernetes python client.

class K8sClient:
    def __init__(self):
        config.load_incluster_config()
        self.core_api = client.CoreV1Api()
        self.custom_api = client.CustomObjectsApi()

    def create_secret(self, client_info, namespace):
        if client_info["clientAuthenticatorType"] != "client-secret":
            return
        secret_data = {"client_secret": client_info["secret"]}
        k8s_secret = client.V1Secret(
            metadata=client.V1ObjectMeta(
                name=f'{client_info["clientId"]}-secret', namespace=namespace
            ),
            type="Opaque",
            data=secret_data,
        )
        self.core_api.create_namespaced_secret(namespace, k8s_secret)

Users, the final piece of the puzzle

…work in progress.