""" Authentication Service Business logic for user synchronization from Authentik. """ import re import uuid from datetime import datetime, timezone from typing import Optional import httpx from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from src.shared.config import get_settings from src.logging_config import get_logger from src.db.models import User, Role, UserPreferences, Group from src.auth.schemas import TokenInfoSchema, UserSchema, RoleSchema, UserPreferencesSchema, UserListItemSchema, BulkSyncResultSchema, GroupListItemSchema logger = get_logger(__name__) settings = get_settings() class AuthService: """ Service for authentication and user synchronization Handles: - Token validation via Authentik userinfo endpoint - User creation/update from OIDC claims - Role synchronization from Authentik groups """ def __init__(self, session: AsyncSession): """ Initialize auth service Args: session: Async database session """ self.session = session self.userinfo_url = f"{settings.authentik_url}/application/o/userinfo/" async def validate_token(self, access_token: str) -> TokenInfoSchema: """ Validate access token via Authentik userinfo endpoint Args: access_token: OIDC access token Returns: Token info containing user claims Raises: ValueError: If token is invalid or expired """ try: async with httpx.AsyncClient(timeout=10.0) as client: response = await client.get( self.userinfo_url, headers={"Authorization": f"Bearer {access_token}"}, ) if response.status_code == 401: raise ValueError("Invalid or expired token") response.raise_for_status() data = response.json() logger.debug(f"Userinfo response: {data}") return TokenInfoSchema( sub=data.get("sub"), email=data.get("email"), name=data.get("name") or data.get("preferred_username"), preferred_username=data.get("preferred_username"), groups=data.get("groups", []), picture=data.get("picture"), ) except httpx.HTTPStatusError as e: logger.error(f"Authentik userinfo request failed: {e}") raise ValueError(f"Token validation failed: {e.response.status_code}") except httpx.RequestError as e: logger.error(f"Authentik userinfo request error: {e}") raise ValueError("Authentication service unavailable") async def get_user_by_email(self, email: str) -> Optional[User]: """ Get user by email address Args: email: User email address Returns: User if found, None otherwise """ stmt = ( select(User) .options(selectinload(User.roles), selectinload(User.preferences)) .where(User.email == email) ) result = await self.session.execute(stmt) return result.scalar_one_or_none() async def get_user_by_authentik_id(self, authentik_id: uuid.UUID) -> Optional[User]: """ Get user by Authentik UUID Args: authentik_id: Authentik user UUID Returns: User if found, None otherwise """ stmt = ( select(User) .options(selectinload(User.roles), selectinload(User.preferences)) .where(User.authentik_id == authentik_id) ) result = await self.session.execute(stmt) return result.scalar_one_or_none() async def sync_user(self, token_info: TokenInfoSchema) -> tuple[User, bool]: """ Create or update user from OIDC token info Args: token_info: Validated token information Returns: Tuple of (User, is_new_user) """ authentik_id = uuid.UUID(token_info.sub) # Try to find existing user stmt = ( select(User) .options(selectinload(User.roles), selectinload(User.preferences)) .where(User.authentik_id == authentik_id) ) result = await self.session.execute(stmt) user = result.scalar_one_or_none() is_new = user is None if is_new: # Create new user user = User( authentik_id=authentik_id, email=token_info.email, name=token_info.name or token_info.email, avatar_url=token_info.picture, last_login=datetime.now(timezone.utc), ) self.session.add(user) await self.session.flush() # Get the user ID # Create default preferences preferences = UserPreferences(user_id=user.id) self.session.add(preferences) logger.info(f"Created new user: {token_info.email}") else: # Update existing user user.email = token_info.email user.name = token_info.name or token_info.email user.avatar_url = token_info.picture user.last_login = datetime.now(timezone.utc) logger.info(f"Updated existing user: {token_info.email}") await self.session.flush() return user, is_new async def sync_roles(self, user: User, groups: list[str]) -> list[Role]: """ Synchronize user roles from Authentik groups Maps Authentik groups (e.g., 'tatlock-control-room-admin') to application roles (e.g., 'control-room:admin'). Args: user: User to sync roles for groups: List of Authentik group names Returns: List of synced Role objects """ # Get all roles that match the user's Authentik groups stmt = select(Role).where(Role.authentik_group.in_(groups)) result = await self.session.execute(stmt) matching_roles = list(result.scalars().all()) # Clear existing roles and set new ones user.roles = matching_roles role_names = [r.name for r in matching_roles] logger.info(f"Synced roles for {user.email}: {role_names}") return matching_roles def user_to_schema(self, user: User) -> UserSchema: """Convert User model to schema""" return UserSchema( id=user.id, authentik_id=user.authentik_id, email=user.email, name=user.name, avatar_url=user.avatar_url, created_at=user.created_at, last_login=user.last_login, ) def roles_to_schema(self, roles: list[Role]) -> list[RoleSchema]: """Convert Role models to schemas""" return [ RoleSchema(name=r.name, domain=r.domain, action=r.action) for r in roles ] def preferences_to_schema(self, preferences: Optional[UserPreferences]) -> UserPreferencesSchema: """Convert UserPreferences model to schema""" if preferences is None: return UserPreferencesSchema() return UserPreferencesSchema( theme=preferences.theme, default_room=preferences.default_room, preferences_json=preferences.preferences_json or {}, ) async def list_users( self, search: Optional[str] = None, offset: int = 0, limit: int = 50, ) -> tuple[list[UserListItemSchema], int]: """ List all users with optional search and pagination Args: search: Optional search query (matches name or email) offset: Number of records to skip limit: Maximum number of records to return Returns: Tuple of (list of user schemas, total count) """ from sqlalchemy import func # Base query with roles loaded base_query = select(User).options(selectinload(User.roles)) # Apply search filter if provided if search: search_filter = f"%{search}%" base_query = base_query.where( (User.name.ilike(search_filter)) | (User.email.ilike(search_filter)) ) # Get total count count_query = select(func.count()).select_from(base_query.subquery()) total_result = await self.session.execute(count_query) total = total_result.scalar() or 0 # Apply pagination and ordering query = base_query.order_by(User.name).offset(offset).limit(limit) result = await self.session.execute(query) users = list(result.scalars().all()) # Convert to schemas items = [ UserListItemSchema( id=user.id, email=user.email, name=user.name, avatar_url=user.avatar_url, created_at=user.created_at, last_login=user.last_login, roles=[role.name for role in user.roles], ) for user in users ] return items, total def _extract_cookie(self, headers: httpx.Headers, cookie_name: str) -> str: """Extract a specific cookie value from Set-Cookie headers""" for header in headers.get_list('set-cookie'): if header.startswith(f'{cookie_name}='): match = re.match(rf'{cookie_name}=([^;]+)', header) if match: return match.group(1) return "" async def _authentik_session_login(self, client: httpx.AsyncClient) -> str: """ Authenticate with Authentik using the flow API to establish a session Authentik's flow API requires: 1. Cookie persistence between requests (manually handled due to domain restrictions) 2. X-authentik-CSRF header set to the authentik_csrf cookie value 3. Multi-stage flow handling (identification -> password -> done) Args: client: httpx client Returns: Session cookie value for subsequent API calls Raises: ValueError: If authentication fails """ flow_url = f"{settings.authentik_url}/api/v3/flows/executor/default-authentication-flow/" # Step 1: Get the initial flow challenge (this sets the session and csrf cookies) resp = await client.get(flow_url, headers={"Accept": "application/json"}) resp.raise_for_status() data = resp.json() # Extract cookies manually from Set-Cookie headers (bypasses domain restrictions) session_cookie = self._extract_cookie(resp.headers, "authentik_session") csrf_cookie = self._extract_cookie(resp.headers, "authentik_csrf") logger.debug(f"Flow initial: component={data.get('component')}, session={bool(session_cookie)}, csrf={bool(csrf_cookie)}") # Build headers with manual cookie and CSRF token def build_headers(): hdrs = { "Accept": "application/json", "Content-Type": "application/json", "Cookie": f"authentik_session={session_cookie}", } if csrf_cookie: hdrs["Cookie"] += f"; authentik_csrf={csrf_cookie}" hdrs["X-authentik-CSRF"] = csrf_cookie return hdrs # Step 2: Handle identification stage - submit username if data.get("component") == "ak-stage-identification": resp = await client.post( flow_url, json={"uid_field": settings.authentik_username}, headers=build_headers(), ) resp.raise_for_status() data = resp.json() # Update session cookie if new one received new_session = self._extract_cookie(resp.headers, "authentik_session") if new_session: session_cookie = new_session logger.debug(f"After username: component={data.get('component')}") # Step 3: Handle password stage if required if data.get("component") == "ak-stage-password": resp = await client.post( flow_url, json={"password": settings.authentik_password}, headers=build_headers(), ) resp.raise_for_status() data = resp.json() # Update session cookie if new one received new_session = self._extract_cookie(resp.headers, "authentik_session") if new_session: session_cookie = new_session logger.debug(f"After password: component={data.get('component')}") # Check for access denied if data.get("component") == "ak-stage-access-denied": raise ValueError("Authentik authentication failed: access denied") # Check for redirect (successful auth) if data.get("component") == "xak-flow-redirect" or data.get("to"): logger.info("Successfully authenticated with Authentik via flow") return session_cookie # If we're still in identification stage, the username might be wrong if data.get("component") == "ak-stage-identification": response_errors = data.get("response_errors", {}) raise ValueError(f"Authentication stuck at identification stage: {response_errors}") logger.info(f"Authentik flow completed with component: {data.get('component')}") return session_cookie async def bulk_sync_from_authentik(self) -> BulkSyncResultSchema: """ Fetch all users from Authentik admin API and sync to local database Returns: BulkSyncResultSchema with counts of created/updated/failed users """ if not settings.authentik_username or not settings.authentik_password: raise ValueError("AUTHENTIK_USERNAME and AUTHENTIK_PASSWORD must be configured") created = 0 updated = 0 failed = 0 errors = [] total_in_authentik = 0 # Step 1: Fetch all user data from Authentik API authentik_users = [] try: async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as client: # Authenticate with Authentik to get session cookie session_cookie = await self._authentik_session_login(client) # Fetch users from Authentik admin API using session cookie response = await client.get( f"{settings.authentik_url}/api/v3/core/users/", params={"page_size": 500}, headers={ "Accept": "application/json", "Cookie": f"authentik_session={session_cookie}", }, ) if response.status_code == 401: raise ValueError("Authentik API token is invalid or expired") response.raise_for_status() data = response.json() authentik_users = data.get("results", []) total_in_authentik = data.get("pagination", {}).get("count", len(authentik_users)) except httpx.HTTPStatusError as e: raise ValueError(f"Authentik API error: {e.response.status_code}") except httpx.RequestError as e: raise ValueError(f"Failed to connect to Authentik: {str(e)}") # Step 2: Sync users to database (outside of httpx context to avoid greenlet issues) for auth_user in authentik_users: try: # Skip service accounts and inactive users if auth_user.get("type") in ("service_account", "internal_service_account"): continue if not auth_user.get("is_active", True): continue # Extract user data from Authentik authentik_id = uuid.UUID(auth_user["uuid"]) email = auth_user.get("email") or f"{auth_user['username']}@local" name = auth_user.get("name") or auth_user.get("username", "Unknown") avatar_url = auth_user.get("avatar") # Get user's groups for role mapping groups = [] groups_summary = auth_user.get("groups_obj", []) for group in groups_summary: groups.append(group.get("name", "")) # Check if user exists stmt = select(User).where(User.authentik_id == authentik_id) result = await self.session.execute(stmt) user = result.scalar_one_or_none() if user is None: # Create new user user = User( authentik_id=authentik_id, email=email, name=name, avatar_url=avatar_url, ) self.session.add(user) await self.session.flush() # Create default preferences preferences = UserPreferences(user_id=user.id) self.session.add(preferences) created += 1 logger.info(f"Created user from Authentik: {email}") else: # Update existing user user.email = email user.name = name user.avatar_url = avatar_url updated += 1 logger.info(f"Updated user from Authentik: {email}") # Sync roles from groups await self.sync_roles(user, groups) except Exception as e: failed += 1 error_msg = f"Failed to sync user {auth_user.get('username', 'unknown')}: {str(e)}" errors.append(error_msg) logger.warning(error_msg) # Commit all changes await self.session.commit() return BulkSyncResultSchema( created=created, updated=updated, failed=failed, total_in_authentik=total_in_authentik, errors=errors, ) async def list_groups( self, search: Optional[str] = None, offset: int = 0, limit: int = 50, ) -> tuple[list[GroupListItemSchema], int]: """ List all groups with optional search and pagination Args: search: Optional search query (matches name) offset: Number of records to skip limit: Maximum number of records to return Returns: Tuple of (list of group schemas, total count) """ from sqlalchemy import func # Base query base_query = select(Group) # Apply search filter if provided if search: search_filter = f"%{search}%" base_query = base_query.where(Group.name.ilike(search_filter)) # Get total count count_query = select(func.count()).select_from(base_query.subquery()) total_result = await self.session.execute(count_query) total = total_result.scalar() or 0 # Apply pagination and ordering query = base_query.order_by(Group.name).offset(offset).limit(limit) result = await self.session.execute(query) groups = list(result.scalars().all()) # Convert to schemas items = [ GroupListItemSchema( id=group.id, authentik_id=group.authentik_id, name=group.name, is_superuser=group.is_superuser, parent_name=group.parent_name, member_count=group.member_count, synced_at=group.synced_at, ) for group in groups ] return items, total async def bulk_sync_groups_from_authentik(self) -> BulkSyncResultSchema: """ Fetch all groups from Authentik admin API and sync to local database Returns: BulkSyncResultSchema with counts of created/updated/failed groups """ if not settings.authentik_username or not settings.authentik_password: raise ValueError("AUTHENTIK_USERNAME and AUTHENTIK_PASSWORD must be configured") created = 0 updated = 0 failed = 0 errors = [] total_in_authentik = 0 # Step 1: Fetch all group data from Authentik API authentik_groups = [] try: async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as client: # Authenticate with Authentik to get session cookie session_cookie = await self._authentik_session_login(client) # Fetch groups from Authentik admin API using session cookie response = await client.get( f"{settings.authentik_url}/api/v3/core/groups/", params={"page_size": 500}, headers={ "Accept": "application/json", "Cookie": f"authentik_session={session_cookie}", }, ) if response.status_code == 401: raise ValueError("Authentik API token is invalid or expired") response.raise_for_status() data = response.json() authentik_groups = data.get("results", []) total_in_authentik = data.get("pagination", {}).get("count", len(authentik_groups)) except httpx.HTTPStatusError as e: raise ValueError(f"Authentik API error: {e.response.status_code}") except httpx.RequestError as e: raise ValueError(f"Failed to connect to Authentik: {str(e)}") # Step 2: Sync groups to database (outside of httpx context to avoid greenlet issues) for auth_group in authentik_groups: try: # Extract group data from Authentik authentik_id = uuid.UUID(auth_group["pk"]) name = auth_group.get("name", "Unknown") is_superuser = auth_group.get("is_superuser", False) parent_name = auth_group.get("parent_name") # users field contains list of user PKs member_count = len(auth_group.get("users", [])) # Check if group exists stmt = select(Group).where(Group.authentik_id == authentik_id) result = await self.session.execute(stmt) group = result.scalar_one_or_none() if group is None: # Create new group group = Group( authentik_id=authentik_id, name=name, is_superuser=is_superuser, parent_name=parent_name, member_count=member_count, ) self.session.add(group) created += 1 logger.info(f"Created group from Authentik: {name}") else: # Update existing group group.name = name group.is_superuser = is_superuser group.parent_name = parent_name group.member_count = member_count updated += 1 logger.info(f"Updated group from Authentik: {name}") except Exception as e: failed += 1 error_msg = f"Failed to sync group {auth_group.get('name', 'unknown')}: {str(e)}" errors.append(error_msg) logger.warning(error_msg) # Commit all changes await self.session.commit() return BulkSyncResultSchema( created=created, updated=updated, failed=failed, total_in_authentik=total_in_authentik, errors=errors, ) # Factory function for dependency injection def get_auth_service(session: AsyncSession) -> AuthService: """Create AuthService instance with database session""" return AuthService(session)