Erstellen Sie mit FastAPI einen perfekten Blog: Fügen Sie ein Benutzersystem hinzu

Im vorherigen Artikel haben wir FastAPI verwendet, um einen einfachen persönlichen Blog zu erstellen und ihn erfolgreich bereitzustellen.

Dieser Blog hat jedoch ein ernstes Sicherheitsproblem: Jeder kann nach Belieben Artikel erstellen.

Im folgenden Tutorial fügen wir diesem Blog ein Benutzer- und Authentifizierungssystem hinzu, um ihn sicherer zu machen.

Beginnen wir ohne weitere Umschweife.

Einführung in Authentifizierungsmethoden

In der Webentwicklung sind die beiden gängigsten Authentifizierungsmethoden tokenbasierte Authentifizierung (z. B. JWT) und sitzungsbasierte Authentifizierung (Cookie).

JWT (JSON Web Tokens): Dies ist derzeit die beliebteste Authentifizierungsmethode. Nachdem sich ein Benutzer angemeldet hat, generiert der Server ein Token und gibt es an den Client zurück. Der Client fügt dieses Token in nachfolgende Anfragen ein, und der Server muss nur das Token validieren, um die Identität des Benutzers zu bestätigen. Da der Server keinen Benutzerstatus speichern muss, ist diese Methode sehr gut für verteilte, horizontal skalierbare Großanwendungen geeignet.
Sitzung-Cookie: Nachdem sich ein Benutzer angemeldet hat, erstellt der Server eine Sitzung und gibt die Sitzungs-ID über ein Cookie an den Browser zurück. Der Browser fügt dieses Cookie automatisch in nachfolgende Anfragen ein. Der Server identifiziert dann den Benutzer, indem er die entsprechenden Sitzungsinformationen anhand der Sitzungs-ID nachschlägt.

In diesem Tutorial werden wir die traditionelle Sitzung-Cookie-Methode wählen. Da unser Blog ein Monolith mit einer einfachen Architektur ist, ist die Verwendung von Sitzung-Cookies für die Authentifizierung der direkteste, klassischste und ausreichend sichere Ansatz.

Schritt 1: Erstellen des Benutzer Moduls

Bevor wir die Authentifizierung handhaben können, benötigen wir zuerst ein Benutzersystem.

1. Erstellen des Benutzerdatenmodells

Öffnen Sie die Datei models.py und fügen Sie das User-Modell oberhalb der Post-Klasse hinzu.

# models.py
import uuid
from datetime import datetime
from typing import Optional
from sqlmodel import Field, SQLModel

class User(SQLModel, table=True):
    id: Optional[uuid.UUID] = Field(default_factory=uuid.uuid4, primary_key=True)
    username: str = Field(unique=True, index=True)
    password: str # Das gespeicherte Passwort ist der verschlüsselte Hash

class Post(SQLModel, table=True):
    id: Optional[uuid.UUID] = Field(default_factory=uuid.uuid4, primary_key=True)
    title: str
    content: str
    createdAt: datetime = Field(default_factory=datetime.utcnow, nullable=False)

Da wir die Funktion create_db_and_tables in main.py im ersten Artikel konfiguriert haben, erkennt sie automatisch alle SQLModel-Modelle und erstellt die entsprechenden Datenbanktabellen beim Start der Anwendung. Daher müssen wir keine SQL-Anweisungen manuell ausführen.

Wenn Sie SQL manuell ausführen müssen und Ihre Datenbank auf Leapcell erstellt wurde,

können Sie SQL-Anweisungen einfach über die grafische Benutzeroberfläche ausführen. Gehen Sie einfach auf die Seite zur Datenbankverwaltung auf der Website, fügen Sie die obigen Anweisungen in die SQL-Oberfläche ein und führen Sie sie aus.

2. Installieren der Passwortverschlüsselungsbibliothek

Aus Sicherheitsgründen dürfen Benutzerpasswörter niemals im Klartext in der Datenbank gespeichert werden. Wir verwenden die Bibliothek bcrypt, um die Passwörter zu hashen.

Fügen Sie zuerst bcrypt zu Ihrer requirements.txt-Datei hinzu:

# requirements.txt
fastapi
uvicorn[standard]
sqlmodel
psycopg2-binary
jinja2
python-dotenv
python-multipart
bcrypt

Führen Sie dann den Installationsbefehl aus:

pip install -r requirements.txt

Schritt 2: Implementieren der Registrierungs- und Validierungslogik für Benutzer

Als Nächstes erstellen wir Funktionen zur Verwaltung von Benutzerdaten und zur Validierung von Passwörtern.

Erstellen Sie eine neue Datei users_service.py im Stammverzeichnis des Projekts, um benutzerbezogene Geschäftslogik zu speichern.

# users_service.py
import bcrypt
from sqlmodel import Session, select
from models import User

def get_user_by_username(username: str, session: Session) -> User | None:
    """Find a user by username"""
    statement = select(User).where(User.username == username)
    return session.exec(statement).first()

def create_user(user_data: dict, session: Session) -> User:
    """Create a new user and hash the password"""
    # Konvertieren Sie das Klartextpasswort in Bytes
    password_bytes = user_data["password"].encode('utf-8')
    # Generieren Sie ein Salt und hashen Sie das Passwort
    salt = bcrypt.gensalt()
    hashed_password = bcrypt.hashpw(password_bytes, salt)
    
    new_user = User(
        username=user_data["username"],
        # Dekodieren Sie das verschlüsselte Passwort (Bytes) in einen String für die Datenbank.
        password=hashed_password.decode('utf-8')
    )
    
    session.add(new_user)
    session.commit()
    session.refresh(new_user)
    return new_user

Erstellen Sie dann eine Datei auth_service.py zur Verwaltung der Benutzerauthentifizierung.

# auth_service.py
import bcrypt
from sqlmodel import Session
from models import User
from users_service import get_user_by_username

def validate_user(username: str, plain_password: str, session: Session) -> User | None:
    """Validate if the username and password match"""
    user = get_user_by_username(username, session)
    if not user:
        return None
        
    # Kodieren Sie sowohl das eingegebene Klartextpasswort als auch das gespeicherte gehashte Passwort in Bytes
    plain_password_bytes = plain_password.encode('utf-8')
    hashed_password_bytes = user.password.encode('utf-8')
    
    # Verwenden Sie bcrypt.checkpw für den Vergleich
    if bcrypt.checkpw(plain_password_bytes, hashed_password_bytes):
        return user # Validierung erfolgreich, Benutzerinformationen zurückgeben
        
    return None # Validierung fehlgeschlagen

Schritt 3: Erstellen von Login- und Registrierungsseiten

Wir müssen eine Schnittstelle für Benutzer zur Registrierung und Anmeldung bereitstellen. Erstellen Sie im Ordner templates die Dateien login.html und register.html.

register.html

{% include "_header.html" %}
<form action="/users/register" method="POST" class="post-form">
  <h2>Register</h2>
  <div class="form-group">
    <label for="username">Username</label>
    <input type="text" id="username" name="username" required />
  </div>
  <div class="form-group">
    <label for="password">Password</label>
    <input type="password" id="password" name="password" required />
  </div>
  <button type="submit">Register</button>
</form>
<p style="text-align: center; margin-top: 1rem;">
  Already have an account? <a href="/auth/login">Login here</a>.
</p>
{% include "_footer.html" %}

login.html

{% include "_header.html" %}
<form action="/auth/login" method="POST" class="post-form">
  <h2>Login</h2>
  <div class="form-group">
    <label for="username">Username</label>
    <input type="text" id="username" name="username" required />
  </div>
  <div class="form-group">
    <label for="password">Password</label>
    <input type="password" id="password" name="password" required />
  </div>
  <button type="submit">Login</button>
</form>
<p style="text-align: center; margin-top: 1rem;">
  Don't have an account? <a href="/users/register">Register here</a>.
</p>
{% include "_footer.html" %}

Aktualisieren wir gleichzeitig _header.html, um Links für Registrierung und Anmeldung oben rechts hinzuzufügen.

<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8" />
  <meta name="viewport" content="width=device-width, initial-scale=1.0" />
  <title>{{ title }}</title>
  <link rel="stylesheet" href="/static/css/style.css" />
</head>
<body>
  <header>
    <h1><a href="/">My Blog</a></h1>
    <nav>
      <a href="/posts/new" class="new-post-btn">New Post</a>
      <a href="/users/register" class="nav-link">Register</a>
      <a href="/auth/login" class="nav-link">Login</a>
    </nav>
  </header>
  <main>

Schritt 4: Implementieren von Routing und Controller-Logik

Nun werden wir die Routing-Logik in verschiedene Dateien aufteilen, um die Projektstruktur zu verdeutlichen.

Erstellen Sie einen Ordner routers im Stammverzeichnis des Projekts.

Schneiden Sie alle Post-bezogenen Routen (@app.get("/posts", ...) etc.) aus main.py aus und fügen Sie sie in eine Datei routers/posts.py ein.

# routers/posts.py
import uuid
from fastapi import APIRouter, Request, Depends, Form
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from sqlmodel import Session, select

from database import get_session
from models import Post

router = APIRouter()
templates = Jinja2Templates(directory="templates")

@router.get("/", response_class=HTMLResponse)
def root():
    return RedirectResponse(url="/posts", status_code=302)

@router.get("/posts", response_class=HTMLResponse)
def get_all_posts(request: Request, session: Session = Depends(get_session)):
    statement = select(Post).order_by(Post.createdAt.desc())
    posts = session.exec(statement).all()
    return templates.TemplateResponse("index.html", {"request": request, "posts": posts, "title": "Home"})

@router.get("/posts/new", response_class=HTMLResponse)
def new_post_form(request: Request):
    return templates.TemplateResponse("new-post.html", {"request": request, "title": "New Post"})

@router.post("/posts", response_class=HTMLResponse)
def create_post(
    title: str = Form(...), 
    content: str = Form(...),
    session: Session = Depends(get_session)
):
    new_post = Post(title=title, content=content)
    session.add(new_post)
    session.commit()
    return RedirectResponse(url="/posts", status_code=302)

@router.get("/posts/{post_id}", response_class=HTMLResponse)
def get_post_by_id(request: Request, post_id: uuid.UUID, session: Session = Depends(get_session)):
    post = session.get(Post, post_id)
    return templates.TemplateResponse("post.html", {"request": request, "post": post, "title": post.title})

Hinweis: Wir ersetzen den @app-Decorator durch @router.

Erstellen Sie im Ordner routers eine Datei users.py zur Behandlung der Benutzerregistrierung.

# routers/users.py
from fastapi import APIRouter, Request, Depends, Form
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from sqlmodel import Session

from database import get_session
import users_service

router = APIRouter()
templates = Jinja2Templates(directory="templates")

@router.get("/users/register", response_class=HTMLResponse)
def show_register_form(request: Request):
    return templates.TemplateResponse("register.html", {"request": request, "title": "Register"})

@router.post("/users/register")
def register_user(
    username: str = Form(...),
    password: str = Form(...),
    session: Session = Depends(get_session)
):
    # Der Einfachheit halber hier keine komplexe Validierung
    users_service.create_user({"username": username, "password": password}, session)
    return RedirectResponse(url="/auth/login", status_code=302)

Erstellen Sie im Ordner routers eine Datei auth.py zur Behandlung der Benutzeranmeldung.

# routers/auth.py
from fastapi import APIRouter, Request, Depends, Form, HTTPException
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from sqlmodel import Session

from database import get_session
import auth_service

router = APIRouter()
templates = Jinja2Templates(directory="templates")

@router.get("/auth/login", response_class=HTMLResponse)
def show_login_form(request: Request):
    return templates.TemplateResponse("login.html", {"request": request, "title": "Login"})

@router.post("/auth/login")
def login(
    username: str = Form(...),
    password: str = Form(...),
    session: Session = Depends(get_session)
):
    user = auth_service.validate_user(username, password, session)
    if not user:
        raise HTTPException(status_code=401, detail="Incorrect username or password")
    
    # Validierung erfolgreich
    return RedirectResponse(url="/posts", status_code=302)

Aktualisieren Sie schließlich main.py, um die alten Routen zu entfernen und die neuen Router-Dateien einzubinden.

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

from database import create_db_and_tables
from routers import posts, users, auth

@asynccontextmanager
async def lifespan(app: FastAPI):
    print("Creating tables..")
    create_db_and_tables()
    yield

app = FastAPI(lifespan=lifespan)

# Mounten Sie das statische Dateien-Verzeichnis
app.mount("/static", StaticFiles(directory="public"), name="static")

# Router einbinden
app.include_router(posts.router)
app.include_router(users.router)
app.include_router(auth.router)

Schritt 5: Testen

An dieser Stelle haben wir eine grundlegende Logik für Benutzerregistrierung und Anmeldeprüfung abgeschlossen.

Starten Sie Ihr Projekt neu:

uvicorn main:app --reload

Besuchen Sie http://localhost:3000/users/register, um sich zu registrieren.

Nach erfolgreicher Registrierung werden Sie automatisch zu http://localhost:3000/auth/login weitergeleitet, um sich anzumelden.

Sie können die Ergebnisse der Eingabe korrekter und falscher Anmeldeinformationen testen. Wenn Sie beispielsweise falsche Informationen eingeben, wird auf der Seite ein 401 Unauthorized-Fehler angezeigt.

Der aktuelle Login ist jedoch nur ein einmaliger Validierungsprozess; der Server "erinnert" sich nicht an den Anmeldestatus des Benutzers. Nach dem Schließen des Browsers oder dem Besuch anderer Seiten befinden Sie sich immer noch in einem nicht authentifizierten Zustand.

Im nächsten Artikel führen wir die Sitzungsverwaltung ein, um eine echte Anmeldestatuspersistenz für Benutzer zu erreichen und den Zugriff auf Seiten und Operationen basierend auf den Benutzerberechtigungen einzuschränken.