Files
MABEA/backend/app/api/v1/endpoints/auth.py
T
patrickandClaude Sonnet 5 446e4f912d
CI / backend-tests (push) Failing after 1s
Deployment-Setup (PostgreSQL+nginx-Installer) + Sprint-0-Fixes aus Subagenten-Reviews
Installer (deploy/):
- install_server.sh: PostgreSQL+nginx auf Debian 13, idempotente Rolle/DB-Anlage,
  pgcrypto-Extension, Speicher-Tuning für 4GB-VPS, Zugangsdaten in chmod-600-Datei
  statt stdout (postgres-expert/owasp-Review)
- nginx-Template mit Security-Headern + Rate-Limit auf /auth/login
- systemd-Unit-Template mit Sandboxing (NoNewPrivileges/ProtectSystem/PrivateTmp)

Backend-Fixes (fastapi-expert-Review):
- get_db: einheitliche commit/rollback-Konvention statt Endpunkt-Copy-Paste
- Test-Fixtures auf SQLAlchemy-2.0-Savepoint-Pattern umgestellt (join_transaction_mode),
  da get_db jetzt selbst committet
- Lifespan-Handler: Startup-Guard gegen JWT-Secret-Platzhalter, engine.dispose() beim Shutdown
- JWT-Payload ohne ungenutztes roles-Claim (Rollen kommen immer frisch aus der DB)

Zusätzlich: Subagenten-Definitionen (~/.claude/agents/) auf lauffähiges Modell fixiert.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01L85hmKbvX7Cqkq47KnQhFt
2026-09-03 23:00:46 +02:00

56 lines
1.6 KiB
Python

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.deps import get_current_user
from app.core.security import create_access_token, verify_password
from app.db.session import get_db
from app.models.auth import Benutzer
router = APIRouter()
class TokenResponse(BaseModel):
access_token: str
token_type: str = "bearer"
class MeResponse(BaseModel):
id: int
name: str
login: str
rollen: list[str]
@router.post("/auth/login", response_model=TokenResponse)
async def login(
form_data: OAuth2PasswordRequestForm = Depends(),
db: AsyncSession = Depends(get_db),
) -> TokenResponse:
result = await db.execute(select(Benutzer).where(Benutzer.login == form_data.username))
benutzer = result.scalar_one_or_none()
if (
benutzer is None
or not benutzer.aktiv
or not verify_password(form_data.password, benutzer.passwort_hash)
):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Login oder Passwort falsch",
headers={"WWW-Authenticate": "Bearer"},
)
token = create_access_token(subject=benutzer.login)
return TokenResponse(access_token=token)
@router.get("/auth/me", response_model=MeResponse)
async def me(current_user: Benutzer = Depends(get_current_user)) -> MeResponse:
return MeResponse(
id=current_user.id,
name=current_user.name,
login=current_user.login,
rollen=current_user.rollen_namen,
)