FastAPI-Sicherheitsauthentifizierung

FastAPI verfügt über integrierte Sicherheitstools, die gängige Authentifizierungs- und Autorisierungsmethoden wie OAuth2, JWT-Token und API-Schlüssel unterstützen. In diesem Abschnitt wird erläutert, wie eine Benutzerauthentifizierung auf Basis von OAuth2 + JWT implementiert wird.


Überblick über Sicherheitsauthentifizierung

Von FastAPI unterstützte Sicherheitsschemata:

LösungAnwendbare SzenarienBeschreibung
HTTP Basic AuthEinfacher interner DienstBenutzername und Passwort werden im Anfrage-Header kodiert, was eine geringere Sicherheit bedeutet
API-SchlüsselDienstübergreifender AufrufDen Schlüssel über Anfrage-Header, Query-Parameter oder Cookies übergeben
OAuth2 + JWTFront-End/Back-End-getrennte AnwendungDer häufigste Ansatz, sicher und flexibel

OAuth2-Passwortmodus + JWT

Dies ist das am häufigsten verwendete Authentifizierungsschema für Anwendungen mit getrenntem Front-End und Back-End. Ablauf:

  1. Der Client sendet den Benutzernamen und das Passwort an/tokenEndpunkt
  2. Der Server überprüft die Anmeldedaten und gibt ein JWT-Zugriffstoken zurück
  3. Der Client übermittelt das Token in nachfolgenden Anfragen (Authorization: Bearer <token>)
  4. Der Server überprüft das Token, um den Benutzer zu identifizieren

1. Abhängigkeiten installieren

pip install "python-jose[cryptography]" passlib[bcrypt]

2. Vollständiges Beispiel

Beispiel

from datetime import datetime, timedelta
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel

# ===== Konfiguration =====
SECRET_KEY = "your-secret-key-keep-it-secret"  # Umgebungsvariablen in der Produktion verwenden
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# ===== Passwort-Hashing =====
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# ===== OAuth2-Schema =====
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# ===== Datenmodelle =====
class Token(BaseModel):
    access_token: str
    token_type: str


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


class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None


class UserInDB(User):
    hashed_password: str


# ===== Mock-Datenbank =====
fake_users_db = {
    "alice": {
        "username": "alice",
        "full_name": "Alice Wonderson",
        "email": "[email protected]",
        "hashed_password": pwd_context.hash("secret"),  # Passwort: secret
        "disabled": False,
    }
}


# ===== Hilfsfunktionen =====
def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Passwort validieren"""
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password: str) -> str:
    """Passwort-Hash generieren"""
    return pwd_context.hash(password)


def get_user(db: dict, username: str) -> UserInDB | None:
    """Benutzer aus der Datenbank abrufen"""
    if username in db:
        return UserInDB(**db[username])
    return None


def authenticate_user(db: dict, username: str, password: str):
    """Benutzer-Anmeldedaten überprüfen"""
    user = get_user(db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


def create_access_token(data: dict, expires_delta: timedelta | None = Keine):
    """JWT-Zugriffstoken erstellen"""
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)


async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    """Aktuellen Benutzer aus Token abrufen (Abhängigkeitsfunktion)"""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail=Anmeldedaten konnten nicht validiert werden,
        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

    user = get_user(fake_users_db, username)
    if user is None:
        raise credentials_exception
    return user


async def get_current_active_user(
    current_user: Annotated[User, Depends(get_current_user)],
):
    """Aktuellen aktiven Benutzer abrufen"""
    if current_user.disabled:
        raise HTTPException(status_code=400, detail=Benutzer wurde deaktiviert)
    return current_user


# ===== Routen =====
app = FastAPI()


@app.post("/token")
async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
    """Anmelden, um ein Token zu erhalten"""
    user = authenticate_user(fake_users_db, form_data.username, form_data.Passwort)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Ungültiger Benutzername oder Passwort",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}


@app.get("/users/me")
async def read_users_me(
    current_user: Annotated[User, Depends(get_current_active_user)],
):
    """Aktuelle Benutzerinformationen abrufen (Authentifizierung erforderlich)"""
    return current_user


@app.get("/users/me/items")
async def read_own_items(
    current_user: Annotated[User, Depends(get_current_active_user)],
):
    """Elemente des aktuellen Benutzers abrufen (Authentifizierung erforderlich)"""
    return [{"item_id": "Foo", "owner": current_user.username}]

Code-Erklärung

Passwort-Hash

NutzungpasslibDer bcrypt-Algorithmus hasht Passwörter, um sicherzustellen, dass Klartextpasswörter nicht in der Datenbank gespeichert werden:

FunktionBeschreibung
pwd_context.hash(password)Klartextpasswort in einen Hashwert umwandeln
pwd_context.verify(plain, hashed)Überprüfen, ob das Klartextpasswort mit dem Hashwert übereinstimmt

JWT-Token

JWT (JSON Web Token) ist ein sicheres Token-Format, das Benutzerinformationen und Ablaufzeit enthält:

AktionFunktionBeschreibung
Token erstellenjwt.encode()Die Daten in einen JWT-String kodieren
Token parsenjwt.decode()JWT-String parsen und verifizieren
Ablaufzeit festlegen"exp": expireToken-Ablaufzeit
Benutzerkennung speichern"sub": usernameDas Subjekt des Tokens (normalerweise der Benutzername)

OAuth2PasswordBearer

FastAPI mitteilen, vonAuthorization: Bearer <token>Token aus dem Request-Header abrufen:

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

tokenUrl="token"Gibt den Endpunktpfad an, über den der Client das Token bezieht; dieser wird in der API-Dokumentation angezeigt.

SECRET_KEYEr muss geheim gehalten werden und ausreichend komplex sein; in Produktionsumgebungen sollten Umgebungsvariablen zur Speicherung verwendet werden. Wenn der geheime Schlüssel preisgegeben wird, können Angreifer Token für jeden Benutzer fälschen.


Test mit der API-Dokumentation

Nach der Konfiguration von OAuth2 zeigt die Swagger UI eine Schaltfläche "Authorize" an:

  1. Klicken Sie auf"Authorize"
  2. Geben Sie den Benutzernamen und das Passwort ein (z. B.alice / secret)
  3. Klicken Sie auf"Authorize"Token abrufen
  4. Alle nachfolgenden Anfragen enthalten automatisch das Token

Zusammenfassung

  • OAuth2 + JWT ist das am häufigsten verwendete Authentifizierungsschema für Anwendungen mit getrenntem Frontend und Backend
  • Passwörter müssen zur Speicherung gehasht werden und dürfen nicht im Klartext gespeichert werden
  • JWT-Tokens enthalten Benutzerkennungen und die Ablaufzeit
  • Mithilfe von Dependency Injection (Depends) zur Implementierung der Wiederverwendung der Authentifizierungslogik
  • In Produktionsumgebungen muss der SECRET_KEY geheim gehalten werden
Weitere Erweiterungen