Implementace JWT ověření v FastAPI: kompletní průvodce s kódem
JWT (JSON Web Token) zajišťuje bezstavové ověření v API. Token se skládá z hlavičky, datové části a podpisu, které jsou zakódované v base64 a oddělené tečkami.
Hlavička určuje algoritmus podepsání:
{
"alg": "HS256",
"typ": "JWT"
}
Datová část obsahuje tvrzení: user_id (sub), čas vydání (iat), platnost (exp).
{
"sub": "user_id_123",
"exp": 1735689600,
"iat": 1735603200
}
Podpis ověřuje integritu prostřednictvím HMAC-SHA256 s tajným klíčem.
Pro produkční prostředí použijte párové tokeny:
- Access (15–30 minut) – pro požadavky na API
- Refresh (7–30 dní) – pro aktualizaci access tokenu
Architektura projektu
Rozdělení do vrstev usnadňuje testování a škálování:
project/
├── app/
│ ├── core/ # config.py, security.py, dependencies.py
│ ├── models/ # user.py (SQLAlchemy)
│ ├── schemas/ # user.py, token.py (Pydantic)
│ ├── routers/ # auth.py, users.py
│ ├── services/ # user.py
│ └── main.py
├── requirements.txt
└── .env
Router zpracovává HTTP požadavky, schémata validují data, modely popisují databázi, služby obsahují obchodní logiku.
Konfigurace a bezpečnostní nástroje
V core/config.py:
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
SECRET_KEY: str = "your-secret-key-change-in-production"
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
REFRESH_TOKEN_EXPIRE_DAYS: int = 7
class Config:
env_file = ".env"
settings = Settings()
Klíčové funkce v core/security.py:
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from .config import settings
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
return pwd_context.hash(password)
def create_access_token(data: dict, expires_delta: timedelta = None) -> str:
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
return encoded_jwt
def create_refresh_token(data: dict) -> str:
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
return encoded_jwt
def decode_token(token: str) -> dict:
try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
return payload
except JWTError:
return None
Schémata Pydantic pro validaci
schemas/user.py:
from pydantic import BaseModel, EmailStr
class UserCreate(BaseModel):
email: EmailStr
password: str
full_name: str
class UserResponse(BaseModel):
id: int
email: EmailStr
full_name: str
class Config:
from_attributes = True
schemas/token.py:
from pydantic import BaseModel
class Token(BaseModel):
access_token: str
refresh_token: str
token_type: str = "bearer"
class TokenData(BaseModel):
user_id: int
Závislost pro aktuálního uživatele
core/dependencies.py implementuje OAuth2PasswordBearer:
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login")
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: Session = Depends(get_db)
) -> User:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
payload = decode_token(token)
if payload is None:
raise credentials_exception
user_id = payload.get("sub")
if user_id is None:
raise credentials_exception
token_data = TokenData(user_id=int(user_id))
user = db.query(User).filter(User.id == token_data.user_id).first()
if user is None:
raise credentials_exception
return user
Endpoints pro ověření
V routers/auth.py:
- POST /api/auth/register – vytvoření uživatele s bcrypt hashováním
- POST /api/auth/login – vrácení access a refresh tokenu
- POST /api/auth/refresh – aktualizace access tokenu
Plný kód pro přihlášení:
@router.post("/login", response_model=Token)
def login(
form_data: OAuth2PasswordRequestForm = Depends(),
db: Session = Depends(get_db)
):
user = db.query(User).filter(User.email == form_data.username).first()
if not user or not verify_password(form_data.password, user.hashed_password):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect email or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token = create_access_token(data={"sub": str(user.id)})
refresh_token = create_refresh_token(data={"sub": str(user.id)})
return {
"access_token": access_token,
"refresh_token": refresh_token,
"token_type": "bearer"
}
Ochranné zdroje
routers/users.py:
@router.get("/me", response_model=UserResponse)
def get_current_user_info(current_user: User = Depends(get_current_user)):
return current_user
Endpoint vyžaduje Bearer token v hlavičce Authorization.
Hlavní aplikace
main.py:
from fastapi import FastAPI
from .routers import auth, users
app = FastAPI(title="FastAPI JWT Auth", version="1.0.0")
app.include_router(auth.router)
app.include_router(users.router)
@app.get("/")
def root():
return {"message": "Welcome to FastAPI JWT Auth API"}
Co je důležité
- Používejte HS256 s silným SECRET_KEY (minimálně 32 znaků)
- Ukládejte refresh tokeny v HttpOnly cookiech pro ochranu před XSS
- Implementujte rotaci refresh tokenů při každém obnovení
- Sledujte neúspěšné pokusy o ověření
- Testujte vypršené tokeny (exp claim)
— Editorial Team
Zatím žádné komentáře.