FastAPI Backend: Part 4 (Authentication)

Logs for implementing authentication

what I want:

Generate a set of 2 tokens on login, access and refresh token
- Access Token: 15 mins expiry
- Refresh Token: 7 days expiry
Refreshing Token:
- Generate new set of tokens
Maintaining refresh token in a separate DB for session
- Using Valkey as “in memory DB” for tracking refresh token
- We can replace it with Redis later if needed just by changing DB client, nothing else
Log out

Login:

So for login, we need to make:

auth routes
auth service
- we will be using combination of user service, security utils and auth DB
security utils:
- for hashing password
- for verifying password
- generate access tokens set
- decode access token and refresh token
- generate “jti” for refresh token to track
Auth DB handler:
- create valkey client
- ping auth db
  - for making sure auth DB is ready before we start the app
- store refresh jti
  - we will be tracking refresh token or session using jti
- get refresh jti with status
- get and delete refresh jti
Guards
- Auth Guard
  - for all the protected routes
- Refresh token Guard
  - for validating refresh token while refreshing tokens

Auth Route:

Python

			
# src/features/users/schemas.py
from pydantic import BaseModel
class LoginUser(BaseModel):
    email: str
    password: str
# src/routes/v1/auth_routes.py
from fastapi import APIRouter
from src.db.session import AsyncSessionDep
from src.features.auth.service import auth_service
from src.features.users.schemas import LoginUser
router = APIRouter(prefix="/auth", tags=["Auth V1"])
@router.post("/login", response_model=StandardResponse[TokensType])
async def login_user(user_data: LoginUser, session: AsyncSessionDep):
    tokens = await auth_service.login_user(user_data=user_data, session=session)
    return {"message": "Login successfull", "data": tokens}

		

Description:

AsyncSessionDep
- We have discussed in DB layer article
- async session from DB pool injected using DI
LoginUser
- DTO to validate the user data sent from the frontend
StandardResponse
- base model/type for our response as defined in Service layer article
TokensType
- this type has to keys, accessToken and refreshToken

Auth Service:

src/features/auth/service.pyPython

			
class AuthService:
    async def login_user(self, user_data: LoginUser, session: AsyncSession):
        user = await user_service.get_user_by_email(
            email=user_data.email, session=session
        )
        is_pwd_valid = Security_utils.verifY_pwd(
            plain_pwd=user_data.password, hashed_pwd=user.password
        )
        if not is_pwd_valid:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials"
            )
        tokens = await Security_utils.gen_tokens(user_data=user)
        return tokens
auth_service = AuthService()

		

Description:

get user
- used “user_service” to get user by email using session
verify user password, using “security utils”
if password is not valid, raise http exception which will be caught by custom exception handler that we created in service layer article
if password is valid, then generate tokens

Security Utils:

src/core/utils/security_utils.pyPython

			
import logging
import uuid
from datetime import datetime, timedelta, timezone
from typing import NamedTuple, TypedDict
import jwt
from fastapi import HTTPException, status
from pwdlib import PasswordHash
from pwdlib.hashers.argon2 import Argon2Hasher
from pwdlib.hashers.bcrypt import BcryptHasher
from src.core.configs.env_config import env_config
from src.core.schemas import UserDataType
from src.db.auth_db import store_refresh_jti
logger = logging.getLogger("security_logger")
class TokensType(TypedDict):
    access_token: str
    refresh_token: str
class RefTokenDecodedValues(NamedTuple):
    user_id: str
    jti: str
class SecurityUtils:
    pwd_hash = PasswordHash(
        (
            Argon2Hasher(),
            BcryptHasher(),
        )
    )
    ACCESS_SECRET_KEY = env_config.JWT_ACCESS_TOKEN_SECRET
    REFRESH_SECRET_KEY = env_config.JWT_REFRESH_TOKEN_SECRET
    ALGORITHM = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES = 15
    REFRESH_TOKEN_EXPIRE_DAYS = 7
    JTI_NAMESPACE = uuid.UUID("15b29e2e-4d50-48f4-b404-e6b763379a12")
    @classmethod
    def get_hashed_pwd(cls, pwd: str) -> str:
        return cls.pwd_hash.hash(pwd)
    @classmethod
    def verifY_pwd(cls, hashed_pwd: str, plain_pwd: str) -> bool:
        return cls.pwd_hash.verify(password=plain_pwd, hash=hashed_pwd)
    @classmethod
    def gen_access_token(cls, user_data: UserDataType) -> str:
        expire = datetime.now(timezone.utc) + timedelta(
            minutes=cls.ACCESS_TOKEN_EXPIRE_MINUTES
        )
        to_encode = {
            "sub": str(user_data.id),
            "exp": int(expire.timestamp()),
            # will add role later
        }
        encoded_jwt = jwt.encode(
            to_encode, cls.ACCESS_SECRET_KEY, algorithm=cls.ALGORITHM
        )
        return encoded_jwt
    @classmethod
    def gen_jti(cls, user_data: UserDataType):
        # Get current time for 'iat' (Issued At)
        # now = datetime.now(timezone.utc)
        # iat_timestamp = int(now.timestamp())
        # jti_seed_name = f"user_{user_data.id}_at_{iat_timestamp}"
        jti_seed_name = f"user_{user_data.id}"
        jti = str(uuid.uuid5(namespace=cls.JTI_NAMESPACE, name=jti_seed_name))
        return jti
    @classmethod
    async def gen_refresh_token(cls, user_data: UserDataType) -> str:
        jti = cls.gen_jti(user_data=user_data)
        expire = datetime.now(timezone.utc) + timedelta(
            days=cls.REFRESH_TOKEN_EXPIRE_DAYS
        )
        to_encode = {
            "sub": str(user_data.id),
            "jti": jti,
            "exp": int(expire.timestamp()),
        }
        # await del_refresh_jti(jti)
        await store_refresh_jti(jti)
        encoded_jwt = jwt.encode(
            to_encode, cls.REFRESH_SECRET_KEY, algorithm=cls.ALGORITHM
        )
        return encoded_jwt
    @classmethod
    async def gen_tokens(cls, user_data: UserDataType) -> TokensType:
        access_token = cls.gen_access_token(user_data=user_data)
        refresh_token = await cls.gen_refresh_token(user_data=user_data)
        return {"access_token": access_token, "refresh_token": refresh_token}
    @classmethod
    def decode_access_token(cls, token: str) -> str | None:
        try:
            payload = jwt.decode(
                token,
                cls.ACCESS_SECRET_KEY,
                algorithms=[cls.ALGORITHM],
                options={"require": ["exp"]},
            )
            user_id = payload.get("sub")
            return user_id
        except jwt.ExpiredSignatureError:
            logger.error("Access Token Expired")
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Your access token has expired",
                # headers={"WWW-Authenticate": "Bearer"},
            )
        except jwt.InvalidTokenError:
            logger.error("Invalid Access Token")
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Could not validate credentials",
            )
    @classmethod
    def decode_refresh_token(cls, token: str) -> RefTokenDecodedValues:
        try:
            payload = jwt.decode(
                token,
                cls.REFRESH_SECRET_KEY,
                algorithms=[cls.ALGORITHM],
                options={"require": ["exp", "jti"]},
            )
            user_id = payload.get("sub")
            jti = payload.get("jti")
            if not (user_id and jti):
                raise jwt.InvalidTokenError("Does not have required data")
            return RefTokenDecodedValues(user_id=user_id, jti=jti)
        except jwt.ExpiredSignatureError:
            logger.error("Refresh Token Expired")
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Your refresh token has expired",
                # headers={"WWW-Authenticate": "Bearer"},
            )
        except jwt.InvalidTokenError:
            logger.error("Invalid Refresh Token")
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid refresh token",
            )
Security_utils = SecurityUtils()

		

Description:

Everything is quite evident and dont need any explanation, but few points need to be mentioned
Password hasher has 2 functions, argon and bcrypt, it will prioritize, argon and then fallback to bcrypt. This will be also the case while verifying password hash as well.
JTI_NAMESPACE is hardocded as of now, but we can also pull it from .env(env_config)
access and refresh token content has one difference, i.e., jti, which is only present in refresh token for tracking session
while decoding both tokens, we make sure to check for “exp” and “jti” as per the tokens so that we can validate token properly.

Auth DB Handler:

src/db/auth_db.pyPython

			
import logging
import valkey.asyncio as aiovalkey
from fastapi import HTTPException, status
from src.core.configs.env_config import env_config
auth_db_host = env_config.VALKEY_HOST
auth_db_port = env_config.VALKEY_PORT
valkey_client = aiovalkey.Valkey(
    host=auth_db_host,
    port=int(auth_db_port),
    decode_responses=True,
)
logger = logging.getLogger("auth_db_logger")
async def ping_auth_db():
    try:
        await valkey_client.ping()
        logger.info("Auth DB ping successfull")
    except Exception as e:
        logger.warning("failed to ping Auth DB")
        logger.warning(e)
        raise
async def close_auth_db():
    await valkey_client.close()
    logger.info("Auth DB closed")
async def store_refresh_jti(jti: str):
    refresh_token_ttl_seconds = 7 * 24 * 60 * 60  # 604,800 seconds
    key = f"refresh_jti:{jti}"
    try:
        await valkey_client.setex(key, refresh_token_ttl_seconds, "active")
    except Exception as e:
        logger.warning(f"Failed to store the refresh token data: {e}")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Authentication service temporarily unavailable.",
        )
async def get_refresh_jti_status(jti: str):
    key = f"refresh_jti:{jti}"
    try:
        token_status = await valkey_client.get(key)
        # if token_status is None:
        #     raise HTTPException(
        #         status_code=status.HTTP_403_FORBIDDEN, detail="You are unauthorized"
        #     )
        return token_status
    except Exception as e:
        logger.warning(f"Failed to get the refresh token data: {e}")
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN, detail="You are unauthorized"
        )
async def del_refresh_jti(jti: str):
    key = f"refresh_jti:{jti}"
    try:
        res = await valkey_client.delete(key)
        return res  # 1= succcess, 0 = no key present
    except Exception as e:
        logger.warning(f"Failed to delete the refresh token data: {e}")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Authentication service temporarily unavailable.",
        )
async def get_and_del_refresh_jti(jti: str):
    key = f"refresh_jti:{jti}"
    try:
        res = await valkey_client.getdel(key)
        if res is None:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN, detail="You are unauthorized"
            )
        if res != "active":
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN, detail="You are unauthorized"
            )
    except Exception as e:
        logger.warning(f"Failed to get the refresh token data: {e}")
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN, detail="You are unauthorized"
        )

		

Description:

created valkey client using aiovalkey
all the code is quite normal
DB ping and close functions for attaching it to the app’s lifespan
storing key in DB with TTL so that it expires and gets removed once time expires. As we are tracking session through refresh token, so TTL is 7 days
the one need attention is “get_and_del_refresh_jti()”
- this will be used in logging out user
- while storing the token, we use “jti” as the key and “active” as the string value
- when certain user need to be banned, we can change this key’s value to be “inactive” or “banned”, so that user cannot login again.
- In primary DB also, I will be changing the user status making it a double write
- But, this will prevent the both DB access for reading requests which will speed up access

Auth Guard:

src/core/guards/auth_guard.pyPython

			
import logging
from typing import Annotated
from fastapi import Depends, HTTPException, Request, status
from src.core.utils.security_utils import Security_utils
logger = logging.getLogger("auth_guard_logger")
def need_auth(req: Request):
    token = req.headers.get("X-Auth-Token")
    if not token:
        logger.error("Custom header unavailable")
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN, detail="Access Denied"
        )
    user_id = Security_utils.decode_access_token(token)
    if user_id is None:
        logger.error("Access token decoding failed")
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN, detail="Access Denied"
        )
    return user_id
AuthGuard = Depends(need_auth)
AuthGuardDep = Annotated[str, AuthGuard]

		

Description:

this is quite simple
I am using a custom header rather than the Bearer Token, coz I am experimenting it
And, the decode access token function to validate the access token

Refresh Token Guard:

src/core/guards/refresh_token_guard.pyPython

			
import logging
from typing import Annotated
from fastapi import Depends, HTTPException, Request, status
from src.core.utils.security_utils import Security_utils
logger = logging.getLogger("refresh_token_guard_logger")
def validate_refresh_token(req: Request):
    token = req.headers.get("X-Auth-Token")
    if not token:
        logger.error("Custom header unavailable")
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN, detail="Access Denied"
        )
    user_id, jti = Security_utils.decode_refresh_token(token)
    # if user_id is None:
    #     logger.error("Refresh token decoding failed")
    #     raise HTTPException(
    #         status_code=status.HTTP_403_FORBIDDEN, detail="Access Denied"
    #     )
    return user_id, jti
RefreshTokenGuard = Depends(validate_refresh_token)
RefreshTokenGuardDep = Annotated[str, RefreshTokenGuard]

		

Description:

Here, we could have also used “user_id” to check if the user is valid or not or has been banned as well to check from our primary DB. But, as we are tracking the user in auth DB, so we have removed the primary DB check from here. This makes the load to primary DB as we are not reaching to the primary DB for every request. We are avoiding the usage of available connections from the pool so that it can be used for actual data fetching rather than just checking the user’s status, making other requests to use the connection pool responsibly.
Token should be the primary attribute to check the user’s authenticity and authorization
The jti returned by this decoding will be used to update auth DB data

This finishes our authentication. I will be implementing RBAC after this in the next article.

FastAPI Backend: Part 4 (Authentication)

what I want:

Login:

Share this:

Leave a comment Cancel reply