# Copyright (c) 2025 Ferrosoft GmbH. All rights reserved.
"""HTTP REST client and request/response dataclasses for the ZITADEL IAM API v2."""
from dataclasses import dataclass, field, fields, is_dataclass
from typing import Any, Dict, List, Optional
from urllib.parse import urljoin, urlencode
import requests
from requests import JSONDecodeError
from ferrosoft.apps.ferrobase.auth.roles import Role
from ferrosoft.config import require_settings
[docs]
def iam_url(path: str) -> str:
"""Return ``path`` joined onto the configured ``OIDC_OP_BASE_URL``."""
settings = require_settings("OIDC_OP_BASE_URL")
return urljoin(settings["OIDC_OP_BASE_URL"], path)
[docs]
def zitadel_logout(request):
"""Build the OIDC end-session URL that signs the user out of Zitadel."""
settings = require_settings("OIDC_RP_CLIENT_ID")
query = urlencode(
{
"client_id": settings["OIDC_RP_CLIENT_ID"],
"post_logout_redirect_uri": request.build_absolute_uri("/"),
"state": "",
}
)
return iam_url("/oidc/v1/end_session?" + query)
def _to_dict(obj: Any) -> Any:
"""Recursively convert dataclasses to dicts, dropping ``None`` and empty lists."""
if is_dataclass(obj):
result: Dict[str, Any] = {}
for f in fields(obj):
val = getattr(obj, f.name)
if val is None:
continue
if isinstance(val, list):
converted = [_to_dict(item) for item in val]
if converted: # keep only non-empty lists
result[f.name] = converted
else:
result[f.name] = _to_dict(val)
return result
if isinstance(obj, dict):
return {k: _to_dict(v) for k, v in obj.items()}
if isinstance(obj, (str, int, float, bool)) or obj is None:
return obj
return obj
[docs]
@dataclass
class EmptyObject:
"""Represents an empty JSON object {}."""
[docs]
def to_dict(self) -> Dict[str, Any]:
"""Serialise as an empty dict."""
return {}
[docs]
@dataclass
class CodeAction:
"""Represents actions like sendCode which may carry an optional urlTemplate."""
urlTemplate: Optional[str] = None
[docs]
@dataclass
class Profile:
"""User profile attributes as expected by the Zitadel ``Human`` payload."""
givenName: str
familyName: str
nickName: Optional[str] = None
displayName: Optional[str] = None
preferredLanguage: Optional[str] = None
gender: Optional[str] = None # e.g. "GENDER_FEMALE"
[docs]
@dataclass
class Email:
"""Email object with optional verification flow descriptors."""
email: str
sendCode: Optional[CodeAction] = None
returnCode: Optional[EmptyObject] = None
isVerified: Optional[bool] = None
[docs]
@dataclass
class Phone:
"""Phone object with optional verification flow descriptors."""
phone: str
sendCode: Optional[CodeAction] = None
returnCode: Optional[EmptyObject] = None
isVerified: Optional[bool] = None
[docs]
@dataclass
class Password:
"""Initial password assignment, optionally forcing change on next login."""
password: str
changeRequired: Optional[bool] = None
[docs]
@dataclass
class IDPLink:
"""Identity-provider linkage referencing an external user."""
idpId: Optional[str] = None
userId: Optional[str] = None
userName: Optional[str] = None
[docs]
@dataclass
class Human:
"""Human-user payload sent when creating Zitadel users."""
email: Email
profile: Profile
userId: Optional[str] = None
username: Optional[str] = None
phone: Optional[Phone] = None
metadata: List[MetadataItem] = field(default_factory=list)
password: Optional[Password] = None
idpLinks: List[IDPLink] = field(default_factory=list)
[docs]
@dataclass
class Admin:
"""Organisation administrator entry (existing ``userId`` or inline ``human``)."""
userId: Optional[str] = None
human: Optional[Human] = None
roles: List[str] = field(default_factory=list)
[docs]
@dataclass
class Organization:
"""Organisation payload combining name, admins and optional ``orgId``."""
name: str
admins: List[Admin] = field(default_factory=list)
orgId: Optional[str] = None
[docs]
def to_dict(self) -> Dict[str, Any]:
"""Serialise to a JSON-ready dict via :func:`_to_dict`."""
return _to_dict(self)
[docs]
class ZitadelClientError(Exception):
"""Raised for non-successful responses or OAuth failures."""
[docs]
class ZitadelClient:
"""
Minimal HTTP REST client for ZITADEL API v2 focusing on "Create an Organization".
Authentication:
- OAuth2 Client Credentials via /oauth/v2/token
- Scopes are passed as a space-separated string; for service users you typically need the audience scope
"urn:zitadel:iam:org:project:id:<PROJECT_ID>:aud" among others, depending on your setup.
Endpoints:
- Token: {base_url}/oauth/v2/token
- Organizations: {base_url}/v2/organizations (Create via POST)
"""
def __init__(
self,
base_url: str,
api_token: str,
timeout: float = 10.0,
session: Optional[requests.Session] = None,
) -> None:
"""
Args:
base_url: Your ZITADEL instance base URL, e.g. "https://your-tenant.zitadel.cloud"
api_token: Bearer API token for a service user/application
timeout: Default request timeout in seconds
session: Optional requests.Session to reuse TCP connections
"""
if not base_url:
raise ValueError("base_url is required")
if not api_token:
raise ValueError("api_token is required")
self.base_url = base_url.rstrip("/")
self.api_token = api_token
self.timeout = timeout
self._session = session or requests.Session()
self._access_token: Optional[str] = None
self._token_expiry_epoch: float = 0.0 # epoch seconds
[docs]
@classmethod
def from_settings(cls, **kwargs):
settings = require_settings("OIDC_OP_BASE_URL", "ZITADEL_API_TOKEN")
return cls(
base_url=settings["OIDC_OP_BASE_URL"],
api_token=settings["ZITADEL_API_TOKEN"],
**kwargs,
)
def _auth_headers(self) -> Dict[str, str]:
"""Return the bearer-token Authorization header."""
return {"Authorization": f"Bearer {self.api_token}"}
def _endpoint_url(self, endpoint: str) -> str:
"""Resolve ``endpoint`` against the configured ``base_url``."""
return urljoin(self.base_url, endpoint)
def _post(self, endpoint: str, json: Dict[str, Any], **kwargs) -> Dict[str, Any]:
"""POST ``json`` to ``endpoint`` and return the parsed JSON response.
Raises:
ZitadelClientError: For any non-200/201 response or invalid JSON.
"""
headers = kwargs.pop("headers", {}) | {
"Accept": "application/json",
"Content-Type": "application/json",
**self._auth_headers(),
}
resp = self._session.post(
endpoint,
json=json,
headers=headers,
timeout=self.timeout,
**kwargs,
)
if resp.status_code not in (200, 201):
# Try to propagate error details
detail = None
try:
detail = resp.json()
except JSONDecodeError:
detail = resp.text
raise ZitadelClientError(
f"Zitadel {endpoint} returned {resp.status_code} {detail}"
)
try:
return resp.json()
except Exception as e:
raise ZitadelClientError(f"Invalid JSON response: {e}") from e
[docs]
def create_organization(
self,
organization: Organization,
) -> Dict[str, Any]:
"""
Create an organization.
Args:
organization: The organization to create.
Returns:
The API response JSON as a dict.
Raises:
ZitadelClientError on HTTP errors or unexpected responses.
"""
return self._post(
self._endpoint_url("/v2/organizations"), organization.to_dict()
)
[docs]
def create_project_grant(
self,
project_id: str,
organization_id: str,
*roles: Role,
):
"""Grant ``roles`` of ``project_id`` to ``organization_id``."""
return self._post(
self._endpoint_url("/management/v1/projects/%s/grants" % project_id),
{
"grantedOrgId": organization_id,
"roleKeys": list([str(role) for role in roles]),
},
)
[docs]
def create_authorization(
self,
user_id: str,
organization_id: str,
project_id: str,
project_grant_id: str,
*role_keys: str,
):
"""Assign ``role_keys`` of a project grant to a specific user."""
return self._post(
self._endpoint_url("/management/v1/users/%s/grants" % user_id),
{
"projectId": project_id,
"projectGrantId": project_grant_id,
"roleKeys": list(role_keys),
},
headers={
"x-zitadel-orgid": organization_id,
},
)