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

# services/keycloak/realm-settings-job.yaml
apiVersion: batch/v1
kind: Job
metadata:
  name: keycloak-realm-settings-10
  namespace: sso
spec:
  backoffLimit: 0
  template:
    spec:
      affinity:
        nodeAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            nodeSelectorTerms:
              - matchExpressions:
                  - key: hardware
                    operator: In
                    values: ["rpi5","rpi4"]
                  - key: node-role.kubernetes.io/worker
                    operator: Exists
      restartPolicy: Never
      containers:
        - name: configure
          image: python:3.11-alpine
          env:
            - name: KEYCLOAK_SERVER
              value: http://keycloak.sso.svc.cluster.local
            - 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 urllib.parse
              import urllib.error
              import urllib.request

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

              def http_json(method: str, url: str, token: str, payload=None):
                  data = None
                  headers = {"Authorization": f"Bearer {token}"}
                  if payload is not None:
                      data = json.dumps(payload).encode()
                      headers["Content-Type"] = "application/json"
                  req = urllib.request.Request(url, data=data, headers=headers, method=method)
                  try:
                      with urllib.request.urlopen(req, timeout=30) as resp:
                          body = resp.read()
                          if not body:
                              return resp.status, None
                          return resp.status, json.loads(body.decode())
                  except urllib.error.HTTPError as exc:
                      raw = exc.read()
                      if not raw:
                          return exc.code, None
                      try:
                          return exc.code, json.loads(raw.decode())
                      except Exception:
                          return exc.code, {"raw": raw.decode(errors="replace")}

              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",
              )
              try:
                  with urllib.request.urlopen(token_req, timeout=10) as resp:
                      token_body = json.loads(resp.read().decode())
              except urllib.error.HTTPError as exc:
                  body = exc.read().decode(errors="replace")
                  raise SystemExit(f"Token request failed: status={exc.code} body={body}")
              access_token = token_body["access_token"]

              # Update realm settings safely by fetching the full realm representation first.
              realm_url = f"{base_url}/admin/realms/{realm}"
              status, realm_rep = http_json("GET", realm_url, access_token)
              if status != 200 or not realm_rep:
                  raise SystemExit(f"Unable to fetch realm {realm} (status={status})")

              realm_rep["resetPasswordAllowed"] = True

              smtp = realm_rep.get("smtpServer") or {}
              smtp.update(
                  {
                      "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",
                  }
              )
              realm_rep["smtpServer"] = smtp

              status, _ = http_json("PUT", realm_url, access_token, realm_rep)
              if status not in (200, 204):
                  raise SystemExit(f"Unexpected realm update response: {status}")

              # Disable Identity Provider Redirector in the browser flow for this realm.
              status, executions = http_json(
                  "GET",
                  f"{base_url}/admin/realms/{realm}/authentication/flows/browser/executions",
                  access_token,
              )
              if status == 200 and executions:
                  for ex in executions:
                      if ex.get("providerId") != "identity-provider-redirector":
                          continue
                      if ex.get("requirement") == "DISABLED":
                          continue
                      ex["requirement"] = "DISABLED"
                      status, _ = http_json(
                          "PUT",
                          f"{base_url}/admin/realms/{realm}/authentication/flows/browser/executions",
                          access_token,
                          ex,
                      )
                      if status not in (200, 204):
                          raise SystemExit(
                              f"Unexpected execution update response for identity-provider-redirector: {status}"
                          )
              PY