""" Authentication Service Business logic for user synchronization from Authentik. """ import hashlib import re import secrets import uuid from datetime import datetime, timezone, timedelta 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.shared.logging import get_logger from src.domains.auth.models import User, Role, UserPreferences, Group, ApiKey from src.domains.auth.schemas import ( TokenInfoSchema, UserSchema, RoleSchema, UserPreferencesSchema, UserListItemSchema, BulkSyncResultSchema, GroupListItemSchema, ApiKeySchema, ApiKeyCreateResponse, ) 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 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) """ # Parse authentik_id - may be UUID or other format try: authentik_id = uuid.UUID(token_info.sub) except ValueError: # If sub is not a valid UUID, derive one deterministically logger.warning(f"sub claim is not a UUID: {token_info.sub}, deriving UUID") authentik_id = uuid.uuid5(uuid.NAMESPACE_OID, 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), ) # Initialize relationships to avoid lazy loading issues in async user.roles = [] self.session.add(user) await self.session.flush() # Get the user ID # Create default preferences and attach to user preferences = UserPreferences(user_id=user.id) user.preferences = preferences 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_user_from_claims( self, authentik_id: uuid.UUID, email: str, name: str, groups: list[str], avatar_url: Optional[str] = None, ) -> tuple[User, bool]: """ Create or update user from forward auth claims (NPM X-authentik-* headers) This is similar to sync_user() but works with raw claims instead of TokenInfoSchema. Used for auto-syncing users on first web login via NPM. Args: authentik_id: The Authentik user UUID (from X-authentik-uid) email: User email (from X-authentik-email) name: User display name (from X-authentik-name) groups: List of group names (from X-authentik-groups) avatar_url: Optional avatar URL Returns: Tuple of (User, is_new_user) """ # 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=email, name=name or email, avatar_url=avatar_url, 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 from forward auth: {email}") else: # Update existing user user.email = email user.name = name or email if avatar_url: user.avatar_url = avatar_url user.last_login = datetime.now(timezone.utc) logger.info(f"Updated existing user from forward auth: {email}") # Sync roles from groups await self.sync_roles(user, groups) await self.session.flush() return user, is_new async def sync_roles(self, user: User, group_names: list[str]) -> list[Role]: """ Synchronize user roles from Authentik groups via group_roles mapping Looks up the user's groups in the database, then retrieves all roles assigned to those groups via the group_roles mapping table. Args: user: User to sync roles for group_names: List of Authentik group names Returns: List of synced Role objects """ # Find local Group records matching the Authentik group names stmt = ( select(Group) .options(selectinload(Group.roles)) .where(Group.name.in_(group_names)) ) result = await self.session.execute(stmt) matching_groups = list(result.scalars().all()) # Collect all unique roles from all matching groups roles_set: dict[uuid.UUID, Role] = {} for group in matching_groups: for role in group.roles: roles_set[role.id] = role matching_roles = list(roles_set.values()) # Clear existing roles and set new ones user.roles = matching_roles role_names = [r.name for r in matching_roles] group_names_found = [g.name for g in matching_groups] logger.info(f"Synced roles for {user.email} via groups {group_names_found}: {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(id=r.id, name=r.name, domain=r.domain, category=r.category, 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 with roles loaded base_query = select(Group).options(selectinload(Group.roles)) # 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, roles=[role.name for role in group.roles], ) for group in groups ] return items, total async def list_roles(self) -> list[Role]: """ List all available roles Returns: List of all Role objects """ stmt = select(Role).order_by(Role.domain, Role.category, Role.action) result = await self.session.execute(stmt) return list(result.scalars().all()) async def get_group_by_id(self, group_id: uuid.UUID) -> Optional[Group]: """ Get a group by its ID with roles loaded Args: group_id: The group's UUID Returns: Group object or None if not found """ stmt = ( select(Group) .options(selectinload(Group.roles)) .where(Group.id == group_id) ) result = await self.session.execute(stmt) return result.scalar_one_or_none() async def assign_role_to_group(self, group_id: uuid.UUID, role_id: uuid.UUID) -> Group: """ Assign a role to a group Args: group_id: The group's UUID role_id: The role's UUID to assign Returns: Updated Group object Raises: ValueError: If group or role not found """ group = await self.get_group_by_id(group_id) if not group: raise ValueError(f"Group not found: {group_id}") stmt = select(Role).where(Role.id == role_id) result = await self.session.execute(stmt) role = result.scalar_one_or_none() if not role: raise ValueError(f"Role not found: {role_id}") # Add role if not already assigned if role not in group.roles: group.roles.append(role) await self.session.flush() logger.info(f"Assigned role {role.name} to group {group.name}") return group async def remove_role_from_group(self, group_id: uuid.UUID, role_id: uuid.UUID) -> Group: """ Remove a role from a group Args: group_id: The group's UUID role_id: The role's UUID to remove Returns: Updated Group object Raises: ValueError: If group or role not found """ group = await self.get_group_by_id(group_id) if not group: raise ValueError(f"Group not found: {group_id}") stmt = select(Role).where(Role.id == role_id) result = await self.session.execute(stmt) role = result.scalar_one_or_none() if not role: raise ValueError(f"Role not found: {role_id}") # Remove role if assigned if role in group.roles: group.roles.remove(role) await self.session.flush() logger.info(f"Removed role {role.name} from group {group.name}") return group 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, ) # ========================================================================= # Phase 4: User Profile & Settings # ========================================================================= async def get_user_by_authentik_id(self, authentik_id: uuid.UUID) -> Optional[User]: """ Get user by Authentik UUID with roles and preferences loaded Args: authentik_id: The Authentik user UUID (from JWT 'sub' claim) Returns: User object or None if not found """ 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 get_user_by_id(self, user_id: uuid.UUID) -> Optional[User]: """ Get user by internal UUID with roles and preferences loaded Args: user_id: The internal user UUID Returns: User object or None if not found """ stmt = ( select(User) .options(selectinload(User.roles), selectinload(User.preferences)) .where(User.id == user_id) ) result = await self.session.execute(stmt) return result.scalar_one_or_none() async def update_preferences( self, user_id: uuid.UUID, theme: Optional[str] = None, default_room: Optional[str] = None, preferences_json: Optional[dict] = None, ) -> UserPreferences: """ Update user preferences Args: user_id: User's UUID theme: New theme value (or None to keep existing) default_room: New default room (or None to keep existing) preferences_json: JSON to merge with existing (or None to keep existing) Returns: Updated UserPreferences object Raises: ValueError: If user not found """ stmt = select(UserPreferences).where(UserPreferences.user_id == user_id) result = await self.session.execute(stmt) prefs = result.scalar_one_or_none() if prefs is None: # Create preferences if they don't exist prefs = UserPreferences(user_id=user_id) self.session.add(prefs) if theme is not None: if theme not in ("system", "light", "dark"): raise ValueError(f"Invalid theme: {theme}") prefs.theme = theme if default_room is not None: prefs.default_room = default_room if preferences_json is not None: # Merge with existing preferences existing = prefs.preferences_json or {} existing.update(preferences_json) prefs.preferences_json = existing await self.session.flush() logger.info(f"Updated preferences for user {user_id}") return prefs # ========================================================================= # Phase 4: API Keys # ========================================================================= def _generate_api_key(self) -> tuple[str, str, str]: """ Generate a new API key Returns: Tuple of (full_key, key_hash, key_prefix) """ # Generate 32 random bytes = 256 bits of entropy random_bytes = secrets.token_bytes(32) # Encode as base64-like string (URL-safe) key_body = secrets.token_urlsafe(32) # Prefix with 'tak_' (tatlock api key) full_key = f"tak_{key_body}" # Hash for storage key_hash = hashlib.sha256(full_key.encode()).hexdigest() # Prefix for identification (first 8 chars after 'tak_') key_prefix = f"tak_{key_body[:4]}" return full_key, key_hash, key_prefix async def create_api_key( self, user_id: uuid.UUID, name: str, scopes: Optional[list[str]] = None, expires_in_days: Optional[int] = None, ) -> tuple[ApiKey, str]: """ Create a new API key for a user Args: user_id: User's UUID name: Human-readable key name scopes: Optional list of scope restrictions expires_in_days: Optional expiration in days Returns: Tuple of (ApiKey object, full key string) The full key is only returned once at creation! Raises: ValueError: If user not found or API keys disabled """ # Check user exists and has API keys enabled user = await self.get_user_by_id(user_id) if user is None: raise ValueError(f"User not found: {user_id}") if not user.api_keys_enabled: raise ValueError("API keys are disabled for this user") # Generate the key full_key, key_hash, key_prefix = self._generate_api_key() # Calculate expiration expires_at = None if expires_in_days: expires_at = datetime.now(timezone.utc) + timedelta(days=expires_in_days) # Create the key record api_key = ApiKey( user_id=user_id, name=name, key_hash=key_hash, key_prefix=key_prefix, scopes=scopes, expires_at=expires_at, ) self.session.add(api_key) await self.session.flush() logger.info(f"Created API key '{name}' for user {user_id}") return api_key, full_key async def list_user_api_keys(self, user_id: uuid.UUID) -> list[ApiKey]: """ List all API keys for a user Args: user_id: User's UUID Returns: List of ApiKey objects (without the actual keys) """ stmt = ( select(ApiKey) .where(ApiKey.user_id == user_id) .order_by(ApiKey.created_at.desc()) ) result = await self.session.execute(stmt) return list(result.scalars().all()) async def delete_api_key(self, user_id: uuid.UUID, key_id: uuid.UUID) -> bool: """ Delete an API key Args: user_id: User's UUID (for authorization) key_id: API key UUID Returns: True if deleted, False if not found Raises: ValueError: If key belongs to different user """ stmt = select(ApiKey).where(ApiKey.id == key_id) result = await self.session.execute(stmt) api_key = result.scalar_one_or_none() if api_key is None: return False if api_key.user_id != user_id: raise ValueError("API key belongs to different user") await self.session.delete(api_key) await self.session.flush() logger.info(f"Deleted API key {key_id} for user {user_id}") return True async def validate_api_key(self, key: str) -> Optional[User]: """ Validate an API key and return the associated user Args: key: The full API key string Returns: User object if valid, None if invalid/expired """ # Hash the provided key key_hash = hashlib.sha256(key.encode()).hexdigest() # Look up by hash stmt = ( select(ApiKey) .options(selectinload(ApiKey.user).selectinload(User.roles)) .where(ApiKey.key_hash == key_hash) ) result = await self.session.execute(stmt) api_key = result.scalar_one_or_none() if api_key is None: return None # Check expiration if api_key.is_expired: logger.warning(f"Expired API key used: {api_key.key_prefix}...") return None # Update last used timestamp api_key.last_used_at = datetime.now(timezone.utc) logger.debug(f"API key authenticated: {api_key.key_prefix}... for user {api_key.user.email}") return api_key.user def api_key_to_schema(self, api_key: ApiKey) -> ApiKeySchema: """Convert ApiKey model to schema""" return ApiKeySchema( id=api_key.id, name=api_key.name, key_prefix=api_key.key_prefix, scopes=api_key.scopes, expires_at=api_key.expires_at, last_used_at=api_key.last_used_at, created_at=api_key.created_at, is_expired=api_key.is_expired, ) # Factory function for dependency injection def get_auth_service(session: AsyncSession) -> AuthService: """Create AuthService instance with database session""" return AuthService(session)