proxmox/scripts/deployment/keycloak-sankofa-ensure-it-admin-group.sh

#!/usr/bin/env bash
# Create Keycloak group sankofa-it-admin (or SANKOFA_IT_ADMIN_GROUP_NAME) and map realm role
# sankofa-it-admin onto it. Run after keycloak-sankofa-ensure-it-admin-role.sh.
#
# Usage: ./scripts/deployment/keycloak-sankofa-ensure-it-admin-group.sh [--dry-run]
set -euo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
PROJECT_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
# shellcheck source=/dev/null
source "${PROJECT_ROOT}/config/ip-addresses.conf" 2>/dev/null || true
if [ -f "$PROJECT_ROOT/.env" ]; then
  set +u
  set -a
  # shellcheck source=/dev/null
  source "$PROJECT_ROOT/.env" 2>/dev/null || true
  set +a
  set -u
fi

PROXMOX_HOST="${PROXMOX_HOST:-${PROXMOX_HOST_R630_01:-192.168.11.11}}"
KEYCLOAK_CT_VMID="${KEYCLOAK_CT_VMID:-${SANKOFA_KEYCLOAK_VMID:-7802}}"
REALM="${KEYCLOAK_REALM:-master}"
ADMIN_USER="${KEYCLOAK_ADMIN:-admin}"
ADMIN_PASS="${KEYCLOAK_ADMIN_PASSWORD:-}"
ROLE_NAME="${SANKOFA_IT_ADMIN_ROLE_NAME:-sankofa-it-admin}"
GROUP_NAME="${SANKOFA_IT_ADMIN_GROUP_NAME:-sankofa-it-admin}"
SSH_OPTS=(-o BatchMode=yes -o StrictHostKeyChecking=accept-new -o ConnectTimeout=15)

DRY=0
[[ "${1:-}" == "--dry-run" ]] && DRY=1

if [ -z "$ADMIN_PASS" ]; then
  echo "KEYCLOAK_ADMIN_PASSWORD is not set in .env" >&2
  exit 1
fi

if [ "$DRY" = 1 ]; then
  echo "[dry-run] Would ensure group ${GROUP_NAME} + map realm role ${ROLE_NAME} in ${REALM}"
  exit 0
fi

ssh "${SSH_OPTS[@]}" "root@${PROXMOX_HOST}" \
  "pct exec ${KEYCLOAK_CT_VMID} -- env KC_PASS=\"${ADMIN_PASS}\" ADMUSER=\"${ADMIN_USER}\" REALM=\"${REALM}\" ROLE_NAME=\"${ROLE_NAME}\" GROUP_NAME=\"${GROUP_NAME}\" python3 -u -" <<'PY'
import json
import os
import urllib.error
import urllib.parse
import urllib.request

base = "http://127.0.0.1:8080"
realm = os.environ["REALM"]
role_name = os.environ["ROLE_NAME"]
group_name = os.environ["GROUP_NAME"]
admin_user = os.environ["ADMUSER"]
password = os.environ["KC_PASS"]


def post_form(url: str, data: dict) -> dict:
    body = urllib.parse.urlencode(data).encode()
    req = urllib.request.Request(url, data=body, method="POST")
    with urllib.request.urlopen(req, timeout=60) as resp:
        return json.loads(resp.read().decode())


def req_json(method: str, url: str, headers: dict, data=None):
    body = None
    hdrs = dict(headers)
    if data is not None:
        body = json.dumps(data).encode()
        hdrs["Content-Type"] = "application/json"
    r = urllib.request.Request(url, data=body, headers=hdrs, method=method)
    with urllib.request.urlopen(r, timeout=120) as resp:
        return resp.read().decode()


def req_json_ignore_404(method: str, url: str, headers: dict, data=None):
    body = None
    hdrs = dict(headers)
    if data is not None:
        body = json.dumps(data).encode()
        hdrs["Content-Type"] = "application/json"
    r = urllib.request.Request(url, data=body, headers=hdrs, method=method)
    try:
        with urllib.request.urlopen(r, timeout=120) as resp:
            return resp.getcode(), resp.read().decode()
    except urllib.error.HTTPError as e:
        return e.code, e.read().decode() if e.fp else ""


tok = post_form(
    f"{base}/realms/master/protocol/openid-connect/token",
    {
        "grant_type": "password",
        "client_id": "admin-cli",
        "username": admin_user,
        "password": password,
    },
)
access = tok.get("access_token")
if not access:
    raise SystemExit(f"token failed: {tok}")

h = {"Authorization": f"Bearer {access}"}

# Role id
role_url = f"{base}/admin/realms/{realm}/roles/{urllib.parse.quote(role_name, safe='')}"
req_r = urllib.request.Request(role_url, headers=h)
try:
    with urllib.request.urlopen(req_r, timeout=60) as resp:
        role = json.loads(resp.read().decode())
except urllib.error.HTTPError as e:
    raise SystemExit(
        f"Realm role {role_name!r} missing; run keycloak-sankofa-ensure-it-admin-role.sh first. HTTP {e.code}"
    ) from e

role_id = role.get("id")
if not role_id:
    raise SystemExit("role JSON missing id")

# Find or create group
list_url = f"{base}/admin/realms/{realm}/groups?search={urllib.parse.quote(group_name)}"
req_g = urllib.request.Request(list_url, headers=h)
with urllib.request.urlopen(req_g, timeout=60) as resp:
    groups = json.loads(resp.read().decode())

gid = None
for g in groups:
    if g.get("name") == group_name:
        gid = g.get("id")
        break

if not gid:
    try:
        req_json(
            "POST",
            f"{base}/admin/realms/{realm}/groups",
            h,
            {"name": group_name},
        )
    except urllib.error.HTTPError as e:
        if e.code != 409:
            raise
    with urllib.request.urlopen(
        urllib.request.Request(list_url, headers=h), timeout=60
    ) as resp:
        groups = json.loads(resp.read().decode())
    for g in groups:
        if g.get("name") == group_name:
            gid = g.get("id")
            break

if not gid:
    raise SystemExit("failed to resolve group id after create")

# Map realm role to group (idempotent: POST may 204 or 409 depending on KC version)
map_url = f"{base}/admin/realms/{realm}/groups/{gid}/role-mappings/realm"
code, _body = req_json_ignore_404(
    "POST",
    map_url,
    h,
    [{"id": role_id, "name": role_name}],
)
if code in (200, 204):
    print(f"Mapped realm role {role_name!r} to group {group_name!r}.", flush=True)
elif code == 409:
    print(f"Role likely already mapped to group {group_name!r}.", flush=True)
else:
    print(f"POST role-mappings HTTP {code}: {_body[:500]}", flush=True)

print(
    f"Add IT users to group {group_name!r} in Admin Console (Groups → Members).",
    flush=True,
)
PY