Controllers¶

def create_auth_controller[UP: UserProtocol[Any], ID](  # noqa: PLR0913
    *,
    backend: AuthenticationBackend[UP, ID],
    rate_limit_config: AuthRateLimitConfig | None = None,
    enable_refresh: bool = False,
    requires_verification: bool = False,
    login_identifier: LoginIdentifier = "email",
    totp_pending_secret: str | None = None,
    totp_pending_lifetime: timedelta = _DEFAULT_PENDING_TOKEN_LIFETIME,
    path: str = "/auth",
) -> type[Controller]:
    """Return a controller subclass bound to the provided backend (DI user manager).

    Args:
        backend: Auth backend used to issue and invalidate tokens.
        rate_limit_config: Optional auth-endpoint rate-limiter configuration.
        enable_refresh: When ``True``, issue refresh tokens on login and add
            ``POST /refresh`` for refresh-token rotation.
        requires_verification: When ``True``, unverified users receive
            ``LOGIN_USER_NOT_VERIFIED`` (400) instead of a token.
        login_identifier: Which user attribute is used to interpret the login
            ``identifier`` field (``email`` or ``username``). Must match
            ``LitestarAuthConfig.login_identifier`` when using the plugin.
        totp_pending_secret: When set, enables 2FA support. Login returns an
            intermediate pending token when the user has TOTP configured.
            Must match the value passed to ``create_totp_controller``.
        totp_pending_lifetime: Maximum age of the intermediate pending token.
        path: Base route prefix for the generated controller.

    Returns:
        Controller subclass with backend-specific login and logout handlers.

    Examples:
        ```python
        from litestar_auth.controllers.auth import create_auth_controller

        AuthController = create_auth_controller(
            backend=backend,
            path="/auth",
        )
        ```
    """
    if totp_pending_secret is not None and not is_testing():
        validate_secret_length(totp_pending_secret, label="totp_pending_secret")
    ctx = _make_auth_controller_context(
        backend=backend,
        rate_limit_config=rate_limit_config,
        enable_refresh=enable_refresh,
        requires_verification=requires_verification,
        login_identifier=login_identifier,
        totp_pending_secret=totp_pending_secret,
        totp_pending_lifetime=totp_pending_lifetime,
    )
    base_cls = _define_auth_controller_class_di(ctx)
    generated_controller: type[Controller] = (
        _define_refresh_auth_controller_class_di(base_cls, ctx) if ctx.refresh_strategy is not None else base_cls
    )
    generated_controller.__name__ = f"{_build_controller_name(backend.name)}AuthController"
    generated_controller.__qualname__ = generated_controller.__name__
    generated_controller.path = path
    return generated_controller

def create_oauth_associate_controller[UP: UserProtocol[Any], ID](  # noqa: PLR0913
    *,
    provider_name: str,
    user_manager: OAuthControllerUserManagerProtocol[UP, ID] | None = None,
    user_manager_dependency_key: str | None = None,
    oauth_client: object,
    redirect_base_url: str,
    path: str = "/auth/associate",
    cookie_secure: bool = True,
) -> type[Controller]:
    """Return a controller for linking an OAuth account to the authenticated user.

    Both /authorize and /callback are protected by is_authenticated. Callback
    upserts the OAuth account for request.user and does not create new users.

    Provide either user_manager (for direct use) or user_manager_dependency_key
    (for plugin use with a request-scoped dependency).

    Returns:
        Generated controller class mounted under the provider-specific path.

    """
    assembly = _build_oauth_controller_assembly(
        provider_name=provider_name,
        oauth_client=oauth_client,
        redirect_base_url=redirect_base_url,
        path=path,
        cookie_secure=cookie_secure,
        state_cookie_prefix=ASSOCIATE_STATE_COOKIE_PREFIX,
        controller_name_suffix="OAuthAssociateController",
        user_manager_binding=_build_associate_user_manager_binding(
            user_manager=user_manager,
            user_manager_dependency_key=user_manager_dependency_key,
        ),
    )
    return _create_oauth_controller_type(
        assembly=assembly,
        authorize_handler=_create_authorize_handler(
            assembly=assembly,
            guards=[is_authenticated],
            allow_scopes=False,
        ),
        callback_handler=_create_associate_callback_handler(assembly=assembly),
        docstring="Provider-specific OAuth associate authorize/callback endpoints.",
    )

def create_oauth_controller[UP: UserProtocol[Any], ID](  # noqa: PLR0913
    *,
    provider_name: str,
    backend: AuthenticationBackend[UP, ID],
    user_manager: OAuthControllerUserManagerProtocol[UP, ID],
    oauth_client: object,
    redirect_base_url: str,
    path: str = "/auth/oauth",
    cookie_secure: bool = True,
    associate_by_email: bool = False,
    trust_provider_email_verified: bool = False,
) -> type[Controller]:
    """Return a controller subclass bound to one OAuth provider.

    Returns:
        Generated controller class mounted under the provider-specific path.
    """
    assembly = _build_oauth_controller_assembly(
        provider_name=provider_name,
        oauth_client=oauth_client,
        redirect_base_url=redirect_base_url,
        path=path,
        cookie_secure=cookie_secure,
        state_cookie_prefix=STATE_COOKIE_PREFIX,
        controller_name_suffix="OAuthController",
        user_manager_binding=_build_direct_user_manager_binding(user_manager),
        associate_by_email=associate_by_email,
        trust_provider_email_verified=trust_provider_email_verified,
    )
    return _create_oauth_controller_type(
        assembly=assembly,
        authorize_handler=_create_authorize_handler(
            assembly=assembly,
            allow_scopes=True,
        ),
        callback_handler=_create_login_callback_handler(
            assembly=assembly,
            backend=backend,
        ),
        docstring="Provider-specific OAuth authorize/callback endpoints.",
    )

def create_register_controller[UP: RegisterControllerUserProtocol[Any], ID](
    *,
    rate_limit_config: AuthRateLimitConfig | None = None,
    path: str = "/auth",
    user_read_schema: type[msgspec.Struct] = UserRead,
    user_create_schema: type[msgspec.Struct] = UserCreate,
) -> type[Controller]:
    """Return a controller subclass that resolves the user manager via Litestar DI.

    Args:
        rate_limit_config: Optional auth-endpoint rate-limiter configuration.
        path: Base route prefix for the generated controller.
        user_read_schema: Custom msgspec struct used for public registration responses.
        user_create_schema: Custom msgspec struct used for registration requests.

    Returns:
        Controller subclass exposing the registration endpoint.

    Examples:
        ```python
        class ExtendedUserCreate(msgspec.Struct):
            email: str
            password: str
            bio: str

        class ExtendedUserRead(msgspec.Struct):
            id: uuid.UUID
            email: str
            is_active: bool
            is_verified: bool
            is_superuser: bool
            bio: str

        controller = create_register_controller(
            user_create_schema=ExtendedUserCreate,
            user_read_schema=ExtendedUserRead,
        )
        ```
    """
    _require_msgspec_struct(user_read_schema, parameter_name="user_read_schema")
    _require_msgspec_struct(user_create_schema, parameter_name="user_create_schema")
    register_rate_limit = rate_limit_config.register if rate_limit_config else None
    user_read_schema_type = user_read_schema
    user_create_schema_type = user_create_schema
    register_before_request = _create_before_request_handler(register_rate_limit)

    class RegisterController(Controller):
        """Endpoints for registering a new user."""

        @post("/register", before_request=register_before_request)
        async def register(  # noqa: PLR6301
            self,
            request: Request[Any, Any, Any],
            data: msgspec.Struct,
            litestar_auth_user_manager: Any,  # noqa: ANN401
        ) -> msgspec.Struct:
            async def _increment_rate_limit() -> None:
                if register_rate_limit is not None:
                    await register_rate_limit.increment(request)

            async with _map_domain_exceptions(
                {
                    UserAlreadyExistsError: (400, ErrorCode.REGISTER_USER_ALREADY_EXISTS),
                    InvalidPasswordError: (400, ErrorCode.REGISTER_INVALID_PASSWORD),
                },
                on_error=_increment_rate_limit,
            ):
                user = await litestar_auth_user_manager.create(data, safe=True)

            if register_rate_limit is not None:
                await register_rate_limit.reset(request)

            return _to_user_schema(user, user_read_schema_type)

    register_cls = RegisterController
    _configure_request_body_handler(register_cls.register, schema=user_create_schema_type)
    register_cls.path = path
    return cast("type[Controller]", register_cls)

def create_reset_password_controller[UP: ResetPasswordControllerUserProtocol[Any], ID](
    *,
    rate_limit_config: AuthRateLimitConfig | None = None,
    path: str = "/auth",
    user_read_schema: type[msgspec.Struct] = UserRead,
) -> type[Controller]:
    """Return a controller subclass that resolves the user manager via Litestar DI.

    Args:
        rate_limit_config: Optional auth-endpoint rate-limiter configuration.
        path: Base route prefix for the generated controller.
        user_read_schema: Custom msgspec struct used for public reset-password responses.

    Returns:
        Controller subclass exposing reset-password endpoints.
    """
    _require_msgspec_struct(user_read_schema, parameter_name="user_read_schema")
    forgot_password_rate_limit = rate_limit_config.forgot_password if rate_limit_config else None
    reset_password_rate_limit = rate_limit_config.reset_password if rate_limit_config else None
    user_read_schema_type = user_read_schema
    forgot_password_rate_limit_before_request = _create_before_request_handler(forgot_password_rate_limit)
    reset_password_rate_limit_before_request = _create_before_request_handler(reset_password_rate_limit)

    reset_password_rate_limit_increment, reset_password_rate_limit_reset = _create_rate_limit_handlers(
        reset_password_rate_limit,
    )

    class ResetPasswordController(Controller):
        """Endpoints for password reset flows."""

        @post(
            "/forgot-password",
            status_code=HTTP_202_ACCEPTED,
            before_request=forgot_password_rate_limit_before_request,
        )
        async def forgot_password(  # noqa: PLR6301
            self,
            request: Request[Any, Any, Any],
            data: ForgotPassword,
            litestar_auth_user_manager: Any,  # noqa: ANN401
        ) -> None:
            await litestar_auth_user_manager.forgot_password(data.email)
            # Rate limit increments only after successful dispatch — intentional.
            # Counting failures would let attackers distinguish "email not found"
            # from "email sent", enabling account enumeration.
            if forgot_password_rate_limit is not None:
                await forgot_password_rate_limit.increment(request)

        @post(
            "/reset-password",
            status_code=HTTP_200_OK,
            before_request=reset_password_rate_limit_before_request,
        )
        async def reset_password(  # noqa: PLR6301
            self,
            request: Request[Any, Any, Any],
            data: ResetPassword,
            litestar_auth_user_manager: Any,  # noqa: ANN401
        ) -> msgspec.Struct:
            async def _increment_rate_limit() -> None:
                await reset_password_rate_limit_increment(request)

            async with _map_domain_exceptions(
                {
                    InvalidResetPasswordTokenError: (400, ErrorCode.RESET_PASSWORD_BAD_TOKEN),
                    InvalidPasswordError: (400, ErrorCode.RESET_PASSWORD_INVALID_PASSWORD),
                },
                on_error=_increment_rate_limit,
            ):
                user = await litestar_auth_user_manager.reset_password(data.token, data.password)

            await reset_password_rate_limit_reset(request)
            return _to_user_schema(user, user_read_schema_type)

    reset_cls = ResetPasswordController
    _configure_request_body_handler(reset_cls.reset_password, schema=ResetPassword)
    reset_cls.path = path
    return reset_cls

def create_totp_controller[UP: UserProtocol[Any], ID](  # noqa: PLR0913
    *,
    backend: AuthenticationBackend[UP, ID],
    user_manager_dependency_key: str,
    used_tokens_store: UsedTotpCodeStore | None = None,
    pending_jti_store: JWTDenylistStore | None = None,
    require_replay_protection: bool = True,
    rate_limit_config: AuthRateLimitConfig | None = None,
    requires_verification: bool = False,
    totp_pending_secret: str,
    totp_enable_requires_password: bool = True,
    totp_issuer: str = "litestar-auth",
    totp_algorithm: TotpAlgorithm = "SHA256",
    totp_pending_lifetime: timedelta | None = None,
    id_parser: Callable[[str], ID] | None = None,
    path: str = "/auth/2fa",
) -> type[Controller]:
    """Return a controller with TOTP enable/verify/disable endpoints.

    Args:
        backend: Auth backend used to issue tokens after successful TOTP verification.
        user_manager_dependency_key: Litestar DI key / handler parameter name for the
            request-scoped user manager.
        used_tokens_store: Optional replay-protection cache for successful `/verify`
            attempts. When omitted, same-window replay protection stays disabled.
        pending_jti_store: Optional denylist store used to reject replayed
            pending-token JTIs after successful `/verify`. When omitted, JTIs
            are still required and validated structurally, but not deduplicated.
        require_replay_protection: When enabled, the controller refuses to start
            without a used-token replay store outside testing mode.
        rate_limit_config: Optional auth-endpoint rate-limiter configuration.
        requires_verification: When ``True``, `/2fa/verify` applies the same
            account-state policy as `/login`, rejecting inactive users and
            users with `is_verified=False`.
        totp_pending_secret: Shared secret for signing and verifying pending-2FA JWTs.
            Must match the value passed to ``create_auth_controller``.
        totp_enable_requires_password: When ``True`` (default), `/enable` requires a JSON body
            with the user's current password and re-authenticates before storing
            a new TOTP secret. Set to ``False`` only if you accept the session-hijack
            escalation risk (not recommended).
        totp_issuer: Issuer label shown inside authenticator-app QR codes.
        totp_algorithm: Hash algorithm used for TOTP generation and verification.
        totp_pending_lifetime: Unused; kept for API symmetry with
            ``create_auth_controller``.
        id_parser: Optional callable that converts the JWT ``sub`` string into the
            application's user ID type (e.g. ``UUID`` for UUID-keyed users).
        path: Base route prefix for the generated controller.

    Returns:
        Controller subclass with TOTP management endpoints.

    Examples:
        ```python
        from litestar_auth.controllers.totp import create_totp_controller

        totp_controller_cls = create_totp_controller(
            backend=backend,
            user_manager_dependency_key="litestar_auth_user_manager",
            totp_pending_secret=settings.totp_pending_secret,
        )
        ```
    """
    del user_manager_dependency_key
    del totp_pending_lifetime  # symmetry param; lifetime is set on the issuer side
    if not is_testing():
        validate_secret_length(totp_pending_secret, label="totp_pending_secret")
    _totp_validate_replay_and_password(
        used_tokens_store=used_tokens_store,
        require_replay_protection=require_replay_protection,
        totp_enable_requires_password=totp_enable_requires_password,
        user_manager=None,
    )
    testing_mode = is_testing()
    effective_pending_jti_store = _totp_resolve_pending_jti_store(pending_jti_store, testing_mode=testing_mode)

    totp_rate_limit = TotpRateLimitOrchestrator(
        enable=rate_limit_config.totp_enable if rate_limit_config else None,
        confirm_enable=rate_limit_config.totp_confirm_enable if rate_limit_config else None,
        verify=rate_limit_config.totp_verify if rate_limit_config else None,
        disable=rate_limit_config.totp_disable if rate_limit_config else None,
    )
    ctx = _TotpControllerContext(
        backend=backend,
        used_tokens_store=used_tokens_store,
        require_replay_protection=require_replay_protection,
        requires_verification=requires_verification,
        totp_enable_requires_password=totp_enable_requires_password,
        totp_issuer=totp_issuer,
        totp_algorithm=totp_algorithm,
        totp_rate_limit=totp_rate_limit,
        totp_pending_secret=totp_pending_secret,
        effective_pending_jti_store=effective_pending_jti_store,
        id_parser=id_parser,
    )

    async def totp_verify_before_request(request: Request[Any, Any, Any]) -> None:
        await totp_rate_limit.before_request("verify", request)

    before = totp_verify_before_request if totp_rate_limit.verify is not None else None
    totp_controller_cls = _define_totp_controller_class_di(
        ctx,
        totp_verify_before_request=before,
    )
    totp_controller_cls.__name__ = "TotpController"
    totp_controller_cls.__qualname__ = "TotpController"
    totp_controller_cls.path = path
    return totp_controller_cls

def create_users_controller[UP: UsersControllerUserProtocol[Any], ID](  # noqa: PLR0913
    *,
    id_parser: Callable[[str], ID] | None = None,
    path: str = "/users",
    default_limit: int = 50,
    max_limit: int = 100,
    hard_delete: bool = False,
    user_read_schema: type[msgspec.Struct] = UserRead,
    user_update_schema: type[msgspec.Struct] = UserUpdate,
) -> type[Controller]:
    """Return a controller subclass that resolves the user manager via Litestar DI.

    Args:
        id_parser: Optional callable that converts path IDs into manager IDs.
        path: Base route prefix for the generated controller.
        default_limit: Default page size for list responses.
        max_limit: Maximum allowed page size for list responses.
        hard_delete: When ``True``, admin deletes remove users permanently.
        user_read_schema: Custom msgspec struct used for public user responses.
        user_update_schema: Custom msgspec struct used for update requests.

    Returns:
        Controller subclass exposing self-service and admin user endpoints.

    Examples:
        ```python
        class ExtendedUserRead(msgspec.Struct):
            id: uuid.UUID
            email: str
            is_active: bool
            is_verified: bool
            is_superuser: bool
            bio: str

        class ExtendedUserUpdate(msgspec.Struct, omit_defaults=True):
            email: str | None = None
            password: str | None = None
            bio: str | None = None

        controller = create_users_controller(
            user_read_schema=ExtendedUserRead,
            user_update_schema=ExtendedUserUpdate,
        )
        ```
    """
    _require_msgspec_struct(user_read_schema, parameter_name="user_read_schema")
    _require_msgspec_struct(user_update_schema, parameter_name="user_update_schema")
    users_page_schema_type = msgspec.defstruct(
        "UsersPageSchema",
        [
            ("items", list[Any]),
            ("total", int),
            ("limit", int),
            ("offset", int),
        ],
    )
    ctx = _UsersControllerContext(
        id_parser=id_parser,
        user_read_schema_type=user_read_schema,
        user_update_schema_type=user_update_schema,
        users_page_schema_type=users_page_schema_type,
        hard_delete=hard_delete,
        default_limit=default_limit,
        max_limit=max_limit,
    )
    controller_cls = _define_users_controller_class_di(ctx)
    controller_cls.path = path
    return controller_cls

def create_verify_controller[UP: VerifyControllerUserProtocol[Any], ID](
    *,
    rate_limit_config: AuthRateLimitConfig | None = None,
    path: str = "/auth",
    user_read_schema: type[msgspec.Struct] = UserRead,
) -> type[Controller]:
    """Return a controller subclass that resolves the user manager via Litestar DI.

    Args:
        rate_limit_config: Optional auth-endpoint rate-limiter configuration. When provided,
            ``request_verify_token`` requests are subject to rate limiting.
        path: Base route prefix for the generated controller.
        user_read_schema: Custom msgspec struct used for public verification responses.

    Returns:
        Controller subclass exposing verification-related endpoints.
    """
    _require_msgspec_struct(user_read_schema, parameter_name="user_read_schema")
    user_read_schema_type = user_read_schema

    verify_rate_limit = rate_limit_config.verify_token if rate_limit_config else None
    request_verify_rate_limit = rate_limit_config.request_verify_token if rate_limit_config else None
    verify_rate_limit_before_request = _create_before_request_handler(verify_rate_limit)
    request_verify_rate_limit_before_request = _create_before_request_handler(request_verify_rate_limit)

    verify_rate_limit_increment, verify_rate_limit_reset = _create_rate_limit_handlers(verify_rate_limit)

    class VerifyController(Controller):
        """Endpoints for email verification."""

        @post(
            "/verify",
            status_code=HTTP_200_OK,
            before_request=verify_rate_limit_before_request,
        )
        async def verify(  # noqa: PLR6301
            self,
            request: Request[Any, Any, Any],
            data: VerifyToken,
            litestar_auth_user_manager: Any,  # noqa: ANN401
        ) -> msgspec.Struct:
            try:
                user = await litestar_auth_user_manager.verify(data.token)
            except InvalidVerifyTokenError as exc:
                await verify_rate_limit_increment(request)
                raise ClientException(
                    status_code=400,
                    detail=str(exc),
                    extra={"code": ErrorCode.VERIFY_USER_BAD_TOKEN},
                ) from exc

            await verify_rate_limit_reset(request)
            return _to_user_schema(user, user_read_schema_type)

        @post(
            "/request-verify-token",
            status_code=HTTP_202_ACCEPTED,
            before_request=request_verify_rate_limit_before_request,
        )
        async def request_verify_token(  # noqa: PLR6301
            self,
            request: Request[Any, Any, Any],
            data: RequestVerifyToken,
            litestar_auth_user_manager: Any,  # noqa: ANN401
        ) -> None:
            await litestar_auth_user_manager.request_verify_token(data.email)
            if request_verify_rate_limit is not None:
                await request_verify_rate_limit.increment(request)

    verify_cls = VerifyController
    verify_cls.path = path
    return cast("type[Controller]", verify_cls)