Decode a JSON Web Token(JWT)

🔁 Quick Recap

🧾 Authentication Flow

1. A user logs in with a username and password.

2. The server validates the credentials.

3. A JWT is returned to the client.

4. The client includes that JWT in the Authorization header for future requests.

5. Every protected endpoint verifies the JWT before processing the request.

Without decoding and validating the JWT, we cannot securely authorize users.

📥 Step 1: Import OAuth2PasswordBearer

Now we need to import :

🗂️ routes/auth.py

from fastapi.security import OAuth2PasswordRequestForm, OAuth2PasswordBearer
from jose import jwt, JWTError

🔎 Why OAuth2PasswordBearer?

This dependency:

• Extracts the Bearer token from the request header

• Automatically reads the Authorization: Bearer <token> header

• Makes the token available for verification

🔗 Step 2: Create the OAuth2 Bearer Dependency

Right below your bcrypt_context, define:

🗂️ routes/auth.py

oauth2_bearer = OAuth2PasswordBearer(tokenUrl="token")

📌 What tokenUrl="token" Means

This tells FastAPI:

• The client must first obtain a token from the /token endpoint.

• The bearer token will be validated against that authentication flow.

This dependency will now be used by any protected endpoint.

🛡 Step 3: Create the Token Verification Function

We now create a function that:

• Decodes the JWT

• Verifies its signature

• Validates the payload

• Raises an exception if invalid

This function is not an API endpoint.

It is a reusable dependency that other endpoints will rely on.

🧠 Implement get_current_user

🗂️ routes/auth.py

from fastapi import HTTPException, status
from typing import Annotated

async def get_current_user(
    token: Annotated[str, Depends(oauth2_bearer)]):
    try:
        # Decode the token
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])

        # Extract values from payload
        username: str = payload.get("sub")
        user_id: int = payload.get("id")

        # Validate payload data
        if username is None or user_id is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Could not validate user"
            )

        return {"username": username, "id": user_id}

    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate user"
        )

What Annotated[str, Depends(oauth2_bearer)] means

It tells FastAPI to inject the JWT token from the request into the token parameter.

Breakdown

1️⃣ str

• The type of the value

• The token will be a string

2️⃣ Depends(oauth2_bearer)

• Declares a dependency

• FastAPI will call oauth2_bearer

• That dependency extracts the token from the Authorization header

Example request header:

Authorization: Bearer eyJhbGciOiJIUzI1NiIs...

FastAPI extracts the token and passes it to token.

3️⃣ Annotated

• A modern Python typing feature

• Allows combining type information (str) + dependency (Depends)

Result

token will contain the JWT token string sent in the request header.

One-line note

Annotated[str, Depends(oauth2_bearer)] tells FastAPI to inject the Bearer token from the Authorization header as a string into the function parameter.

✅ Step 1: Extract Token from Header

FastAPI automatically reads:

Authorization: Bearer <JWT>

Thanks to OAuth2PasswordBearer.

✅ Step 2: Decode the Token

payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])

This verifies:

• The signature matches the secret key

• The algorithm is correct

• The token has not expired

• The token has not been modified

If someone attempts to forge a token without knowing your secret key, verification will fail.

✅ Step 3: Extract the Payload

We retrieve:

username = payload.get("sub")
user_id = payload.get("id")

If either value is missing → the token is invalid.

✅ Step 4: Handle Errors

If:

• The token is malformed

• The signature is invalid

• The token is expired

• The payload is incomplete

We raise:

HTTPException(status_code=401)

This ensures unauthorized users cannot access protected routes.

🔐 Why This Function Is Critical

This function becomes the security gatekeeper of your application.

Any future endpoint that requires authentication will use:

Depends(get_current_user)

This ensures:

• Only valid users can access protected resources

• Invalid or expired tokens are rejected

• Security is centralized and reusable

from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException
from datetime import datetime, timedelta, timezone
from pydantic import BaseModel
from sqlalchemy.orm import Session
from starlette import status
from database import SessionLocal
from models import Users
from passlib.context import CryptContext
from fastapi.security import OAuth2PasswordRequestForm, OAuth2PasswordBearer
from jose import jwt, JWTError

router = APIRouter()

SECRET_KEY = "735d62ded4a5f73eec7de13a6fa3dd309a575d8feaccb644cf34e4841a88317a"
ALGORITHM = "HS256"

# Password hashing setup
bcrypt_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_bearer = OAuth2PasswordBearer(tokenUrl="token")

# Pydantic schema for user creation
class CreateUserRequest(BaseModel):
    username: str
    email: str
    first_name: str
    last_name: str
    password: str
    role: str

class Token(BaseModel):
    access_token: str
    token_type: str

# Database dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

db_dependency = Annotated[Session, Depends(get_db)]

def authenticate_user(username: str, password: str, db: Session):
    user = db.query(Users).filter(Users.username == username).first()

    if not user:
        return False

    if not bcrypt_context.verify(password, user.hashed_password):
        return False

    return user

def create_access_token(username: str, user_id: int, expires_delta: timedelta):
    encode = {
        "sub": username,
        "id": user_id
    }

    expires = datetime.now(timezone.utc) + expires_delta
    encode.update({"exp": expires})

    return jwt.encode(encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(
    token: Annotated[str, Depends(oauth2_bearer)]):
    try:
        # Decode the token
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])

        # Extract values from payload
        username: str = payload.get("sub")
        user_id: int = payload.get("id")

        # Validate payload data
        if username is None or user_id is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Could not validate user"
            )

        return {"username": username, "id": user_id}

    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Could not validate user"
        )

# Create user endpoint
@router.post("/auth/", status_code=status.HTTP_201_CREATED)
async def create_user(
    create_user_request: CreateUserRequest,
    db: db_dependency
		):

    # Create a new user model
    create_user_model = Users(
        email=create_user_request.email,
        username=create_user_request.username,
        first_name=create_user_request.first_name,
        last_name=create_user_request.last_name,
        role=create_user_request.role,
        hashed_password=bcrypt_context.hash(create_user_request.password),
        is_active=True
    )

    # Save to database
    db.add(create_user_model)
    db.commit()

@router.post("/token", response_model=Token)
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: db_dependency):
    user = authenticate_user(form_data.username, form_data.password, db)

    if not user:
        return "Failed Authentication"

    token = create_access_token(
        user.username,
        user.id,
        timedelta(minutes=20)
    )

    return {
        "access_token": token,
        "token_type": "bearer"
    }