next-tjwater-server/scripts/provision_integration_ui_client.py

#!/usr/bin/env python3
"""Provision the isolated public Keycloak client used by the 15173 UI."""
from __future__ import annotations

import argparse
import getpass
import json
import os
from pathlib import Path
import sys

REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
if str(REPOSITORY_ROOT) not in sys.path:
    sys.path.insert(0, str(REPOSITORY_ROOT))

from scripts.provision_operational_worker_identity import (  # noqa: E402
    KeycloakProvisioner,
    ProvisionError,
    _request,
)


REDIRECT_URIS = [
    "http://127.0.0.1:15173/*",
    "http://localhost:15173/*",
]
WEB_ORIGINS = [
    "http://127.0.0.1:15173",
    "http://localhost:15173",
]


def ensure_public_client(
    keycloak: KeycloakProvisioner, client_id: str
) -> tuple[bool, dict]:
    matches = _request(
        keycloak.client,
        "GET",
        f"{keycloak.admin_base}/clients",
        params={"clientId": client_id},
    ).json()
    exact = [item for item in matches if item.get("clientId") == client_id]
    if len(exact) > 1:
        raise ProvisionError("multiple Keycloak clients use the integration client id")
    created = not exact
    if created:
        _request(
            keycloak.client,
            "POST",
            f"{keycloak.admin_base}/clients",
            json={
                "clientId": client_id,
                "name": "TJWater isolated integration UI",
                "description": "Local 15173 UI; authorization code flow with PKCE only",
                "enabled": True,
                "protocol": "openid-connect",
                "publicClient": True,
                "bearerOnly": False,
                "serviceAccountsEnabled": False,
                "standardFlowEnabled": True,
                "implicitFlowEnabled": False,
                "directAccessGrantsEnabled": False,
                "fullScopeAllowed": False,
                "redirectUris": REDIRECT_URIS,
                "webOrigins": WEB_ORIGINS,
                "attributes": {"pkce.code.challenge.method": "S256"},
            },
        )
        matches = _request(
            keycloak.client,
            "GET",
            f"{keycloak.admin_base}/clients",
            params={"clientId": client_id},
        ).json()
        exact = [item for item in matches if item.get("clientId") == client_id]
    if len(exact) != 1 or not isinstance(exact[0].get("id"), str):
        raise ProvisionError("could not resolve the integration UI client")
    internal_id = exact[0]["id"]
    representation = _request(
        keycloak.client,
        "GET",
        f"{keycloak.admin_base}/clients/{internal_id}",
    ).json()
    attributes = dict(representation.get("attributes") or {})
    attributes["pkce.code.challenge.method"] = "S256"
    representation.update(
        {
            "enabled": True,
            "publicClient": True,
            "bearerOnly": False,
            "serviceAccountsEnabled": False,
            "standardFlowEnabled": True,
            "implicitFlowEnabled": False,
            "directAccessGrantsEnabled": False,
            "fullScopeAllowed": False,
            "redirectUris": REDIRECT_URIS,
            "webOrigins": WEB_ORIGINS,
            "attributes": attributes,
        }
    )
    _request(
        keycloak.client,
        "PUT",
        f"{keycloak.admin_base}/clients/{internal_id}",
        json=representation,
    )
    verified = _request(
        keycloak.client,
        "GET",
        f"{keycloak.admin_base}/clients/{internal_id}",
    ).json()
    expected = {
        "enabled": True,
        "publicClient": True,
        "bearerOnly": False,
        "serviceAccountsEnabled": False,
        "standardFlowEnabled": True,
        "implicitFlowEnabled": False,
        "directAccessGrantsEnabled": False,
        "fullScopeAllowed": False,
    }
    if any(verified.get(key) != value for key, value in expected.items()):
        raise ProvisionError("integration UI client flow settings did not persist")
    if set(verified.get("redirectUris") or []) != set(REDIRECT_URIS):
        raise ProvisionError("integration UI redirect URI allowlist did not persist")
    if set(verified.get("webOrigins") or []) != set(WEB_ORIGINS):
        raise ProvisionError("integration UI web origin allowlist did not persist")
    if (verified.get("attributes") or {}).get("pkce.code.challenge.method") != "S256":
        raise ProvisionError("integration UI PKCE policy did not persist")
    return created, {
        "client_id": client_id,
        "public_client": True,
        "standard_flow": True,
        "pkce": "S256",
        "implicit_flow": False,
        "direct_grants": False,
        "service_accounts": False,
        "redirect_uris": REDIRECT_URIS,
        "web_origins": WEB_ORIGINS,
    }


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--keycloak-url", default="http://192.168.1.114:8088")
    parser.add_argument("--realm", default="tjwater")
    parser.add_argument("--admin-username", default="admin")
    parser.add_argument("--client-id", default="tjwater-integration-ui")
    args = parser.parse_args()
    password = os.environ.get("KEYCLOAK_ADMIN_PASSWORD") or getpass.getpass(
        "Keycloak admin password: "
    )
    if not password:
        raise SystemExit("administrator password is required")
    try:
        keycloak = KeycloakProvisioner(
            args.keycloak_url, args.realm, args.admin_username, password
        )
        try:
            created, result = ensure_public_client(keycloak, args.client_id)
        finally:
            keycloak.close()
    except ProvisionError as exc:
        raise SystemExit(f"provisioning failed: {exc}") from None
    print(json.dumps({"created": created, **result}, sort_keys=True))


if __name__ == "__main__":
    main()