authentication.py

import inspect
import logging
from abc import ABC, abstractmethod
from functools import lru_cache
from logging import Logger
from typing import Any, Sequence, Type

from rodi import ContainerProtocol

from guardpost.abc import BaseStrategy
from guardpost.protection import InvalidCredentialsError, RateLimiter


class Identity:
    """
    Represents the characteristics of a person or a thing in the context of an
    application. It can be a user interacting with an app, or a technical account.
    """

    def __init__(
        self,
        claims: dict | None = None,
        authentication_mode: str | None = None,
    ):
        self.claims = claims or {}
        self.authentication_mode = authentication_mode
        self.access_token: str | None = None
        self.refresh_token: str | None = None

    @property
    def sub(self) -> str | None:
        return self.get("sub")

    @property
    def roles(self) -> str | None:
        return self.get("roles")

    def is_authenticated(self) -> bool:
        return bool(self.authentication_mode)

    def get(self, key: str):
        return self.claims.get(key)

    def __getitem__(self, item):
        return self.claims[item]

    def has_claim(self, name: str) -> bool:
        return name in self.claims

    def has_claim_value(self, name: str, value: str) -> bool:
        return self.claims.get(name) == value

    def has_role(self, name: str) -> bool:
        if not self.roles:
            return False
        return name in self.roles


class User(Identity):
    @property
    def id(self) -> str | None:
        return self.get("id") or self.sub

    @property
    def name(self) -> str | None:
        return self.get("name")

    @property
    def email(self) -> str | None:
        return self.get("email")


class AuthenticationHandler(ABC):
    """Base class for types that implement authentication logic."""

    @property
    def scheme(self) -> str:
        """Returns the name of the Authentication Scheme used by this handler."""
        return self.__class__.__name__

    @abstractmethod
    def authenticate(self, context: Any) -> Identity | None:
        """Obtains an identity from a context."""


@lru_cache(maxsize=None)
def _is_async_handler(handler_type: Type[AuthenticationHandler]) -> bool:
    # Faster alternative to using inspect.iscoroutinefunction without caching
    # Note: this must be used on Types - not instances!
    return inspect.iscoroutinefunction(handler_type.authenticate)


AuthenticationHandlerConfType = AuthenticationHandler | Type[AuthenticationHandler]


class AuthenticationSchemesNotFound(ValueError):
    def __init__(
        self, configured_schemes: Sequence[str], required_schemes: Sequence[str]
    ):
        super().__init__(
            "Could not find authentication handlers for required schemes: "
            f'{", ".join(required_schemes)}. '
            f'Configured schemes are: {", ".join(configured_schemes)}'
        )


class AuthenticationStrategy(BaseStrategy):
    def __init__(
        self,
        *handlers: AuthenticationHandlerConfType,
        container: ContainerProtocol | None = None,
        rate_limiter: RateLimiter | None = None,
        logger: Logger | None = None,
    ):
        """
        Initializes an AuthenticationStrategy instance.

        Args:
            *handlers: One or more authentication handler instances or types to be used
                for authentication.
            container: Optional dependency injection container for resolving handler
                instances.
            rate_limiter: Optional RateLimiter to apply rate limiting to authentication
                attempts.
            logger: Optional logger instance for logging authentication events. If not
                provided, defaults to `logging.getLogger("guardpost")`
        """
        super().__init__(container)
        self.handlers = list(handlers)
        self._logger = logger or logging.getLogger("guardpost")
        self._rate_limiter = rate_limiter

    def add(self, handler: AuthenticationHandlerConfType) -> "AuthenticationStrategy":
        self.handlers.append(handler)
        return self

    def __iadd__(
        self, handler: AuthenticationHandlerConfType
    ) -> "AuthenticationStrategy":
        self.handlers.append(handler)
        return self

    def _get_handlers_by_schemes(
        self,
        authentication_schemes: Sequence[str] | None = None,
        context: Any = None,
    ) -> list[AuthenticationHandler]:
        if not authentication_schemes:
            return list(self._get_instances(self.handlers, context))

        handlers = [
            handler
            for handler in self._get_instances(self.handlers, context)
            if handler.scheme in authentication_schemes
        ]

        if not handlers:
            raise AuthenticationSchemesNotFound(
                [
                    handler.scheme
                    for handler in self._get_instances(self.handlers, context)
                ],
                authentication_schemes,
            )

        return handlers

    async def authenticate(
        self, context: Any, authentication_schemes: Sequence[str] | None = None
    ) -> Identity | None:
        """
        Tries to obtain the user for a context, applying authentication rules and
        optional rate limiting.
        """
        if not context:
            raise ValueError("Missing context to evaluate authentication")

        if self._rate_limiter:
            await self._rate_limiter.validate_authentication_attempt(context)

        identity = None
        for handler in self._get_handlers_by_schemes(authentication_schemes, context):
            try:
                identity = await self._authenticate_with_handler(handler, context)
            except InvalidCredentialsError as invalid_credentials_error:
                # A client provided credentials of a given type, and they were invalid.
                # Store the information, so later calls can be validated without
                # attempting authentication.
                self._logger.info(
                    "Invalid credentials received from client IP %s for scheme: %s",
                    invalid_credentials_error.client_ip,
                    handler.scheme,
                )
                if self._rate_limiter:
                    await self._rate_limiter.store_authentication_failure(
                        invalid_credentials_error
                    )

            if identity:
                try:
                    context.identity = identity
                except AttributeError:
                    pass
                return identity
            else:
                try:
                    if context.identity is None:
                        context.identity = Identity()
                except AttributeError:
                    pass
        return None

    async def _authenticate_with_handler(self, handler: AuthenticationHandler, context):
        if _is_async_handler(type(handler)):
            return await handler.authenticate(context)  # type: ignore
        else:
            return handler.authenticate(context)