smueller/TK-Wiki-Newsletter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Enhance CSRF protection and user management features

Browse Source 
- Introduce CSRF token generation and management improvements, ensuring tokens are consistently retrieved from cookies or request state.
- Update user creation logic to handle validation errors more gracefully, providing user feedback for invalid input and existing email addresses.
- Revise user management UI to display success and error messages, improving user experience during user creation.
- Refactor admin user access checks to include additional role validation.
This commit is contained in:
smueller
2026-07-03 14:54:14 +02:00
parent 3f3223f17c
commit b35c524ba5
4 changed files with 94 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

89
app/main.py
View File

@@ -1,5 +1,7 @@
from pathlib import Path
from pydantic import ValidationError
from fastapi import Depends, FastAPI, Form, HTTPException, Request, Response, status
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
@@ -17,7 +19,7 @@ from app.models.user import User
from app.schemas.user import UserCreate
from app.services.auth import get_admin_user, get_current_user, get_optional_user, require_editor_or_admin
from app.services.config_store import get_config, set_config
from app.services.csrf import CSRF_COOKIE_NAME, ensure_csrf_cookie, validate_csrf
from app.services.csrf import CSRF_COOKIE_NAME, ensure_csrf_cookie, generate_csrf_token, validate_csrf
from app.services.newsletter import (
DEFAULT_DISPLAY,
article_url,
@@ -62,13 +64,22 @@ def _ensure_schema_upgrades() -> None:
@app.middleware("http")
async def add_security_headers(request: Request, call_next):
cookie_token = request.cookies.get(CSRF_COOKIE_NAME)
request.state.csrf_token = cookie_token or generate_csrf_token()
response: Response = await call_next(request)
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "DENY"
response.headers["Referrer-Policy"] = "same-origin"
response.headers["Content-Security-Policy"] = "default-src 'self'; style-src 'self'; script-src 'self';"
if not request.cookies.get(CSRF_COOKIE_NAME):
response.set_cookie(CSRF_COOKIE_NAME, ensure_csrf_cookie(request), httponly=True, secure=cookie_secure(request), samesite="strict")
if not cookie_token:
response.set_cookie(
CSRF_COOKIE_NAME,
request.state.csrf_token,
httponly=True,
secure=cookie_secure(request),
samesite="strict",
)
return response
@@ -99,7 +110,7 @@ def _base_context(request: Request, user: User, db: Session) -> dict:
active_nav = "users"
return {
"user": user,
"csrf_token": request.cookies.get(CSRF_COOKIE_NAME) or ensure_csrf_cookie(request),
"csrf_token": ensure_csrf_cookie(request),
"smtp": get_smtp_settings(db),
"active_nav": active_nav,
}
@@ -138,7 +149,7 @@ def login_page(request: Request, db: Session = Depends(get_db)):
request,
"login.html",
{
"csrf_token": request.cookies.get(CSRF_COOKIE_NAME) or ensure_csrf_cookie(request),
"csrf_token": ensure_csrf_cookie(request),
"error_message": None,
},
)
@@ -159,7 +170,7 @@ def login(
request,
"login.html",
{
"csrf_token": request.cookies.get(CSRF_COOKIE_NAME) or ensure_csrf_cookie(request),
"csrf_token": ensure_csrf_cookie(request),
"error_message": "Sitzung abgelaufen. Bitte erneut anmelden.",
},
)
@@ -169,7 +180,7 @@ def login(
request,
"login.html",
{
"csrf_token": request.cookies.get(CSRF_COOKIE_NAME) or ensure_csrf_cookie(request),
"csrf_token": ensure_csrf_cookie(request),
"error_message": "Ungültige Zugangsdaten.",
},
)
@@ -325,14 +336,41 @@ def update_smtp_settings(
return RedirectResponse(url="/dashboard", status_code=status.HTTP_302_FOUND)
@app.get("/admin/users", response_class=HTMLResponse)
def users_page(request: Request, db: Session = Depends(get_db), admin: User = Depends(get_admin_user)):
def _users_context(request: Request, admin: User, db: Session, **extra) -> dict:
context = _base_context(request, admin, db)
context["users"] = db.query(User).order_by(User.created_at.desc()).all()
return _render(request, "users.html", context)
context.setdefault("error_message", None)
context.setdefault("success_message", None)
context.update(extra)
return context
@app.post("/admin/users")
def _format_validation_error(exc: ValidationError) -> str:
messages: list[str] = []
for err in exc.errors():
msg = err.get("msg", "Ungültige Eingabe")
if msg.startswith("Value error, "):
msg = msg[13:]
messages.append(msg)
return " ".join(messages) if messages else "Ungültige Eingabe."
@app.get("/admin/users", response_class=HTMLResponse)
def users_page(request: Request, db: Session = Depends(get_db), admin: User = Depends(get_admin_user)):
created = request.query_params.get("created") == "1"
return _render(
request,
"users.html",
_users_context(
request,
admin,
db,
success_message="Benutzer wurde erfolgreich angelegt." if created else None,
),
)
@app.post("/admin/users", response_class=HTMLResponse)
def create_user(
request: Request,
csrf_token: str = Form(...),
@@ -343,11 +381,32 @@ def create_user(
db: Session = Depends(get_db),
admin: User = Depends(get_admin_user),
):
validate_csrf(request, csrf_token)
payload = UserCreate(email=email, full_name=full_name, password=password, role=role)
try:
validate_csrf(request, csrf_token)
except HTTPException:
return _render(
request,
"users.html",
_users_context(request, admin, db, error_message="Sitzung abgelaufen. Bitte Seite neu laden und erneut versuchen."),
)
try:
payload = UserCreate(email=email.strip(), full_name=full_name.strip(), password=password, role=role)
except ValidationError as exc:
return _render(
request,
"users.html",
_users_context(request, admin, db, error_message=_format_validation_error(exc)),
)
existing = db.query(User).filter(User.email == payload.email).first()
if existing:
raise HTTPException(status_code=400, detail="E-Mail existiert bereits.")
return _render(
request,
"users.html",
_users_context(request, admin, db, error_message="Diese E-Mail-Adresse ist bereits registriert."),
)
user = User(
email=payload.email,
full_name=payload.full_name,
@@ -358,4 +417,4 @@ def create_user(
)
db.add(user)
db.commit()
return RedirectResponse(url="/admin/users", status_code=status.HTTP_302_FOUND)
return RedirectResponse(url="/admin/users?created=1", status_code=status.HTTP_302_FOUND)

2
app/services/auth.py
View File

@@ -36,7 +36,7 @@ def get_current_user(request: Request, db: Session = Depends(get_db)) -> User:
def get_admin_user(current_user: User = Depends(get_current_user)) -> User:
if current_user.role != "admin":
if current_user.role != "admin" and not current_user.is_admin:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin-Rechte erforderlich.")
return current_user

16
app/services/csrf.py
View File

@@ -7,13 +7,21 @@ CSRF_COOKIE_NAME = "csrf_token"
CSRF_FORM_FIELD = "csrf_token"
def ensure_csrf_cookie(request: Request) -> str:
token = request.cookies.get(CSRF_COOKIE_NAME)
if token:
return token
def generate_csrf_token() -> str:
return secrets.token_urlsafe(32)
def ensure_csrf_cookie(request: Request) -> str:
"""Liefert das CSRF-Token für die aktuelle Anfrage (Cookie oder request.state)."""
state_token = getattr(request.state, "csrf_token", None)
if state_token:
return state_token
cookie_token = request.cookies.get(CSRF_COOKIE_NAME)
if cookie_token:
return cookie_token
return generate_csrf_token()
def validate_csrf(request: Request, form_token: str) -> None:
cookie_token = request.cookies.get(CSRF_COOKIE_NAME)
if not cookie_token or not form_token or not secrets.compare_digest(cookie_token, form_token):

8
app/templates/users.html
View File

@@ -13,6 +13,12 @@
<div class="card-header">
<h3>Neuen Benutzer anlegen</h3>
</div>
{% if error_message %}
<p class="error">{{ error_message }}</p>
{% endif %}
{% if success_message %}
<p class="hint" style="color:var(--tk-success);font-weight:600;">{{ success_message }}</p>
{% endif %}
<form method="post" action="/admin/users" class="form-grid">
<input type="hidden" name="csrf_token" value="{{ csrf_token }}">
<div class="form-row">
@@ -25,7 +31,7 @@
</div>
<div class="form-row">
<label>Passwort
<input type="password" name="password" required minlength="10">
<input type="password" name="password" required minlength="10" placeholder="Mindestens 10 Zeichen">
</label>
<label>Rolle
<select name="role">