patrickandClaude Sonnet 5
48b6d6a5be
CI / backend-tests (push) Failing after 1s
- Vollständiges Ziel-Schema aus Prompt 20 als initiale Migration (inkl. Karte-13-Vorbereitung: UUID-PKs, systemknoten) - Seed-Migration für Hauptserver-Datensatz (Sprintplan E6) - JWT-Login + /auth/me, require_roles-Dependency (Prompt 05 Berechtigungsmatrix) - Tests: health, login/me, Rollen-Ablehnung/-Zulassung (S0-Abnahmekriterien) - Gitea-Actions-CI: install -> migrate -> pytest mit Coverage-Gate 50% Nur Code/Config erzeugt, nicht lokal installiert oder ausgeführt (Deployment-Regel). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01L85hmKbvX7Cqkq47KnQhFt
53 lines
1.7 KiB
Python
53 lines
1.7 KiB
Python
from collections.abc import Callable
|
|
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import OAuth2PasswordBearer
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.security import InvalidTokenError, decode_access_token
|
|
from app.db.session import get_db
|
|
from app.models.auth import Benutzer, RolleTyp
|
|
|
|
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")
|
|
|
|
|
|
async def get_current_user(
|
|
token: str = Depends(oauth2_scheme),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> Benutzer:
|
|
credentials_error = HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Ungültiger oder abgelaufener Token",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
try:
|
|
payload = decode_access_token(token)
|
|
except InvalidTokenError as exc:
|
|
raise credentials_error from exc
|
|
|
|
login = payload.get("sub")
|
|
if login is None:
|
|
raise credentials_error
|
|
|
|
result = await db.execute(select(Benutzer).where(Benutzer.login == login))
|
|
benutzer = result.scalar_one_or_none()
|
|
if benutzer is None or not benutzer.aktiv:
|
|
raise credentials_error
|
|
return benutzer
|
|
|
|
|
|
def require_roles(*erlaubte_rollen: RolleTyp) -> Callable:
|
|
"""Prompt 05 Berechtigungsmatrix: zentrale Rollenprüfung als FastAPI-Dependency."""
|
|
|
|
async def checker(current_user: Benutzer = Depends(get_current_user)) -> Benutzer:
|
|
besitzt = {RolleTyp(r) for r in current_user.rollen_namen}
|
|
if not besitzt.intersection(erlaubte_rollen):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Für diese Aktion fehlt die erforderliche Rolle",
|
|
)
|
|
return current_user
|
|
|
|
return checker
|