member/member-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(idp): add keycloak-first support with authentik fallback

Browse Source
This commit is contained in:
Chris
2026-04-01 00:41:38 +08:00
parent f6105f079d
commit c7ed517ed2
16 changed files with 435 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
app/api/admin_catalog.py
View File

@@ -136,12 +136,12 @@ def _generate_api_key() -> str:
def _sync_member_to_authentik(
*,
user_sub: str | None,
idp_user_id: int | None,
idp_user_id: str | None,
username: str | None,
email: str | None,
display_name: str | None,
is_active: bool,
) -> dict[str, str | int]:
) -> dict[str, str]:
if not email:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="email_required_for_authentik_sync")
settings = get_settings()
@@ -602,7 +602,7 @@ def upsert_member(
display_name=payload.display_name,
is_active=payload.is_active,
)
idp_user_id = int(sync["idp_user_id"])
idp_user_id = str(sync["idp_user_id"])
if sync.get("user_sub"):
resolved_sub = str(sync["user_sub"])
if not resolved_sub:
@@ -651,7 +651,7 @@ def update_member(
display_name=next_display_name,
is_active=next_is_active,
)
idp_user_id = int(sync["idp_user_id"])
idp_user_id = str(sync["idp_user_id"])
row = users_repo.upsert_by_sub(
user_sub=row.user_sub,

41
app/api/auth.py
View File

@@ -1,6 +1,5 @@
import logging
import secrets
from urllib.parse import urljoin
import httpx
from fastapi import APIRouter, HTTPException, status
@@ -18,6 +17,9 @@ logger = logging.getLogger(__name__)
def _resolve_username_by_email(settings, email: str) -> str | None:
# Authentik-only helper. Keycloak does not need this path.
if settings.use_keycloak:
return None
if not settings.authentik_base_url or not settings.authentik_admin_token:
return None
@@ -51,19 +53,17 @@ def _resolve_username_by_email(settings, email: str) -> str | None:
@router.post("/login", response_model=LoginResponse)
def login(payload: LoginRequest) -> LoginResponse:
settings = get_settings()
client_id = settings.authentik_client_id or settings.authentik_audience
client_id = settings.idp_client_id or settings.idp_audience
if not settings.authentik_base_url or not client_id or not settings.authentik_client_secret:
if not settings.idp_base_url or not client_id or not settings.idp_client_secret:
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="authentik_login_not_configured")
token_endpoint = settings.authentik_token_endpoint or urljoin(
settings.authentik_base_url.rstrip("/") + "/", "application/o/token/"
)
token_endpoint = settings.idp_token_endpoint
form = {
"grant_type": "password",
"client_id": client_id,
"client_secret": settings.authentik_client_secret,
"client_secret": settings.idp_client_secret,
"username": payload.username,
"password": payload.password,
"scope": "openid profile email",
@@ -74,7 +74,7 @@ def login(payload: LoginRequest) -> LoginResponse:
token_endpoint,
data=form_data,
timeout=10,
verify=settings.authentik_verify_tls,
verify=settings.idp_verify_tls,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
return resp
@@ -95,7 +95,7 @@ def login(payload: LoginRequest) -> LoginResponse:
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="authentik_unreachable") from exc
if resp.status_code >= 400:
logger.warning("authentik password grant failed: status=%s body=%s", resp.status_code, resp.text)
logger.warning("idp password grant failed: status=%s body=%s", resp.status_code, resp.text)
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid_username_or_password")
data = resp.json()
@@ -116,13 +116,14 @@ def get_oidc_authorize_url(
redirect_uri: str,
login_hint: str | None = None,
prompt: str = "login",
idp_hint: str | None = None,
) -> OIDCAuthUrlResponse:
settings = get_settings()
client_id = settings.authentik_client_id or settings.authentik_audience
if not settings.authentik_base_url or not client_id:
client_id = settings.idp_client_id or settings.idp_audience
if not settings.idp_base_url or not client_id:
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="authentik_login_not_configured")
authorize_endpoint = urljoin(settings.authentik_base_url.rstrip("/") + "/", "application/o/authorize/")
authorize_endpoint = settings.idp_authorize_endpoint
state = secrets.token_urlsafe(24)
query = {
"client_id": client_id,
@@ -134,6 +135,8 @@ def get_oidc_authorize_url(
}
if login_hint:
query["login_hint"] = login_hint
if idp_hint and settings.use_keycloak:
query["kc_idp_hint"] = idp_hint
params = httpx.QueryParams(query)
return OIDCAuthUrlResponse(authorize_url=f"{authorize_endpoint}?{params}")
@@ -142,17 +145,15 @@ def get_oidc_authorize_url(
@router.post("/oidc/exchange", response_model=LoginResponse)
def exchange_oidc_code(payload: OIDCCodeExchangeRequest) -> LoginResponse:
settings = get_settings()
client_id = settings.authentik_client_id or settings.authentik_audience
if not settings.authentik_base_url or not client_id or not settings.authentik_client_secret:
client_id = settings.idp_client_id or settings.idp_audience
if not settings.idp_base_url or not client_id or not settings.idp_client_secret:
raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="authentik_login_not_configured")
token_endpoint = settings.authentik_token_endpoint or urljoin(
settings.authentik_base_url.rstrip("/") + "/", "application/o/token/"
)
token_endpoint = settings.idp_token_endpoint
form = {
"grant_type": "authorization_code",
"client_id": client_id,
"client_secret": settings.authentik_client_secret,
"client_secret": settings.idp_client_secret,
"code": payload.code,
"redirect_uri": payload.redirect_uri,
}
@@ -162,14 +163,14 @@ def exchange_oidc_code(payload: OIDCCodeExchangeRequest) -> LoginResponse:
token_endpoint,
data=form,
timeout=10,
verify=settings.authentik_verify_tls,
verify=settings.idp_verify_tls,
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
except Exception as exc:
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="authentik_unreachable") from exc
if resp.status_code >= 400:
logger.warning("authentik auth-code exchange failed: status=%s body=%s", resp.status_code, resp.text)
logger.warning("idp auth-code exchange failed: status=%s body=%s", resp.status_code, resp.text)
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="authentik_code_exchange_failed")
data = resp.json()

2
app/api/internal.py
View File

@@ -57,6 +57,8 @@ def get_permission_snapshot(
@router.post("/authentik/users/ensure", response_model=AuthentikEnsureUserResponse)
@router.post("/idp/users/ensure", response_model=AuthentikEnsureUserResponse)
@router.post("/keycloak/users/ensure", response_model=AuthentikEnsureUserResponse)
def ensure_authentik_user(
payload: AuthentikEnsureUserRequest,
db: Session = Depends(get_db),

88
app/core/config.py
View File

@@ -27,6 +27,20 @@ class Settings(BaseSettings):
authentik_client_secret: str = ""
authentik_token_endpoint: str = ""
authentik_userinfo_endpoint: str = ""
# Keycloak (preferred when configured)
keycloak_base_url: str = ""
keycloak_realm: str = ""
keycloak_verify_tls: bool = True
keycloak_issuer: str = ""
keycloak_jwks_url: str = ""
keycloak_audience: str = ""
keycloak_client_id: str = ""
keycloak_client_secret: str = ""
keycloak_token_endpoint: str = ""
keycloak_userinfo_endpoint: str = ""
keycloak_admin_client_id: str = ""
keycloak_admin_client_secret: str = ""
keycloak_admin_realm: str = ""
public_frontend_origins: Annotated[list[str], NoDecode] = ["https://member.ose.tw"]
internal_shared_secret: str = ""
@@ -57,6 +71,80 @@ class Settings(BaseSettings):
f"{self.db_user}:{self.db_password}@{self.db_host}:{self.db_port}/{self.db_name}"
)
@property
def use_keycloak(self) -> bool:
return bool(self.keycloak_base_url and self.keycloak_realm)
@property
def idp_base_url(self) -> str:
if self.use_keycloak:
return self.keycloak_base_url.rstrip("/")
return self.authentik_base_url.rstrip("/")
@property
def idp_verify_tls(self) -> bool:
if self.use_keycloak:
return self.keycloak_verify_tls
return self.authentik_verify_tls
@property
def idp_issuer(self) -> str:
if self.use_keycloak:
if self.keycloak_issuer:
return self.keycloak_issuer.rstrip("/")
return f"{self.idp_base_url}/realms/{self.keycloak_realm}"
return self.authentik_issuer.rstrip("/")
@property
def idp_jwks_url(self) -> str:
if self.use_keycloak:
if self.keycloak_jwks_url:
return self.keycloak_jwks_url
return f"{self.idp_issuer}/protocol/openid-connect/certs"
return self.authentik_jwks_url
@property
def idp_audience(self) -> str:
if self.use_keycloak:
return self.keycloak_audience or self.keycloak_client_id
return self.authentik_audience or self.authentik_client_id
@property
def idp_client_id(self) -> str:
if self.use_keycloak:
return self.keycloak_client_id
return self.authentik_client_id
@property
def idp_client_secret(self) -> str:
if self.use_keycloak:
return self.keycloak_client_secret
return self.authentik_client_secret
@property
def idp_token_endpoint(self) -> str:
if self.use_keycloak:
if self.keycloak_token_endpoint:
return self.keycloak_token_endpoint
return f"{self.idp_issuer}/protocol/openid-connect/token"
return self.authentik_token_endpoint or (f"{self.idp_base_url}/application/o/token/" if self.idp_base_url else "")
@property
def idp_userinfo_endpoint(self) -> str:
if self.use_keycloak:
if self.keycloak_userinfo_endpoint:
return self.keycloak_userinfo_endpoint
return f"{self.idp_issuer}/protocol/openid-connect/userinfo"
return self.authentik_userinfo_endpoint or (
f"{self.idp_base_url}/application/o/userinfo/" if self.idp_base_url else ""
)
@property
def idp_authorize_endpoint(self) -> str:
if self.use_keycloak:
return f"{self.idp_issuer}/protocol/openid-connect/auth"
return f"{self.idp_base_url}/application/o/authorize/" if self.idp_base_url else ""
@lru_cache
def get_settings() -> Settings:

4
app/models/user.py
View File

@@ -1,7 +1,7 @@
from datetime import datetime
from uuid import uuid4
from sqlalchemy import Boolean, DateTime, Integer, String, func
from sqlalchemy import Boolean, DateTime, String, func
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column
@@ -13,7 +13,7 @@ class User(Base):
id: Mapped[str] = mapped_column(UUID(as_uuid=False), primary_key=True, default=lambda: str(uuid4()))
user_sub: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True)
idp_user_id: Mapped[int | None] = mapped_column(Integer)
idp_user_id: Mapped[str | None] = mapped_column(String(128))
username: Mapped[str | None] = mapped_column(String(255), unique=True)
email: Mapped[str | None] = mapped_column(String(320))
display_name: Mapped[str | None] = mapped_column(String(255))

2
app/repositories/users_repo.py
View File

@@ -53,7 +53,7 @@ class UsersRepository:
email: str | None,
display_name: str | None,
is_active: bool,
idp_user_id: int | None = None,
idp_user_id: str | None = None,
) -> User:
user = self.get_by_sub(user_sub)
if user is None:

2
app/schemas/authentik_admin.py
View File

@@ -10,5 +10,5 @@ class AuthentikEnsureUserRequest(BaseModel):
class AuthentikEnsureUserResponse(BaseModel):
idp_user_id: int
idp_user_id: str
action: str

2
app/schemas/internal.py
View File

@@ -78,7 +78,7 @@ class InternalMemberListResponse(BaseModel):
class InternalUpsertUserBySubResponse(BaseModel):
id: str
user_sub: str
idp_user_id: int | None = None
idp_user_id: str | None = None
username: str | None = None
email: str | None = None
display_name: str | None = None

20
app/security/authentik_jwt.py
View File

@@ -50,16 +50,18 @@ class AuthentikTokenVerifier:
@staticmethod
def _infer_userinfo_endpoint(issuer: str | None, base_url: str | None) -> str | None:
if base_url:
return base_url.rstrip("/") + "/application/o/userinfo/"
if issuer:
normalized = issuer.rstrip("/")
if "/realms/" in normalized:
return normalized + "/protocol/openid-connect/userinfo"
marker = "/application/o/"
marker_index = normalized.find(marker)
if marker_index != -1:
root = normalized[:marker_index]
return root + marker + "userinfo/"
return normalized + "/userinfo/"
if base_url:
return base_url.rstrip("/") + "/application/o/userinfo/"
return None
def _enrich_from_userinfo(self, principal: AuthentikPrincipal, token: str) -> AuthentikPrincipal:
@@ -156,13 +158,13 @@ class AuthentikTokenVerifier:
def _get_verifier() -> AuthentikTokenVerifier:
settings = get_settings()
return AuthentikTokenVerifier(
issuer=settings.authentik_issuer,
jwks_url=settings.authentik_jwks_url,
audience=settings.authentik_audience,
client_secret=settings.authentik_client_secret,
base_url=settings.authentik_base_url,
userinfo_endpoint=settings.authentik_userinfo_endpoint,
verify_tls=settings.authentik_verify_tls,
issuer=settings.idp_issuer,
jwks_url=settings.idp_jwks_url,
audience=settings.idp_audience,
client_secret=settings.idp_client_secret,
base_url=settings.idp_base_url,
userinfo_endpoint=settings.idp_userinfo_endpoint,
verify_tls=settings.idp_verify_tls,
)

297
app/services/authentik_admin_service.py
View File

@@ -12,46 +12,47 @@ from app.core.config import Settings
@dataclass
class AuthentikSyncResult:
user_id: int
user_id: str
action: str
user_sub: str | None = None
@dataclass
class AuthentikPasswordResetResult:
user_id: int
user_id: str
temporary_password: str
@dataclass
class AuthentikDeleteResult:
action: str
user_id: int | None = None
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.base_url = settings.authentik_base_url.rstrip("/")
self.admin_token = settings.authentik_admin_token
self.verify_tls = settings.authentik_verify_tls
self.settings = settings
self.is_keycloak = settings.use_keycloak
self.verify_tls = settings.idp_verify_tls
if not self.base_url or not self.admin_token:
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail="authentik_admin_not_configured",
)
def _client(self) -> httpx.Client:
return httpx.Client(
base_url=self.base_url,
headers={
"Authorization": f"Bearer {self.admin_token}",
"Accept": "application/json",
"Content-Type": "application/json",
},
timeout=10,
verify=self.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:
@@ -71,7 +72,48 @@ class AuthentikAdminService:
results = data.get("results") if isinstance(data, dict) else None
return results[0] if isinstance(results, list) and results else None
def _lookup_user_by_id(self, client: httpx.Client, user_id: int) -> dict | 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
@@ -79,7 +121,7 @@ class AuthentikAdminService:
raise HTTPException(status_code=502, detail="authentik_lookup_failed")
return resp.json()
def _lookup_user_by_email_or_username(
def _ak_lookup_user_by_email_or_username(
self, client: httpx.Client, *, email: str | None, username: str | None
) -> dict | None:
if email:
@@ -89,7 +131,6 @@ class AuthentikAdminService:
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:
@@ -97,7 +138,34 @@ class AuthentikAdminService:
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(
@@ -108,81 +176,188 @@ class AuthentikAdminService:
username: str | None,
display_name: str | None,
is_active: bool = True,
idp_user_id: int | None = None,
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)
with self._client() as client:
existing = None
if idp_user_id is not None:
existing = self._lookup_user_by_id(client, idp_user_id)
if existing is None:
existing = self._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"))
if existing and existing.get("pk") is not None:
user_pk = int(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"))
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=int(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: int | None,
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 is not None:
existing = self._lookup_user_by_id(client, idp_user_id)
if idp_user_id:
existing = self._ak_lookup_user_by_id(client, idp_user_id)
if existing is None:
existing = self._lookup_user_by_email_or_username(client, email=email, username=username)
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 = int(existing["pk"])
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: int | None,
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 is not None:
existing = self._lookup_user_by_id(client, idp_user_id)
if idp_user_id:
existing = self._ak_lookup_user_by_id(client, idp_user_id)
if existing is None:
existing = self._lookup_user_by_email_or_username(client, email=email, username=username)
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 = int(existing["pk"])
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(