Keycloak on Kubernetes with operators, step by step

# "demo" realm: users, groups and clients used by the Envoy Gateway examples.
# KeycloakRealmImport only creates a realm that does not exist yet: it neither
# updates nor deletes it. Secrets come from the "demo-realm-secrets" Secret
# through placeholders, they never appear in this file.
apiVersion: k8s.keycloak.org/v2beta1
kind: KeycloakRealmImport
metadata:
  name: demo
  namespace: keycloak
spec:
  keycloakCRName: keycloak
  placeholders:
    ALICE_PASSWORD:
      secret:
        name: demo-realm-secrets
        key: alice-password
    BOB_PASSWORD:
      secret:
        name: demo-realm-secrets
        key: bob-password
    INTRANET_CLIENT_SECRET:
      secret:
        name: demo-realm-secrets
        key: intranet-client-secret
    REPORTING_JOB_SECRET:
      secret:
        name: demo-realm-secrets
        key: reporting-job-secret
    INVENTORY_SYNC_SECRET:
      secret:
        name: demo-realm-secrets
        key: inventory-sync-secret
    AUDIT_JOB_SECRET:
      secret:
        name: demo-realm-secrets
        key: audit-job-secret
  realm:
    realm: demo
    displayName: Demo
    enabled: true
    sslRequired: external
    bruteForceProtected: true

    # User profile: two business attributes on top of the default ones,
    # managed by administrators and visible to the user.
    components:
      org.keycloak.userprofile.UserProfileProvider:
        - providerId: declarative-user-profile
          subComponents: {}
          config:
            kc.user.profile.config:
              - |
                {
                  "attributes": [
                    {"name": "username", "displayName": "Username",
                     "validations": {"length": {"min": 3, "max": 255}, "username-prohibited-characters": {}, "up-username-not-idn-homograph": {}},
                     "permissions": {"view": ["admin", "user"], "edit": ["admin", "user"]}, "multivalued": false},
                    {"name": "email", "displayName": "Email",
                     "validations": {"email": {}, "length": {"max": 255}},
                     "required": {"roles": ["user"]},
                     "permissions": {"view": ["admin", "user"], "edit": ["admin", "user"]}, "multivalued": false},
                    {"name": "firstName", "displayName": "First name",
                     "validations": {"length": {"max": 255}, "person-name-prohibited-characters": {}},
                     "required": {"roles": ["user"]},
                     "permissions": {"view": ["admin", "user"], "edit": ["admin", "user"]}, "multivalued": false},
                    {"name": "lastName", "displayName": "Last name",
                     "validations": {"length": {"max": 255}, "person-name-prohibited-characters": {}},
                     "required": {"roles": ["user"]},
                     "permissions": {"view": ["admin", "user"], "edit": ["admin", "user"]}, "multivalued": false},
                    {"name": "employee_id", "displayName": "Employee ID",
                     "permissions": {"view": ["admin", "user"], "edit": ["admin"]}, "multivalued": false},
                    {"name": "department", "displayName": "Department",
                     "permissions": {"view": ["admin", "user"], "edit": ["admin"]}, "multivalued": false}
                  ],
                  "groups": [
                    {"name": "user-metadata", "displayHeader": "User metadata", "displayDescription": "Attributes, which refer to user metadata"}
                  ]
                }

    roles:
      client:
        intranet:
          - name: user
            description: Access to the intranet application
        echo-api:
          - name: read
            description: Read items
          - name: write
            description: Create items

    # Members of "staff" get the intranet "user" role, contractors do not.
    groups:
      - name: staff
        clientRoles:
          intranet:
            - user
      - name: contractors

    users:
      - username: alice
        enabled: true
        email: alice@example.com
        emailVerified: true
        firstName: Alice
        lastName: Martin
        attributes:
          employee_id:
            - E-1001
          department:
            - Finance
        credentials:
          - type: password
            value: "${ALICE_PASSWORD}"
            temporary: false
        groups:
          - /staff
      - username: bob
        enabled: true
        email: bob@example.com
        emailVerified: true
        firstName: Bob
        lastName: Durand
        attributes:
          employee_id:
            - C-2002
          department:
            - IT
        credentials:
          - type: password
            value: "${BOB_PASSWORD}"
            temporary: false
        groups:
          - /contractors
      # Service accounts of the two machine clients, with their API roles.
      - username: service-account-reporting-job
        enabled: true
        serviceAccountClientId: reporting-job
        clientRoles:
          echo-api:
            - read
      - username: service-account-inventory-sync
        enabled: true
        serviceAccountClientId: inventory-sync
        clientRoles:
          echo-api:
            - read
            - write
      - username: service-account-audit-job
        enabled: true
        serviceAccountClientId: audit-job

    clients:
      # Confidential client used by the OIDC filter of Envoy Gateway.
      - clientId: intranet
        name: Intranet (legacy application behind Envoy Gateway)
        enabled: true
        publicClient: false
        clientAuthenticatorType: client-secret
        secret: "${INTRANET_CLIENT_SECRET}"
        standardFlowEnabled: true
        implicitFlowEnabled: false
        directAccessGrantsEnabled: false
        serviceAccountsEnabled: false
        redirectUris:
          - https://intranet.kube.localhost/oauth2/callback
        attributes:
          post.logout.redirect.uris: https://intranet.kube.localhost/*
          # The OIDC filter of Envoy sends a PKCE challenge: make it mandatory.
          pkce.code.challenge.method: S256
        protocolMappers:
          # The access token is meant for the intranet: aud = intranet.
          - name: audience-intranet
            protocol: openid-connect
            protocolMapper: oidc-audience-mapper
            config:
              included.client.audience: intranet
              access.token.claim: "true"
              id.token.claim: "false"
              introspection.token.claim: "true"
          # Groups and HR attributes are only returned by the UserInfo endpoint:
          # tokens stay small (Envoy stores them in cookies).
          - name: groups-userinfo
            protocol: openid-connect
            protocolMapper: oidc-group-membership-mapper
            config:
              claim.name: groups
              full.path: "false"
              access.token.claim: "false"
              id.token.claim: "false"
              userinfo.token.claim: "true"
              introspection.token.claim: "false"
          - name: employee-id-userinfo
            protocol: openid-connect
            protocolMapper: oidc-usermodel-attribute-mapper
            config:
              user.attribute: employee_id
              claim.name: employee_id
              jsonType.label: String
              access.token.claim: "false"
              id.token.claim: "false"
              userinfo.token.claim: "true"
              introspection.token.claim: "false"
          - name: department-userinfo
            protocol: openid-connect
            protocolMapper: oidc-usermodel-attribute-mapper
            config:
              user.attribute: department
              claim.name: department
              jsonType.label: String
              access.token.claim: "false"
              id.token.claim: "false"
              userinfo.token.claim: "true"
              introspection.token.claim: "false"

      # Resource server: no login flow, it only defines the API roles.
      # Tokens carrying one of its roles get aud = echo-api.
      - clientId: echo-api
        name: Echo API
        enabled: true
        publicClient: false
        standardFlowEnabled: false
        implicitFlowEnabled: false
        directAccessGrantsEnabled: false
        serviceAccountsEnabled: false

      # Machine clients (client credentials grant). audit-job has no API role.
      - clientId: reporting-job
        name: Reporting job (read only)
        enabled: true
        publicClient: false
        clientAuthenticatorType: client-secret
        secret: "${REPORTING_JOB_SECRET}"
        standardFlowEnabled: false
        implicitFlowEnabled: false
        directAccessGrantsEnabled: false
        serviceAccountsEnabled: true
      - clientId: inventory-sync
        name: Inventory synchronization (read and write)
        enabled: true
        publicClient: false
        clientAuthenticatorType: client-secret
        secret: "${INVENTORY_SYNC_SECRET}"
        standardFlowEnabled: false
        implicitFlowEnabled: false
        directAccessGrantsEnabled: false
        serviceAccountsEnabled: true
      - clientId: audit-job
        name: Audit job (no API role)
        enabled: true
        publicClient: false
        clientAuthenticatorType: client-secret
        secret: "${AUDIT_JOB_SECRET}"
        standardFlowEnabled: false
        implicitFlowEnabled: false
        directAccessGrantsEnabled: false
        serviceAccountsEnabled: true

# Workloads of the demo: ordinary Deployments and Services, with no
# authentication code or configuration. The security lives in the routes
# (intranet-route.yaml, echo-api-route.yaml).
# userinfo-authz: external authorization service (ext_authz). No route
# points to it: only the Envoy proxies call it.
apiVersion: apps/v1
kind: Deployment
metadata:
  name: userinfo-authz
  labels:
    app.kubernetes.io/name: userinfo-authz
spec:
  replicas: 1
  selector:
    matchLabels:
      app.kubernetes.io/name: userinfo-authz
  template:
    metadata:
      labels:
        app.kubernetes.io/name: userinfo-authz
    spec:
      automountServiceAccountToken: false
      securityContext:
        runAsNonRoot: true
        runAsUser: 10001
        runAsGroup: 10001
        seccompProfile:
          type: RuntimeDefault
      containers:
        - name: app
          image: python:3.14.8-alpine3.24
          command: ["python", "-u", "/app/userinfo_authz.py"]
          env:
            # Internal URL: the call does not leave the cluster.
            - name: USERINFO_URL
              value: http://keycloak-service.keycloak.svc.cluster.local:8080/realms/demo/protocol/openid-connect/userinfo
            - name: CACHE_TTL_SECONDS
              value: "10"
            - name: PYTHONDONTWRITEBYTECODE
              value: "1"
          ports:
            - name: http
              containerPort: 8080
          readinessProbe:
            httpGet:
              path: /healthz
              port: http
          resources:
            requests:
              cpu: 10m
              memory: 32Mi
            limits:
              memory: 64Mi
          securityContext:
            allowPrivilegeEscalation: false
            readOnlyRootFilesystem: true
            capabilities:
              drop: ["ALL"]
          volumeMounts:
            - name: app
              mountPath: /app
              readOnly: true
      volumes:
        - name: app
          configMap:
            name: demo-apps
---
apiVersion: v1
kind: Service
metadata:
  name: userinfo-authz
spec:
  selector:
    app.kubernetes.io/name: userinfo-authz
  ports:
    - name: http
      port: 8080
      targetPort: http
---
# intranet: the legacy application.
apiVersion: apps/v1
kind: Deployment
metadata:
  name: intranet
  labels:
    app.kubernetes.io/name: intranet
spec:
  replicas: 1
  selector:
    matchLabels:
      app.kubernetes.io/name: intranet
  template:
    metadata:
      labels:
        app.kubernetes.io/name: intranet
    spec:
      automountServiceAccountToken: false
      securityContext:
        runAsNonRoot: true
        runAsUser: 10001
        runAsGroup: 10001
        seccompProfile:
          type: RuntimeDefault
      containers:
        - name: app
          image: python:3.14.8-alpine3.24
          command: ["python", "-u", "/app/intranet.py"]
          env:
            - name: PYTHONDONTWRITEBYTECODE
              value: "1"
          ports:
            - name: http
              containerPort: 8080
          readinessProbe:
            httpGet:
              path: /healthz
              port: http
          resources:
            requests:
              cpu: 10m
              memory: 32Mi
            limits:
              memory: 64Mi
          securityContext:
            allowPrivilegeEscalation: false
            readOnlyRootFilesystem: true
            capabilities:
              drop: ["ALL"]
          volumeMounts:
            - name: app
              mountPath: /app
              readOnly: true
      volumes:
        - name: app
          configMap:
            name: demo-apps
---
apiVersion: v1
kind: Service
metadata:
  name: intranet
spec:
  selector:
    app.kubernetes.io/name: intranet
  ports:
    - name: http
      port: 8080
      targetPort: http
---
# echo-api: the REST API.
apiVersion: apps/v1
kind: Deployment
metadata:
  name: echo-api
  labels:
    app.kubernetes.io/name: echo-api
spec:
  replicas: 1
  selector:
    matchLabels:
      app.kubernetes.io/name: echo-api
  template:
    metadata:
      labels:
        app.kubernetes.io/name: echo-api
    spec:
      automountServiceAccountToken: false
      securityContext:
        runAsNonRoot: true
        runAsUser: 10001
        runAsGroup: 10001
        seccompProfile:
          type: RuntimeDefault
      containers:
        - name: app
          image: python:3.14.8-alpine3.24
          command: ["python", "-u", "/app/echo_api.py"]
          env:
            - name: PYTHONDONTWRITEBYTECODE
              value: "1"
          ports:
            - name: http
              containerPort: 8080
          readinessProbe:
            httpGet:
              path: /healthz
              port: http
          resources:
            requests:
              cpu: 10m
              memory: 32Mi
            limits:
              memory: 64Mi
          securityContext:
            allowPrivilegeEscalation: false
            readOnlyRootFilesystem: true
            capabilities:
              drop: ["ALL"]
          volumeMounts:
            - name: app
              mountPath: /app
              readOnly: true
      volumes:
        - name: app
          configMap:
            name: demo-apps
---
apiVersion: v1
kind: Service
metadata:
  name: echo-api
spec:
  selector:
    app.kubernetes.io/name: echo-api
  ports:
    - name: http
      port: 8080
      targetPort: http

"""Legacy intranet application.

It has no OIDC code. Like an application behind an Apache mod_auth_* module or a
web access management agent, it trusts the identity headers set by the gateway
in front of it, and only works when it is reached through that gateway.
"""
import html
import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

IDENTITY_HEADERS = [
    "X-Remote-User",
    "X-Remote-Name",
    "X-Remote-Email",
    "X-Remote-Groups",
    "X-Remote-Employee-Id",
    "X-Remote-Department",
]


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/healthz":
            return self.reply(200, "text/plain", "ok\n")

        identity = {name: self.headers.get(name) for name in IDENTITY_HEADERS}

        # Raw view of what the application receives, used by the tests.
        if self.path == "/whoami":
            body = {
                "identity": identity,
                "authorization_header": "Authorization" in self.headers,
                "x_forwarded_for": self.headers.get("X-Forwarded-For"),
                "headers": sorted(self.headers.keys()),
            }
            return self.reply(200, "application/json", json.dumps(body, indent=2) + "\n")

        if not identity["X-Remote-User"]:
            return self.reply(401, "text/plain", "No identity header: use the gateway.\n")

        rows = "".join(
            f"<tr><th>{html.escape(name)}</th><td>{html.escape(value or '-')}</td></tr>"
            for name, value in identity.items()
        )
        page = f"""<!doctype html>
<html><head><meta charset="utf-8"><title>Legacy intranet</title>
<style>body{{font-family:sans-serif;margin:2rem}}th{{text-align:left;padding-right:1rem}}</style>
</head><body>
<h1>Welcome {html.escape(identity["X-Remote-Name"] or identity["X-Remote-User"])}</h1>
<p>This application has no OpenID Connect code: it reads the headers set by Envoy Gateway.</p>
<table>{rows}</table>
<p><a href="/logout">Sign out</a></p>
</body></html>
"""
        return self.reply(200, "text/html; charset=utf-8", page)

    def reply(self, status, content_type, body):
        data = body.encode()
        self.send_response(status)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)


if __name__ == "__main__":
    ThreadingHTTPServer(("", 8080), Handler).serve_forever()

"""External authorization service for Envoy Gateway (HTTP ext_authz).

For each request, Envoy sends the Authorization header to this service. The
service calls the OpenID Connect UserInfo endpoint of Keycloak with the access
token and answers:
  - 200 with X-Remote-* headers built from the UserInfo claims: Envoy copies
    them to the request sent to the application (headersToBackend);
  - 401 when Keycloak rejects the token, for example because the session was
    ended: the request does not reach the application.
Answers are cached for CACHE_TTL_SECONDS per token.
"""
import hashlib
import json
import os
import time
import urllib.error
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

USERINFO_URL = os.environ["USERINFO_URL"]
CACHE_TTL_SECONDS = int(os.environ.get("CACHE_TTL_SECONDS", "10"))

# UserInfo claim -> request header
CLAIM_HEADERS = {
    "groups": "X-Remote-Groups",
    "employee_id": "X-Remote-Employee-Id",
    "department": "X-Remote-Department",
}

cache = {}  # sha256(token) -> (expiry, headers)


def userinfo_headers(token):
    key = hashlib.sha256(token.encode()).hexdigest()
    cached = cache.get(key)
    if cached and cached[0] > time.monotonic():
        return cached[1]

    request = urllib.request.Request(
        USERINFO_URL,
        headers={"Authorization": f"Bearer {token}", "Accept": "application/json"},
    )
    with urllib.request.urlopen(request, timeout=3) as response:
        claims = json.load(response)

    headers = {}
    for claim, header in CLAIM_HEADERS.items():
        value = claims.get(claim)
        if isinstance(value, list):
            value = ",".join(str(v) for v in value)
        if value is not None:
            headers[header] = str(value)

    for k in [k for k, (expiry, _) in cache.items() if expiry <= time.monotonic()]:
        cache.pop(k, None)
    cache[key] = (time.monotonic() + CACHE_TTL_SECONDS, headers)
    return headers


class Handler(BaseHTTPRequestHandler):
    def check(self):
        if self.path == "/healthz":
            return self.reply(200, {})
        if not self.path.startswith("/authz"):
            return self.reply(404, {})

        authorization = self.headers.get("Authorization", "")
        if not authorization.lower().startswith("bearer "):
            return self.reply(401, {}, "Missing access token.\n")
        try:
            headers = userinfo_headers(authorization[7:])
        except urllib.error.HTTPError as error:
            if error.code in (401, 403):
                return self.reply(401, {}, "Session ended. Sign in again: /logout\n")
            return self.reply(503, {}, "UserInfo endpoint error.\n")
        except (urllib.error.URLError, TimeoutError, ValueError):
            return self.reply(503, {}, "UserInfo endpoint unreachable.\n")
        return self.reply(200, headers)

    # Envoy forwards the method of the original request.
    do_GET = do_POST = do_PUT = do_PATCH = do_DELETE = do_HEAD = do_OPTIONS = check

    def reply(self, status, headers, body=""):
        data = body.encode()
        self.send_response(status)
        for name, value in headers.items():
            self.send_header(name, value)
        self.send_header("Content-Type", "text/plain")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        if self.command != "HEAD":
            self.wfile.write(data)


if __name__ == "__main__":
    ThreadingHTTPServer(("", 8080), Handler).serve_forever()