Developer
193 lines
8.4 KiB
Python
193 lines
8.4 KiB
Python
"""Read-only Authentik directory client.
|
|
|
|
The client normalizes the subset of Authentik core data that Manage displays.
|
|
It deliberately does not fetch individual users or expose policy/provider data.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
import requests
|
|
|
|
from media_library_viewer_api.clients.http_timeout import DEFAULT_READ_TIMEOUT, http_timeout
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_MAX_COLLECTION_ITEMS = 10_000
|
|
_PAGE_SIZE = 100
|
|
|
|
|
|
def _text(value: Any) -> str:
|
|
return str(value).strip() if value is not None else ""
|
|
|
|
|
|
def _identifier(item: dict[str, Any]) -> str:
|
|
for key in ("pk", "id", "uuid"):
|
|
value = _text(item.get(key))
|
|
if value:
|
|
return value
|
|
return ""
|
|
|
|
|
|
def _page_total(payload: dict[str, Any], fallback: int) -> int:
|
|
pagination = payload.get("pagination")
|
|
if isinstance(pagination, dict):
|
|
try:
|
|
return max(0, int(pagination.get("count") or fallback))
|
|
except (TypeError, ValueError):
|
|
pass
|
|
return fallback
|
|
|
|
|
|
class AuthentikClient:
|
|
"""Small wrapper around Authentik's read-only core API."""
|
|
|
|
def __init__(self, base_url: str, api_token: str, timeout: float = DEFAULT_READ_TIMEOUT):
|
|
if not base_url:
|
|
raise ValueError("Authentik base_url is required")
|
|
if not api_token:
|
|
raise ValueError("Authentik API token is required")
|
|
|
|
self.base_url = base_url.rstrip("/")
|
|
if self.base_url.endswith("/api/v3"):
|
|
self.base_url = self.base_url[:-7]
|
|
self.api_token = api_token
|
|
self.timeout = http_timeout(timeout)
|
|
self.session = requests.Session()
|
|
self.session.headers.update({"Authorization": f"Bearer {api_token}", "Accept": "application/json"})
|
|
|
|
def get(self, path: str, **params: Any) -> Any:
|
|
"""GET an Authentik endpoint and include useful response text on errors."""
|
|
clean_params = {key: value for key, value in params.items() if value is not None and value != ""}
|
|
logger.debug("Authentik GET %s params=%s", path, sorted(clean_params.keys()))
|
|
response = self.session.get(f"{self.base_url}/api/v3{path}", params=clean_params, timeout=self.timeout)
|
|
try:
|
|
response.raise_for_status()
|
|
except requests.HTTPError as exc:
|
|
detail = response.text[:500]
|
|
logger.warning("Authentik GET %s failed status=%s url=%s", path, response.status_code, response.url)
|
|
raise requests.HTTPError(f"{response.status_code} for {response.url}: {detail}", response=response) from exc
|
|
return response.json()
|
|
|
|
def users(self, search: str | None = None, page: int = 1, page_size: int = 50) -> dict[str, Any]:
|
|
"""Return one raw user page for the directory and messaging surfaces."""
|
|
payload = self.get("/core/users/", search=search, page=page, page_size=page_size)
|
|
if not isinstance(payload, dict):
|
|
logger.warning("Authentik users payload was not a dict: %s", type(payload).__name__)
|
|
return {"items": [], "total": 0, "page": page, "page_size": page_size}
|
|
results = payload.get("results")
|
|
items = [item for item in results if isinstance(item, dict)] if isinstance(results, list) else []
|
|
return {"items": items, "total": _page_total(payload, len(items)), "page": page, "page_size": page_size}
|
|
|
|
def _collection(self, path: str, limit: int = _MAX_COLLECTION_ITEMS) -> dict[str, Any]:
|
|
"""Read a paginated core collection with a hard cap and loop protection."""
|
|
try:
|
|
requested = max(1, min(int(limit), _MAX_COLLECTION_ITEMS))
|
|
except (TypeError, ValueError):
|
|
requested = _MAX_COLLECTION_ITEMS
|
|
items: list[dict[str, Any]] = []
|
|
page = 1
|
|
total = 0
|
|
while len(items) < requested:
|
|
payload = self.get(path, page=page, page_size=min(_PAGE_SIZE, requested - len(items)))
|
|
if not isinstance(payload, dict):
|
|
logger.warning("Authentik %s payload was not a dict: %s", path, type(payload).__name__)
|
|
break
|
|
results = payload.get("results")
|
|
page_items = [item for item in results if isinstance(item, dict)] if isinstance(results, list) else []
|
|
total = _page_total(payload, len(items) + len(page_items))
|
|
items.extend(page_items[: requested - len(items)])
|
|
if not page_items or len(items) >= total:
|
|
break
|
|
page += 1
|
|
if page > 100: # defensive limit for malformed pagination responses
|
|
logger.warning("Authentik %s pagination stopped after 100 pages", path)
|
|
break
|
|
return {"items": items, "total": total or len(items)}
|
|
|
|
@staticmethod
|
|
def _normalize_group(item: dict[str, Any]) -> dict[str, str] | None:
|
|
group_id = _identifier(item)
|
|
if not group_id:
|
|
return None
|
|
name = _text(item.get("name") or item.get("display_name") or item.get("slug"))
|
|
return {"id": group_id, "name": name or f"Unnamed group ({group_id})"}
|
|
|
|
@staticmethod
|
|
def _normalize_application(item: dict[str, Any]) -> dict[str, str]:
|
|
app_id = _identifier(item)
|
|
return {
|
|
"id": app_id,
|
|
"name": _text(item.get("name") or item.get("slug") or item.get("meta_name")) or "Unnamed application",
|
|
"slug": _text(item.get("slug")),
|
|
"launch_url": _text(item.get("launch_url") or item.get("meta_launch_url")),
|
|
}
|
|
|
|
def groups(self, limit: int = _MAX_COLLECTION_ITEMS) -> dict[str, Any]:
|
|
"""Return normalized groups; only display-safe identifiers and names are retained."""
|
|
raw = self._collection("/core/groups/", limit)
|
|
items = [normalized for item in raw["items"] if (normalized := self._normalize_group(item)) is not None]
|
|
return {"items": items, "total": raw["total"]}
|
|
|
|
def applications(self, limit: int = _MAX_COLLECTION_ITEMS) -> dict[str, Any]:
|
|
"""Return normalized applications without provider, policy, or secret fields."""
|
|
raw = self._collection("/core/applications/", limit)
|
|
return {"items": [self._normalize_application(item) for item in raw["items"]], "total": raw["total"]}
|
|
|
|
@staticmethod
|
|
def _group_references(user: dict[str, Any]) -> list[str]:
|
|
"""Extract group ids from release-dependent user reference shapes."""
|
|
raw = user.get("groups", user.get("group", []))
|
|
if not isinstance(raw, list):
|
|
raw = [raw] if raw is not None else []
|
|
ids: list[str] = []
|
|
for reference in raw:
|
|
if isinstance(reference, dict):
|
|
group_id = _identifier(reference)
|
|
else:
|
|
group_id = _text(reference)
|
|
if group_id and group_id not in ids:
|
|
ids.append(group_id)
|
|
return ids
|
|
|
|
def access_summaries(
|
|
self,
|
|
search: str | None = None,
|
|
page: int = 1,
|
|
page_size: int = 50,
|
|
) -> dict[str, Any]:
|
|
"""Summarize user group references and privileged flags without N+1 user reads.
|
|
|
|
This is directory metadata only: group membership plus the explicit
|
|
``is_superuser`` and ``is_staff`` fields. It does not evaluate policies
|
|
or claim to calculate effective authorization.
|
|
"""
|
|
users = self.users(search=search, page=page, page_size=page_size)
|
|
groups = self.groups()
|
|
group_names = {group["id"]: group["name"] for group in groups["items"]}
|
|
summaries: list[dict[str, Any]] = []
|
|
for user in users["items"]:
|
|
group_ids = self._group_references(user)
|
|
summaries.append(
|
|
{
|
|
"id": _identifier(user),
|
|
"username": _text(user.get("username")),
|
|
"name": _text(user.get("name")),
|
|
"email": _text(user.get("email")),
|
|
"is_active": bool(user.get("is_active", True)),
|
|
"is_superuser": bool(user.get("is_superuser", False)),
|
|
"is_staff": bool(user.get("is_staff", False)),
|
|
"groups": [
|
|
{
|
|
"id": group_id,
|
|
"name": group_names.get(group_id, f"Unknown group ({group_id})"),
|
|
"known": group_id in group_names,
|
|
}
|
|
for group_id in group_ids
|
|
],
|
|
}
|
|
)
|
|
return {"items": summaries, "total": users["total"], "page": users["page"], "page_size": users["page_size"]}
|