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/.gitignore

@@ -0,0 +1,9 @@
+__pycache__/
+*.pyc
+.venv/
+venv/
+.env
+.pytest_cache/
+.coverage
+htmlcov/
+*.egg-info/

backend/README.md

@@ -0,0 +1,35 @@
+# MABEA Backend – Sprint 0
+
+FastAPI-Skeleton, Datenbankschema (Alembic), Auth (JWT), Rollen-Dependency. Details: `ergebnisse/sprintplan.md`, `ergebnisse/19_technische_architektur.md`, `ergebnisse/20_datenbank_schema.md`.
+
+## Setup (auf dem Zielsystem, nicht lokal)
+
+```bash
+python3 -m venv .venv
+source .venv/bin/activate
+pip install -e ".[dev]"
+
+cp example.env .env
+# .env editieren: DATABASE_URL, JWT_SECRET_KEY, SYSTEMKNOTEN_ID
+
+alembic upgrade head
+uvicorn app.main:app --host 0.0.0.0 --port 8000
+```
+
+## Tests (CI oder Zielsystem)
+
+```bash
+pytest
+```
+
+Erwartet eine per Alembic migrierte PostgreSQL-Test-Datenbank (`DATABASE_URL` zeigt darauf). CI-Workflow: `.gitea/workflows/ci.yml`.
+
+## Struktur
+
+- `app/core/app_settings.py` – Konfiguration aus Umgebungsvariablen
+- `app/core/security.py` – Passwort-Hashing, JWT
+- `app/db/` – SQLAlchemy Engine/Session
+- `app/models/` – ORM-Modelle (Sprint 0: nur Auth-relevante Tabellen; weitere Modelle folgen Sprint 1+)
+- `app/api/` – FastAPI-Router, Dependencies (u.a. `require_roles` für Prompt-05-Berechtigungsmatrix)
+- `alembic/versions/0001_initial_schema.py` – vollständiges Ziel-Schema (Prompt 20), auch für Tabellen, die erst spätere Sprints per API befüllen
+- `alembic/versions/0002_seed_hauptserver.py` – Sprintplan E6: seedet den einen `systemknoten`-Datensatz (`typ='haupt'`)

backend/alembic.ini

@@ -0,0 +1,38 @@
+[alembic]
+script_location = alembic
+prepend_sys_path = .
+sqlalchemy.url =
+
+[loggers]
+keys = root,sqlalchemy,alembic
+
+[handlers]
+keys = console
+
+[formatters]
+keys = generic
+
+[logger_root]
+level = WARN
+handlers = console
+qualname =
+
+[logger_sqlalchemy]
+level = WARN
+handlers =
+qualname = sqlalchemy.engine
+
+[logger_alembic]
+level = INFO
+handlers =
+qualname = alembic
+
+[handler_console]
+class = StreamHandler
+args = (sys.stderr,)
+level = NOTSET
+formatter = generic
+
+[formatter_generic]
+format = %(levelname)-5.5s [%(name)s] %(message)s
+datefmt = %H:%M:%S

backend/alembic/env.py

@@ -0,0 +1,51 @@
+import asyncio
+from logging.config import fileConfig
+
+from alembic import context
+from sqlalchemy import pool
+from sqlalchemy.ext.asyncio import create_async_engine
+
+from app.core.app_settings import settings
+from app.db.base import Base
+import app.models  # noqa: F401  (registriert alle Modelle an Base.metadata)
+
+config = context.config
+
+if config.config_file_name is not None:
+    fileConfig(config.config_file_name)
+
+target_metadata = Base.metadata
+
+
+def get_url() -> str:
+    return settings.database_url
+
+
+def run_migrations_offline() -> None:
+    context.configure(
+        url=get_url(),
+        target_metadata=target_metadata,
+        literal_binds=True,
+        dialect_opts={"paramstyle": "named"},
+    )
+    with context.begin_transaction():
+        context.run_migrations()
+
+
+def do_run_migrations(connection) -> None:
+    context.configure(connection=connection, target_metadata=target_metadata)
+    with context.begin_transaction():
+        context.run_migrations()
+
+
+async def run_migrations_online() -> None:
+    connectable = create_async_engine(get_url(), poolclass=pool.NullPool)
+    async with connectable.connect() as connection:
+        await connection.run_sync(do_run_migrations)
+    await connectable.dispose()
+
+
+if context.is_offline_mode():
+    run_migrations_offline()
+else:
+    asyncio.run(run_migrations_online())

backend/alembic/script.py.mako

@@ -0,0 +1,26 @@
+"""${message}
+
+Revision ID: ${up_revision}
+Revises: ${down_revision | comma,n}
+Create Date: ${create_date}
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+${imports if imports else ""}
+
+# revision identifiers, used by Alembic.
+revision: str = ${repr(up_revision)}
+down_revision: Union[str, None] = ${repr(down_revision)}
+branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
+depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}
+
+
+def upgrade() -> None:
+    ${upgrades if upgrades else "pass"}
+
+
+def downgrade() -> None:
+    ${downgrades if downgrades else "pass"}

backend/alembic/versions/0001_initial_schema.py

@@ -0,0 +1,291 @@
+"""Initiales Schema (Prompt 20 / ergebnisse/20_datenbank_schema.md)
+
+Revision ID: 0001_initial_schema
+Revises:
+Create Date: 2026-09-03
+
+Vollständiges Ziel-Schema aus Prompt 20 (inkl. Karte-13-Vorbereitung: UUID auf
+sync-relevanten Tabellen, systemknoten). Sprint 0 legt das komplette Schema an,
+auch wenn erst spätere Sprints die zugehörigen Endpunkte/Business-Logik bauen –
+so entfällt ein späterer Schema-Umbau (Designziel, Prompt 06/19).
+"""
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "0001_initial_schema"
+down_revision: Union[str, None] = None
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+UPGRADE_SQL = """
+-- 1. Stammdaten
+CREATE TABLE bereich (
+    id              SERIAL PRIMARY KEY,
+    name            TEXT NOT NULL UNIQUE,
+    beschreibung    TEXT
+);
+
+CREATE TABLE kategorie (
+    id                  SERIAL PRIMARY KEY,
+    bereich_id          INTEGER NOT NULL REFERENCES bereich(id),
+    name                TEXT NOT NULL,
+    ueberkategorie_id   INTEGER REFERENCES kategorie(id),
+    UNIQUE (bereich_id, name)
+);
+
+CREATE TABLE standort (
+    id      SERIAL PRIMARY KEY,
+    name    TEXT NOT NULL UNIQUE,
+    adresse TEXT
+);
+
+CREATE TABLE objekttyp (
+    id          SERIAL PRIMARY KEY,
+    bereich_id  INTEGER NOT NULL REFERENCES bereich(id),
+    kategorie_id INTEGER REFERENCES kategorie(id),
+    name        TEXT NOT NULL,
+    UNIQUE (bereich_id, name)
+);
+
+CREATE TYPE materialtyp AS ENUM ('standard', 'ablauf_charge', 'geraet_sn');
+
+CREATE TABLE material (
+    id              SERIAL PRIMARY KEY,
+    name            TEXT NOT NULL,
+    artikelnummer   TEXT,
+    einheit         TEXT NOT NULL,
+    materialtyp     materialtyp NOT NULL,
+    kategorie_id    INTEGER REFERENCES kategorie(id),
+    hersteller      TEXT,
+    beschreibung    TEXT,
+    code            TEXT UNIQUE,
+    warnzeitraum_tage INTEGER,
+    aktiv           BOOLEAN NOT NULL DEFAULT TRUE
+);
+
+-- 2. Vorlagen
+CREATE TYPE vorlage_status AS ENUM ('aktiv', 'veraltet');
+
+CREATE TABLE beladungsvorlage (
+    id              SERIAL PRIMARY KEY,
+    objekttyp_id    INTEGER NOT NULL REFERENCES objekttyp(id),
+    name            TEXT NOT NULL,
+    version         INTEGER NOT NULL,
+    gueltig_ab      TIMESTAMPTZ NOT NULL DEFAULT now(),
+    status          vorlage_status NOT NULL DEFAULT 'aktiv',
+    UNIQUE (objekttyp_id, name, version)
+);
+
+CREATE TABLE vorlagenposition (
+    id              SERIAL PRIMARY KEY,
+    vorlage_id      INTEGER NOT NULL REFERENCES beladungsvorlage(id),
+    material_id     INTEGER NOT NULL REFERENCES material(id),
+    fach            TEXT,
+    sollmenge       NUMERIC NOT NULL,
+    UNIQUE (vorlage_id, material_id)
+);
+
+-- 3. Objekte
+CREATE TYPE knoten_typ AS ENUM ('haupt', 'satellit');
+
+CREATE TABLE systemknoten (
+    id              SERIAL PRIMARY KEY,
+    name            TEXT NOT NULL UNIQUE,
+    typ             knoten_typ NOT NULL DEFAULT 'satellit'
+);
+
+CREATE TYPE objekt_status AS ENUM ('aktiv', 'ausser_dienst');
+
+CREATE TABLE objekt (
+    id                      SERIAL PRIMARY KEY,
+    code                    TEXT NOT NULL UNIQUE,
+    name                    TEXT NOT NULL,
+    objekttyp_id            INTEGER NOT NULL REFERENCES objekttyp(id),
+    vorlage_id              INTEGER REFERENCES beladungsvorlage(id),
+    standort_id             INTEGER NOT NULL REFERENCES standort(id),
+    status                  objekt_status NOT NULL DEFAULT 'aktiv',
+    zustaendiger_server_id  INTEGER NOT NULL REFERENCES systemknoten(id)
+);
+
+CREATE TYPE objektposition_status AS ENUM ('aktiv', 'entfernt');
+
+CREATE TABLE objektposition (
+    id                  UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    objekt_id           INTEGER NOT NULL REFERENCES objekt(id),
+    material_id         INTEGER NOT NULL REFERENCES material(id),
+    sollmenge_override  NUMERIC,
+    ist_status          objektposition_status NOT NULL DEFAULT 'aktiv',
+    istmenge            NUMERIC NOT NULL DEFAULT 0,
+    seriennummer        TEXT,
+    ablaufdatum         DATE,
+    chargennummer       TEXT,
+    zuletzt_geaendert_am TIMESTAMPTZ NOT NULL DEFAULT now(),
+    UNIQUE (objekt_id, material_id)
+);
+
+-- 4. Zuständigkeiten, Benutzer, Rollen
+CREATE TABLE benutzer (
+    id              SERIAL PRIMARY KEY,
+    name            TEXT NOT NULL,
+    login           TEXT NOT NULL UNIQUE,
+    passwort_hash   TEXT NOT NULL,
+    aktiv           BOOLEAN NOT NULL DEFAULT TRUE
+);
+
+CREATE TYPE rolle_typ AS ENUM ('mitarbeiter', 'materialverantwortlicher', 'leitungsverantwortlicher', 'administration');
+
+CREATE TABLE benutzer_rolle (
+    benutzer_id     INTEGER NOT NULL REFERENCES benutzer(id),
+    rolle           rolle_typ NOT NULL,
+    PRIMARY KEY (benutzer_id, rolle)
+);
+
+CREATE TABLE zustaendigkeit (
+    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    benutzer_id     INTEGER NOT NULL REFERENCES benutzer(id),
+    standort_id     INTEGER REFERENCES standort(id),
+    objekt_id       INTEGER REFERENCES objekt(id),
+    CHECK (standort_id IS NOT NULL OR objekt_id IS NOT NULL)
+);
+
+CREATE TABLE kontrollverantwortung (
+    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    objekt_id       INTEGER NOT NULL REFERENCES objekt(id),
+    benutzer_id     INTEGER REFERENCES benutzer(id),
+    gruppe          TEXT
+);
+
+-- 5. Kontrolle
+CREATE TYPE kontroll_status AS ENUM ('nicht_gestartet', 'in_bearbeitung', 'abgeschlossen', 'abgebrochen');
+
+CREATE TABLE kontrolle (
+    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    erzeugt_von_server_id INTEGER NOT NULL REFERENCES systemknoten(id),
+    objekt_id       INTEGER NOT NULL REFERENCES objekt(id),
+    benutzer_id     INTEGER NOT NULL REFERENCES benutzer(id),
+    status          kontroll_status NOT NULL DEFAULT 'in_bearbeitung',
+    gestartet_am    TIMESTAMPTZ NOT NULL DEFAULT now(),
+    beendet_am      TIMESTAMPTZ,
+    abbruch_grund   TEXT,
+    signatur        BYTEA
+);
+
+CREATE TABLE kontrollposition (
+    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    kontrolle_id    UUID NOT NULL REFERENCES kontrolle(id),
+    material_id     INTEGER NOT NULL REFERENCES material(id),
+    sollmenge_snapshot  NUMERIC NOT NULL,
+    istmenge_erfasst    NUMERIC NOT NULL,
+    abweichung          BOOLEAN NOT NULL
+);
+
+-- 6. Fehlbestand, Nachfüllung, Mindermenge
+CREATE TYPE fehlbestand_status AS ENUM ('offen', 'in_bearbeitung', 'nachgefuellt_teilweise', 'erledigt');
+
+CREATE TABLE fehlbestand (
+    id                  UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    erzeugt_von_server_id INTEGER NOT NULL REFERENCES systemknoten(id),
+    objekt_id           INTEGER NOT NULL REFERENCES objekt(id),
+    material_id         INTEGER NOT NULL REFERENCES material(id),
+    standort_id         INTEGER NOT NULL REFERENCES standort(id),
+    sollmenge           NUMERIC NOT NULL,
+    istmenge             NUMERIC NOT NULL,
+    fehlmenge            NUMERIC NOT NULL,
+    entstanden_am         TIMESTAMPTZ NOT NULL DEFAULT now(),
+    festgestellt_von      INTEGER NOT NULL REFERENCES benutzer(id),
+    kontrolle_id          UUID REFERENCES kontrolle(id),
+    ursache               TEXT,
+    verantwortlicher_id   INTEGER REFERENCES benutzer(id),
+    status                fehlbestand_status NOT NULL DEFAULT 'offen',
+    erledigt_am           TIMESTAMPTZ
+);
+
+CREATE TABLE nachfuellung (
+    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    fehlbestand_id  UUID REFERENCES fehlbestand(id),
+    objekt_id       INTEGER NOT NULL REFERENCES objekt(id),
+    material_id     INTEGER NOT NULL REFERENCES material(id),
+    menge           NUMERIC NOT NULL,
+    benutzer_id     INTEGER NOT NULL REFERENCES benutzer(id),
+    zeitpunkt       TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TYPE mindermenge_status AS ENUM ('aktiv', 'abgelaufen', 'beendet_durch_erledigung');
+
+CREATE TABLE mindermengen_genehmigung (
+    id                  UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    fehlbestand_id      UUID NOT NULL REFERENCES fehlbestand(id),
+    genehmigt_von        INTEGER NOT NULL REFERENCES benutzer(id),
+    begruendung          TEXT NOT NULL,
+    genehmigt_am          TIMESTAMPTZ NOT NULL DEFAULT now(),
+    ausloesende_kontrolle_id UUID NOT NULL REFERENCES kontrolle(id),
+    status                mindermenge_status NOT NULL DEFAULT 'aktiv',
+    beendet_am            TIMESTAMPTZ,
+    beendende_kontrolle_id UUID REFERENCES kontrolle(id)
+);
+
+-- 7. Historie/Audit
+CREATE TABLE historie (
+    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    erzeugt_von_server_id INTEGER NOT NULL REFERENCES systemknoten(id),
+    zeitpunkt       TIMESTAMPTZ NOT NULL DEFAULT now(),
+    benutzer_id     INTEGER REFERENCES benutzer(id),
+    ereignistyp     TEXT NOT NULL,
+    entitaet_typ    TEXT NOT NULL,
+    entitaet_id     TEXT NOT NULL,
+    alter_wert      JSONB,
+    neuer_wert      JSONB,
+    begruendung     TEXT
+);
+
+-- 8. Indizes
+CREATE INDEX idx_fehlbestand_status ON fehlbestand(status);
+CREATE INDEX idx_fehlbestand_objekt ON fehlbestand(objekt_id);
+CREATE INDEX idx_fehlbestand_entstanden ON fehlbestand(entstanden_am);
+CREATE INDEX idx_objektposition_objekt ON objektposition(objekt_id);
+CREATE INDEX idx_historie_entitaet ON historie(entitaet_typ, entitaet_id);
+CREATE INDEX idx_zustaendigkeit_benutzer ON zustaendigkeit(benutzer_id);
+"""
+
+DOWNGRADE_SQL = """
+DROP TABLE IF EXISTS historie;
+DROP TABLE IF EXISTS mindermengen_genehmigung;
+DROP TABLE IF EXISTS nachfuellung;
+DROP TABLE IF EXISTS fehlbestand;
+DROP TABLE IF EXISTS kontrollposition;
+DROP TABLE IF EXISTS kontrolle;
+DROP TABLE IF EXISTS kontrollverantwortung;
+DROP TABLE IF EXISTS zustaendigkeit;
+DROP TABLE IF EXISTS benutzer_rolle;
+DROP TABLE IF EXISTS benutzer;
+DROP TABLE IF EXISTS objektposition;
+DROP TABLE IF EXISTS objekt;
+DROP TABLE IF EXISTS systemknoten;
+DROP TABLE IF EXISTS vorlagenposition;
+DROP TABLE IF EXISTS beladungsvorlage;
+DROP TABLE IF EXISTS material;
+DROP TABLE IF EXISTS objekttyp;
+DROP TABLE IF EXISTS standort;
+DROP TABLE IF EXISTS kategorie;
+DROP TABLE IF EXISTS bereich;
+
+DROP TYPE IF EXISTS mindermenge_status;
+DROP TYPE IF EXISTS fehlbestand_status;
+DROP TYPE IF EXISTS kontroll_status;
+DROP TYPE IF EXISTS rolle_typ;
+DROP TYPE IF EXISTS objektposition_status;
+DROP TYPE IF EXISTS objekt_status;
+DROP TYPE IF EXISTS knoten_typ;
+DROP TYPE IF EXISTS vorlage_status;
+DROP TYPE IF EXISTS materialtyp;
+"""
+
+
+def upgrade() -> None:
+    op.execute(UPGRADE_SQL)
+
+
+def downgrade() -> None:
+    op.execute(DOWNGRADE_SQL)

backend/alembic/versions/0002_seed_hauptserver.py

@@ -0,0 +1,25 @@
+"""Seed: genau ein systemknoten-Datensatz mit typ='haupt' (Sprintplan E6)
+
+Revision ID: 0002_seed_hauptserver
+Revises: 0001_initial_schema
+Create Date: 2026-09-03
+"""
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "0002_seed_hauptserver"
+down_revision: Union[str, None] = "0001_initial_schema"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.execute(
+        "INSERT INTO systemknoten (name, typ) VALUES ('Hauptserver', 'haupt') "
+        "ON CONFLICT (name) DO NOTHING;"
+    )
+
+
+def downgrade() -> None:
+    op.execute("DELETE FROM systemknoten WHERE name = 'Hauptserver' AND typ = 'haupt';")

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")

backend/example.env

@@ -0,0 +1,9 @@
+# Kopieren nach .env (auf dem Zielsystem) und Werte setzen. .env selbst nie committen.
+
+DATABASE_URL=postgresql+asyncpg://mabea:changeme@localhost:5432/mabea
+JWT_SECRET_KEY=change-me-to-a-long-random-value
+JWT_ALGORITHM=HS256
+ACCESS_TOKEN_EXPIRE_MINUTES=480
+
+# ID des systemknoten-Datensatzes mit typ='haupt' (Sprintplan E6, siehe alembic/versions/0002_seed_hauptserver.py)
+SYSTEMKNOTEN_ID=1

backend/pyproject.toml

@@ -0,0 +1,31 @@
+[project]
+name = "mabea-backend"
+version = "0.1.0"
+description = "MABEA – Digitales Materialmanagement Rettungsdienst/KatS, Backend (Sprint 0)"
+requires-python = ">=3.11"
+dependencies = [
+    "fastapi>=0.115,<0.116",
+    "uvicorn[standard]>=0.32,<0.33",
+    "sqlalchemy>=2.0,<2.1",
+    "asyncpg>=0.30,<0.31",
+    "alembic>=1.13,<1.14",
+    "pydantic-settings>=2.6,<2.7",
+    "passlib[bcrypt]>=1.7,<1.8",
+    "pyjwt>=2.9,<2.10",
+    "python-multipart>=0.0.12,<0.1",
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest>=8.3,<8.4",
+    "pytest-asyncio>=0.24,<0.25",
+    "pytest-cov>=5.0,<6.0",
+    "httpx>=0.27,<0.28",
+]
+
+[build-system]
+requires = ["setuptools>=68"]
+build-backend = "setuptools.build_meta"
+
+[tool.setuptools.packages.find]
+include = ["app*"]

backend/pytest.ini

@@ -0,0 +1,4 @@
+[pytest]
+asyncio_mode = auto
+addopts = --cov=app --cov-report=term-missing --cov-fail-under=50
+testpaths = tests

backend/tests/conftest.py

@@ -0,0 +1,70 @@
+import pytest
+import pytest_asyncio
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+
+from app.core.app_settings import settings
+from app.core.security import hash_password
+from app.db.session import get_db
+from app.main import app
+from app.models.auth import Benutzer, BenutzerRolle, RolleTyp
+
+# Erwartet eine bereits per Alembic migrierte Test-Datenbank (CI: install -> migrate -> pytest,
+# siehe testphasen.md Phase 0). Jeder Test läuft in einer Transaktion, die am Ende zurückgerollt
+# wird, damit Tests sich nicht gegenseitig beeinflussen.
+engine = create_async_engine(settings.database_url, pool_pre_ping=True)
+TestSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
+
+
+@pytest_asyncio.fixture
+async def db_session():
+    async with engine.connect() as connection:
+        transaction = await connection.begin()
+        session = AsyncSession(bind=connection, expire_on_commit=False)
+        try:
+            yield session
+        finally:
+            await session.close()
+            await transaction.rollback()
+
+
+@pytest_asyncio.fixture
+async def client(db_session):
+    async def _get_db_override():
+        yield db_session
+
+    app.dependency_overrides[get_db] = _get_db_override
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as ac:
+        yield ac
+    app.dependency_overrides.clear()
+
+
+@pytest_asyncio.fixture
+async def mitarbeiter_user(db_session):
+    benutzer = Benutzer(
+        name="Test Mitarbeiter",
+        login="mitarbeiter1",
+        passwort_hash=hash_password("test-passwort-123"),
+        aktiv=True,
+    )
+    db_session.add(benutzer)
+    await db_session.flush()
+    db_session.add(BenutzerRolle(benutzer_id=benutzer.id, rolle=RolleTyp.mitarbeiter))
+    await db_session.flush()
+    return benutzer
+
+
+@pytest_asyncio.fixture
+async def admin_user(db_session):
+    benutzer = Benutzer(
+        name="Test Administration",
+        login="admin1",
+        passwort_hash=hash_password("test-passwort-123"),
+        aktiv=True,
+    )
+    db_session.add(benutzer)
+    await db_session.flush()
+    db_session.add(BenutzerRolle(benutzer_id=benutzer.id, rolle=RolleTyp.administration))
+    await db_session.flush()
+    return benutzer

backend/tests/test_auth.py

@@ -0,0 +1,45 @@
+import pytest
+
+
+@pytest.mark.asyncio
+async def test_login_success(client, mitarbeiter_user):
+    response = await client.post(
+        "/api/v1/auth/login",
+        data={"username": "mitarbeiter1", "password": "test-passwort-123"},
+    )
+    assert response.status_code == 200
+    body = response.json()
+    assert body["token_type"] == "bearer"
+    assert body["access_token"]
+
+
+@pytest.mark.asyncio
+async def test_login_wrong_password(client, mitarbeiter_user):
+    response = await client.post(
+        "/api/v1/auth/login",
+        data={"username": "mitarbeiter1", "password": "falsch"},
+    )
+    assert response.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_me_requires_token(client):
+    response = await client.get("/api/v1/auth/me")
+    assert response.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_me_returns_current_user(client, mitarbeiter_user):
+    login_response = await client.post(
+        "/api/v1/auth/login",
+        data={"username": "mitarbeiter1", "password": "test-passwort-123"},
+    )
+    token = login_response.json()["access_token"]
+
+    response = await client.get(
+        "/api/v1/auth/me", headers={"Authorization": f"Bearer {token}"}
+    )
+    assert response.status_code == 200
+    body = response.json()
+    assert body["login"] == "mitarbeiter1"
+    assert body["rollen"] == ["mitarbeiter"]

backend/tests/test_health.py

@@ -0,0 +1,8 @@
+import pytest
+
+
+@pytest.mark.asyncio
+async def test_health(client):
+    response = await client.get("/api/v1/health")
+    assert response.status_code == 200
+    assert response.json() == {"status": "ok"}

backend/tests/test_roles.py

@@ -0,0 +1,28 @@
+import pytest
+
+
+async def _login(client, username: str) -> str:
+    response = await client.post(
+        "/api/v1/auth/login",
+        data={"username": username, "password": "test-passwort-123"},
+    )
+    return response.json()["access_token"]
+
+
+@pytest.mark.asyncio
+async def test_admin_only_rejects_mitarbeiter(client, mitarbeiter_user):
+    token = await _login(client, "mitarbeiter1")
+    response = await client.get(
+        "/api/v1/health/admin-only", headers={"Authorization": f"Bearer {token}"}
+    )
+    assert response.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_admin_only_allows_administration(client, admin_user):
+    token = await _login(client, "admin1")
+    response = await client.get(
+        "/api/v1/health/admin-only", headers={"Authorization": f"Bearer {token}"}
+    )
+    assert response.status_code == 200
+    assert response.json()["login"] == "admin1"