Sprint 0: FastAPI-Skeleton, vollständiges DB-Schema (Alembic), Auth (JWT/bcrypt), Rollen-Dependency, CI

- 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

backend/app/api/deps.py

@@ -0,0 +1,52 @@
+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

backend/app/api/v1/api.py

@@ -0,0 +1,7 @@
+from fastapi import APIRouter
+
+from app.api.v1.endpoints import auth, health
+
+api_router = APIRouter()
+api_router.include_router(health.router, tags=["health"])
+api_router.include_router(auth.router, tags=["auth"])

backend/app/api/v1/endpoints/auth.py

@@ -0,0 +1,55 @@
+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, roles=benutzer.rollen_namen)
+    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,
+    )

backend/app/api/v1/endpoints/health.py

@@ -0,0 +1,19 @@
+from fastapi import APIRouter, Depends
+
+from app.api.deps import require_roles
+from app.models.auth import Benutzer, RolleTyp
+
+router = APIRouter()
+
+
+@router.get("/health")
+async def health() -> dict:
+    return {"status": "ok"}
+
+
+@router.get("/health/admin-only")
+async def health_admin_only(
+    current_user: Benutzer = Depends(require_roles(RolleTyp.administration)),
+) -> dict:
+    """Nachweis, dass die Rollen-Dependency (Prompt 05) greift – kein Fachfeature."""
+    return {"status": "ok", "login": current_user.login}

backend/app/core/app_settings.py

@@ -0,0 +1,18 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    """Konfiguration aus Umgebungsvariablen (.env auf dem Zielsystem, nie committen)."""
+
+    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
+
+    database_url: str = "postgresql+asyncpg://mabea:changeme@localhost:5432/mabea"
+    jwt_secret_key: str = "change-me-to-a-long-random-value"
+    jwt_algorithm: str = "HS256"
+    access_token_expire_minutes: int = 480
+
+    # ID des systemknoten-Datensatzes mit typ='haupt' (Sprintplan E6).
+    systemknoten_id: int = 1
+
+
+settings = Settings()

backend/app/core/security.py

@@ -0,0 +1,33 @@
+from datetime import datetime, timedelta, timezone
+
+import jwt
+from passlib.context import CryptContext
+
+from app.core.app_settings import settings
+
+_pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+
+def hash_password(password: str) -> str:
+    return _pwd_context.hash(password)
+
+
+def verify_password(plain_password: str, password_hash: str) -> bool:
+    return _pwd_context.verify(plain_password, password_hash)
+
+
+def create_access_token(*, subject: str, roles: list[str]) -> str:
+    expire = datetime.now(timezone.utc) + timedelta(minutes=settings.access_token_expire_minutes)
+    payload = {"sub": subject, "roles": roles, "exp": expire}
+    return jwt.encode(payload, settings.jwt_secret_key, algorithm=settings.jwt_algorithm)
+
+
+class InvalidTokenError(Exception):
+    pass
+
+
+def decode_access_token(token: str) -> dict:
+    try:
+        return jwt.decode(token, settings.jwt_secret_key, algorithms=[settings.jwt_algorithm])
+    except jwt.PyJWTError as exc:
+        raise InvalidTokenError(str(exc)) from exc

backend/app/db/base.py

@@ -0,0 +1,5 @@
+from sqlalchemy.orm import DeclarativeBase
+
+
+class Base(DeclarativeBase):
+    """Gemeinsame Basisklasse aller ORM-Modelle (app/models/)."""

backend/app/db/session.py

@@ -0,0 +1,13 @@
+from collections.abc import AsyncGenerator
+
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+
+from app.core.app_settings import settings
+
+engine = create_async_engine(settings.database_url, pool_pre_ping=True)
+SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+
+
+async def get_db() -> AsyncGenerator[AsyncSession, None]:
+    async with SessionLocal() as session:
+        yield session

backend/app/main.py

@@ -0,0 +1,6 @@
+from fastapi import FastAPI
+
+from app.api.v1.api import api_router
+
+app = FastAPI(title="MABEA", version="0.1.0")
+app.include_router(api_router, prefix="/api/v1")

backend/app/models/__init__.py

@@ -0,0 +1,3 @@
+from app.models.auth import Benutzer, BenutzerRolle, RolleTyp, Systemknoten, KnotenTyp
+
+__all__ = ["Benutzer", "BenutzerRolle", "RolleTyp", "Systemknoten", "KnotenTyp"]

backend/app/models/auth.py

@@ -0,0 +1,60 @@
+import enum
+
+from sqlalchemy import Boolean, ForeignKey, String
+from sqlalchemy.dialects.postgresql import ENUM as PgEnum
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from app.db.base import Base
+
+
+class KnotenTyp(str, enum.Enum):
+    haupt = "haupt"
+    satellit = "satellit"
+
+
+class RolleTyp(str, enum.Enum):
+    mitarbeiter = "mitarbeiter"
+    materialverantwortlicher = "materialverantwortlicher"
+    leitungsverantwortlicher = "leitungsverantwortlicher"
+    administration = "administration"
+
+
+# create_type=False: die PostgreSQL-ENUM-Typen werden ausschließlich per Alembic-Migration
+# angelegt (0001_initial_schema), damit Migration und ORM-Modell nicht auseinanderlaufen.
+knoten_typ_pg = PgEnum(KnotenTyp, name="knoten_typ", create_type=False)
+rolle_typ_pg = PgEnum(RolleTyp, name="rolle_typ", create_type=False)
+
+
+class Systemknoten(Base):
+    """Hauptserver oder Satellit (Karte 13). V1 nutzt ausschließlich den Hauptserver-Datensatz."""
+
+    __tablename__ = "systemknoten"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column(String, unique=True, nullable=False)
+    typ: Mapped[KnotenTyp] = mapped_column(knoten_typ_pg, nullable=False, default=KnotenTyp.satellit)
+
+
+class Benutzer(Base):
+    __tablename__ = "benutzer"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column(String, nullable=False)
+    login: Mapped[str] = mapped_column(String, unique=True, nullable=False)
+    passwort_hash: Mapped[str] = mapped_column(String, nullable=False)
+    aktiv: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
+
+    rollen: Mapped[list["BenutzerRolle"]] = relationship(back_populates="benutzer", lazy="selectin")
+
+    @property
+    def rollen_namen(self) -> list[str]:
+        return [r.rolle.value for r in self.rollen]
+
+
+class BenutzerRolle(Base):
+    __tablename__ = "benutzer_rolle"
+
+    benutzer_id: Mapped[int] = mapped_column(ForeignKey("benutzer.id"), primary_key=True)
+    rolle: Mapped[RolleTyp] = mapped_column(rolle_typ_pg, primary_key=True)
+
+    benutzer: Mapped["Benutzer"] = relationship(back_populates="rollen")