Skip to Content

JWT Login Route: OAuth2PasswordRequestForm + Scopes

Data Flow

JWT LOGIN (with scopes)  —  Data Flow

   POST /auth/jwt/login (form: username, password)
        │
        ▼  SELECT user WHERE email = username   ── unique constraint
        │
        ▼  verify_password(form.password, user.hashed_password)    bcrypt
        │
        ▼  issue_access_token(
        │      sub = str(user.id),
        │      email = user.email,
        │      role  = user.role,
        │      scopes = user.scopes.split(" "),
        │      iat = now,
        │      exp = now + access_token_expire_minutes,
        │      algorithm = settings.jwt_algorithm,           HS256 default
        │      secret    = settings.secret_key,
        │  )
        │
        ▼
   { access_token, scopes, token_type="bearer" }

What? (Concept Overview)

The /auth/login endpoint is the canonical OAuth2 password-credentials flow: it accepts an OAuth2PasswordRequestForm (which is application/x-www-form-urlencoded), looks up the customer by username (mapped to email), verifies the bcrypt password hash, and issues a JWT with sub, id, role, and scopes claims. The endpoint is exposed at the root level (/auth/login), NOT under /api/v1, so it works with any auth-aware client tooling.

Project Context

In the fca project context -> app/api/routes/auth.py, the login route uses OAuth2PasswordRequestForm’s username field (OAuth2 form convention) and maps it to the email column. The JWT carries sub (business ID), id (PK), role (RBAC), scopes (permissions) so downstream get_current_active_user can validate and route. The route returns a typed Token Pydantic model. The endpoint deliberately returns “Incorrect username or password” for both wrong-email and wrong-password so attackers can’t enumerate users.

How? (Quick Reference Blocks)

3.1 The Login Route

# app/api/routes/auth.py
from datetime import timedelta
from typing import Any
 
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel
 
from app.database import get_db
from app.config import settings
from app.services.security_service import SecurityService
from app.models.customer import Customer
 
router = APIRouter(prefix="/auth", tags=["Authentication"])
security_service = SecurityService()
 
class Token(BaseModel):
    access_token: str
    token_type: str
 
@router.post("/login", response_model=Token)
async def login_for_access_token(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: AsyncSession = Depends(get_db),
) -> Any:
    """OAuth2 password login. username field maps to email."""
    query = select(Customer).where(Customer.email == form_data.username)
    result = await db.execute(query)
    user = result.scalar_one_or_none()
 
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
 
    if (not user.hashed_password
            or not security_service.verify_password(
                form_data.password, user.hashed_password)):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
 
    access_token_expires = timedelta(minutes=settings.access_token_expire_minutes)
 
    access_token = security_service.create_access_token(
        data={
            "sub": str(user.customer_id),     # business ID
            "id": user.id,                    # primary key
            "role": user.role,                # RBAC role
            "scopes": user.scopes,            # permissions
        },
        expires_delta=access_token_expires,
    )
 
    return {"access_token": access_token, "token_type": "bearer"}

3.2 Token Wire Format

# example payload (decoded) — NOT a secret
{
    "sub": "CUST-000001",         # business ID (string per JWT spec)
    "id": 1,                      # primary key
    "role": "user",
    "scopes": "read:accounts",
    "exp": 1735000000,            # expiry as Unix seconds
    "iat": 1734997000             # issued-at
}

3.3 What’s in SecurityService

# app/services/security_service.py — JWT ops
from jose import jwt
from datetime import datetime, timedelta
from typing import Optional, Dict, Any
 
class SecurityService:
    def create_access_token(
        self, data: dict, expires_delta: Optional[timedelta] = None,
    ) -> str:
        to_encode = data.copy()
        expire = datetime.utcnow() + (
            expires_delta or timedelta(minutes=self.expire_minutes)
        )
        to_encode.update({"exp": expire})
        return jwt.encode(to_encode, self.secret_key, algorithm=self.algorithm)
 
    def decode_token(self, token: str) -> Optional[Dict[str, Any]]:
        try:
            return jwt.decode(
                token, self.secret_key, algorithms=[self.algorithm]
            )
        except Exception:    # JWTError
            return None

Why? (Parameter Breakdown

Common Pitfalls

  1. Returning different error messages for “user not found” vs “wrong password”. Tells attackers which emails exist. Always use a single generic 401 message.
  2. Embedding sensitive data in the JWT — JWTs are NOT encrypted by default (they’re signed). Don’t put PII (email, phone), account numbers, or anything an exfiltrated JWT would leak. The current schema (sub, id, role, scopes) is appropriately minimal.
  3. Not setting WWW-Authenticate header — OAuth2-aware clients can’t tell what scheme to retry with. Standard header for HTTP-Bearer auth.

Real-World Interview Prep

Q1: How would you add rate-limiting to /auth/login?

A: Two layers. (1) App layer — slowapi’s Limiter.limit("5/minute") keyed by client IP. (2) Database — track failed attempts per email; lock account after 5 failures within 15 minutes; notify user via email/SMS. Use both: app-layer stops botnets, DB-layer catches targeted account attacks. The Pyproject config (rate_limit_calls=10, rate_limit_period=60) suggests a global rate limit; add a tighter per-IP cap on /auth/login specifically.

Q2: How do you revoke a JWT issued to a former employee without rotating the secret?

A: Use a token-blacklist (Redis SET of revoked jti claims) checked inside get_current_user. Pattern: at logout, push the JWT’s jti into revoked_tokens with TTL = remaining token lifetime. get_current_user does if payload["jti"] in redis.smembers("revoked_tokens"): raise 401. Alternative: short access tokens (e.g., 5 min) with refresh tokens that DO check the blacklist; rotation frequency reduces blast radius. The current architecture uses 3000-minute single-token — too long for revocation-only.

Q3: Why split id (DB PK) and sub (business ID) in the JWT?

A: Two roles, different lifecycles. Customer.id is the DB primary key (FK target for joins, fast lookups) — but DB IDs can change during migrations (table rebuilds, schema switches). sub is the business-facing identifier (CUST-000001) — stable across system changes; the customer_id column is the public contract. Storing both means: (a) downstream code uses id for performance-critical DB joins, (b) external APIs and the JWT subject use sub for stability. A token with only id would lock you out during a migration-reroll.

Top-to-Bottom Code Walkthrough (app/api/routes/auth.py)

The login endpoint is the single source of truth for “who is the caller and what can they do”. It pairs OAuth2PasswordRequestForm (a FastAPI built-in) with python-jose JWTs and a scope-aware payload.

OAuth2PasswordRequestForm = Depends()

Tells FastAPI: parse the request as if it were a HTML form (application/x-www-form-urlencoded), not JSON. The form must contain:

The route body, line by line

  1. Look up the user: user = await db.execute(select(Customer).where(Customer.email == form.username)). form.username is the email address (matches our User model since email is unique).
  2. Constant-time verify: if not security_service.verify_password(form.password, user.hashed_password): raise HTTPException(401). Why constant time (handled inside verify_password): timing attacks reveal valid usernames by measuring how long the hash compare takes.
  3. Build the JWT payload: 
    payload = {
  "sub": str(user.id),         # standard "subject" claim
  "id": user.id,
  "email": user.email,
  "role": user.role,           # "user" | "admin"
  "scopes": user.scopes.split(" "),  # e.g. ["read:accounts", "write:messages"]
  "type": "access",
  "iat": datetime.utcnow(),    # issued-at
  "exp": datetime.utcnow() + timedelta(minutes=settings.access_token_expire_minutes),
}
  4. Sign the token: token = jwt.encode(payload, settings.secret_key, algorithm=settings.jwt_algorithm). With jwt_algorithm = HS256, the same secret_key signs and verifies. The secret must be at least 32 chars (enforced by Settings).
  5. Return: {”access_token”: token, ”token_type”: “bearer”}. Never include the password or the hashed password in the response.

Security(scope="...") usage

Once the token contains scopes, downstream endpoints declare their needs:

@router.post("/admin/seed-db", dependencies=[Security(get_current_active_user, scopes=["admin"])])
async def seed_db(...):

The Security(...) factory raises 403 automatically if the JWT lacks admin.

OpenAPI docs integration

OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login") shows a “Authorize” button on /docs that lets developers paste username/password and call protected endpoints directly. Without it, every test requires manual Authorization: Bearer ... headers.

Common Pitfalls

Returning a non-encoded token ({”token”: str(token)} after jwt.encode returns a str in v3.x, bytes in older versions). Always test the type — wrap with jwt.encode(..., algorithm=...) and confirm with headers["Content-Type"] == "application/json".

Trusting form.username as a real username when the field could be an email; match your DB schema.

Long-lived access tokens (e.g. 7 days) without refresh — gives attackers a 7-day window. Default to 5-60 minutes and pair with /auth/refresh.

Real-World Interview Prep

Q1: What’s the difference between HS256 and RS256 JWT algorithms?

A: HS256 (HMAC): symmetric — same secret_key signs and verifies. Faster, simpler, but only the server can verify. RS256 (RSA): asymmetric — the server has a private key to sign, anyone has the public key to verify. Choose RS256 when you have multiple services that need to verify the token without holding the secret (microservices).

Q2: Why include a type claim in the JWT?

A: To prevent confusion with refresh tokens. An attacker stealing a refresh token shouldn’t be able to use it as an access token (and vice versa). Adding type: "access" (or "refresh") locks each token’s purpose.

Q3: How do you revoke a leaked JWT before its expiry?

A: JWTs are stateless — you cannot revoke them by definition. Practical mitigations: short expiry, server-side block-list of revoked jti (JWT id) claims, or refresh-token rotation with a server-side revocation list.

🛠️ Production Failure Playbook

🚨 Failure 1 — Token expires after 50 hours — long replay window

🚨 Failure 2 — bcrypt rounds unchanged since 2018 — GPU cracking

🚨 Failure 3 — JWT secret leaked via exception trace

Advertisement