Authentication and JWT — A Secure API Identity System
A JWT is like a hotel room key—the Access Token is a daily pass (valid for a short period), and the Refresh Token is a monthly pass (valid for a long time but can be revoked at any time); the front desk (authentication service) is responsible for issuing and verifying the keys.
1. What You'll Learn
- JWT Principles and Structure: An Explanation of Header, Payload, and Signature
python-joseIssuing and Verifying Access Tokens and Refresh Tokens (Dual Tokens)OAuth2PasswordBearerIntegration with FastAPI Security Tools- Password Hashing:
passlib+bcryptBest Practices - Alice Scenario: SaaS Multi-Tenant Authentication for PriceTracker—Permissions Matrix for Different Subscription Plans
2. Alice's True Story
(1) Pain Point: Unauthenticated APIs Being Maliciously Crawled
Alice's PriceTracker API is completely open, and anyone can call all of its endpoints. A competitor wrote a script that scrapes a million price data entries every minute. To make matters worse, someone used the API to submit fake price data, corrupting the entire database. Alice needs to implement user registration and login, API authentication, and access control based on different subscription tiers, but she doesn't know how to do so securely.
(2) Solutions for JWT Authentication
JWT (JSON Web Token) is a stateless authentication solution: Users obtain a token upon login, include the token in subsequent requests, and the server verifies the signature—there is no need to store sessions, and it inherently supports distributed systems.
PYTHON
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
@app.get("/me")
async def get_me(token: str = Depends(oauth2_scheme)):
user = verify_token(token)
return user
(3) Revenue
Once API authentication is enabled, unregistered users will be unable to access protected endpoints, and malicious scraping will be blocked by both rate-limiting middleware and authentication. Pro users can import data in bulk, while Free users are limited to 1,000 records; competitors can only view public data.
3. JWT Principles and Structure
(1) The Three-Part Structure of a JWT
A JWT consists of three parts: Header (algorithm), Payload (data), and Signature, separated by ..
sequenceDiagram
participant Client as Bob Frontend
participant Auth as /auth/login
participant API as Protected API
participant Verify as Token Verifier
Client->>Auth: POST email + password
Auth->>Auth: Verify credentials
Auth-->>Client: Access Token + Refresh Token
Client->>API: GET /products (Bearer Token)
API->>Verify: Decode & verify signature
Verify-->>API: User payload
API-->>Client: 200 OK + data
Note over Client,Auth: Access Token expires after 30 min
Client->>Auth: POST /auth/refresh (Refresh Token)
Auth-->>Client: New Access Token
| Section | Content | Example |
|---|---|---|
| Header | Algorithm Type | {"alg": "HS256", "typ": "JWT"} |
| Payload | User Data (Claims) | {"sub": "alice", "role": "admin", "exp": 1700000000} |
| Signature | Signature | HMACSHA256(header.payload, secret) |
(1) ▶ Example: JWT Encoding and Decoding
PYTHON
from jose import jwt, JWTError
from datetime import datetime, timedelta
SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
def create_access_token(data: dict, expires_delta: timedelta | None = 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) -> dict:
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return payload
except JWTError:
raise ValueError("Invalid token")
# Test
token = create_access_token({"sub": "alice", "role": "admin"})
print(f"Token: {token[:50]}...")
payload = verify_token(token)
print(f"Payload: {payload}")
Output:
TEXT
# Function defined successfully
(2) Access Token vs. Refresh Token
| Dimension | Access Token | Refresh Token |
|---|---|---|
| Validity Period | 30 minutes | 7 days |
| Purpose | Access API Resources | Refresh Access Token |
| Storage | Memory (Front End) | HttpOnly Cookie |
| Exposure Risk | High (included in every request) | Low (used only during refreshes) |
| Revocation Method | Wait for Expiration | Server Blacklist |
4. OAuth2PasswordBearer Integration
(1) ▶ Example: Complete Authentication Configuration
PYTHON
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt, JWTError
from datetime import datetime, timedelta
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE = timedelta(minutes=30)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
app = FastAPI()
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=401,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
# In production: query user from database
user = {"username": username, "role": payload.get("role", "user")}
return user
@app.post("/auth/login")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
# In production: verify password from database
if form_data.username != "alice" or form_data.password != "secret":
raise HTTPException(status_code=401, detail="Incorrect credentials")
access_token = create_access_token(
data={"sub": form_data.username, "role": "admin"},
expires_delta=ACCESS_TOKEN_EXPIRE,
)
return {"access_token": access_token, "token_type": "bearer"}
@app.get("/me")
async def read_me(current_user: dict = Depends(get_current_user)):
return current_user
Output:
TEXT
# Function defined successfully
5. Password Hashing
(1) passlib + bcrypt
Passwords are never stored in plain text; they are salted and hashed using the bcrypt algorithm.
(1) ▶ Example: Password Hashing and Authentication
PYTHON
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def hash_password(password: str) -> str:
return pwd_context.hash(password)
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
# Test
hashed = hash_password("my-secret-password")
print(f"Hashed: {hashed[:30]}...")
print(f"Verify correct: {verify_password('my-secret-password', hashed)}")
print(f"Verify wrong: {verify_password('wrong-password', hashed)}")
Output:
TEXT
Hashed: $2b$12$K3YQ8z9wB5eF7gH1j...
Verify correct: True
Verify wrong: False
(2) Best Practices for Password Security
| Practice | Description |
|---|---|
| Using bcrypt | Adaptive salted hashing to prevent rainbow tables |
| Does not use MD5/SHA256 | No salt; calculations are too fast, making it vulnerable to brute-force attacks |
| Password length ≥ 8 | Enforce minimum length |
| Hash before verification | Prevent timing attacks (built into passlib) |
| SECRET_KEY is long enough | A random string of at least 32 characters; store it in an environment variable |
6. SaaS Multi-Tenant Permissions Matrix
(1) Subscription-Level Permission Design
(1) ▶ Example: Subscription-Based Permission Dependencies
PYTHON
from fastapi import Depends, HTTPException
SUBSCRIPTION_LIMITS = {
"free": {"max_import": 1000, "websocket": False, "export": False},
"pro": {"max_import": 100000, "websocket": True, "export": True},
"enterprise": {"max_import": 1000000, "websocket": True, "export": True},
}
def require_subscription(min_level: str):
LEVELS = {"free": 0, "pro": 1, "enterprise": 2}
async def check_subscription(user: dict = Depends(get_current_user)):
user_level = user.get("subscription", "free")
if LEVELS.get(user_level, 0) < LEVELS.get(min_level, 0):
raise HTTPException(
status_code=403,
detail=f"Requires {min_level} subscription. Current: {user_level}",
)
return user
return check_subscription
@app.get("/api/v1/analytics")
async def get_analytics(user=Depends(require_subscription("pro"))):
return {"total_products": 1000000, "active_users": 5000}
@app.post("/api/v1/prices/bulk")
async def bulk_import(
prices: list[PriceCreate],
user=Depends(require_subscription("free")),
):
limits = SUBSCRIPTION_LIMITS[user.get("subscription", "free")]
if len(prices) > limits["max_import"]:
raise HTTPException(
status_code=403,
detail=f"Import limit: {limits['max_import']} for {user['subscription']} plan",
)
return {"imported": len(prices)}
Output:
TEXT
# Function defined successfully
| Endpoint | Free | Pro | Enterprise |
|---|---|---|---|
| GET /products | 1,000/day | Unlimited | Unlimited |
| POST /prices | 1,000/batch | 100,000/batch | 1,000,000/batch |
| WebSocket /ws/prices | - | ✓ | ✓ |
| GET /analytics | - | ✓ | ✓ |
| CSV Export | - | ✓ | ✓ |
| API Rate Limiting | 100/minute | 1,000/minute | Unlimited |
❓ FAQ
Q What is the difference between JWT and Session?
A JWT is stateless (not stored on the server) and is suitable for distributed systems and microservices; Session is stateful (stored on the server) and is suitable for single-server applications. JWT is recommended for API services.
Q What should I do if the token expires?
A When the Access Token expires, use the Refresh Token to obtain a new Access Token. If the Refresh Token expires, you'll need to log in again.
Q How should SECRET_KEY be managed?
A In a production environment, use environment variables or a key management service (such as AWS KMS or HashiCorp Vault); never hard-code it into the code. It must be at least 32 characters long.
Q Which is better, bcrypt or Argon2?
A Argon2 is the winner of the cryptographic hash competition and offers stronger resistance to GPU and ASIC attacks. However, bcrypt is time-tested and has a mature ecosystem; both are far superior to MD5 and SHA.
Q What is the difference between OAuth2PasswordBearer and HTTPBearer?
A OAuth2PasswordBearer implements the OAuth2 password grant and automatically displays a login form in Swagger UI; HTTPBearer is a generic Bearer token extractor. We recommend the former.
Q Can a JWT be revoked immediately?
A Since JWTs are stateless by nature, they cannot be revoked immediately. Solution: Set a short expiration time + use a Redis blacklist (to store the IDs of revoked tokens until they expire).
📖 Summary
- A JWT consists of three parts: Header, Payload, and Signature. Stateless authentication inherently supports distributed systems.
- The Access Token (short-term) is used for API access, while the Refresh Token (long-term) is used for refreshing; this two-token system enhances security.
OAuth2PasswordBearerIntegrates FastAPI security tools; Swagger UI automatically displays the login form- Passwords are stored as bcrypt hashes; passlib handles the salting and verification logic.
- PriceTracker Three-Tier Subscription Permissions Matrix: Free/Pro/Enterprise, implemented via
require_subscriptionDI
📝 Exercises
- Basic Exercise (Difficulty ⭐): Implement the
/auth/loginendpoint to receive a username and password; after authentication, return a JWT Access Token (valid for 30 minutes). UseDepends(oauth2_scheme)to parse the user at the/meendpoint. Hint:OAuth2PasswordRequestForm+jwt.encode - Advanced Exercise (Difficulty ⭐⭐): Add a refresh token mechanism. The access token is valid for 15 minutes, and the refresh token is valid for 7 days. Implement the
/auth/refreshendpoint to exchange the refresh token for a new access token. Hint: Twocreate_tokenfunctions + different expiration times - Challenge (Difficulty ⭐⭐⭐): Implement a complete registration/login/permissions system—store passwords for the registration endpoint using bcrypt hashing; use the
require_subscription(min_level)DI chain to check subscription levels; limit Free users to importing 1,000 records, while Pro users have no limit. Hint:hash_password+verify_password+require_subscriptionDI
---|
