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ösung | Anwendbare Szenarien | Beschreibung |
|---|---|---|
| HTTP Basic Auth | Einfacher interner Dienst | Benutzername und Passwort werden im Anfrage-Header kodiert, was eine geringere Sicherheit bedeutet |
| API-Schlüssel | Dienstübergreifender Aufruf | Den Schlüssel über Anfrage-Header, Query-Parameter oder Cookies übergeben |
| OAuth2 + JWT | Front-End/Back-End-getrennte Anwendung | Der 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:
- Der Client sendet den Benutzernamen und das Passwort an
/tokenEndpunkt - Der Server überprüft die Anmeldedaten und gibt ein JWT-Zugriffstoken zurück
- Der Client übermittelt das Token in nachfolgenden Anfragen (
Authorization: Bearer <token>) - 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 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:
| Funktion | Beschreibung |
|---|---|
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:
| Aktion | Funktion | Beschreibung |
|---|---|---|
| Token erstellen | jwt.encode() | Die Daten in einen JWT-String kodieren |
| Token parsen | jwt.decode() | JWT-String parsen und verifizieren |
| Ablaufzeit festlegen | "exp": expire | Token-Ablaufzeit |
| Benutzerkennung speichern | "sub": username | Das 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:
- Klicken Sie auf"Authorize"
- Geben Sie den Benutzernamen und das Passwort ein (z. B.
alice/secret) - Klicken Sie auf"Authorize"Token abrufen
- 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