568 lines
25 KiB
Python
568 lines
25 KiB
Python
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")
|