feat: Phase 0 – Gerüst (FastAPI+SQLite, Multi-User-Login, React-App-Shell, Deploy)

Grundgerüst für Certain Finance:
- Backend: FastAPI + SQLAlchemy + SQLite, Alembic-Migrationen (Batch-Mode),
  argon2-Passwort-Hashing, server-seitige Sitzungen im Cookie, Erst-Admin,
  Benutzerverwaltung mit Temporär-Passwort, Einstellungen je Benutzer.
- Ein Prozess, ein Port: uvicorn liefert API (/api) und gebautes SPA aus,
  inkl. SPA-Deeplink-Fallback.
- Frontend: React 18 + Vite + JSX, Design-Tokens (Light/Dark), i18n DE/EN,
  Inline-SVG-Icons (keine Emojis), Login, Passwort-ändern, Dashboard-Platzhalter,
  Benutzer- und Einstellungsseiten.
- Deploy: nativer LXC-Installer + Update-Skript (systemd), env-Vorlage, Doku.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Guido
2026-08-19 18:27:42 +02:00
co-authored by Claude Opus 4.8
commit 2c795a57c8
59 changed files with 6419 additions and 0 deletions
+42
View File
@@ -0,0 +1,42 @@
# Alembic-Konfiguration. Die Datenbank-URL wird NICHT hier gesetzt, sondern in
# migrations/env.py aus app.database (SQLite, aus der App-Konfiguration) gelesen –
# so gibt es genau eine Quelle der Wahrheit.
[alembic]
script_location = migrations
prepend_sys_path = .
path_separator = os
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARNING
handlers = console
qualname =
[logger_sqlalchemy]
level = WARNING
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
View File
+21
View File
@@ -0,0 +1,21 @@
"""Zentrale Namens-/Versionskonstanten (Backend-Seite).
Der Anzeigename der Anwendung steht bewusst NUR hier (und im Frontend in
``web/src/brand.js``). Überall sonst werden generische Bezeichner verwendet
(Paket ``app``, Env-Präfix ``APP_``), damit ein späteres Umbenennen genau eine
Konstante betrifft.
"""
import os
# Anzeigename – die einzige Stelle im Backend, an der der Produktname steht.
APP_NAME = "Certain Finance"
# Programmversion. Wird bei /api/version ausgegeben.
VERSION = "0.1.0"
def git_sha() -> str | None:
"""Kurzer Git-Commit, falls vom Build/Env gesetzt (``APP_GIT_SHA``)."""
sha = os.environ.get("APP_GIT_SHA", "").strip()
return sha or None
+55
View File
@@ -0,0 +1,55 @@
"""Konfiguration über Umgebungsvariablen (Präfix ``APP_``).
Beispiel: ``APP_DATA_DIR=/var/lib/certain-finance`` setzt ``data_dir``.
Die Standardwerte sind bewusst entwicklungsfreundlich (relatives Datenverzeichnis,
unsichere Cookies erlaubt). Für den Produktivbetrieb setzt das Install-Skript die
sicheren Werte in ``/etc/certain-finance/env`` (siehe ``deploy/env.example``).
"""
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="APP_", env_file=".env", extra="ignore")
# --- Ablageort für Datenbank und Backups ---------------------------------
# Produktiv per Env auf einen absoluten Pfad setzen (z. B. /var/lib/...).
data_dir: str = "./data"
# Leer = SQLite-Datei unter data_dir/app.db. Sonst vollständige URL.
database_url: str = ""
# --- Sitzungen / Cookies -------------------------------------------------
session_secret: str = "change-me-in-production"
session_ttl_days: int = 30
cookie_name: str = "cf_session"
# Standard False, damit die Anmeldung auch über http://localhost im Dev
# funktioniert. Produktiv hinter HTTPS (Nginx Proxy Manager) auf true setzen.
cookie_secure: bool = False
cookie_samesite: str = "lax"
# --- Erster Admin, nur beim allerersten Start angelegt -------------------
admin_username: str = "admin"
admin_password: str = "changeme"
# --- Währung -------------------------------------------------------------
default_base_currency: str = "EUR"
# Beim Start die aktuellen EZB-Kurse nachladen (benötigt Internet).
ecb_fetch_on_start: bool = True
# --- CORS ----------------------------------------------------------------
# Leer = kein Cross-Origin (gleiche Herkunft, da FastAPI das SPA selbst
# ausliefert). Kommagetrennte Liste oder "*" nur bei Bedarf.
cors_origins: str = ""
@property
def effective_database_url(self) -> str:
if self.database_url:
return self.database_url
return f"sqlite:///{self.data_dir.rstrip('/')}/app.db"
@lru_cache
def get_settings() -> Settings:
return Settings()
+62
View File
@@ -0,0 +1,62 @@
"""Datenbank-Anbindung (SQLite).
SQLite braucht ein paar PRAGMAs, damit es sich unter FastAPIs Thread-Pool
vernünftig verhält: WAL für gleichzeitige Leser, ``foreign_keys=ON`` (sonst
greifen die ON-DELETE-Regeln nicht), und ein Busy-Timeout, damit kurze
Schreibsperren nicht sofort krachen.
"""
import os
from collections.abc import Generator
from sqlalchemy import create_engine, event
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
from .config import get_settings
settings = get_settings()
_url = settings.effective_database_url
# Für die SQLite-Datei das Datenverzeichnis sicherstellen.
if _url.startswith("sqlite:///"):
_path = _url.replace("sqlite:///", "", 1)
_dir = os.path.dirname(_path)
if _dir:
os.makedirs(_dir, exist_ok=True)
connect_args = {"check_same_thread": False} if _url.startswith("sqlite") else {}
engine = create_engine(
_url,
connect_args=connect_args,
pool_pre_ping=True,
future=True,
)
@event.listens_for(engine, "connect")
def _sqlite_pragmas(dbapi_connection, _record):
"""Bei jeder neuen SQLite-Verbindung sinnvolle Voreinstellungen setzen."""
if not _url.startswith("sqlite"):
return
cur = dbapi_connection.cursor()
cur.execute("PRAGMA journal_mode=WAL")
cur.execute("PRAGMA foreign_keys=ON")
cur.execute("PRAGMA busy_timeout=5000")
cur.close()
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
class Base(DeclarativeBase):
pass
def get_db() -> Generator[Session, None, None]:
db = SessionLocal()
try:
yield db
finally:
db.close()
+36
View File
@@ -0,0 +1,36 @@
"""Gemeinsame FastAPI-Abhängigkeiten: aktueller Benutzer (Cookie), Admin-Pflicht.
Die Anmeldung läuft über eine server-seitige Sitzung im Cookie (siehe
``services/sessions``). ``get_current_user`` liest das Cookie, prüft die Sitzung
und liefert den Benutzer – oder 401.
"""
from fastapi import Depends, HTTPException, Request, status
from sqlalchemy.orm import Session
from .config import get_settings
from .database import get_db
from .models import Role, User
from .services import sessions
_credentials_exc = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Nicht angemeldet oder Sitzung abgelaufen",
)
def get_current_user(request: Request, db: Session = Depends(get_db)) -> User:
token = request.cookies.get(get_settings().cookie_name, "")
user = sessions.validate(db, token)
if user is None:
raise _credentials_exc
return user
def require_admin(current_user: User = Depends(get_current_user)) -> User:
if current_user.role != Role.admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Nur Administratoren dürfen diese Aktion ausführen",
)
return current_user
+97
View File
@@ -0,0 +1,97 @@
"""FastAPI-Anwendung.
Ein einzelner Prozess liefert sowohl die JSON-API unter ``/api/*`` als auch das
gebaute Frontend (``web/dist``) aus – hinter dem Nginx Proxy Manager des Nutzers.
Beim Start werden die Alembic-Migrationen angewandt und der erste Admin angelegt.
"""
from contextlib import asynccontextmanager
from pathlib import Path
from fastapi import APIRouter, FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from .branding import APP_NAME, VERSION, git_sha
from .config import get_settings
from .database import SessionLocal
from .migrate import upgrade_to_head
from .routers import auth, settings as settings_router, users
from .seed import ensure_first_admin
settings = get_settings()
# Pfad zum gebauten Frontend (im Dev von Vite selbst ausgeliefert -> fehlt hier).
REPO_ROOT = Path(__file__).resolve().parent.parent.parent
DIST_DIR = REPO_ROOT / "web" / "dist"
@asynccontextmanager
async def lifespan(app: FastAPI):
# Schema auf den aktuellen Stand bringen (idempotent).
upgrade_to_head()
db = SessionLocal()
try:
ensure_first_admin(db)
finally:
db.close()
yield
app = FastAPI(title=APP_NAME, version=VERSION, lifespan=lifespan)
# CORS nur, wenn ausdrücklich konfiguriert (Standard: gleiche Herkunft, aus).
if settings.cors_origins.strip():
origins = (
["*"]
if settings.cors_origins.strip() == "*"
else [o.strip() for o in settings.cors_origins.split(",") if o.strip()]
)
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# --- API -------------------------------------------------------------------
api = APIRouter(prefix="/api")
@api.get("/health", tags=["meta"])
def health() -> dict:
return {"status": "ok"}
@api.get("/version", tags=["meta"])
def version() -> dict:
return {"name": APP_NAME, "version": VERSION, "git_sha": git_sha()}
api.include_router(auth.router)
api.include_router(users.router)
api.include_router(settings_router.router)
app.include_router(api)
# --- Frontend (SPA) --------------------------------------------------------
# Gebaute, gehashte Assets unter /assets aggressiv cachebar ausliefern; alle
# übrigen (nicht-/api) Pfade auf index.html abbilden, damit Client-seitige
# Routen wie /accounts direkt aufrufbar sind (verlinkbar).
if DIST_DIR.exists():
assets_dir = DIST_DIR / "assets"
if assets_dir.exists():
app.mount("/assets", StaticFiles(directory=assets_dir), name="assets")
@app.get("/{full_path:path}", include_in_schema=False)
async def spa(full_path: str):
# /api-Pfade, die keine echte Route trafen, sind echte 404 (kein SPA).
if full_path.startswith("api"):
raise HTTPException(status_code=404, detail="Not Found")
candidate = DIST_DIR / full_path
if full_path and candidate.is_file():
return FileResponse(candidate)
return FileResponse(DIST_DIR / "index.html", headers={"Cache-Control": "no-cache"})
+25
View File
@@ -0,0 +1,25 @@
"""Programmatisches Anwenden der Alembic-Migrationen.
Wird beim App-Start aufgerufen, damit ein frisch geklontes/gestartetes System
sofort das aktuelle Schema hat – ohne dass man ``alembic`` von Hand ausführen muss.
Das Install-/Update-Skript ruft ``alembic upgrade head`` zusätzlich auf; das ist
idempotent.
"""
from pathlib import Path
from alembic import command
from alembic.config import Config
# backend/ (eine Ebene über app/)
BACKEND_DIR = Path(__file__).resolve().parent.parent
def _config() -> Config:
cfg = Config(str(BACKEND_DIR / "alembic.ini"))
cfg.set_main_option("script_location", str(BACKEND_DIR / "migrations"))
return cfg
def upgrade_to_head() -> None:
command.upgrade(_config(), "head")
+98
View File
@@ -0,0 +1,98 @@
"""SQLAlchemy-Modelle.
Konventionen (wie in der Vorlage): ``Mapped[...] = mapped_column(...)``, Geld als
Ganzzahl in Cent/Rappen, Enums als kurze VARCHAR-Strings (``native_enum=False``,
ohne DB-CHECK) – das hält Alembic auf SQLite einfach.
Phase 0 umfasst nur Benutzer, Sitzungen und Benutzer-Einstellungen. Fachliche
Tabellen (Konten, Kategorien, Transaktionen, Budget …) kommen in späteren Phasen
hinzu und werden dann alle über ``owner_id`` je Benutzer getrennt.
"""
from __future__ import annotations
import enum
from datetime import datetime, timezone
from sqlalchemy import (
Boolean,
DateTime,
Enum as SAEnum,
ForeignKey,
Integer,
String,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from .database import Base
def _now() -> datetime:
return datetime.now(timezone.utc)
def _enum(enum_cls: type[enum.Enum]):
"""Enum als VARCHAR ohne DB-CHECK – Alembic-freundlich auf SQLite."""
return SAEnum(enum_cls, native_enum=False, create_constraint=False, length=32)
class Role(str, enum.Enum):
admin = "admin"
user = "user"
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
username: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
password_hash: Mapped[str] = mapped_column(String(255), nullable=False)
role: Mapped[Role] = mapped_column(_enum(Role), nullable=False, default=Role.user)
display_name: Mapped[str | None] = mapped_column(String(120), nullable=True)
is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
# Vom Admin mit Temporär-Passwort angelegt: beim ersten Login ändern lassen.
must_change_password: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
sessions: Mapped[list[Session]] = relationship(
back_populates="user", cascade="all, delete-orphan"
)
settings: Mapped[list[UserSetting]] = relationship(
back_populates="user", cascade="all, delete-orphan"
)
class Session(Base):
"""Server-seitige Sitzung. Im Cookie liegt nur das Klartext-Token, hier der Hash."""
__tablename__ = "sessions"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
user_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
)
token_hash: Mapped[str] = mapped_column(String(64), unique=True, nullable=False, index=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
last_seen_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
user_agent: Mapped[str | None] = mapped_column(String(255), nullable=True)
user: Mapped[User] = relationship(back_populates="sessions")
class UserSetting(Base):
"""Freie Schlüssel-Wert-Einstellungen je Benutzer.
Bekannte Schlüssel: ``base_currency`` (EUR|CHF), ``budget_start_day`` (int),
``locale`` (de|en), ``theme`` (system|light|dark), ``date_format``.
"""
__tablename__ = "user_settings"
user_id: Mapped[int] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"), primary_key=True
)
key: Mapped[str] = mapped_column(String(64), primary_key=True)
value: Mapped[str | None] = mapped_column(String(255), nullable=True)
user: Mapped[User] = relationship(back_populates="settings")
View File
+82
View File
@@ -0,0 +1,82 @@
"""Anmeldung über server-seitige Sitzung im Cookie."""
from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from ..config import get_settings
from ..database import get_db
from ..deps import get_current_user
from ..models import User
from ..schemas import PasswordChange, UserOut
from ..security import hash_password, verify_password
from ..services import sessions
router = APIRouter(prefix="/auth", tags=["auth"])
def _set_session_cookie(response: Response, raw_token: str) -> None:
settings = get_settings()
response.set_cookie(
key=settings.cookie_name,
value=raw_token,
max_age=settings.session_ttl_days * 86400,
httponly=True,
secure=settings.cookie_secure,
samesite=settings.cookie_samesite,
path="/",
)
@router.post("/login", response_model=UserOut)
def login(
response: Response,
request: Request,
form: OAuth2PasswordRequestForm = Depends(),
db: Session = Depends(get_db),
) -> User:
user = db.query(User).filter(User.username == form.username).first()
if user is None or not verify_password(form.password, user.password_hash):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Benutzername oder Passwort falsch",
)
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Dieses Konto ist deaktiviert",
)
raw = sessions.create_session(db, user, request.headers.get("user-agent"))
_set_session_cookie(response, raw)
return user
@router.post("/logout", status_code=status.HTTP_204_NO_CONTENT)
def logout(request: Request, response: Response, db: Session = Depends(get_db)) -> None:
token = request.cookies.get(get_settings().cookie_name, "")
if token:
sessions.revoke(db, token)
response.delete_cookie(get_settings().cookie_name, path="/")
@router.get("/me", response_model=UserOut)
def me(current_user: User = Depends(get_current_user)) -> User:
return current_user
@router.post("/change-password", response_model=UserOut)
def change_password(
payload: PasswordChange,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db),
) -> User:
if not verify_password(payload.current_password, current_user.password_hash):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Aktuelles Passwort ist falsch",
)
current_user.password_hash = hash_password(payload.new_password)
current_user.must_change_password = False
db.commit()
db.refresh(current_user)
return current_user
+67
View File
@@ -0,0 +1,67 @@
"""Anzeige-/Verhaltens-Einstellungen je Benutzer (Schlüssel-Wert).
Jeder Benutzer hat seine eigenen Einstellungen. Unbekannte Schlüssel werden
ebenfalls gespeichert; die hier bekannten liefern sinnvolle Vorgaben.
"""
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from ..config import get_settings
from ..database import get_db
from ..deps import get_current_user
from ..models import User, UserSetting
from ..schemas import SettingOut
router = APIRouter(prefix="/settings", tags=["settings"])
BASE_CURRENCY_KEY = "base_currency"
BUDGET_START_DAY_KEY = "budget_start_day"
LOCALE_KEY = "locale"
THEME_KEY = "theme"
DATE_FORMAT_KEY = "date_format"
def known_defaults() -> dict[str, str]:
return {
BASE_CURRENCY_KEY: get_settings().default_base_currency,
BUDGET_START_DAY_KEY: "1",
LOCALE_KEY: "de",
THEME_KEY: "system",
DATE_FORMAT_KEY: "de",
}
def get_user_setting(db: Session, user: User, key: str) -> str | None:
row = db.get(UserSetting, (user.id, key))
if row is not None and row.value is not None:
return row.value
return known_defaults().get(key)
@router.get("", response_model=list[SettingOut])
def list_settings(
db: Session = Depends(get_db), user: User = Depends(get_current_user)
) -> list[SettingOut]:
rows = db.query(UserSetting).filter(UserSetting.user_id == user.id).all()
merged: dict[str, str | None] = dict(known_defaults())
for r in rows:
merged[r.key] = r.value
return [SettingOut(key=k, value=v) for k, v in merged.items()]
@router.put("/{key}", response_model=SettingOut)
def set_setting(
key: str,
value: str,
db: Session = Depends(get_db),
user: User = Depends(get_current_user),
) -> SettingOut:
row = db.get(UserSetting, (user.id, key))
if row is None:
row = UserSetting(user_id=user.id, key=key, value=value)
db.add(row)
else:
row.value = value
db.commit()
return SettingOut(key=key, value=value)
+78
View File
@@ -0,0 +1,78 @@
"""Benutzerverwaltung (nur Admin). Neue Nutzer bekommen ein Temporär-Passwort."""
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from ..database import get_db
from ..deps import require_admin
from ..models import Role, User
from ..schemas import UserCreate, UserOut, UserUpdate
from ..security import hash_password
router = APIRouter(prefix="/users", tags=["users"], dependencies=[Depends(require_admin)])
@router.get("", response_model=list[UserOut])
def list_users(db: Session = Depends(get_db)) -> list[User]:
return db.query(User).order_by(User.username).all()
@router.post("", response_model=UserOut, status_code=status.HTTP_201_CREATED)
def create_user(payload: UserCreate, db: Session = Depends(get_db)) -> User:
if db.query(User).filter(User.username == payload.username).first():
raise HTTPException(status.HTTP_409_CONFLICT, "Benutzername existiert bereits")
user = User(
username=payload.username,
password_hash=hash_password(payload.password),
role=payload.role,
display_name=payload.display_name,
is_active=True,
# Vom Admin gesetztes Temporär-Passwort: beim ersten Login ändern lassen.
must_change_password=True,
)
db.add(user)
db.commit()
db.refresh(user)
return user
@router.patch("/{user_id}", response_model=UserOut)
def update_user(user_id: int, payload: UserUpdate, db: Session = Depends(get_db)) -> User:
user = db.get(User, user_id)
if user is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "Benutzer nicht gefunden")
if payload.password is not None:
user.password_hash = hash_password(payload.password)
# Vom Admin gesetztes Passwort gilt als temporär.
user.must_change_password = True
if payload.role is not None:
user.role = payload.role
if payload.display_name is not None:
user.display_name = payload.display_name
if payload.is_active is not None:
user.is_active = payload.is_active
db.commit()
db.refresh(user)
return user
@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(
user_id: int,
db: Session = Depends(get_db),
current_admin: User = Depends(require_admin),
) -> None:
user = db.get(User, user_id)
if user is None:
raise HTTPException(status.HTTP_404_NOT_FOUND, "Benutzer nicht gefunden")
if user.id == current_admin.id:
raise HTTPException(status.HTTP_400_BAD_REQUEST, "Du kannst dich nicht selbst löschen")
if user.role == Role.admin:
admin_count = db.query(User).filter(User.role == Role.admin).count()
if admin_count <= 1:
raise HTTPException(
status.HTTP_400_BAD_REQUEST,
"Der letzte Administrator kann nicht gelöscht werden",
)
db.delete(user)
db.commit()
+43
View File
@@ -0,0 +1,43 @@
"""Pydantic-Schemas für Anfragen/Antworten (Phase 0: Auth, Benutzer, Einstellungen)."""
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
from .models import Role
class UserOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
username: str
role: Role
display_name: str | None = None
is_active: bool
must_change_password: bool
created_at: datetime
class UserCreate(BaseModel):
username: str = Field(min_length=1, max_length=64)
password: str = Field(min_length=1, max_length=1024)
role: Role = Role.user
display_name: str | None = Field(default=None, max_length=120)
class UserUpdate(BaseModel):
password: str | None = Field(default=None, min_length=1, max_length=1024)
role: Role | None = None
display_name: str | None = Field(default=None, max_length=120)
is_active: bool | None = None
class PasswordChange(BaseModel):
current_password: str = Field(min_length=1, max_length=1024)
new_password: str = Field(min_length=1, max_length=1024)
class SettingOut(BaseModel):
key: str
value: str | None = None
+43
View File
@@ -0,0 +1,43 @@
"""Passwort-Hashing (argon2id) und Token-Helfer für Sitzungen.
argon2id ist der aktuelle Empfehlungsstandard und hat – anders als bcrypt – keine
72-Byte-Grenze. Die Sitzungs-Tokens selbst werden nur als SHA-256-Hash gespeichert,
damit ein Blick in die Datenbank keine gültigen Cookies verrät.
"""
import hashlib
import secrets
from argon2 import PasswordHasher
from argon2.exceptions import VerifyMismatchError, VerificationError, InvalidHashError
_hasher = PasswordHasher()
def hash_password(password: str) -> str:
return _hasher.hash(password)
def verify_password(password: str, password_hash: str) -> bool:
try:
return _hasher.verify(password_hash, password)
except (VerifyMismatchError, VerificationError, InvalidHashError, TypeError):
return False
def needs_rehash(password_hash: str) -> bool:
"""True, wenn der Hash mit veralteten Parametern erzeugt wurde."""
try:
return _hasher.check_needs_rehash(password_hash)
except (InvalidHashError, TypeError):
return False
def generate_session_token() -> str:
"""Zufälliges, langlebiges Sitzungs-Token (Klartext, nur im Cookie)."""
return secrets.token_urlsafe(32)
def hash_session_token(token: str) -> str:
"""SHA-256 – schnell genug, um bei jeder Anfrage geprüft zu werden."""
return hashlib.sha256(token.encode("utf-8")).hexdigest()
+26
View File
@@ -0,0 +1,26 @@
"""Startup-Seeds. Phase 0: nur der erste Admin-Benutzer.
Fachliche Startdaten (Standard-Kategorien usw.) folgen in späteren Phasen.
"""
from sqlalchemy.orm import Session
from .config import get_settings
from .models import Role, User
from .security import hash_password
def ensure_first_admin(db: Session) -> None:
"""Legt einen Admin an, solange noch gar kein Benutzer existiert."""
if db.query(User).first() is not None:
return
settings = get_settings()
admin = User(
username=settings.admin_username,
password_hash=hash_password(settings.admin_password),
role=Role.admin,
is_active=True,
must_change_password=False,
)
db.add(admin)
db.commit()
View File
+78
View File
@@ -0,0 +1,78 @@
"""Server-seitige Sitzungen: anlegen, prüfen, widerrufen.
Im Cookie liegt nur das Klartext-Token; in der Datenbank nur dessen SHA-256-Hash.
So lässt sich eine Sitzung serverseitig beenden (Logout, „überall abmelden"),
und ein Blick in die DB verrät kein gültiges Cookie.
"""
from datetime import datetime, timedelta, timezone
from sqlalchemy.orm import Session as DBSession
from ..config import get_settings
from ..models import Session as UserSession, User
from ..security import generate_session_token, hash_session_token
def _now() -> datetime:
return datetime.now(timezone.utc)
def _aware(dt: datetime) -> datetime:
"""SQLite gibt datetimes teils ohne tzinfo zurück – als UTC interpretieren."""
return dt if dt.tzinfo is not None else dt.replace(tzinfo=timezone.utc)
def create_session(db: DBSession, user: User, user_agent: str | None = None) -> str:
"""Legt eine Sitzung an und gibt das Klartext-Token zurück (nur hier sichtbar)."""
settings = get_settings()
raw = generate_session_token()
entry = UserSession(
user_id=user.id,
token_hash=hash_session_token(raw),
expires_at=_now() + timedelta(days=settings.session_ttl_days),
last_seen_at=_now(),
user_agent=(user_agent or "")[:255] or None,
)
db.add(entry)
db.commit()
return raw
def validate(db: DBSession, raw_token: str) -> User | None:
"""Gibt den Benutzer zurück, wenn das Token gültig und nicht abgelaufen ist."""
if not raw_token:
return None
entry = (
db.query(UserSession)
.filter(UserSession.token_hash == hash_session_token(raw_token))
.first()
)
if entry is None:
return None
if _aware(entry.expires_at) < _now():
db.delete(entry)
db.commit()
return None
user = db.get(User, entry.user_id)
if user is None or not user.is_active:
return None
entry.last_seen_at = _now()
db.commit()
return user
def revoke(db: DBSession, raw_token: str) -> None:
entry = (
db.query(UserSession)
.filter(UserSession.token_hash == hash_session_token(raw_token))
.first()
)
if entry is not None:
db.delete(entry)
db.commit()
def revoke_all(db: DBSession, user: User) -> None:
db.query(UserSession).filter(UserSession.user_id == user.id).delete()
db.commit()
+50
View File
@@ -0,0 +1,50 @@
"""Alembic-Umgebung.
Nutzt die Engine und die Metadaten der Anwendung (SQLite), damit Migrationen und
Laufzeit garantiert dieselbe Datenbank/dasselbe Schema verwenden. ``render_as_batch``
ist für SQLite Pflicht, damit ALTER-TABLE über Tabellen-Kopien läuft.
"""
from logging.config import fileConfig
from alembic import context
from app.database import Base, engine
from app import models # noqa: F401 – registriert alle Tabellen an Base.metadata
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
target_metadata = Base.metadata
def run_migrations_offline() -> None:
context.configure(
url=str(engine.url),
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
render_as_batch=True,
compare_type=True,
)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online() -> None:
with engine.connect() as connection:
context.configure(
connection=connection,
target_metadata=target_metadata,
render_as_batch=True,
compare_type=True,
)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()
+23
View File
@@ -0,0 +1,23 @@
"""${message}
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
from alembic import op
import sqlalchemy as sa
${imports if imports else ""}
# revision identifiers, used by Alembic.
revision = ${repr(up_revision)}
down_revision = ${repr(down_revision)}
branch_labels = ${repr(branch_labels)}
depends_on = ${repr(depends_on)}
def upgrade() -> None:
${upgrades if upgrades else "pass"}
def downgrade() -> None:
${downgrades if downgrades else "pass"}
@@ -0,0 +1,71 @@
"""Initiales Schema: Benutzer, Sitzungen, Benutzer-Einstellungen.
Revision ID: 0001
Revises:
Create Date: 2026-08-19
"""
from alembic import op
import sqlalchemy as sa
revision = "0001"
down_revision = None
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
"users",
sa.Column("id", sa.Integer(), primary_key=True),
sa.Column("username", sa.String(length=64), nullable=False),
sa.Column("password_hash", sa.String(length=255), nullable=False),
sa.Column("role", sa.String(length=32), nullable=False),
sa.Column("display_name", sa.String(length=120), nullable=True),
sa.Column("is_active", sa.Boolean(), nullable=False, server_default=sa.text("1")),
sa.Column(
"must_change_password",
sa.Boolean(),
nullable=False,
server_default=sa.text("0"),
),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=True),
sa.UniqueConstraint("username", name="uq_users_username"),
)
op.create_table(
"sessions",
sa.Column("id", sa.Integer(), primary_key=True),
sa.Column(
"user_id",
sa.Integer(),
sa.ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column("token_hash", sa.String(length=64), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("user_agent", sa.String(length=255), nullable=True),
)
op.create_index("ix_sessions_user_id", "sessions", ["user_id"])
op.create_index("ix_sessions_token_hash", "sessions", ["token_hash"], unique=True)
op.create_table(
"user_settings",
sa.Column(
"user_id",
sa.Integer(),
sa.ForeignKey("users.id", ondelete="CASCADE"),
primary_key=True,
),
sa.Column("key", sa.String(length=64), primary_key=True),
sa.Column("value", sa.String(length=255), nullable=True),
)
def downgrade() -> None:
op.drop_table("user_settings")
op.drop_index("ix_sessions_token_hash", table_name="sessions")
op.drop_index("ix_sessions_user_id", table_name="sessions")
op.drop_table("sessions")
op.drop_table("users")
+11
View File
@@ -0,0 +1,11 @@
fastapi==0.115.6
uvicorn[standard]==0.34.0
SQLAlchemy==2.0.36
alembic==1.14.0
pydantic==2.10.4
pydantic-settings==2.7.1
argon2-cffi==23.1.0
python-multipart==0.0.20
httpx==0.28.1
python-dateutil==2.9.0.post0
pytest==8.3.4