refactor(keycloak): remove authentik naming and switch to keycloak-only paths
This commit is contained in:
Chris
209
app/services/idp_admin_service.py
Normal file
209
app/services/idp_admin_service.py
Normal file
@@ -0,0 +1,209 @@
|
||||
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 KeycloakSyncResult:
|
||||
user_id: str
|
||||
action: str
|
||||
user_sub: str | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class KeycloakPasswordResetResult:
|
||||
user_id: str
|
||||
temporary_password: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class KeycloakDeleteResult:
|
||||
action: str
|
||||
user_id: str | None = None
|
||||
|
||||
|
||||
class KeycloakAdminService:
|
||||
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_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,
|
||||
idp_user_id: str | None = None,
|
||||
) -> KeycloakSyncResult:
|
||||
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, idp_user_id) if idp_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 KeycloakSyncResult(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 KeycloakSyncResult(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,
|
||||
) -> KeycloakPasswordResetResult:
|
||||
with self._client() as client:
|
||||
existing = self._lookup_user_by_id(client, idp_user_id) if idp_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 KeycloakPasswordResetResult(user_id=user_id, temporary_password=temp_password)
|
||||
|
||||
def delete_user(
|
||||
self,
|
||||
*,
|
||||
idp_user_id: str | None,
|
||||
email: str | None,
|
||||
username: str | None,
|
||||
) -> KeycloakDeleteResult:
|
||||
with self._client() as client:
|
||||
existing = self._lookup_user_by_id(client, idp_user_id) if idp_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 KeycloakDeleteResult(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 KeycloakDeleteResult(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 KeycloakDeleteResult(action="deleted", user_id=user_id)
|
||||
Reference in New Issue
Block a user