Credential Stuffing in Fastapi with Jwt Tokens

Credential stuffing is an automated attack where previously breached username and password pairs are systematically attempted against a login endpoint to exploit reused credentials. In FastAPI applications that use JWT tokens for authentication, the presence of a predictable or weakly protected token issuance flow can turn a successful credential stuffing attempt into a high-impact account takeover.

JWT tokens are typically issued after a user provides valid credentials. If FastAPI does not enforce strong rate limiting on the token issuance endpoint, an attacker can use credential stuffing to obtain valid JWTs for a subset of user accounts. Even when JWTs are cryptographically signed, a stolen or guessed JWT grants access according to the token’s embedded claims, which often include scopes and expiration but not adaptive risk checks.

Additionally, if FastAPI services accept JWTs via an Authorization: Bearer header and do not adequately validate token provenance, session fixation or token replay becomes feasible when credentials are paired with a valid JWT from another source. Common misconfigurations include missing token binding to the originating client context, lack of one-time use constraints, and permissive CORS that allows JWTs to be reused across origins.

Because this scanning methodology tests the unauthenticated attack surface, middleBrick examines whether the token issuance path is exposed to brute-volume attempts and whether JWT validation is strict enough to prevent acceptance of tokens issued to other users. Findings in this area highlight risks where attackers could chain credential stuffing with valid JWTs to impersonate users without triggering account lockout mechanisms.

Remediation focuses on strengthening the token issuance and validation pipeline in FastAPI. Implement robust rate limiting on authentication endpoints, tie token verification to client context, and enforce strict token validation practices.

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
import jwt
from datetime import datetime, timedelta
from typing import Optional
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
import time

app = FastAPI()

# Simple in-memory rate limiting for authentication endpoints
login_attempts = {}

def rate_limit(username: str, window: int = 300, max_attempts: int = 5) -> bool:
    now = time.time()
    attempts = login_attempts.get(username, [])
    recent = [t for t in attempts if now - t < window]
    if len(recent) >= max_attempts:
        return False
    recent.append(now)
    login_attempts[username] = recent
    return True

SECRET_KEY = "your-strong-secret-key-change-this"
ALGORITHM = "HS256"

class TokenData(BaseModel):
    username: Optional[str] = None

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=30))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def verify_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
        return username
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    if not rate_limit(form_data.username):
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail="Too many login attempts. Try again later."
        )
    # Replace with actual user verification and password hashing
    if form_data.username == "valid_user" and form_data.password == "correct_password":
        access_token = create_access_token(data={"sub": form_data.username})
        return {"access_token": access_token, "token_type": "bearer"}
    raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect credentials")

@app.get("/users/me")
async def read_users_me(token: str = Depends(oauth2_scheme)):
    username = verify_token(token)
    return {"username": username}

Key remediation practices shown:

• Rate limiting on /token to mitigate credential stuffing by capping attempts per username within a sliding window.
• Strict JWT validation with expiration and signature checks, rejecting malformed or expired tokens.
• Avoiding token reuse by not embedding sensitive operations in the token payload without additional context checks.

For production, use a robust secret management approach, prefer asymmetric algorithms (RS256) with public/private keys, bind tokens to client fingerprints where appropriate, and integrate continuous security scanning using tools like middleBrick to detect misconfigurations in authentication flows.