118 lines
4.0 KiB
Python
118 lines
4.0 KiB
Python
from dataclasses import dataclass
|
|
from urllib.parse import quote
|
|
from uuid import UUID
|
|
|
|
import httpx
|
|
|
|
|
|
class KeycloakUserSyncConfigurationError(ValueError):
|
|
pass
|
|
|
|
|
|
class KeycloakUserSyncRequestError(RuntimeError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class KeycloakUserSnapshot:
|
|
keycloak_id: UUID
|
|
username: str
|
|
email: str | None
|
|
service_account_client_id: str | None = None
|
|
|
|
|
|
def _keycloak_admin_users_url(issuer: str) -> str:
|
|
issuer = issuer.rstrip("/")
|
|
marker = "/realms/"
|
|
if not issuer or marker not in issuer:
|
|
raise KeycloakUserSyncConfigurationError(
|
|
"Keycloak token issuer is not a complete realm URL"
|
|
)
|
|
server_url, realm = issuer.rsplit(marker, 1)
|
|
if not server_url or not realm:
|
|
raise KeycloakUserSyncConfigurationError(
|
|
"Keycloak token issuer is not a complete realm URL"
|
|
)
|
|
return f"{server_url}/admin/realms/{quote(realm, safe='')}/users"
|
|
|
|
|
|
async def fetch_keycloak_users(
|
|
*,
|
|
issuer: str,
|
|
bearer_token: str,
|
|
client: httpx.AsyncClient | None = None,
|
|
) -> list[KeycloakUserSnapshot]:
|
|
admin_users_url = _keycloak_admin_users_url(issuer)
|
|
if not bearer_token:
|
|
raise KeycloakUserSyncConfigurationError(
|
|
"Keycloak administrator access token is missing"
|
|
)
|
|
owns_client = client is None
|
|
if client is None:
|
|
client = httpx.AsyncClient(timeout=15.0)
|
|
|
|
try:
|
|
try:
|
|
snapshots: list[KeycloakUserSnapshot] = []
|
|
first = 0
|
|
page_size = 100
|
|
while True:
|
|
response = await client.get(
|
|
admin_users_url,
|
|
params={"first": first, "max": page_size},
|
|
headers={
|
|
"Accept": "application/json",
|
|
"Authorization": f"Bearer {bearer_token}",
|
|
},
|
|
)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
if not isinstance(payload, list):
|
|
raise KeycloakUserSyncRequestError(
|
|
"Keycloak users response is not a list"
|
|
)
|
|
for item in payload:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
try:
|
|
keycloak_id = UUID(str(item.get("id", "")))
|
|
except ValueError:
|
|
continue
|
|
username = str(item.get("username") or "").strip()
|
|
if not username:
|
|
continue
|
|
email_value = item.get("email")
|
|
email = str(email_value).strip() if email_value else None
|
|
service_account_value = item.get("serviceAccountClientId")
|
|
snapshots.append(
|
|
KeycloakUserSnapshot(
|
|
keycloak_id=keycloak_id,
|
|
username=username,
|
|
email=email,
|
|
service_account_client_id=(
|
|
str(service_account_value)
|
|
if service_account_value
|
|
else None
|
|
),
|
|
)
|
|
)
|
|
if len(payload) < page_size:
|
|
break
|
|
first += page_size
|
|
return snapshots
|
|
except httpx.HTTPStatusError as exc:
|
|
if exc.response.status_code == 403:
|
|
raise KeycloakUserSyncRequestError(
|
|
"Keycloak administrator requires query-users and view-users roles"
|
|
) from exc
|
|
raise KeycloakUserSyncRequestError(
|
|
"Keycloak user directory request failed"
|
|
) from exc
|
|
except httpx.HTTPError as exc:
|
|
raise KeycloakUserSyncRequestError(
|
|
"Keycloak user directory request failed"
|
|
) from exc
|
|
finally:
|
|
if owns_client:
|
|
await client.aclose()
|