from __future__ import annotations from dataclasses import dataclass import secrets import string import httpx from fastapi import HTTPException, status from app.core.config import Settings @dataclass class ProviderSyncResult: user_id: str action: str user_sub: str | None = None @dataclass class ProviderPasswordResetResult: user_id: str temporary_password: str @dataclass class ProviderDeleteResult: action: str user_id: str | None = None @dataclass class ProviderGroupSyncResult: group_id: str action: str @dataclass class ProviderRoleSyncResult: role_name: str action: str class ProviderAdminService: def __init__(self, settings: Settings) -> None: self.base_url = settings.keycloak_base_url.rstrip("/") self.realm = settings.keycloak_realm self.admin_realm = settings.keycloak_admin_realm or settings.keycloak_realm self.admin_client_id = settings.keycloak_admin_client_id self.admin_client_secret = settings.keycloak_admin_client_secret self.verify_tls = settings.keycloak_verify_tls if not self.base_url or not self.realm or not self.admin_client_id or not self.admin_client_secret: raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="idp_admin_not_configured") @staticmethod def _safe_username(sub: str | None, email: str) -> str: if email and "@" in email: return email.split("@", 1)[0] if sub: return sub.replace("|", "_")[:150] return "member-user" @staticmethod def _generate_temporary_password(length: int = 14) -> str: alphabet = string.ascii_letters + string.digits + "!@#$%^&*" return "".join(secrets.choice(alphabet) for _ in range(length)) def _get_admin_token(self) -> str: token_endpoint = f"{self.base_url}/realms/{self.admin_realm}/protocol/openid-connect/token" try: resp = httpx.post( token_endpoint, data={ "grant_type": "client_credentials", "client_id": self.admin_client_id, "client_secret": self.admin_client_secret, }, timeout=10, verify=self.verify_tls, headers={"Content-Type": "application/x-www-form-urlencoded"}, ) except Exception as exc: raise HTTPException(status_code=502, detail="idp_lookup_failed") from exc if resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_lookup_failed") token = resp.json().get("access_token") if not token: raise HTTPException(status_code=502, detail="idp_lookup_failed") return str(token) def _client(self) -> httpx.Client: return httpx.Client( base_url=self.base_url, headers={ "Authorization": f"Bearer {self._get_admin_token()}", "Accept": "application/json", "Content-Type": "application/json", }, timeout=10, verify=self.verify_tls, ) def _lookup_user_by_id(self, client: httpx.Client, user_id: str) -> dict | None: resp = client.get(f"/admin/realms/{self.realm}/users/{user_id}") if resp.status_code == 404: return None if resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_lookup_failed") return resp.json() def _lookup_group_by_id(self, client: httpx.Client, group_id: str) -> dict | None: resp = client.get(f"/admin/realms/{self.realm}/groups/{group_id}") if resp.status_code == 404: return None if resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_group_lookup_failed") payload = resp.json() if resp.content else {} return payload if isinstance(payload, dict) else None def _lookup_group_by_name(self, client: httpx.Client, *, name: str, parent_group_id: str | None) -> dict | None: if parent_group_id: resp = client.get( f"/admin/realms/{self.realm}/groups/{parent_group_id}/children", params={"search": name, "briefRepresentation": "false"}, ) if resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_group_lookup_failed") matches = resp.json() if isinstance(resp.json(), list) else [] for row in matches: if isinstance(row, dict) and str(row.get("name", "")).strip() == name: return row return None resp = client.get( f"/admin/realms/{self.realm}/groups", params={"search": name, "exact": "true", "briefRepresentation": "false"}, ) if resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_group_lookup_failed") matches = resp.json() if isinstance(resp.json(), list) else [] for row in matches: if not isinstance(row, dict): continue if str(row.get("name", "")).strip() != name: continue parent_id = row.get("parentId") if parent_group_id: if str(parent_id or "") == parent_group_id: return row elif not parent_id: return row return None @staticmethod def _normalize_group_attributes(attributes: dict[str, str | list[str]] | None) -> dict[str, list[str]]: if not attributes: return {} output: dict[str, list[str]] = {} for key, value in attributes.items(): normalized_key = str(key).strip() if not normalized_key: continue if isinstance(value, list): output[normalized_key] = [str(v) for v in value if str(v)] elif value is not None and str(value): output[normalized_key] = [str(value)] return output def _lookup_user_by_email_or_username( self, client: httpx.Client, *, email: str | None, username: str | None ) -> dict | None: if email: resp = client.get(f"/admin/realms/{self.realm}/users", params={"email": email, "exact": "true"}) if resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_lookup_failed") matches = resp.json() if isinstance(resp.json(), list) else [] if matches: return matches[0] if username: resp = client.get(f"/admin/realms/{self.realm}/users", params={"username": username, "exact": "true"}) if resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_lookup_failed") matches = resp.json() if isinstance(resp.json(), list) else [] if matches: return matches[0] return None def ensure_user( self, *, sub: str | None, email: str, username: str | None, display_name: str | None, is_active: bool = True, provider_user_id: str | None = None, ) -> ProviderSyncResult: resolved_username = username or self._safe_username(sub=sub, email=email) first_name = display_name or resolved_username payload = { "username": resolved_username, "email": email, "enabled": is_active, "emailVerified": True, "firstName": first_name, "attributes": {"user_sub": [sub]} if sub else {}, } with self._client() as client: existing = self._lookup_user_by_id(client, provider_user_id) if provider_user_id else None if existing is None: existing = self._lookup_user_by_email_or_username(client, email=email, username=resolved_username) if existing and existing.get("id"): user_id = str(existing["id"]) put_resp = client.put(f"/admin/realms/{self.realm}/users/{user_id}", json=payload) if put_resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_update_failed") return ProviderSyncResult(user_id=user_id, action="updated", user_sub=user_id) create_resp = client.post(f"/admin/realms/{self.realm}/users", json=payload) if create_resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_create_failed") location = create_resp.headers.get("Location", "") user_id = location.rstrip("/").split("/")[-1] if location and "/" in location else "" if not user_id: found = self._lookup_user_by_email_or_username(client, email=email, username=resolved_username) user_id = str(found["id"]) if found and found.get("id") else "" if not user_id: raise HTTPException(status_code=502, detail="idp_create_failed") return ProviderSyncResult(user_id=user_id, action="created", user_sub=user_id) def ensure_group( self, *, name: str, group_id: str | None = None, parent_group_id: str | None = None, attributes: dict[str, str | list[str]] | None = None, ) -> ProviderGroupSyncResult: if not name: raise HTTPException(status_code=400, detail="idp_group_name_required") normalized_attrs = self._normalize_group_attributes(attributes) with self._client() as client: existing = self._lookup_group_by_id(client, group_id) if group_id else None if existing is None: existing = self._lookup_group_by_name(client, name=name, parent_group_id=parent_group_id) if existing and existing.get("id"): resolved_id = str(existing["id"]) payload = {"name": name, "attributes": normalized_attrs} put_resp = client.put(f"/admin/realms/{self.realm}/groups/{resolved_id}", json=payload) if put_resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_group_update_failed") return ProviderGroupSyncResult(group_id=resolved_id, action="updated") payload = {"name": name, "attributes": normalized_attrs} if parent_group_id: create_resp = client.post(f"/admin/realms/{self.realm}/groups/{parent_group_id}/children", json=payload) else: create_resp = client.post(f"/admin/realms/{self.realm}/groups", json=payload) if create_resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_group_create_failed") location = create_resp.headers.get("Location", "") resolved_id = location.rstrip("/").split("/")[-1] if location and "/" in location else "" if not resolved_id: found = self._lookup_group_by_name(client, name=name, parent_group_id=parent_group_id) resolved_id = str(found.get("id")) if found and found.get("id") else "" if not resolved_id: raise HTTPException(status_code=502, detail="idp_group_create_failed") return ProviderGroupSyncResult(group_id=resolved_id, action="created") def delete_group(self, *, group_id: str | None) -> ProviderDeleteResult: if not group_id: return ProviderDeleteResult(action="not_found") with self._client() as client: resp = client.delete(f"/admin/realms/{self.realm}/groups/{group_id}") if resp.status_code in {204, 404}: return ProviderDeleteResult(action="deleted" if resp.status_code == 204 else "not_found") if resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_group_delete_failed") return ProviderDeleteResult(action="deleted") def reset_password( self, *, provider_user_id: str | None, email: str | None, username: str | None, ) -> ProviderPasswordResetResult: with self._client() as client: existing = self._lookup_user_by_id(client, provider_user_id) if provider_user_id else None if existing is None: existing = self._lookup_user_by_email_or_username(client, email=email, username=username) if not existing or not existing.get("id"): raise HTTPException(status_code=404, detail="idp_user_not_found") user_id = str(existing["id"]) temp_password = self._generate_temporary_password() resp = client.put( f"/admin/realms/{self.realm}/users/{user_id}/reset-password", json={"type": "password", "value": temp_password, "temporary": True}, ) if resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_set_password_failed") return ProviderPasswordResetResult(user_id=user_id, temporary_password=temp_password) def delete_user( self, *, provider_user_id: str | None, email: str | None, username: str | None, ) -> ProviderDeleteResult: with self._client() as client: existing = self._lookup_user_by_id(client, provider_user_id) if provider_user_id else None if existing is None: existing = self._lookup_user_by_email_or_username(client, email=email, username=username) if not existing or not existing.get("id"): return ProviderDeleteResult(action="not_found") user_id = str(existing["id"]) resp = client.delete(f"/admin/realms/{self.realm}/users/{user_id}") if resp.status_code in {204, 404}: return ProviderDeleteResult(action="deleted" if resp.status_code == 204 else "not_found", user_id=user_id) if resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_delete_failed") return ProviderDeleteResult(action="deleted", user_id=user_id) def list_groups_tree(self) -> list[dict]: with self._client() as client: resp = client.get( f"/admin/realms/{self.realm}/groups", params={"first": 0, "max": 5000, "briefRepresentation": "false"}, ) if resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_group_lookup_failed") payload = resp.json() if resp.content else [] return payload if isinstance(payload, list) else [] def list_users(self) -> list[dict]: users: list[dict] = [] first = 0 page_size = 200 with self._client() as client: while True: resp = client.get( f"/admin/realms/{self.realm}/users", params={"first": first, "max": page_size}, ) if resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_lookup_failed") batch = resp.json() if isinstance(resp.json(), list) else [] users.extend([row for row in batch if isinstance(row, dict)]) if len(batch) < page_size: break first += page_size return users def list_clients(self) -> list[dict]: clients: list[dict] = [] first = 0 page_size = 200 with self._client() as client: while True: resp = client.get( f"/admin/realms/{self.realm}/clients", params={"first": first, "max": page_size}, ) if resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_lookup_failed") batch = resp.json() if isinstance(resp.json(), list) else [] clients.extend([row for row in batch if isinstance(row, dict)]) if len(batch) < page_size: break first += page_size return clients def list_client_roles(self, client_uuid: str) -> list[dict]: with self._client() as client: resp = client.get( f"/admin/realms/{self.realm}/clients/{client_uuid}/roles", params={"first": 0, "max": 5000}, ) if resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_lookup_failed") payload = resp.json() if resp.content else [] return payload if isinstance(payload, list) else [] def _resolve_client_uuid(self, client: httpx.Client, provider_client_id: str) -> str: if not provider_client_id: raise HTTPException(status_code=400, detail="provider_client_id_required") # provider_client_id stores Keycloak clientId (not internal UUID). resp = client.get( f"/admin/realms/{self.realm}/clients", params={"clientId": provider_client_id}, ) if resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_lookup_failed") payload = resp.json() if resp.content else [] rows = payload if isinstance(payload, list) else [] for row in rows: if not isinstance(row, dict): continue if str(row.get("clientId", "")).strip() != provider_client_id: continue client_uuid = str(row.get("id", "")).strip() if client_uuid: return client_uuid raise HTTPException(status_code=404, detail="provider_client_not_found") def _get_client_role_representation(self, client: httpx.Client, *, client_uuid: str, role_name: str) -> dict: resp = client.get(f"/admin/realms/{self.realm}/clients/{client_uuid}/roles/{role_name}") if resp.status_code == 404: raise HTTPException(status_code=404, detail=f"provider_role_not_found:{role_name}") if resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_lookup_failed") payload = resp.json() if resp.content else {} if not isinstance(payload, dict): raise HTTPException(status_code=502, detail="idp_lookup_failed") return payload def set_group_client_roles(self, *, group_id: str, provider_client_id: str, role_names: list[str]) -> None: if not group_id: raise HTTPException(status_code=400, detail="provider_group_id_required") desired_names = [name.strip() for name in role_names if isinstance(name, str) and name.strip()] desired_name_set = set(desired_names) with self._client() as client: client_uuid = self._resolve_client_uuid(client, provider_client_id) current_resp = client.get(f"/admin/realms/{self.realm}/groups/{group_id}/role-mappings/clients/{client_uuid}") if current_resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_group_role_mapping_lookup_failed") current_payload = current_resp.json() if current_resp.content else [] current_rows = current_payload if isinstance(current_payload, list) else [] current_map: dict[str, dict] = {} for row in current_rows: if not isinstance(row, dict): continue name = str(row.get("name", "")).strip() if name: current_map[name] = row to_add_names = sorted(desired_name_set - set(current_map.keys())) to_remove_names = sorted(set(current_map.keys()) - desired_name_set) if to_add_names: add_payload = [ self._get_client_role_representation(client, client_uuid=client_uuid, role_name=role_name) for role_name in to_add_names ] add_resp = client.post( f"/admin/realms/{self.realm}/groups/{group_id}/role-mappings/clients/{client_uuid}", json=add_payload, ) if add_resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_group_role_mapping_add_failed") if to_remove_names: remove_payload = [current_map[name] for name in to_remove_names] remove_resp = client.request( "DELETE", f"/admin/realms/{self.realm}/groups/{group_id}/role-mappings/clients/{client_uuid}", json=remove_payload, ) if remove_resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_group_role_mapping_remove_failed") def ensure_client_role( self, *, provider_client_id: str, provider_role_name: str, description: str | None = None, ) -> ProviderRoleSyncResult: role_name = provider_role_name.strip() if not role_name: raise HTTPException(status_code=400, detail="provider_role_name_required") with self._client() as client: client_uuid = self._resolve_client_uuid(client, provider_client_id) get_resp = client.get(f"/admin/realms/{self.realm}/clients/{client_uuid}/roles/{role_name}") if get_resp.status_code == 404: create_resp = client.post( f"/admin/realms/{self.realm}/clients/{client_uuid}/roles", json={"name": role_name, "description": description or ""}, ) if create_resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_role_create_failed") return ProviderRoleSyncResult(role_name=role_name, action="created") if get_resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_lookup_failed") payload = get_resp.json() if get_resp.content else {} update_payload = { "name": role_name, "description": description if description is not None else payload.get("description", ""), "attributes": payload.get("attributes") if isinstance(payload, dict) else {}, } put_resp = client.put( f"/admin/realms/{self.realm}/clients/{client_uuid}/roles/{role_name}", json=update_payload, ) if put_resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_role_update_failed") return ProviderRoleSyncResult(role_name=role_name, action="updated") def update_client_role( self, *, provider_client_id: str, old_provider_role_name: str, new_provider_role_name: str, description: str | None = None, ) -> ProviderRoleSyncResult: old_name = old_provider_role_name.strip() new_name = new_provider_role_name.strip() if not old_name or not new_name: raise HTTPException(status_code=400, detail="provider_role_name_required") with self._client() as client: client_uuid = self._resolve_client_uuid(client, provider_client_id) get_resp = client.get(f"/admin/realms/{self.realm}/clients/{client_uuid}/roles/{old_name}") if get_resp.status_code == 404: # If old role missing, create the new one to self-heal drift. create_resp = client.post( f"/admin/realms/{self.realm}/clients/{client_uuid}/roles", json={"name": new_name, "description": description or ""}, ) if create_resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_role_create_failed") return ProviderRoleSyncResult(role_name=new_name, action="created") if get_resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_lookup_failed") payload = get_resp.json() if get_resp.content else {} update_payload = { "name": new_name, "description": description if description is not None else payload.get("description", ""), "attributes": payload.get("attributes") if isinstance(payload, dict) else {}, } put_resp = client.put( f"/admin/realms/{self.realm}/clients/{client_uuid}/roles/{old_name}", json=update_payload, ) if put_resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_role_update_failed") return ProviderRoleSyncResult(role_name=new_name, action="updated") def delete_client_role(self, *, provider_client_id: str, provider_role_name: str) -> ProviderDeleteResult: role_name = provider_role_name.strip() if not role_name: return ProviderDeleteResult(action="not_found") with self._client() as client: client_uuid = self._resolve_client_uuid(client, provider_client_id) resp = client.delete(f"/admin/realms/{self.realm}/clients/{client_uuid}/roles/{role_name}") if resp.status_code in {204, 404}: return ProviderDeleteResult(action="deleted" if resp.status_code == 204 else "not_found") if resp.status_code >= 400: raise HTTPException(status_code=502, detail="idp_role_delete_failed") return ProviderDeleteResult(action="deleted")