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 AuthentikSyncResult: user_id: str action: str user_sub: str | None = None @dataclass class AuthentikPasswordResetResult: user_id: str temporary_password: str @dataclass class AuthentikDeleteResult: action: str user_id: str | None = None class AuthentikAdminService: """ Backward-compatible service name. Supports Keycloak (preferred, when KEYCLOAK_* configured) and Authentik. """ def __init__(self, settings: Settings) -> None: self.settings = settings self.is_keycloak = settings.use_keycloak self.verify_tls = settings.idp_verify_tls if self.is_keycloak: 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 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="authentik_admin_not_configured") else: self.base_url = settings.authentik_base_url.rstrip("/") self.admin_token = settings.authentik_admin_token if not self.base_url or not self.admin_token: raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="authentik_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)) @staticmethod def _extract_first_result(data: dict) -> dict | None: results = data.get("results") if isinstance(data, dict) else None return results[0] if isinstance(results, list) and results else None def _get_keycloak_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="authentik_lookup_failed") from exc if resp.status_code >= 400: raise HTTPException(status_code=502, detail="authentik_lookup_failed") token = resp.json().get("access_token") if not token: raise HTTPException(status_code=502, detail="authentik_lookup_failed") return str(token) def _client(self) -> httpx.Client: if self.is_keycloak: bearer_token = self._get_keycloak_admin_token() else: bearer_token = self.admin_token return httpx.Client( base_url=self.base_url, headers={ "Authorization": f"Bearer {bearer_token}", "Accept": "application/json", "Content-Type": "application/json", }, timeout=10, verify=self.verify_tls, ) # -------- Authentik lookups -------- def _ak_lookup_user_by_id(self, client: httpx.Client, user_id: str) -> dict | None: resp = client.get(f"/api/v3/core/users/{user_id}/") if resp.status_code == 404: return None if resp.status_code >= 400: raise HTTPException(status_code=502, detail="authentik_lookup_failed") return resp.json() def _ak_lookup_user_by_email_or_username( self, client: httpx.Client, *, email: str | None, username: str | None ) -> dict | None: if email: resp = client.get("/api/v3/core/users/", params={"email": email}) if resp.status_code >= 400: raise HTTPException(status_code=502, detail="authentik_lookup_failed") existing = self._extract_first_result(resp.json()) if existing: return existing if username: resp = client.get("/api/v3/core/users/", params={"username": username}) if resp.status_code >= 400: raise HTTPException(status_code=502, detail="authentik_lookup_failed") existing = self._extract_first_result(resp.json()) if existing: return existing return None # -------- Keycloak lookups -------- def _kc_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="authentik_lookup_failed") return resp.json() def _kc_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="authentik_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="authentik_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, idp_user_id: str | None = None, ) -> AuthentikSyncResult: resolved_username = username or self._safe_username(sub=sub, email=email) with self._client() as client: if self.is_keycloak: return self._ensure_user_keycloak( client, sub=sub, email=email, resolved_username=resolved_username, display_name=display_name, is_active=is_active, idp_user_id=idp_user_id, ) return self._ensure_user_authentik( client, sub=sub, email=email, resolved_username=resolved_username, display_name=display_name, is_active=is_active, idp_user_id=idp_user_id, ) def _ensure_user_authentik( self, client: httpx.Client, *, sub: str | None, email: str, resolved_username: str, display_name: str | None, is_active: bool, idp_user_id: str | None, ) -> AuthentikSyncResult: payload = { "username": resolved_username, "name": display_name or email, "email": email, "is_active": is_active, } existing = None if idp_user_id: existing = self._ak_lookup_user_by_id(client, idp_user_id) if existing is None: existing = self._ak_lookup_user_by_email_or_username(client, email=email, username=resolved_username) if existing and existing.get("pk") is not None: user_pk = str(existing["pk"]) patch_resp = client.patch(f"/api/v3/core/users/{user_pk}/", json=payload) if patch_resp.status_code >= 400: raise HTTPException(status_code=502, detail="authentik_update_failed") return AuthentikSyncResult(user_id=user_pk, action="updated", user_sub=existing.get("uid")) create_resp = client.post("/api/v3/core/users/", json=payload) if create_resp.status_code >= 400: raise HTTPException(status_code=502, detail="authentik_create_failed") created = create_resp.json() return AuthentikSyncResult(user_id=str(created["pk"]), action="created", user_sub=created.get("uid")) def _ensure_user_keycloak( self, client: httpx.Client, *, sub: str | None, email: str, resolved_username: str, display_name: str | None, is_active: bool, idp_user_id: str | None, ) -> AuthentikSyncResult: 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 {}, } existing = None if idp_user_id: existing = self._kc_lookup_user_by_id(client, idp_user_id) if existing is None: existing = self._kc_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="authentik_update_failed") return AuthentikSyncResult(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="authentik_create_failed") user_id: str | None = None location = create_resp.headers.get("Location", "") if location and "/" in location: user_id = location.rstrip("/").split("/")[-1] if not user_id: found = self._kc_lookup_user_by_email_or_username(client, email=email, username=resolved_username) user_id = str(found["id"]) if found and found.get("id") else None if not user_id: raise HTTPException(status_code=502, detail="authentik_create_failed") return AuthentikSyncResult(user_id=user_id, action="created", user_sub=user_id) def reset_password( self, *, idp_user_id: str | None, email: str | None, username: str | None, ) -> AuthentikPasswordResetResult: with self._client() as client: if self.is_keycloak: existing = None if idp_user_id: existing = self._kc_lookup_user_by_id(client, idp_user_id) if existing is None: existing = self._kc_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="authentik_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="authentik_set_password_failed") return AuthentikPasswordResetResult(user_id=user_id, temporary_password=temp_password) existing = None if idp_user_id: existing = self._ak_lookup_user_by_id(client, idp_user_id) if existing is None: existing = self._ak_lookup_user_by_email_or_username(client, email=email, username=username) if not existing or existing.get("pk") is None: raise HTTPException(status_code=404, detail="authentik_user_not_found") user_pk = str(existing["pk"]) temp_password = self._generate_temporary_password() set_pwd_resp = client.post(f"/api/v3/core/users/{user_pk}/set_password/", json={"password": temp_password}) if set_pwd_resp.status_code >= 400: raise HTTPException(status_code=502, detail="authentik_set_password_failed") return AuthentikPasswordResetResult(user_id=user_pk, temporary_password=temp_password) def delete_user( self, *, idp_user_id: str | None, email: str | None, username: str | None, ) -> AuthentikDeleteResult: with self._client() as client: if self.is_keycloak: existing = None if idp_user_id: existing = self._kc_lookup_user_by_id(client, idp_user_id) if existing is None: existing = self._kc_lookup_user_by_email_or_username(client, email=email, username=username) if not existing or not existing.get("id"): return AuthentikDeleteResult(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 AuthentikDeleteResult(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="authentik_delete_failed") return AuthentikDeleteResult(action="deleted", user_id=user_id) existing = None if idp_user_id: existing = self._ak_lookup_user_by_id(client, idp_user_id) if existing is None: existing = self._ak_lookup_user_by_email_or_username(client, email=email, username=username) if not existing or existing.get("pk") is None: return AuthentikDeleteResult(action="not_found") user_pk = str(existing["pk"]) delete_resp = client.delete(f"/api/v3/core/users/{user_pk}/") if delete_resp.status_code in {204, 404}: return AuthentikDeleteResult( action="deleted" if delete_resp.status_code == 204 else "not_found", user_id=user_pk, ) if delete_resp.status_code >= 400: raise HTTPException(status_code=502, detail="authentik_delete_failed") return AuthentikDeleteResult(action="deleted", user_id=user_pk)