blog.pyaqa.ru/app/infrastructure/auth/client.py

"""Keycloak authentication client.

This module provides a client for Keycloak authentication operations
including token introspection and user info retrieval.
"""

import time

import httpx

from app.infrastructure.auth.models import KeycloakUser, TokenInfo
from app.infrastructure.config.settings import Settings


class KeycloakAuthClient:
    """Client for Keycloak authentication operations.

    Handles token validation via introspection and user info retrieval.
    Implements token caching to reduce Keycloak server load.

    Attributes:
        _settings: Application settings with Keycloak config.
        _base_url: Keycloak realm base URL.
        _client_id: OAuth client identifier.
        _client_secret: OAuth client secret.
        _cache: Token info cache for performance.
        _cache_ttl: Cache time-to-live in seconds.

    Example:
        >>> client = KeycloakAuthClient(settings)
        >>> token_info = await client.introspect_token(token)
    """

    def __init__(self, settings: Settings) -> None:
        """Initialize Keycloak client with settings.

        Args:
            settings: Application settings with Keycloak configuration.
        """
        self._settings = settings
        self._base_url = f"{settings.kc.server_url}/realms/{settings.kc.realm}"
        self._client_id = settings.kc.client_id
        self._client_secret = settings.kc.client_secret
        self._cache: dict[str, tuple[TokenInfo, float]] = {}
        self._cache_ttl = settings.kc.token_cache_ttl

    def _get_introspection_url(self) -> str:
        """Get token introspection endpoint URL.

        Returns:
            Full URL for token introspection endpoint.
        """
        return f"{self._base_url}/protocol/openid-connect/token/introspection"

    def _get_userinfo_url(self) -> str:
        """Get userinfo endpoint URL.

        Returns:
            Full URL for userinfo endpoint.
        """
        return f"{self._base_url}/protocol/openid-connect/userinfo"

    def _get_cached_token(self, token: str) -> TokenInfo | None:
        """Get cached token info if valid.

        Args:
            token: Access token string.

        Returns:
            Cached TokenInfo if valid and not expired, None otherwise.
        """
        if token not in self._cache:
            return None

        token_info, cached_at = self._cache[token]
        if time.time() - cached_at > self._cache_ttl:
            del self._cache[token]
            return None

        return token_info

    def _cache_token(self, token: str, token_info: TokenInfo) -> None:
        """Cache token info.

        Args:
            token: Access token string as cache key.
            token_info: TokenInfo to cache.
        """
        self._cache[token] = (token_info, time.time())
        current_time = time.time()
        expired_keys = [
            k for k, (_, t) in self._cache.items() if current_time - t > self._cache_ttl
        ]
        for k in expired_keys:
            del self._cache[k]

    async def introspect_token(self, token: str) -> TokenInfo:
        """Introspect access token using Keycloak.

        Validates token with Keycloak server and extracts user information.
        Uses cache to reduce server requests for recently validated tokens.

        Args:
            token: Access token to validate.

        Returns:
            TokenInfo with validation result and user claims.
        """
        cached = self._get_cached_token(token)
        if cached:
            return cached

        data = {
            "token": token,
            "client_id": self._client_id,
            "client_secret": self._client_secret,
        }

        try:
            async with httpx.AsyncClient() as client:
                response = await client.post(
                    self._get_introspection_url(),
                    data=data,
                    headers={"Content-Type": "application/x-www-form-urlencoded"},
                    timeout=10.0,
                )
                response.raise_for_status()
                result = response.json()
        except httpx.HTTPError as e:
            return TokenInfo(active=False, raw_claims={"error": str(e)})

        if not result.get("active", False):
            return TokenInfo(active=False, raw_claims=result)

        roles: list[str] = []
        realm_access = result.get("realm_access", {})
        if isinstance(realm_access, dict):
            roles.extend(realm_access.get("roles", []))

        token_info = TokenInfo(
            active=True,
            user_id=result.get("sub", ""),
            username=result.get("preferred_username", ""),
            email=result.get("email", ""),
            roles=roles,
            raw_claims=result,
        )

        self._cache_token(token, token_info)

        return token_info

    async def get_userinfo(self, token: str) -> KeycloakUser | None:
        """Get user information from Keycloak using access token.

        Fetches detailed user profile from Keycloak userinfo endpoint.

        Args:
            token: Valid access token.

        Returns:
            KeycloakUser with profile data, or None on error.
        """
        try:
            async with httpx.AsyncClient() as client:
                response = await client.get(
                    self._get_userinfo_url(),
                    headers={"Authorization": f"Bearer {token}"},
                    timeout=10.0,
                )
                response.raise_for_status()
                data = response.json()
        except httpx.HTTPError:
            return None

        return KeycloakUser(
            id=data.get("sub", ""),
            username=data.get("preferred_username", ""),
            email=data.get("email", ""),
            first_name=data.get("given_name", ""),
            last_name=data.get("family_name", ""),
            roles=data.get("realm_access", {}).get("roles", [])
            if isinstance(data.get("realm_access"), dict)
            else [],
        )