from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError
from jose.exceptions import ExpiredSignatureError
SECRET_KEY = "change-me-in-production"
ALGORITHM = "HS256"
ACCESS_TTL = timedelta(minutes=15)
REFRESH_TTL = timedelta(days=7)
class TokenError(Exception):
pass
class TokenService:
def _encode(self, subject: str, token_type: str, ttl: timedelta) -> str:
now = datetime.now(timezone.utc)
payload = {
"sub": subject,
"type": token_type,
"iat": now,
"exp": now + ttl,
}
return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
def create_access(self, subject: str) -> str:
return self._encode(subject, "access", ACCESS_TTL)
def create_refresh(self, subject: str) -> str:
return self._encode(subject, "refresh", REFRESH_TTL)
def _decode(self, token: str, expected_type: str) -> dict:
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
except ExpiredSignatureError:
raise TokenError("token expired")
except JWTError:
raise TokenError("invalid token")
if payload.get("type") != expected_type:
raise TokenError("wrong token type")
return payload
def decode_access(self, token: str) -> dict:
return self._decode(token, "access")
def decode_refresh(self, token: str) -> dict:
return self._decode(token, "refresh")
token_service = TokenService()
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from .security import token_service, TokenError
from .users import User, get_user_by_id
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")
CREDENTIALS_EXC = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
try:
payload = token_service.decode_access(token)
except TokenError:
raise CREDENTIALS_EXC
user_id = payload.get("sub")
if user_id is None:
raise CREDENTIALS_EXC
user = get_user_by_id(user_id)
if user is None or not user.is_active:
raise CREDENTIALS_EXC
return user
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from pydantic import BaseModel
from .security import token_service, TokenError
from .deps import get_current_user
from .users import User, authenticate
router = APIRouter(prefix="/auth", tags=["auth"])
class TokenPair(BaseModel):
access_token: str
refresh_token: str
token_type: str = "bearer"
class RefreshRequest(BaseModel):
refresh_token: str
@router.post("/login", response_model=TokenPair)
def login(form: OAuth2PasswordRequestForm = Depends()):
user = authenticate(form.username, form.password)
if user is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
)
return TokenPair(
access_token=token_service.create_access(user.id),
refresh_token=token_service.create_refresh(user.id),
)
@router.post("/refresh", response_model=TokenPair)
def refresh(body: RefreshRequest):
try:
payload = token_service.decode_refresh(body.refresh_token)
except TokenError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid refresh token",
)
subject = payload["sub"]
return TokenPair(
access_token=token_service.create_access(subject),
refresh_token=token_service.create_refresh(subject),
)
@router.get("/me", response_model=dict)
def me(user: User = Depends(get_current_user)):
return {"id": user.id, "email": user.email}
This snippet shows how signed JWTs are issued and verified in a FastAPI service, with token minting isolated from the request-time verification logic. The core idea is that authentication state travels inside a self-contained, signed token rather than in a server-side session, so any process holding the secret can validate a request without a shared session store. The trade-off is that stateless tokens cannot be revoked instantly; short access-token lifetimes and a separate refresh token mitigate that.
In security.py, the TokenService centralizes all cryptographic work using jose.jwt. _encode sets the standard registered claims — sub for the subject, iat for issued-at, exp for expiry, and a custom type claim used to distinguish access tokens from refresh tokens. Encoding type into the payload is what prevents a refresh token from being replayed as an access token: decode_access explicitly rejects any token whose type is not access. Verification catches ExpiredSignatureError and the generic JWTError separately so callers can surface a precise 401, and it always specifies algorithms=[ALGORITHM] to avoid the classic algorithm-confusion attack where an attacker downgrades to none.
In deps.py, verification is expressed as a FastAPI dependency. OAuth2PasswordBearer extracts the bearer token from the Authorization header and also wires up the interactive docs. get_current_user decodes the access token, pulls the sub claim, loads the corresponding user, and raises a 401 with a WWW-Authenticate header on any failure. Because it is a dependency, any route can require authentication simply by declaring user: User = Depends(get_current_user), keeping the check declarative and testable — the dependency can be overridden wholesale in tests.
In routes.py, /login verifies credentials and returns both tokens, while /refresh consumes a refresh token via decode_refresh and mints a fresh access token, so clients rotate credentials without re-sending a password. The protected /me route demonstrates the payoff: it receives an already-authenticated User with no auth boilerplate in its body. A common pitfall this design avoids is trusting client-supplied identity; the sub is only ever read back out of a token the server itself signed.
Related snips
payload = {
sub: user.id,
iss: 'https://auth.example.com',
aud: 'codesnips-api',
exp: 15.minutes.from_now.to_i,
iat: Time.now.to_i,
JWT issuance and verification without common footguns
package com.example.myapp
import android.app.Application
import dagger.hilt.android.HiltAndroidApp
@HiltAndroidApp
Dependency injection with Hilt
#!/usr/bin/env bash
set -euo pipefail
export VAULT_ADDR="https://vault.internal:8200"
export VAULT_TOKEN="${VAULT_TOKEN:?missing VAULT_TOKEN}"
Secrets management with environment isolation and Vault
<?php
namespace App\Providers;
use App\Contracts\PaymentGateway;
use App\Services\StripePaymentGateway;
Laravel service container and dependency injection
import { randomBytes, createHash } from "crypto";
import jwt from "jsonwebtoken";
import { RefreshTokenStore } from "./store";
const ACCESS_SECRET = process.env.ACCESS_SECRET!;
const ACCESS_TTL = "15m";
JWT access + refresh token rotation (conceptual)
package files
import (
"context"
"time"
Presigned S3 upload URLs (AWS SDK v2)
Share this code
Here's the card — post it anywhere.