titan-iac/services/keycloak/realm-settings-job.yaml

# services/keycloak/realm-settings-job.yaml
apiVersion: batch/v1
kind: Job
metadata:
  name: keycloak-realm-settings-1
  namespace: sso
spec:
  backoffLimit: 2
  template:
    spec:
      restartPolicy: OnFailure
      containers:
        - name: configure
          image: python:3.11-alpine
          env:
            - name: KEYCLOAK_URL
              value: http://keycloak.sso.svc.cluster.local:8080
            - name: KEYCLOAK_REALM
              value: atlas
            - name: KEYCLOAK_ADMIN_USER
              valueFrom:
                secretKeyRef:
                  name: keycloak-admin
                  key: username
            - name: KEYCLOAK_ADMIN_PASSWORD
              valueFrom:
                secretKeyRef:
                  name: keycloak-admin
                  key: password
            - name: KEYCLOAK_SMTP_HOST
              value: mailu-front.mailu-mailserver.svc.cluster.local
            - name: KEYCLOAK_SMTP_PORT
              value: "25"
            - name: KEYCLOAK_SMTP_FROM
              value: no-reply@bstein.dev
            - name: KEYCLOAK_SMTP_FROM_NAME
              value: Atlas SSO
            - name: KEYCLOAK_SMTP_REPLY_TO
              value: no-reply@bstein.dev
            - name: KEYCLOAK_SMTP_REPLY_TO_NAME
              value: Atlas SSO
          command: ["/bin/sh", "-c"]
          args:
            - |
              set -euo pipefail
              python - <<'PY'
              import json
              import os
              import time
              import urllib.error
              import urllib.parse
              import urllib.request

              base_url = os.environ["KEYCLOAK_URL"].rstrip("/")
              realm = os.environ["KEYCLOAK_REALM"]
              admin_user = os.environ["KEYCLOAK_ADMIN_USER"]
              admin_password = os.environ["KEYCLOAK_ADMIN_PASSWORD"]

              smtp_defaults = {
                  "host": os.environ["KEYCLOAK_SMTP_HOST"],
                  "port": os.environ["KEYCLOAK_SMTP_PORT"],
                  "from": os.environ["KEYCLOAK_SMTP_FROM"],
                  "fromDisplayName": os.environ["KEYCLOAK_SMTP_FROM_NAME"],
                  "replyTo": os.environ["KEYCLOAK_SMTP_REPLY_TO"],
                  "replyToDisplayName": os.environ["KEYCLOAK_SMTP_REPLY_TO_NAME"],
                  "auth": "false",
                  "starttls": "false",
                  "ssl": "false",
              }

              def request(path, method="GET", data=None, headers=None):
                  if headers is None:
                      headers = {}
                  payload = None
                  if data is not None:
                      payload = json.dumps(data).encode()
                      headers = dict(headers)
                      headers["Content-Type"] = "application/json"
                  req = urllib.request.Request(
                      f"{base_url}{path}",
                      data=payload,
                      headers=headers,
                      method=method,
                  )
                  return urllib.request.urlopen(req, timeout=10)

              for _ in range(60):
                  try:
                      with request("/health/ready") as resp:
                          if resp.status == 200:
                              break
                  except Exception:
                      time.sleep(2)
              else:
                  raise SystemExit("Keycloak API did not become ready in time")

              token_data = urllib.parse.urlencode(
                  {
                      "grant_type": "password",
                      "client_id": "admin-cli",
                      "username": admin_user,
                      "password": admin_password,
                  }
              ).encode()
              token_req = urllib.request.Request(
                  f"{base_url}/realms/master/protocol/openid-connect/token",
                  data=token_data,
                  headers={"Content-Type": "application/x-www-form-urlencoded"},
                  method="POST",
              )
              with urllib.request.urlopen(token_req, timeout=10) as resp:
                  token_body = json.loads(resp.read().decode())
              access_token = token_body["access_token"]
              auth_headers = {"Authorization": f"Bearer {access_token}"}

              with request(f"/admin/realms/{realm}", headers=auth_headers) as resp:
                  realm_data = json.loads(resp.read().decode())

              changed = False
              if not realm_data.get("resetPasswordAllowed", False):
                  realm_data["resetPasswordAllowed"] = True
                  changed = True

              smtp = realm_data.get("smtpServer") or {}
              if not smtp.get("host"):
                  smtp.update(smtp_defaults)
                  realm_data["smtpServer"] = smtp
                  changed = True

              if not changed:
                  raise SystemExit(0)

              with request(
                  f"/admin/realms/{realm}",
                  method="PUT",
                  data=realm_data,
                  headers=auth_headers,
              ) as resp:
                  if resp.status not in (200, 204):
                      raise SystemExit(f"Unexpected response: {resp.status}")
              PY