JWT-Autorisierung FastAPI: Schritt-für-Schritt-Code

JWT-Authentifizierung in FastAPI implementieren: Ein umfassender Entwicklerleitfaden

JWT (JSON Web Token) ist ein kompaktes Token zur Übertragung von Ansprüchen zwischen Parteien. Das Token besteht aus drei Base64-kodierten Teilen: Header, Payload und Signatur, getrennt durch Punkte.

Header enthält den Signaturalgorithmus (z.B. HS256) und den Tokentyp (JWT):

{"alg": "HS256", "typ": "JWT"}

Payload speichert Ansprüche: sub (Benutzerkennung), iat (Ausstellungszeit), exp (Ablaufzeit):

{"sub": "alex", "name": "Alex Tomchok", "iat": 1516239022, "exp": 1516242622}

Signatur wird als HMAC-SHA256 von (Header.Payload + Geheimschlüssel) berechnet. Der Server überprüft die Signatur bei der Validierung: Jede Änderung am Payload macht das Token ungültig. Daten sind lesbar, aber vor Manipulation geschützt.

Zugriffstokens haben eine kurze Lebensdauer (15–30 Minuten), während Aktualisierungstokens langlebig sind (Tage/Monate). Dieses Beispiel verwendet nur Zugriffstokens.

Projekteinrichtung und Abhängigkeiten

Erstellen Sie die Struktur:

fastapi-jwt-auth/
├── app/
│   ├── main.py
│   ├── core/
│   │   ├── config.py
│   │   └── security.py
│   ├── schemas/
│   │   ├── token.py
│   │   └── user.py
│   ├── api/
│   │   └── v1/
│   │       ├── endpoints/
│   │       │   ├── auth.py
│   │       │   └── users.py
│   │       └── dependencies.py
│   └── services/
│       └── user_service.py
└── requirements.txt

requirements.txt:

fastapi==0.104.1
uvicorn[standard]==0.24.0
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
python-multipart==0.0.6
pydantic[email]==2.5.0

Installieren: pip install -r requirements.txt.

Konfiguration und Pydantic-Schemata

In core/config.py definieren Sie Einstellungen mit Pydantic:

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    SECRET_KEY: str = "your-super-secret-key-change-this-in-production"
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30

    class Config:
        env_file = ".env"

settings = Settings()

Schemata in schemas/user.py:

from pydantic import BaseModel, EmailStr

class UserBase(BaseModel):
    username: str
    email: EmailStr

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool = True

    class Config:
        from_attributes = True

class UserInDB(User):
    hashed_password: str

schemas/token.py:

from pydantic import BaseModel

class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"

class TokenData(BaseModel):
    username: str | None = None

Sicherheit: Hashing und JWT-Operationen

In core/security.py implementieren Sie Passwort-Hashing (bcrypt) und JWT-Funktionen:

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) -> str:
    to_encode = data.copy()
    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 decode_access_token(token: str) -> dict:
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        return payload
    except JWTError:
        return None

Benutzerdienst und Fake-Datenbank

services/user_service.py mit In-Memory-Speicher:

from app.schemas.user import UserInDB
from app.core.security import verify_password, get_password_hash

fake_users_db = {
    "alex": {
        "id": 1,
        "username": "alex",
        "email": "[email protected]",
        "hashed_password": get_password_hash("secret123"),
        "is_active": True,
    }
}

def get_user_by_username(username: str) -> UserInDB | None:
    if username in fake_users_db:
        return UserInDB(**fake_users_db[username])
    return None

def authenticate_user(username: str, password: str) -> UserInDB | None:
    user = get_user_by_username(username)
    if not user:
        return None
    if not verify_password(password, user.hashed_password):
        return None
    return user

Abhängigkeiten für Autorisierung

In api/v1/dependencies.py verwenden Sie OAuth2PasswordBearer:

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from app.core.security import decode_access_token
from app.services.user_service import get_user_by_username
from app.schemas.token import TokenData
from app.schemas.user import User

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Anmeldedaten konnten nicht validiert werden",
        headers={"WWW-Authenticate": "Bearer"},
    )
    payload = decode_access_token(token)
    if payload is None:
        raise credentials_exception
    username: str = payload.get("sub")
    if username is None:
        raise credentials_exception
    user = get_user_by_username(username)
    if user is None:
        raise credentials_exception
    return user

async def get_current_active_user(current_user: User = Depends(get_current_user)) -> User:
    if not current_user.is_active:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Inaktiver Benutzer")
    return current_user

Endpunkte: Login und geschützte Routen

api/v1/endpoints/auth.py:

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from app.schemas.token import Token
from app.services.user_service import authenticate_user
from app.core.security import create_access_token

router = APIRouter(prefix="/auth", tags=["authentication"])

@router.post("/login", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Falscher Benutzername oder Passwort",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(data={"sub": user.username})
    return Token(access_token=access_token, token_type="bearer")

api/v1/endpoints/users.py:

from fastapi import APIRouter, Depends
from app.schemas.user import User
from app.api.v1.dependencies import get_current_active_user

router = APIRouter(prefix="/users", tags=["users"])

@router.get("/me", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user

Verbinden Sie Router in main.py und starten Sie uvicorn app.main:app.

Wichtige Punkte

• JWT-Payload ist nicht verschlüsselt, nur signiert – verwenden Sie nur nicht sensible Daten.
• Speichern Sie SECRET_KEY in Umgebungsvariablen, generieren Sie ihn mit 32+ Zeichen.
• In der Produktion fügen Sie Aktualisierungstokens und eine Datenbank hinzu (SQLAlchemy + Alembic).
• FastAPI-Abhängigkeiten ermöglichen die Wiederverwendung von Autorisierungslogik ohne Duplizierung.
• Testen Sie Endpunkte über Swagger UI: /docs.

— Editorial Team