Auth Manager

The BaseAuthManager is responsible for managing authentication, authorization, and ORM models management. This is layer between client and Database. It uses Repositories to interact with the database.

Creation

To create AuthManager, simply inherit BaseAuthManager from fastauth.manager module. After inhering we need to override parse_id(self, pk: str) method. This method is used to convert user primary key from string(which we get from client or token) to the type stored in ORM model. We can pass ORM models to the BaseAuthManager Generic for type hinting.

import uuid
from fastauth.manager import BaseAuthManager
from fastapi import Request


class AuthManager(BaseAuthManager[User, uuid.UUID, Role, Permission, OAuthAccount]):

    def parse_id(self, pk: str):
        return uuid.UUID(pk)

Also we use AuthManager to override event methods like on_after_login, on_after_register etc. Supported events are:

import uuid
from fastauth.manager import BaseAuthManager
from fastapi import Request


class AuthManager(BaseAuthManager[User, uuid.UUID, Role, Permission, OAuthAccount]):

    def parse_id(self, pk: str):
        return uuid.UUID(pk)

    async def on_after_login(self, user, request: Request | None = None):
        print(f"User {user.email} logged in")

    async def on_after_register(self, user, request: Request | None = None):
        print(f"User {user.email} registered")

Supported events

• on_after_login - called when user logged in. Provide User model and Request

Dependency

We can define dependency callable for AuthManager, to passed it in FastAuth class. The main differeces from other dependencies callable, is than callable take FastAuthConfig instance as first argument. So if we use SQLAlchemy ORM, we need to create Repositories which need AsyncSession dependency, and pass it to AuthManager:

async def get_auth_manager(
    config: FastAuthConfig,
    session: AsyncSession = Depends(get_db),
):
    return AuthManager(
        config,
        UserRepository(session),
        RBACRepository(session),
        OAuthRepository(session),
    )

Auth manager also get password helper and token encoder instances, which can be passed to __init__ method.

async def get_auth_manager(
    config: FastAuthConfig,
    session: AsyncSession = Depends(get_db),
):
    return AuthManager(
        config,
        UserRepository(session),
        RBACRepository(session),
        OAuthRepository(session),
        password_helper=PasswordHelper(),
    )

Password Helper

Password helper is used to hash and verify passwords. It should implement PasswordHelperProtocol from fastauth.utils.password module. By default BaseAuthManager uses PasswordHelper class, which uses argon2 and bcrypt hashing algorithms, provided by pwdlib package

class PasswordHelperProtocol(Protocol):
    def verify_and_update(
        self, plain_password: str, hashed_password: str
    ) -> tuple[bool, str | None]: ...  # pragma: no cover

    def hash(self, password: str) -> str: ...  # pragma: no cover

    def generate(self) -> str: ...  # pragma: no cover

PasswordHelper

class PasswordHelper(PasswordHelperProtocol):
    def __init__(self, password_hash: PasswordHash | None = None) -> None:
        if password_hash is None:
            self.password_hash = PasswordHash(
                (
                    Argon2Hasher(),
                    BcryptHasher(),
                )
            )
        else:
            self.password_hash = password_hash  # pragma: no cover

    def verify_and_update(
        self, plain_password: str, hashed_password: str
    ) -> tuple[bool, str | None]:
        return self.password_hash.verify_and_update(plain_password, hashed_password)

    def hash(self, password: str) -> str:
        return self.password_hash.hash(password)

    def generate(self) -> str:
        return secrets.token_urlsafe()

Token Encoder

Token encoder is used to encode and decode tokens, as well as TokenStrategy but on lower level. It should implement TokenEncoderProtocol from fastauth.utils.jwt_helper module. By default it use JWT class from fastauth.utils.jwt_helper module.

class TokenHelperProtocol(Protocol):
    def decode_token(self, token: str, *args, **kwargs) -> dict[str, Any]:
        raise NotImplementedError

    def encode_token(
        self, payload: dict[str, Any], token_type: TokenType, *args, **kwargs
    ) -> str:
        raise NotImplementedError

JWT helper class

class JWTHelper:
    def __init__(self, secretkey: str, algorithm: str):
        self._secretkey = secretkey
        self._algorithm = algorithm

    def decode_token(
        self, token: str, audience: StringOrSequence | None = None, **kwargs
    ):
        return jwt.decode(
            token,
            key=self._secretkey,
            algorithms=[self._algorithm],
            audience=audience,
            **kwargs,
        )

    def encode_token(
        self,
        payload: dict[str, Any],
        token_type: TokenType,
        max_age: int | None = None,
        audience: StringOrSequence | None = None,
        headers: dict[str, Any] | None = None,
        **kwargs,
    ):
        payload["type"] = payload.get("type", token_type)
        payload["aud"] = payload.get("aud", audience)
        payload["iat"] = payload.get("iat", datetime.now(timezone.utc))
        payload["exp"] = payload.get(
            "exp", payload.get("iat") + timedelta(seconds=max_age)
        )
        return jwt.encode(
            payload,
            key=self._secretkey,
            algorithm=self._algorithm,
            headers=headers,
            **kwargs,
        )