Developer
691d78ff06
Extract the duplicated _resolve_service_record helper (identical in routers/monitoring.py and routers/authentik_users.py) into a shared services/service_resolution.py module. Both routers now import resolve_service_record from the shared module. The authentik router previously hardcoded service_type='authentik' in its local copy; the shared helper takes service_type as a param (same as monitoring's did). Tests updated: test_api.py patches now target the correct module paths (resolve_service_record on the monitoring module where it's imported, build_service_record on the service_resolution module). 283 backend tests pass; ruff clean.
123 lines
4.5 KiB
Python
123 lines
4.5 KiB
Python
"""Authentik directory + messaging router.
|
|
|
|
Resolves an ``authentik`` service instance from the registry, builds an
|
|
:class:`AuthentikClient` from its config + decrypted ``api_token`` secret, and
|
|
proxies paginated directory queries plus message-compose (email enqueue).
|
|
Graceful "not configured" / "unreachable" payloads (matching the monitoring
|
|
router's pattern) so the UI always renders.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends
|
|
from pydantic import BaseModel
|
|
|
|
from media_library_viewer_api.clients.authentik import AuthentikClient
|
|
from media_library_viewer_api.config import get_settings
|
|
from media_library_viewer_api.dependencies import get_mail_queue, get_settings_store
|
|
from media_library_viewer_api.services.mail_queue import MailQueue
|
|
from media_library_viewer_api.services.mailer import validate_smtp_settings
|
|
from media_library_viewer_api.services.service_resolution import resolve_service_record
|
|
from media_library_viewer_api.services.settings_store import SettingsStore
|
|
from media_library_viewer_api.widgets.sources import ServiceRecord
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/api/services/authentik", tags=["authentik"])
|
|
|
|
|
|
class MessageRequest(BaseModel):
|
|
"""Compose-request body for the Authentik messaging endpoint."""
|
|
|
|
recipient_emails: list[str]
|
|
subject: str
|
|
html_body: str
|
|
|
|
|
|
def _build_client(service: ServiceRecord) -> AuthentikClient:
|
|
base_url = str(service.config.get("base_url") or "").rstrip("/")
|
|
api_token = str(service.secrets.get("api_token") or "")
|
|
try:
|
|
timeout = float(service.config.get("timeout_seconds") or 10)
|
|
except (TypeError, ValueError):
|
|
timeout = 10.0
|
|
return AuthentikClient(base_url=base_url, api_token=api_token, timeout=timeout)
|
|
|
|
|
|
def _empty(error: str) -> dict[str, Any]:
|
|
return {"items": [], "total": 0, "page": 1, "page_size": 50, "error": error}
|
|
|
|
|
|
@router.get("/{service_id}/users")
|
|
def get_authentik_users(
|
|
service_id: str,
|
|
search: str | None = None,
|
|
page: int = 1,
|
|
page_size: int = 50,
|
|
store: SettingsStore = Depends(get_settings_store),
|
|
) -> dict[str, Any]:
|
|
"""Paginated Authentik user directory for a specific service instance."""
|
|
service = resolve_service_record(store, "authentik", service_id)
|
|
if service is None:
|
|
logger.info("Authentik users requested but no enabled authentik service for id=%s", service_id)
|
|
return _empty("Authentik service not configured")
|
|
|
|
try:
|
|
client = _build_client(service)
|
|
return client.users(search=search, page=page, page_size=page_size)
|
|
except Exception:
|
|
logger.exception("Authentik users query failed for service %s", service_id)
|
|
return _empty("Authentik is unreachable")
|
|
|
|
|
|
@router.get("/{service_id}/message/status")
|
|
def get_authentik_message_status(
|
|
service_id: str,
|
|
store: SettingsStore = Depends(get_settings_store),
|
|
mail_queue: MailQueue = Depends(get_mail_queue),
|
|
) -> dict[str, Any]:
|
|
"""Mail-queue status snapshot for the Authentik messaging tab."""
|
|
service = resolve_service_record(store, "authentik", service_id)
|
|
if service is None:
|
|
return {"state": "stopped", "worker_running": False, "error": "Authentik service not configured"}
|
|
return mail_queue.status()
|
|
|
|
|
|
@router.post("/{service_id}/message")
|
|
def post_authentik_message(
|
|
service_id: str,
|
|
body: MessageRequest,
|
|
store: SettingsStore = Depends(get_settings_store),
|
|
mail_queue: MailQueue = Depends(get_mail_queue),
|
|
) -> dict[str, Any]:
|
|
"""Enqueue an email to Authentik-sourced recipients via the mail queue."""
|
|
service = resolve_service_record(store, "authentik", service_id)
|
|
if service is None:
|
|
return {"status": "error", "error": "Authentik service not configured"}
|
|
|
|
recipients = [r.strip() for r in body.recipient_emails if r.strip()]
|
|
if not recipients:
|
|
return {"status": "error", "error": "No recipients with valid email addresses."}
|
|
|
|
settings = get_settings()
|
|
try:
|
|
validate_smtp_settings(settings)
|
|
except ValueError as exc:
|
|
return {"status": "error", "error": f"SMTP settings invalid: {exc}"}
|
|
|
|
request_id = mail_queue.enqueue(
|
|
settings=settings,
|
|
recipients=recipients,
|
|
subject=body.subject,
|
|
html_body=body.html_body,
|
|
)
|
|
logger.info("Authentik message enqueued for service %s (%d recipients)", service_id, len(recipients))
|
|
return {
|
|
"status": "queued",
|
|
"request_id": request_id,
|
|
"recipient_count": len(recipients),
|
|
}
|