265 lines
11 KiB
Python
265 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
import time
|
|
|
|
import httpx
|
|
|
|
from ..settings import settings
|
|
|
|
|
|
class KeycloakAdminClient:
|
|
"""Call the Keycloak admin API for user, group, and attribute updates."""
|
|
|
|
def __init__(self) -> None:
|
|
self._token: str = ""
|
|
self._expires_at: float = 0.0
|
|
self._group_id_cache: dict[str, str] = {}
|
|
|
|
@staticmethod
|
|
def _safe_update_payload(full: dict[str, Any]) -> dict[str, Any]:
|
|
payload: dict[str, Any] = {}
|
|
username = full.get("username")
|
|
if isinstance(username, str):
|
|
payload["username"] = username
|
|
enabled = full.get("enabled")
|
|
if isinstance(enabled, bool):
|
|
payload["enabled"] = enabled
|
|
email = full.get("email")
|
|
if isinstance(email, str):
|
|
payload["email"] = email
|
|
email_verified = full.get("emailVerified")
|
|
if isinstance(email_verified, bool):
|
|
payload["emailVerified"] = email_verified
|
|
first_name = full.get("firstName")
|
|
if isinstance(first_name, str):
|
|
payload["firstName"] = first_name
|
|
last_name = full.get("lastName")
|
|
if isinstance(last_name, str):
|
|
payload["lastName"] = last_name
|
|
|
|
actions = full.get("requiredActions")
|
|
if isinstance(actions, list):
|
|
payload["requiredActions"] = [a for a in actions if isinstance(a, str)]
|
|
|
|
attrs = full.get("attributes")
|
|
payload["attributes"] = attrs if isinstance(attrs, dict) else {}
|
|
return payload
|
|
|
|
def ready(self) -> bool:
|
|
return bool(settings.keycloak_admin_client_id and settings.keycloak_admin_client_secret)
|
|
|
|
def _get_token(self) -> str:
|
|
if not self.ready():
|
|
raise RuntimeError("keycloak admin client not configured")
|
|
|
|
now = time.time()
|
|
if self._token and now < self._expires_at - 30:
|
|
return self._token
|
|
|
|
token_url = (
|
|
f"{settings.keycloak_admin_url}/realms/{settings.keycloak_admin_realm}/protocol/openid-connect/token"
|
|
)
|
|
data = {
|
|
"grant_type": "client_credentials",
|
|
"client_id": settings.keycloak_admin_client_id,
|
|
"client_secret": settings.keycloak_admin_client_secret,
|
|
}
|
|
with httpx.Client(timeout=10.0) as client:
|
|
resp = client.post(token_url, data=data)
|
|
resp.raise_for_status()
|
|
payload = resp.json()
|
|
token = payload.get("access_token") or ""
|
|
if not token:
|
|
raise RuntimeError("no access_token in response")
|
|
expires_in = int(payload.get("expires_in") or 60)
|
|
self._token = token
|
|
self._expires_at = now + expires_in
|
|
return token
|
|
|
|
def _headers(self) -> dict[str, str]:
|
|
return {"Authorization": f"Bearer {self._get_token()}"}
|
|
|
|
def headers(self) -> dict[str, str]:
|
|
return self._headers()
|
|
|
|
def find_user(self, username: str) -> dict[str, Any] | None:
|
|
url = f"{settings.keycloak_admin_url}/admin/realms/{settings.keycloak_realm}/users"
|
|
params = {"username": username, "exact": "true", "max": "1"}
|
|
with httpx.Client(timeout=10.0) as client:
|
|
resp = client.get(url, params=params, headers=self._headers())
|
|
resp.raise_for_status()
|
|
users = resp.json()
|
|
if not isinstance(users, list) or not users:
|
|
return None
|
|
user = users[0]
|
|
return user if isinstance(user, dict) else None
|
|
|
|
def find_user_by_email(self, email: str) -> dict[str, Any] | None:
|
|
email = (email or "").strip()
|
|
if not email:
|
|
return None
|
|
|
|
url = f"{settings.keycloak_admin_url}/admin/realms/{settings.keycloak_realm}/users"
|
|
params = {"email": email, "exact": "true", "max": "2"}
|
|
email_norm = email.lower()
|
|
with httpx.Client(timeout=10.0) as client:
|
|
resp = client.get(url, params=params, headers=self._headers())
|
|
resp.raise_for_status()
|
|
users = resp.json()
|
|
if not isinstance(users, list) or not users:
|
|
return None
|
|
for user in users:
|
|
if not isinstance(user, dict):
|
|
continue
|
|
candidate = user.get("email")
|
|
if isinstance(candidate, str) and candidate.strip().lower() == email_norm:
|
|
return user
|
|
return None
|
|
|
|
def get_user(self, user_id: str) -> dict[str, Any]:
|
|
url = f"{settings.keycloak_admin_url}/admin/realms/{settings.keycloak_realm}/users/{user_id}"
|
|
with httpx.Client(timeout=10.0) as client:
|
|
resp = client.get(url, headers=self._headers())
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
if not isinstance(data, dict):
|
|
raise RuntimeError("unexpected user payload")
|
|
return data
|
|
|
|
def update_user(self, user_id: str, payload: dict[str, Any]) -> None:
|
|
url = f"{settings.keycloak_admin_url}/admin/realms/{settings.keycloak_realm}/users/{user_id}"
|
|
with httpx.Client(timeout=10.0) as client:
|
|
resp = client.put(url, headers={**self._headers(), "Content-Type": "application/json"}, json=payload)
|
|
resp.raise_for_status()
|
|
|
|
def update_user_safe(self, user_id: str, payload: dict[str, Any]) -> None:
|
|
full = self.get_user(user_id)
|
|
merged = self._safe_update_payload(full)
|
|
for key, value in payload.items():
|
|
if key == "attributes":
|
|
attrs = merged.get("attributes")
|
|
if not isinstance(attrs, dict):
|
|
attrs = {}
|
|
if isinstance(value, dict):
|
|
attrs.update(value)
|
|
merged["attributes"] = attrs
|
|
continue
|
|
merged[key] = value
|
|
self.update_user(user_id, merged)
|
|
|
|
def create_user(self, payload: dict[str, Any]) -> str:
|
|
url = f"{settings.keycloak_admin_url}/admin/realms/{settings.keycloak_realm}/users"
|
|
with httpx.Client(timeout=10.0) as client:
|
|
resp = client.post(url, headers={**self._headers(), "Content-Type": "application/json"}, json=payload)
|
|
resp.raise_for_status()
|
|
location = resp.headers.get("Location") or ""
|
|
if location:
|
|
return location.rstrip("/").split("/")[-1]
|
|
raise RuntimeError("failed to determine created user id")
|
|
|
|
def reset_password(self, user_id: str, password: str, temporary: bool = False) -> None:
|
|
url = (
|
|
f"{settings.keycloak_admin_url}/admin/realms/{settings.keycloak_realm}"
|
|
f"/users/{user_id}/reset-password"
|
|
)
|
|
payload = {"type": "password", "value": password, "temporary": bool(temporary)}
|
|
with httpx.Client(timeout=10.0) as client:
|
|
resp = client.put(url, headers={**self._headers(), "Content-Type": "application/json"}, json=payload)
|
|
resp.raise_for_status()
|
|
|
|
def set_user_attribute(self, username: str, key: str, value: str) -> None:
|
|
user = self.find_user(username)
|
|
if not user:
|
|
raise RuntimeError("user not found")
|
|
user_id = user.get("id") or ""
|
|
if not user_id:
|
|
raise RuntimeError("user id missing")
|
|
|
|
full = self.get_user(user_id)
|
|
payload = self._safe_update_payload(full)
|
|
attrs = payload.get("attributes")
|
|
if not isinstance(attrs, dict):
|
|
attrs = {}
|
|
attrs[key] = [value]
|
|
payload["attributes"] = attrs
|
|
self.update_user(user_id, payload)
|
|
|
|
def get_group_id(self, group_name: str) -> str | None:
|
|
cached = self._group_id_cache.get(group_name)
|
|
if cached:
|
|
return cached
|
|
|
|
url = f"{settings.keycloak_admin_url}/admin/realms/{settings.keycloak_realm}/groups"
|
|
params = {"search": group_name}
|
|
with httpx.Client(timeout=10.0) as client:
|
|
resp = client.get(url, params=params, headers=self._headers())
|
|
resp.raise_for_status()
|
|
items = resp.json()
|
|
if not isinstance(items, list):
|
|
return None
|
|
for item in items:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
if item.get("name") == group_name and item.get("id"):
|
|
gid = str(item["id"])
|
|
self._group_id_cache[group_name] = gid
|
|
return gid
|
|
return None
|
|
|
|
def add_user_to_group(self, user_id: str, group_id: str) -> None:
|
|
url = (
|
|
f"{settings.keycloak_admin_url}/admin/realms/{settings.keycloak_realm}"
|
|
f"/users/{user_id}/groups/{group_id}"
|
|
)
|
|
with httpx.Client(timeout=10.0) as client:
|
|
resp = client.put(url, headers=self._headers())
|
|
resp.raise_for_status()
|
|
|
|
def iter_users(self, page_size: int = 200, brief: bool = False) -> list[dict[str, Any]]:
|
|
url = f"{settings.keycloak_admin_url}/admin/realms/{settings.keycloak_realm}/users"
|
|
users: list[dict[str, Any]] = []
|
|
first = 0
|
|
while True:
|
|
params = {"first": str(first), "max": str(page_size)}
|
|
if not brief:
|
|
params["briefRepresentation"] = "false"
|
|
with httpx.Client(timeout=10.0) as client:
|
|
resp = client.get(url, params=params, headers=self._headers())
|
|
resp.raise_for_status()
|
|
payload = resp.json()
|
|
if not isinstance(payload, list) or not payload:
|
|
return users
|
|
for item in payload:
|
|
if isinstance(item, dict):
|
|
users.append(item)
|
|
if len(payload) < page_size:
|
|
return users
|
|
first += page_size
|
|
|
|
def list_groups(self, max_groups: int = 200) -> list[dict[str, Any]]:
|
|
url = f"{settings.keycloak_admin_url}/admin/realms/{settings.keycloak_realm}/groups"
|
|
params = {"max": str(max_groups)}
|
|
with httpx.Client(timeout=10.0) as client:
|
|
resp = client.get(url, params=params, headers=self._headers())
|
|
resp.raise_for_status()
|
|
payload = resp.json()
|
|
if not isinstance(payload, list):
|
|
return []
|
|
return [item for item in payload if isinstance(item, dict)]
|
|
|
|
def list_group_names(self, exclude: set[str] | None = None) -> list[str]:
|
|
exclude = {name.strip() for name in (exclude or set()) if name.strip()}
|
|
names: list[str] = []
|
|
for group in self.list_groups():
|
|
name = group.get("name")
|
|
if isinstance(name, str) and name.strip():
|
|
normalized = name.strip()
|
|
if normalized in exclude:
|
|
continue
|
|
names.append(normalized)
|
|
return sorted(set(names))
|
|
|
|
|
|
keycloak_admin = KeycloakAdminClient()
|