""" Authentik API Client Provides methods for interacting with Authentik Identity Provider API. Used for managing applications, providers, and authentication flows. """ import httpx from typing import Dict, List, Any, Optional from functools import lru_cache from src.shared.logging import get_logger logger = get_logger(__name__) class AuthentikClient: """Client for Authentik API operations""" def __init__(self, base_url: str, api_token: str): """ Initialize Authentik client Args: base_url: Authentik base URL (e.g., http://authentik-server:9000) api_token: API token for authentication """ self.base_url = base_url.rstrip('/') self.api_token = api_token self.client = httpx.AsyncClient(timeout=30.0) async def _request(self, method: str, endpoint: str, **kwargs) -> Dict: """Make authenticated API request using token auth""" headers = kwargs.pop("headers", {}) headers["Authorization"] = f"Bearer {self.api_token}" response = await self.client.request( method, f"{self.base_url}/api/v3/{endpoint.lstrip('/')}", headers=headers, **kwargs ) if not response.is_success: logger.error(f"API request failed: {response.status_code}") logger.error(f"Response body: {response.text}") response.raise_for_status() return response.json() async def health_check(self) -> bool: """Check if Authentik is accessible""" try: response = await self.client.get(f"{self.base_url}/-/health/live/") return response.status_code == 200 except Exception as e: logger.error(f"Authentik health check failed: {e}") return False async def create_oauth2_provider( self, name: str, client_id: str, redirect_uris: List[str], authorization_flow_slug: str = "default-provider-authorization-implicit-consent", signing_key: Optional[str] = None ) -> Dict: """ Create an OAuth2/OIDC provider Args: name: Provider name client_id: OAuth2 client ID redirect_uris: List of allowed redirect URIs authorization_flow_slug: Authorization flow slug (will be resolved to UUID) signing_key: Signing key UUID (defaults to auto-selected) Returns: Created provider data including client_secret """ # Get authorization flow UUID from slug flows = await self.list_flows() auth_flow_uuid = None invalidation_flow_uuid = None for flow in flows: if flow.get("slug") == authorization_flow_slug: auth_flow_uuid = flow.get("pk") if flow.get("slug") == "default-provider-invalidation-flow": invalidation_flow_uuid = flow.get("pk") if not auth_flow_uuid: raise ValueError(f"Authorization flow '{authorization_flow_slug}' not found") if not invalidation_flow_uuid: raise ValueError("Invalidation flow not found") # Get signing key if not provided if not signing_key: keys = await self._request("GET", "crypto/certificatekeypairs/") # Find the self-signed cert for key in keys.get("results", []): if "authentik" in key.get("name", "").lower(): signing_key = key.get("pk") break if not signing_key and keys.get("results"): signing_key = keys["results"][0]["pk"] # Format redirect URIs as objects with matching_mode formatted_redirect_uris = [ {"url": uri, "matching_mode": "strict"} for uri in redirect_uris ] provider_data = { "name": name, "authorization_flow": auth_flow_uuid, "invalidation_flow": invalidation_flow_uuid, "client_type": "confidential", "client_id": client_id, "redirect_uris": formatted_redirect_uris, "signing_key": signing_key, "sub_mode": "hashed_user_id", "include_claims_in_id_token": True, "issuer_mode": "per_provider", "access_token_validity": "minutes=60", "refresh_token_validity": "days=30", "property_mappings": [] # Will use default mappings } result = await self._request("POST", "providers/oauth2/", json=provider_data) logger.info(f"Created OAuth2 provider: {name} (ID: {result.get('pk')})") return result async def create_application( self, name: str, slug: str, provider_pk: int, launch_url: Optional[str] = None, icon_url: Optional[str] = None ) -> Dict: """ Create an application Args: name: Application display name slug: Application slug (URL-safe identifier) provider_pk: Primary key of the provider to use launch_url: Optional launch URL icon_url: Optional icon URL Returns: Created application data """ app_data = { "name": name, "slug": slug, "provider": provider_pk, "meta_launch_url": launch_url or "", "meta_icon": icon_url or "", "policy_engine_mode": "any", "open_in_new_tab": False } result = await self._request("POST", "core/applications/", json=app_data) logger.info(f"Created application: {name} (slug: {slug})") return result async def get_provider_by_name(self, name: str) -> Optional[Dict]: """Get OAuth2 provider by name""" providers = await self._request("GET", "providers/oauth2/", params={"name": name}) results = providers.get("results", []) return results[0] if results else None async def get_application_by_slug(self, slug: str) -> Optional[Dict]: """Get application by slug""" apps = await self._request("GET", "core/applications/", params={"slug": slug}) results = apps.get("results", []) return results[0] if results else None async def list_flows(self) -> List[Dict]: """List all authentication flows""" result = await self._request("GET", "flows/instances/") return result.get("results", []) async def create_proxy_provider( self, name: str, external_host: str, authorization_flow_slug: str = "default-provider-authorization-implicit-consent", mode: str = "forward_single", token_validity: int = 480 # 8 hours in minutes ) -> Dict: """ Create a Proxy Provider for forward authentication Args: name: Provider name external_host: External URL (e.g., https://auth.schweitz.net) authorization_flow_slug: Authorization flow slug mode: Proxy mode (forward_single for forward auth) token_validity: Token validity in minutes (default: 480 = 8 hours) Returns: Created provider data """ # Get authorization flow UUID from slug flows = await self.list_flows() auth_flow_uuid = None invalidation_flow_uuid = None for flow in flows: if flow.get("slug") == authorization_flow_slug: auth_flow_uuid = flow.get("pk") if flow.get("slug") == "default-provider-invalidation-flow": invalidation_flow_uuid = flow.get("pk") if not auth_flow_uuid: raise ValueError(f"Authorization flow '{authorization_flow_slug}' not found") if not invalidation_flow_uuid: raise ValueError("Invalidation flow not found") provider_data = { "name": name, "authorization_flow": auth_flow_uuid, "invalidation_flow": invalidation_flow_uuid, "mode": mode, "external_host": external_host, "access_token_validity": f"minutes={token_validity}", "refresh_token_validity": f"minutes={token_validity}", "session_duration": f"seconds={token_validity * 60}", "cookie_domain": "", # Will use the domain of each proxied site "property_mappings": [] } result = await self._request("POST", "providers/proxy/", json=provider_data) logger.info(f"Created Proxy provider: {name} (ID: {result.get('pk')})") return result async def get_provider_by_name_proxy(self, name: str) -> Optional[Dict]: """Get Proxy provider by name""" providers = await self._request("GET", "providers/proxy/", params={"name": name}) results = providers.get("results", []) return results[0] if results else None async def create_outpost( self, name: str, type: str, providers: List[int], config: Optional[Dict] = None ) -> Dict: """ Create an Authentik Outpost Args: name: Outpost name type: Outpost type (e.g., "proxy") providers: List of provider PKs config: Optional configuration overrides Returns: Created outpost data """ outpost_data = { "name": name, "type": type, "providers": providers, "config": config or {}, "service_connection": None # Will use local Docker } result = await self._request("POST", "outposts/instances/", json=outpost_data) logger.info(f"Created outpost: {name} (ID: {result.get('pk')})") return result async def get_outpost_by_name(self, name: str) -> Optional[Dict]: """Get outpost by name""" outposts = await self._request("GET", "outposts/instances/", params={"name": name}) results = outposts.get("results", []) return results[0] if results else None async def close(self): """Close HTTP client""" await self.client.aclose() @lru_cache() def get_authentik_client() -> AuthentikClient: """Get cached Authentik client instance""" # Import credentials from gitignored module try: from src.credentials import AUTHENTIK_URL, AUTHENTIK_CORE_API_TOKEN except ImportError: # Fallback to environment variables if credentials.py doesn't exist import os AUTHENTIK_URL = os.getenv("AUTHENTIK_URL", "http://authentik-server:9000") AUTHENTIK_CORE_API_TOKEN = os.getenv("AUTHENTIK_API_TOKEN", "") return AuthentikClient( base_url=AUTHENTIK_URL, api_token=AUTHENTIK_CORE_API_TOKEN )