feat: implement realm setup job and enhance realm configuration for Keycloak

| Portal backend checks bearer tokens and roles | Done (audience check still to switch on) |

| Portal real login (Authorization Code + PKCE, server-side session, logout) | Next: the Portal still has a placeholder login screen |

| API products: roles on `oeg`, scope gate, provisioner client (realm import) | Done for new realms |

| API products: provisioner permissions, registration policies, existing realms | Next: a realm job |

| API products: provisioner permissions, registration policies, later product changes | Next: a job that runs on every install and upgrade |

| OEG checks tokens and reads `organization` | Next: OEG does not check tokens yet |

| HTTPS / Ingress for Keycloak | Not yet: `http` + NodePort today; `keycloak.hostname` can already point at an HTTPS address behind your own ingress |

"""Brings realm oop to the state ADR-0048 needs, whether or not Keycloak imported it from this

chart. Every step only does what is missing, so running the job again changes nothing.

"""Finishes realm oop after Keycloak imported it from this chart (ADR-0048).

Reads /config/config.json (rendered by the chart) and these environment variables:

KEYCLOAK_URL, KEYCLOAK_ADMIN_USERNAME, KEYCLOAK_ADMIN_PASSWORD, PROVISIONER_CLIENT_SECRET.

The import cannot create the provisioner's fine-grained permissions or change Keycloak's

registration policies, and never runs again, so a product added to global.apiProducts later

would not reach the realm. This does all three. Every step only does what is missing, so

running it again changes nothing.

Reads /config/config.json (rendered by the chart) and the environment variables KEYCLOAK_URL,

KEYCLOAK_ADMIN_USERNAME and KEYCLOAK_ADMIN_PASSWORD.

"""

import json

import os

import sys

import time

import urllib.error

import urllib.parse

import urllib.request

from typing import Any

CONFIG_FILE = os.environ.get("CONFIG_FILE", "/config/config.json")

# What the provisioner may hold in realm-management, and nothing else.

PROVISIONER_ROLES = {"create-client", "view-users", "query-users"}

POLICY_TYPE = "org.keycloak.services.clientregistration.policy.ClientRegistrationPolicy"

# What the portal puts on the clients it registers: the `oeg` audience, and the claims that

# tie a client to its creator.

REGISTRATION_MAPPER_TYPES = ["oidc-audience-mapper", "oidc-hardcoded-claim-mapper"]

class Keycloak:

    def __init__(self, url: str, realm: str, admin: str, password: str) -> None:

        self.url, self.realm, self._admin, self._password = url.rstrip("/"), realm, admin, password

        self.url, self.realm = url.rstrip("/"), realm

        self._admin, self._password = admin, password

        self._token = ""

        self._renew_at = 0.0

            self._renew_at = time.monotonic() + 30

        return self._token

    def call(self, method: str, path: str, body: object = None) -> object:

    def call(self, method: str, path: str, body: Any = None) -> Any:

        request = urllib.request.Request(

            f"{self.url}/admin/realms/{self.realm}{path}",

            data=None if body is None else json.dumps(body).encode(),

            raise SystemExit(f"{method} {path} -> {exc.code} {exc.read().decode()[:300]}") from exc

        return json.loads(raw) if raw else None

    def get(self, path: str) -> object:

    def get(self, path: str) -> Any:

        return self.call("GET", path)

    def internal_id_of(self, client_id: str) -> str:

        """The internal id of a client; the job cannot do anything useful without it."""

        found = self.get(f"/clients?clientId={urllib.parse.quote(client_id)}")

        if not found:

            raise SystemExit(f"client {client_id} is missing from realm {self.realm}")

        return found[0]["id"]

    def wait_until_ready(self) -> None:

        for _ in range(120):

            try:

    print(f"{action:8} {what}", flush=True)

def ensure_oeg(kc: Keycloak, cfg: dict) -> str:

    """The `oeg` client, under the id the portal is configured with. Returns that id."""

    found = kc.get("/clients?clientId=oeg")

    if not found:

        kc.call(

            "POST",

            "/clients",

            {

                "id": cfg["oegId"],

                "clientId": "oeg",

                "name": "Open Exposure Gateway",

                "publicClient": False,

                "standardFlowEnabled": False,

                "implicitFlowEnabled": False,

                "directAccessGrantsEnabled": False,

                "serviceAccountsEnabled": False,

            },

        )

        say("created", f"client oeg ({cfg['oegId']})")

    elif found[0]["id"] != cfg["oegId"]:

def check_oeg(kc: Keycloak, cfg: dict) -> str:

    """The `oeg` client must have the id the portal is configured with."""

    found = kc.internal_id_of("oeg")

    if found != cfg["oegId"]:

        raise SystemExit(

            f"client oeg has the id {found[0]['id']}, not {cfg['oegId']}: "

            f"set keycloak.realms.oop.oeg.internalId to {found[0]['id']}"

            f"client oeg has the id {found}, not {cfg['oegId']}: "

            f"set keycloak.realms.oop.oeg.internalId to {found}"

        )

    else:

        say("ok", "client oeg")

    return cfg["oegId"]

    return found

def ensure_product_roles(kc: Keycloak, oeg: str, cfg: dict) -> None:

        if product["name"] in have:

            say("ok", f"role {product['name']}")

            continue

        description = f"API product {product['name']} (ADR-0048)"

        kc.call(

            "POST",

            f"/clients/{oeg}/roles",

            {"name": product["name"], "description": f"API product {product['name']} (ADR-0048)"},

            "POST", f"/clients/{oeg}/roles", {"name": product["name"], "description": description}

        )

        say("created", f"role {product['name']}")

def ensure_scopes_gated(kc: Keycloak, oeg: str, cfg: dict) -> None:

    """Every CAMARA scope exists, is a default scope of the portal client, and is issued only

    to holders of its product's role."""

    portal = kc.get(f"/clients?clientId={cfg['portalClientId']}")[0]["id"]

    portal = kc.internal_id_of(cfg["portalClientId"])

    defaults = {s["name"] for s in kc.get(f"/clients/{portal}/default-client-scopes")}

    scope_ids = {s["name"]: s["id"] for s in kc.get("/client-scopes")}

    for product in cfg["products"]:

                        "attributes": {"include.in.token.scope": "true"},

                    },

                )

                scope_ids[scope] = next(

                    s["id"] for s in kc.get("/client-scopes") if s["name"] == scope

                )

                scope_ids = {s["name"]: s["id"] for s in kc.get("/client-scopes")}

                say("created", f"scope {scope}")

            mapped = {r["name"] for r in kc.get(

                f"/client-scopes/{scope_ids[scope]}/scope-mappings/clients/{oeg}"

            )}

            if product["name"] not in mapped:

                kc.call(

                    "POST",

                    f"/client-scopes/{scope_ids[scope]}/scope-mappings/clients/{oeg}",

                    [role],

                )

            mappings = f"/client-scopes/{scope_ids[scope]}/scope-mappings/clients/{oeg}"

            if product["name"] not in {r["name"] for r in kc.get(mappings)}:

                kc.call("POST", mappings, [role])

                say("gated", f"{scope} -> {product['name']}")

            if scope not in defaults:

                kc.call("PUT", f"/clients/{portal}/default-client-scopes/{scope_ids[scope]}")

                say("added", f"{scope} to {cfg['portalClientId']}")

PROVISIONER_ROLES = {"create-client", "view-users", "query-users"}

# Would expose every client's secret, the oop-portal-admin one included.

FORBIDDEN_PROVISIONER_ROLES = {"view-clients"}

def ensure_provisioner(kc: Keycloak, cfg: dict, secret: str) -> str:

    """The client the portal creates API clients with, and exactly the roles it may hold.

    Returns the client's id."""

    client_id = cfg["provisionerClientId"]

    found = kc.get(f"/clients?clientId={client_id}")

    if found:

        say("ok", f"client {client_id}")

        provisioner = found[0]["id"]

    else:

        kc.call(

            "POST",

            "/clients",

            {

                "clientId": client_id,

                "name": "OOP Portal API client provisioning",

                "description": "Service account for creating API clients (ADR-0048); no login flows.",

                "publicClient": False,

                "clientAuthenticatorType": "client-secret",

                "secret": secret,

                "standardFlowEnabled": False,

                "implicitFlowEnabled": False,

                "directAccessGrantsEnabled": False,

                "serviceAccountsEnabled": True,

            },

        )

        provisioner = kc.get(f"/clients?clientId={client_id}")[0]["id"]

        say("created", f"client {client_id}")

def ensure_provisioner_roles(kc: Keycloak, provisioner: str) -> None:

    """The provisioner holds the roles it needs and no other realm-management role: view-clients,

    for one, would expose every client's secret."""

    account = kc.get(f"/clients/{provisioner}/service-account-user")["id"]

    management = kc.get("/clients?clientId=realm-management")[0]["id"]

    management = kc.internal_id_of("realm-management")

    path = f"/users/{account}/role-mappings/clients/{management}"

    held = {role["name"]: role for role in kc.get(path)}

    available = {role["name"]: role for role in kc.get(f"/clients/{management}/roles")}

    for name in sorted(PROVISIONER_ROLES - held.keys()):

        kc.call("POST", path, [available[name]])

        say("granted", f"{name} to {client_id}")

    for name in sorted(FORBIDDEN_PROVISIONER_ROLES & held.keys()):

        say("granted", f"{name} to the provisioner")

    for name in sorted(held.keys() - PROVISIONER_ROLES):

        kc.call("DELETE", path, [held[name]])

        say("removed", f"{name} from {client_id}")

    return provisioner

        say("removed", f"{name} from the provisioner")

def check_provisioner_secret(kc: Keycloak, cfg: dict, secret: str) -> None:

    body = urllib.parse.urlencode(

        {

            "grant_type": "client_credentials",

            "client_id": cfg["provisionerClientId"],

            "client_secret": secret,

def ensure_permissions(kc: Keycloak, oeg: str, provisioner: str, cfg: dict) -> None:

    """Fine-grained admin permissions: the provisioner may give and take the product roles,

    and read the `oeg` client (to get the roles' ids), and nothing more."""

    permissions_client = kc.internal_id_of("admin-permissions")

    authz = f"/clients/{permissions_client}/authz/resource-server"

    def find(name: str) -> dict | None:

        found = kc.get(f"{authz}/policy?name={urllib.parse.quote(name)}&first=0&max=50")

        return next((p for p in found if p["name"] == name), None)

    who = find("is-provisioner")

    if who is None:

        kc.call(

            "POST", f"{authz}/policy/client", {"name": "is-provisioner", "clients": [provisioner]}

        )

        who = find("is-provisioner")

        say("created", "policy is-provisioner")

    products = {product["name"] for product in cfg["products"]}

    role_ids = [role["id"] for role in kc.get(f"/clients/{oeg}/roles") if role["name"] in products]

    wanted: dict[str, dict] = {

        "provisioner-map-roles-on-users": {"resourceType": "Users", "scopes": ["map-roles"]},

        "provisioner-map-product-roles": {

            "resourceType": "Roles",

            "resources": role_ids,

            "scopes": ["map-role"],

        },

        "provisioner-view-oeg": {"resourceType": "Clients", "resources": [oeg], "scopes": ["view"]},

    }

    ).encode()

    try:

        urllib.request.urlopen(

            f"{kc.url}/realms/{kc.realm}/protocol/openid-connect/token", body, timeout=10

        ).close()

    except urllib.error.HTTPError as exc:

        raise SystemExit(

            f"{cfg['provisionerClientId']} refuses the secret in the credentials Secret "

            f"({exc.code}): change the client's secret in Keycloak to match it"

        ) from exc

    say("ok", "provisioner secret")

    for name, permission in wanted.items():

        body = {**permission, "name": name, "policies": [who["id"]]}

        existing = find(name)

        if existing is None:

            kc.call("POST", f"{authz}/permission/scope", body)

            say("created", f"permission {name}")

            continue

        resources = permission.get("resources")

        current = [r["name"] for r in kc.get(f"{authz}/policy/{existing['id']}/resources")]

        if resources is not None and sorted(current) != sorted(resources):

            kc.call(

                "PUT", f"{authz}/permission/scope/{existing['id']}", {**body, "id": existing["id"]}

            )

            say("updated", f"permission {name}")

        else:

            say("ok", f"permission {name}")

def ensure_registration_policies(kc: Keycloak, cfg: dict) -> None:

    """Registration with a token (the provisioner's) may use our mappers and CAMARA scopes.

    Adds to Keycloak's own policies and keeps everything they already allow."""

    components = kc.get(f"/components?type={urllib.parse.quote(POLICY_TYPE)}")

    def allow(provider: str, key: str, wanted: list[str], what: str) -> None:

        policy = next(

            c for c in components if c["providerId"] == provider and c["subType"] == "authenticated"

        )

        have = policy["config"].get(key, [])

        missing = [item for item in wanted if item not in have]

        if not missing:

            say("ok", f"registration allows {what}")

            return

        policy["config"][key] = have + missing

        kc.call("PUT", f"/components/{policy['id']}", policy)

        say("allowed", f"{len(missing)} {what} for registration")

    scopes = [scope for product in cfg["products"] for scope in product["scopes"]]

    allow(

        "allowed-protocol-mappers",

        "allowed-protocol-mapper-types",

        REGISTRATION_MAPPER_TYPES,

        "mapper types",

    )

    allow("allowed-client-templates", "allowed-client-scopes", scopes, "scopes")

def main() -> None:

    cfg = json.load(open(CONFIG_FILE))

    with open(CONFIG_FILE) as file:

        cfg = json.load(file)

    kc = Keycloak(

        os.environ["KEYCLOAK_URL"],

        cfg["realm"],

        os.environ["KEYCLOAK_ADMIN_PASSWORD"],

    )

    kc.wait_until_ready()

    oeg = ensure_oeg(kc, cfg)

    oeg = check_oeg(kc, cfg)

    ensure_product_roles(kc, oeg, cfg)

    ensure_scopes_gated(kc, oeg, cfg)

    secret = os.environ["PROVISIONER_CLIENT_SECRET"]

    ensure_provisioner(kc, cfg, secret)

    check_provisioner_secret(kc, cfg, secret)

    provisioner = kc.internal_id_of(cfg["provisionerClientId"])

    ensure_provisioner_roles(kc, provisioner)

    ensure_permissions(kc, oeg, provisioner, cfg)

    ensure_registration_policies(kc, cfg)

    say("done", "")

if __name__ == "__main__":

    sys.exit(main())

    main()

{{/*

Pod spec of the realm setup job. It waits for Keycloak itself, so it can start with the rollout.

*/}}

{{- define "keycloak.realmSetupJobSpec" -}}

backoffLimit: 6

activeDeadlineSeconds: 1800

template:

  metadata:

    labels:

      app.kubernetes.io/name: {{ .Values.name }}-realm-setup

      app.kubernetes.io/instance: {{ .Release.Name }}

  spec:

    restartPolicy: OnFailure

    containers:

      - name: realm-setup

        image: {{ .Values.realmSetup.image | quote }}

        imagePullPolicy: IfNotPresent

        command: ["python", "/config/realm-setup.py"]

        env:

          - name: PYTHONDONTWRITEBYTECODE

            value: "1"

          - name: KEYCLOAK_URL

            value: "http://{{ include "keycloak.serviceName" . }}:{{ include "keycloak.servicePort" . }}"

          - name: KEYCLOAK_ADMIN_USERNAME

            valueFrom:

              secretKeyRef:

                name: {{ .Values.secretName }}

                key: admin-username

          - name: KEYCLOAK_ADMIN_PASSWORD

            valueFrom:

              secretKeyRef:

                name: {{ .Values.secretName }}

                key: admin-password

        securityContext:

          runAsNonRoot: true

          runAsUser: 1000

          allowPrivilegeEscalation: false

          readOnlyRootFilesystem: true

          capabilities:

            drop: ["ALL"]

        volumeMounts:

          - name: config

            mountPath: /config

            readOnly: true

    volumes:

      - name: config

        configMap:

          name: {{ .Values.name }}-realm-setup

{{- end }}

{{- /*

Finishes realm oop after the import (files/realm-setup.py): the provisioner's permissions,

Keycloak's registration policies, and any product added to global.apiProducts later. Runs on

every install and upgrade.

*/ -}}

{{- if and (include "keycloak.authEnabled" .) .Values.realms.oop.enabled }}

{{- with .Values.realms.oop }}

{{- $config := dict

      "realm" "oop"

      "portalClientId" .portal.clientId

      "oegId" .oeg.internalId

      "provisionerClientId" .provisioner.clientId

      "products" (include "keycloak.apiProducts" $ | fromJsonArray)

    | toPrettyJson }}

{{- $spec := include "keycloak.realmSetupJobSpec" $ }}

apiVersion: v1

kind: ConfigMap

metadata:

  name: {{ $.Values.name }}-realm-setup

  namespace: {{ $.Release.Namespace }}

  labels:

    {{- include "keycloak.labels" $ | nindent 4 }}

data:

  config.json: |

    {{- $config | nindent 4 }}

  realm-setup.py: |

    {{- $.Files.Get "files/realm-setup.py" | nindent 4 }}

---

apiVersion: batch/v1

kind: Job

metadata:

  # The revision makes every install, upgrade and rollback run it again.

  name: {{ $.Values.name }}-realm-setup-{{ printf "%s%d" $spec ($.Release.Revision | int) | sha256sum | trunc 8 }}

  namespace: {{ $.Release.Namespace }}

  labels:

    {{- include "keycloak.labels" $ | nindent 4 }}

spec:

  {{- $spec | nindent 2 }}

{{- end }}

{{- end }}

  {{- if .enabled }}

  {{- $portalUrl := include "keycloak.portalBaseUrl" $ }}

  {{- $apiProducts := include "keycloak.apiProducts" $ | fromJsonArray }}

  {{- $productNames := list }}

  {{- range $apiProducts }}{{ $productNames = append $productNames .name }}{{ end }}

  oop.json: |

    {

      "realm": "oop",

          "lastName": "Admin",

          "attributes": { "organization": [ {{ .sampleUsers.organization | toJson }} ] },

          "credentials": [ { "type": "password", "value": "${DEV_USER_PASSWORD}", "temporary": false } ],

          "realmRoles": ["default-roles-oop", "platform-admin"]

          "realmRoles": ["default-roles-oop", "platform-admin"],

          "clientRoles": { "oeg": {{ $productNames | toJson }} }

        },

        {

          "username": "dev-provider",

          "lastName": "Provider",

          "attributes": { "organization": [ {{ .sampleUsers.organization | toJson }} ] },

          "credentials": [ { "type": "password", "value": "${DEV_USER_PASSWORD}", "temporary": false } ],

          "realmRoles": ["default-roles-oop"]

          "realmRoles": ["default-roles-oop"],

          "clientRoles": { "oeg": {{ $productNames | toJson }} }

        },

        {

          "username": "dev-operator",

          "lastName": "Operator",

          "attributes": { "organization": [ {{ .sampleUsers.organization | toJson }} ] },

          "credentials": [ { "type": "password", "value": "${DEV_USER_PASSWORD}", "temporary": false } ],

          "realmRoles": ["default-roles-oop", "operator"]

          "realmRoles": ["default-roles-oop", "operator"],

          "clientRoles": { "oeg": {{ $productNames | toJson }} }

        }

        {{- end }}

      ]

resources: {}

# The job that finishes realm oop after the import (files/realm-setup.py). Any image with

# Python 3 will do: the script uses only its standard library.

realmSetup:

  image: python:3.13.16-alpine

# OpenOP login pages (themes/oop): the Portal's look with the OpenOP logo, mounted from a

# ConfigMap. Realm oop is imported with it; a realm that already exists keeps its old theme

# until you pick `oop` under Realm settings → Themes → Login theme.