Support authentication via API key?

[aknoerig]

In addition to logging in to the system, I'd like to offer users to call my API programmatically. For that purpose, it seems a bit cumbersome to go through the login/JWT process.

So I was thinking to generate an individual API key for each user, that lets them make API requests directly. Conveniently, FastAPI already has the validation built-in (see fastapi.security.api_key).
Now I'm looking for a way to marry that to fastapi-users as an alternative strategy.

Has somebody tried this before, or has any thoughts on this? Wouldn't this be a useful general addition to fastapi-users?
Any pointers are appreciated!

[caniko]

@frankie567?

[lemanjo]

Has somebody tried this before, or has any thoughts on this? Wouldn't this be a useful general addition to fastapi-users? Any pointers are appreciated!

Old question, but if someone is still reading this and trying to find a solution. You can define multiple authentication backends and just pass those to your FastAPIUsers instance. Check https://fastapi-users.github.io/fastapi-users/12.1/configuration/authentication/backend/

[tim-habitat]

hey, thanks for your guidance - would you say the common way to do this would be to define a backend using a bearer transport and a database strategy ?

[lemanjo]

Common way for just API authentication? I guess it is, I'm not the most qualified person to answer that.

I personally have implemented in a way that the users have a UI to enable it and regenerate/disable tokens etc.

[tim-habitat]

Thanks for the prompt reply!
Yeah i think that's where i would like to go as well - were you using fastapi-users then ?

[lemanjo]

Yes.

[patrickwasp]

I'm having trouble implementing this, can you provide any tips?

when I add the api_key_backend, the swagger authorize button changes from jwt to api_key (Token URL: auth/api_key/login) and longer works to log in using a username/password. I'm looking for this functionality in swagger, https://swagger.io/docs/specification/authentication/api-keys/ using request headers.

fastapi_users_router = fastapi_users.FastAPIUsers[models.users.User, uuid.UUID](
    get_user_manager,
    [api.authentication.jwt_backend, api.authentication.api_key_backend],
)

models/users.py

import datetime
import typing

import fastapi_users_db_sqlmodel
import pydantic
import sqlmodel

from models import mixins

class User(
    fastapi_users_db_sqlmodel.SQLModelBaseUserDB, mixins.TimeStampMixin, table=True
):
    email: str
    name: str

    api_keys: list["ApiKey"] = sqlmodel.Relationship(back_populates="user")


class ApiKey(mixins.TimeStampMixin, table=True):
    key: str = sqlmodel.Field(primary_key=True)
    name: str | None
    expired_at: datetime.datetime | None

    user_id: pydantic.UUID4 = sqlmodel.Field(default=None, foreign_key="user.id")
    user: User = sqlmodel.Relationship(back_populates="api_keys")

dependencies/auth.py

import typing

import fastapi
import fastapi.security
import fastapi_users.db
import sqlalchemy.ext.asyncio
import sqlmodel

import models.users
from config.database import get_async_session


async def get_user_db(
    session: sqlalchemy.ext.asyncio.AsyncSession = fastapi.Depends(get_async_session),
):
    yield fastapi_users.db.SQLAlchemyUserDatabase(session, models.users.User)


class ApiKeyDatabase:
    def __init__(self, session: sqlalchemy.ext.asyncio.AsyncSession):
        self.session = session

    async def get_by_key(self, key: str) -> models.users.ApiKey | None:
        result = await self.session.execute(
            sqlmodel.select(models.users.ApiKey).where(models.users.ApiKey.key == key)
        )
        return result.scalars().first()

    async def create(self, create_dict: dict[str, typing.Any]) -> models.users.ApiKey:
        api_key = models.users.ApiKey(**create_dict)
        self.session.add(api_key)
        await self.session.commit()
        await self.session.refresh(api_key)
        return api_key

    async def delete(self, api_key: models.users.ApiKey) -> None:
        await self.session.delete(api_key)
        await self.session.commit()


async def get_api_key_db(
    session: sqlalchemy.ext.asyncio.AsyncSession = fastapi.Depends(get_async_session),
):
    yield ApiKeyDatabase(session)

authentication.py

import datetime

import fastapi
import fastapi_users.authentication
import fastapi_users.exceptions
import fastapi_users.models
from fastapi_users.authentication.strategy.base import Strategy
from fastapi_users.manager import BaseUserManager

import api.dependencies.auth
from config.settings import settings

SECRET = settings.JWT_SECRET

# JWT Backend
bearer_transport = fastapi_users.authentication.BearerTransport(
    tokenUrl="auth/jwt/login"
)


def get_jwt_strategy() -> fastapi_users.authentication.JWTStrategy:
    return fastapi_users.authentication.JWTStrategy(
        secret=SECRET, lifetime_seconds=3600
    )


jwt_backend = fastapi_users.authentication.AuthenticationBackend(
    name="jwt",
    transport=bearer_transport,
    get_strategy=get_jwt_strategy,
)


# API Key Strategy
class ApiKeyStrategy(Strategy[fastapi_users.models.UP, fastapi_users.models.ID]):
    def __init__(self, api_key_db: api.dependencies.auth.ApiKeyDatabase):
        self.api_key_db = api_key_db

    async def read_token(
        self,
        token: str | None,
        user_manager: BaseUserManager[fastapi_users.models.UP, fastapi_users.models.ID],
    ) -> fastapi_users.models.UP | None:
        if token is None:
            return None

        api_key = await self.api_key_db.get_by_key(token)
        if (
            api_key is None
            or api_key.expired_at is not None
            and api_key.expired_at < datetime.datetime.utcnow()
        ):
            return None

        try:
            parsed_id = user_manager.parse_id(api_key.user_id)
            return await user_manager.get(parsed_id)
        except (
            fastapi_users.exceptions.UserNotExists,
            fastapi_users.exceptions.InvalidID,
        ):
            return None

    async def write_token(self, user: fastapi_users.models.UP) -> str:
        raise NotImplementedError("API keys are not generated automatically")

    async def destroy_token(self, token: str, user: fastapi_users.models.UP) -> None:
        api_key = await self.api_key_db.get_by_key(token)
        if api_key is not None:
            await self.api_key_db.delete(api_key)


def get_api_key_strategy(
    api_key_db: api.dependencies.auth.ApiKeyDatabase = fastapi.Depends(
        api.dependencies.auth.get_api_key_db
    ),
) -> ApiKeyStrategy:
    return ApiKeyStrategy(api_key_db)


api_key_transport = fastapi_users.authentication.BearerTransport(
    tokenUrl="auth/api_key/login"
)

api_key_backend = fastapi_users.authentication.AuthenticationBackend(
    name="api_key",
    transport=api_key_transport,
    get_strategy=get_api_key_strategy,
)

routes/users.py

import typing
import uuid

import fastapi
import fastapi.security
import fastapi_users
import fastapi_users.authentication
import fastapi_users.db
import fastapi_users.schemas
import sqlalchemy

import api.authentication
import api.schemas.users
import models.users
from config.database import async_engine

async_session_maker = sqlalchemy.ext.asyncio.async_sessionmaker(
    async_engine, expire_on_commit=False
)


# User Manager Definition
class UserManager(
    fastapi_users.UUIDIDMixin,
    fastapi_users.BaseUserManager[models.users.User, uuid.UUID],
):
    async def on_after_register(
        self, user: models.users.User, request: typing.Optional[fastapi.Request] = None
    ):
        print(f"User {user.id} has registered.")

    async def on_after_forgot_password(
        self,
        user: models.users.User,
        token: str,
        request: typing.Optional[fastapi.Request] = None,
    ):
        print(f"User {user.id} has forgotten their password. Reset token: {token}")

    async def on_after_request_verify(
        self,
        user: models.users.User,
        token: str,
        request: typing.Optional[fastapi.Request] = None,
    ):
        print(f"Verification requested for user {user.id}. Verification token: {token}")


# User Manager and FastAPI Users Initialization
async def get_async_session() -> (
    typing.AsyncGenerator[sqlalchemy.ext.asyncio.AsyncSession, None]
):
    async with async_session_maker() as session:
        yield session


async def get_user_db(
    session: sqlalchemy.ext.asyncio.AsyncSession = fastapi.Depends(get_async_session),
):
    yield fastapi_users.db.SQLAlchemyUserDatabase(session, models.users.User)


async def get_user_manager(
    user_db: fastapi_users.db.SQLAlchemyUserDatabase = fastapi.Depends(get_user_db),
):
    yield UserManager(user_db)


fastapi_users_router = fastapi_users.FastAPIUsers[models.users.User, uuid.UUID](
    get_user_manager,
    [api.authentication.jwt_backend, api.authentication.api_key_backend],
)

current_superuser = fastapi_users_router.current_user(active=True, superuser=True)
current_user = fastapi_users_router.current_user(active=True)


# Router Configuration
users_router = fastapi.APIRouter()
auth_router = fastapi.APIRouter()

auth_router.include_router(
    fastapi_users_router.get_auth_router(api.authentication.jwt_backend),
    prefix="/auth/jwt",
    tags=["auth"],
)

auth_router.include_router(
    fastapi_users_router.get_auth_router(api.authentication.api_key_backend),
    prefix="/auth/api_key",
    tags=["auth"],
)

auth_router.include_router(
    fastapi_users_router.get_reset_password_router(),
    prefix="/auth",
    tags=["auth"],
)
auth_router.include_router(
    fastapi_users_router.get_verify_router(api.schemas.users.UserRead),
    prefix="/auth",
    tags=["auth"],
)
users_router.include_router(
    fastapi_users_router.get_users_router(
        api.schemas.users.UserRead, api.schemas.users.UserUpdate
    ),
    prefix="/users",
    tags=["users"],
)

[lemanjo]

Not completely sure since I don't care about the Swagger functionality. I've never looked at what's going on in there. Complete mystery to me if Swagger even supports multiple authentication methods to the docs it creates. Nor have I touched that part of my code in a long time.

I'm wondering why are you rewriting the whole strategy. Could you use the existing database strategy with the API key implementation?

from fastapi_users.authentication.strategy.db import DatabaseStrategy

auth_backend_token = AuthenticationBackend(
    name="database",
    transport=bearer_transport,
    get_strategy=get_database_strategy,
)

def get_database_strategy(
    access_token_db: AccessTokenDatabase[AccessToken] = Depends(
        get_access_token_db),
) -> DatabaseStrategy:
    return DatabaseStrategy(access_token_db)

[frankvp11]

Hello!
I too am trying to make it so that you can use API keys (in some sense) alongside the JWT secret. I too would like to use the Database strategy for this.
I've managed to get it so that I can have two auth backends in the FastAPI Users, however when I register a new user with the register route, it doesn't add a token to the Access Token, and it doesn't look like the Model ever gets used.
Furthermore, I'd like to be able to check in some sense if the user making a request to one of my endpoints has one of these "API Keys" in my database. How can I check this, similar to the fastapi.Depends(current_user) you can do with the JWT secret.

Here is my code:

import typing
import uuid

import fastapi
import fastapi_users
import fastapi_users.schemas
import sqlalchemy

import models.users
import api.schemas.users
from config.database import async_engine
from config.settings import settings

async_session_maker = sqlalchemy.ext.asyncio.async_sessionmaker(
    async_engine, expire_on_commit=False
)


# User Manager Definition
class UserManager(
    fastapi_users.UUIDIDMixin,
    fastapi_users.BaseUserManager[models.users.User, uuid.UUID],
):
    async def on_after_register(
        self, user: models.users.User, request: typing.Optional[fastapi.Request] = None
    ):
        print(f"User {user.id} has registered.")

    async def on_after_forgot_password(
        self,
        user: models.users.User,
        token: str,
        request: typing.Optional[fastapi.Request] = None,
    ):
        print(f"User {user.id} has forgotten their password. Reset token: {token}")

    async def on_after_request_verify(
        self,
        user: models.users.User,
        token: str,
        request: typing.Optional[fastapi.Request] = None,
    ):
        print(f"Verification requested for user {user.id}. Verification token: {token}")


# Authentication Setup
bearer_transport = fastapi_users.authentication.BearerTransport(
    tokenUrl="auth/jwt/login"
)


# User Manager and FastAPI Users Initialization
async def get_async_session() -> (
    typing.AsyncGenerator[sqlalchemy.ext.asyncio.AsyncSession, None]
):
    async with async_session_maker() as session:
        yield session


def get_jwt_strategy() -> fastapi_users.authentication.JWTStrategy:
    return fastapi_users.authentication.JWTStrategy(
        secret=settings.JWT_SECRET, lifetime_seconds=3600
    )  # 1 hour


auth_backend = fastapi_users.authentication.AuthenticationBackend(
    name="jwt",
    transport=bearer_transport,
    get_strategy=get_jwt_strategy,
)

import fastapi_users_db_sqlmodel.access_token


async def get_access_token_db(
    session: sqlalchemy.ext.asyncio.AsyncSession = fastapi.Depends(get_async_session),
):
    yield fastapi_users_db_sqlmodel.access_token.SQLModelAccessTokenDatabaseAsync(
        session, models.users.AccessToken
    )


def get_database_strategy(
    access_token_db: fastapi_users_db_sqlmodel.access_token.SQLModelAccessTokenDatabaseAsync[
        models.users.AccessToken
    ] = fastapi.Depends(
        get_access_token_db
    ),
) -> fastapi_users.authentication.strategy.db:
    return fastapi_users.authentication.strategy.db.DatabaseStrategy(access_token_db)


auth_backend_token = fastapi_users.authentication.AuthenticationBackend(
    name="database",
    transport=bearer_transport,
    get_strategy=get_database_strategy,
)


async def get_user_db(
    session: sqlalchemy.ext.asyncio.AsyncSession = fastapi.Depends(get_async_session),
):
    yield fastapi_users.db.SQLAlchemyUserDatabase(session, models.users.User)


async def get_user_manager(
    user_db: fastapi_users.db.SQLAlchemyUserDatabase = fastapi.Depends(get_user_db),
):
    yield UserManager(user_db)


fastapi_users_router = fastapi_users.FastAPIUsers[models.users.User, uuid.UUID](
    get_user_manager, [auth_backend, auth_backend_token]
)

current_superuser = fastapi_users_router.current_user(active=True, superuser=True)
current_user = fastapi_users_router.current_user(active=True)


# Router Configuration
users_router = fastapi.APIRouter()
auth_router = fastapi.APIRouter()

auth_router.include_router(
    fastapi_users_router.get_auth_router(auth_backend),
    prefix="/auth/jwt",
    tags=["auth"],
)

auth_router.include_router(
    fastapi_users_router.get_reset_password_router(),
    prefix="/auth",
    tags=["auth"],
)
auth_router.include_router(
    fastapi_users_router.get_verify_router(api.schemas.users.UserRead),
    prefix="/auth",
    tags=["auth"],
)
users_router.include_router(
    fastapi_users_router.get_users_router(
        api.schemas.users.UserRead, api.schemas.users.UserUpdate
    ),
    prefix="/users",
    tags=["users"],
)
auth_router.include_router(
    fastapi_users_router.get_register_router(
        api.schemas.users.UserRead, api.schemas.users.UserCreate
    ),
    prefix="/auth",
    tags=["auth"],
)

Here is my model:

import typing

import fastapi_users_db_sqlmodel
import pydantic
import sqlmodel



class User(fastapi_users_db_sqlmodel.SQLModelBaseUserDB, table=True):
    email: str
    name: str



import fastapi_users_db_sqlmodel.access_token


class AccessToken(
    fastapi_users_db_sqlmodel.access_token.SQLModelBaseAccessToken, table=True
):
    pass

[lemanjo]

You may need to create the API key separately in UserManager on_after_register or then write custom logic for the user creation. I did this in a way that users enable API key from app settings so I generate the key at that point.

I think fastapi.Depends(current_user) should work with the API key or JWT out of the box when you just get the key in, since you've defined both backends. No need to create separate logic. You can also specify endpoints to use one of the authentication backends: https://fastapi-users.github.io/fastapi-users/10.0/usage/current-user/