feat: 2FA, Passwortrichtlinie und wiederherstellbares JSON-Backup
Zwei-Faktor per TOTP mit zehn Einmal-Wiederherstellungscodes. Der Zwischenzustand "Passwort stimmt, Code fehlt" haengt an der vorhandenen Sitzungstabelle (pending_2fa, zehn Minuten Laufzeit) und gilt nirgends als angemeldet; nach dem Code wird das Cookie getauscht statt weiterverwendet. Der Wiederverwendungsschutz merkt sich das tatsaechlich getroffene Zeitfenster, nicht die aktuelle Uhrzeit - sonst liesse sich ein Code aus dem naechsten Fenster gleich darauf nochmal einloesen. Passwortrichtlinie (Laenge, Zeichenklassen, MFA-Pflicht) in der neuen globalen Tabelle app_settings, durchgesetzt beim Anlegen, Zuruecksetzen und Aendern von Passwoertern. Bestehende Passwoerter lassen sich nicht gegen eine neue Richtlinie pruefen - der Hash gibt nichts her. Beim Anmelden liegt das Passwort aber kurz im Klartext vor: dort wird geprueft und bei Verstoss ein Wechsel verlangt, mit passendem Grund fuer den Hinweistext. Nebenbei geschlossen: must_change_password war bisher nur im Frontend erzwungen, wer die API direkt ansprach kam vorbei. enforce_account_state haengt jetzt am /api-Sammelrouter und laesst nur noch durch, was man zum Erfuellen der Pflicht braucht. Das JSON-Backup liess sich bisher grundsaetzlich nicht zurueckladen: keine Version, Kategorien nur ueber mehrdeutige Namen, sieben Tabellen fehlten. Daneben steht nun ein Format v2, das Beziehungen ueber lokale IDs abbildet, je Benutzer oder systemweit. Zurueckgeladen wird nach einer Pruefansicht, wahlweise ersetzend oder ergaenzend; vor dem Ersetzen legt der Server einen Schnappschuss der Datenbank an. Der alte Ansichts-Export bleibt fuer Tabellenkalkulationen bestehen und wird beim Hochladen ausdruecklich abgewiesen, statt halb zu raten. Frontend: wiederverwendbares Passwortfeld mit Augensymbol, animierte Anforderungsliste, 2FA-Seite, Admin-Karte fuer die Richtlinie und Karte zum Zurueckladen. Neue Bewegung ist in prefers-reduced-motion hinterlegt; Secret und Codes sind markierbar, sonst verhinderte das globale user-select das Kopieren. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Scarriffle
co-authored by
Claude Opus 5
Claude Opus 5
parent
1d5b23c127
commit
7fc8763d57
+71
-1
@@ -3,6 +3,11 @@
|
||||
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.
|
||||
|
||||
``enforce_account_state`` hängt am zentralen ``/api``-Router und sperrt Benutzer
|
||||
aus, die erst noch etwas erledigen müssen (Passwort wechseln, zweiten Faktor
|
||||
einrichten). Bisher wurde das nur im Frontend umgeleitet – wer die API direkt
|
||||
ansprach, kam daran vorbei.
|
||||
"""
|
||||
|
||||
from fastapi import Depends, HTTPException, Request, status
|
||||
@@ -11,6 +16,8 @@ from sqlalchemy.orm import Session
|
||||
from .config import get_settings
|
||||
from .database import get_db
|
||||
from .models import Role, User
|
||||
from .schemas import UserOut
|
||||
from .services import policy as policy_service
|
||||
from .services import sessions
|
||||
|
||||
_credentials_exc = HTTPException(
|
||||
@@ -18,15 +25,71 @@ _credentials_exc = HTTPException(
|
||||
detail="Nicht angemeldet oder Sitzung abgelaufen",
|
||||
)
|
||||
|
||||
# Pfade, die trotz offener Pflicht erreichbar bleiben müssen – sonst könnte
|
||||
# niemand die Pflicht je erfüllen. Präfix-Vergleich, ohne führendes /api.
|
||||
_ALWAYS_ALLOWED = (
|
||||
"/health",
|
||||
"/version",
|
||||
"/auth/me",
|
||||
"/auth/logout",
|
||||
"/auth/login",
|
||||
"/auth/change-password",
|
||||
"/auth/2fa",
|
||||
)
|
||||
# Die Richtlinie darf gelesen werden (die Checkliste braucht sie), aber ein
|
||||
# Admin mit offener Pflicht soll sie nicht erst einmal entschärfen können.
|
||||
_READ_ONLY_ALLOWED = ("/policy",)
|
||||
|
||||
def get_current_user(request: Request, db: Session = Depends(get_db)) -> User:
|
||||
|
||||
def _session_user(request: Request, db: Session) -> User | None:
|
||||
"""Validiert die Sitzung genau einmal pro Anfrage und merkt sie sich.
|
||||
|
||||
``enforce_account_state`` läuft vor den Endpunkt-Abhängigkeiten; ohne diesen
|
||||
Zwischenspeicher würde ``sessions.validate`` (und damit der Schreibzugriff
|
||||
auf ``last_seen_at``) pro Anfrage zweimal laufen."""
|
||||
if hasattr(request.state, "session_user"):
|
||||
return request.state.session_user
|
||||
token = request.cookies.get(get_settings().cookie_name, "")
|
||||
user = sessions.validate(db, token)
|
||||
request.state.session_user = user
|
||||
return user
|
||||
|
||||
|
||||
def get_current_user(request: Request, db: Session = Depends(get_db)) -> User:
|
||||
user = _session_user(request, db)
|
||||
if user is None:
|
||||
raise _credentials_exc
|
||||
return user
|
||||
|
||||
|
||||
def pending_reason(db: Session, user: User) -> str | None:
|
||||
"""Was der Benutzer erledigen muss, bevor er die App benutzen darf."""
|
||||
if user.must_change_password:
|
||||
return "password_change_required"
|
||||
if policy_service.load(db).mfa_required and not user.totp_enabled:
|
||||
return "mfa_setup_required"
|
||||
return None
|
||||
|
||||
|
||||
def enforce_account_state(request: Request, db: Session = Depends(get_db)) -> None:
|
||||
"""Router-weite Sperre für Konten mit offener Pflicht (403, sonst Durchlass).
|
||||
|
||||
Nicht angemeldete Anfragen laufen hier ungehindert durch – um die kümmert
|
||||
sich ``get_current_user`` am jeweiligen Endpunkt."""
|
||||
pfad = request.url.path
|
||||
rest = pfad[4:] if pfad.startswith("/api") else pfad
|
||||
if rest.startswith(_ALWAYS_ALLOWED):
|
||||
return
|
||||
if request.method == "GET" and rest.startswith(_READ_ONLY_ALLOWED):
|
||||
return
|
||||
user = _session_user(request, db)
|
||||
if user is None:
|
||||
return
|
||||
grund = pending_reason(db, user)
|
||||
if grund is not None:
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=grund)
|
||||
|
||||
|
||||
def require_admin(current_user: User = Depends(get_current_user)) -> User:
|
||||
if current_user.role != Role.admin:
|
||||
raise HTTPException(
|
||||
@@ -36,6 +99,13 @@ def require_admin(current_user: User = Depends(get_current_user)) -> User:
|
||||
return current_user
|
||||
|
||||
|
||||
def user_out(db: Session, user: User) -> UserOut:
|
||||
"""``UserOut`` inklusive des berechneten ``mfa_setup_required``."""
|
||||
out = UserOut.model_validate(user)
|
||||
out.mfa_setup_required = policy_service.load(db).mfa_required and not user.totp_enabled
|
||||
return out
|
||||
|
||||
|
||||
def get_owned(db: Session, model, obj_id: int, user: User):
|
||||
"""Holt eine Zeile und stellt sicher, dass sie dem Nutzer gehört – sonst 404.
|
||||
|
||||
|
||||
+8
-2
@@ -9,7 +9,7 @@ import threading
|
||||
from contextlib import asynccontextmanager
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import APIRouter, FastAPI, HTTPException
|
||||
from fastapi import APIRouter, Depends, FastAPI, HTTPException
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.responses import FileResponse
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
@@ -17,6 +17,7 @@ from fastapi.staticfiles import StaticFiles
|
||||
from .branding import APP_NAME, VERSION, git_sha
|
||||
from .config import get_settings
|
||||
from .database import SessionLocal
|
||||
from .deps import enforce_account_state
|
||||
from .migrate import upgrade_to_head
|
||||
from .routers import (
|
||||
accounts,
|
||||
@@ -27,6 +28,7 @@ from .routers import (
|
||||
dashboard,
|
||||
documents,
|
||||
imports,
|
||||
policy,
|
||||
rates,
|
||||
settings as settings_router,
|
||||
transactions,
|
||||
@@ -85,7 +87,10 @@ if settings.cors_origins.strip():
|
||||
)
|
||||
|
||||
# --- API -------------------------------------------------------------------
|
||||
api = APIRouter(prefix="/api")
|
||||
# Die Abhängigkeit hängt am Sammel-Router, nicht an jedem einzelnen Endpunkt:
|
||||
# Konten mit offener Pflicht (Passwortwechsel, 2FA-Einrichtung) kommen so an
|
||||
# keiner Stelle vorbei, auch nicht an neu hinzukommenden.
|
||||
api = APIRouter(prefix="/api", dependencies=[Depends(enforce_account_state)])
|
||||
|
||||
|
||||
@api.get("/health", tags=["meta"])
|
||||
@@ -101,6 +106,7 @@ def version() -> dict:
|
||||
api.include_router(auth.router)
|
||||
api.include_router(users.router)
|
||||
api.include_router(settings_router.router)
|
||||
api.include_router(policy.router)
|
||||
api.include_router(accounts.router)
|
||||
api.include_router(categories.router)
|
||||
api.include_router(transactions.router)
|
||||
|
||||
@@ -119,14 +119,33 @@ class User(Base):
|
||||
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)
|
||||
# Warum der Wechsel fällig ist: "temp" (Admin-Passwort) oder "policy"
|
||||
# (Richtlinie wurde verschärft). Steuert nur den Hinweistext.
|
||||
must_change_password_reason: Mapped[str | None] = mapped_column(String(16), nullable=True)
|
||||
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
|
||||
|
||||
# --- Zwei-Faktor (TOTP) ---
|
||||
# Das Secret liegt im Klartext. Verschlüsseln brächte hier nichts: der
|
||||
# Schlüssel läge auf derselben Maschine wie die Datenbank. Wer die DB-Datei
|
||||
# hat, hätte dann auch den Schlüssel. Schutz kommt aus Dateirechten.
|
||||
totp_secret: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
||||
totp_enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
|
||||
totp_confirmed_at: Mapped[datetime | None] = mapped_column(
|
||||
DateTime(timezone=True), nullable=True
|
||||
)
|
||||
# Zuletzt verbrauchtes 30-Sekunden-Zeitfenster – verhindert, dass derselbe
|
||||
# Code innerhalb seiner Gültigkeit ein zweites Mal benutzt wird.
|
||||
totp_last_step: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
||||
|
||||
sessions: Mapped[list[Session]] = relationship(
|
||||
back_populates="user", cascade="all, delete-orphan"
|
||||
)
|
||||
settings: Mapped[list[UserSetting]] = relationship(
|
||||
back_populates="user", cascade="all, delete-orphan"
|
||||
)
|
||||
recovery_codes: Mapped[list[RecoveryCode]] = relationship(
|
||||
back_populates="user", cascade="all, delete-orphan"
|
||||
)
|
||||
|
||||
|
||||
class Session(Base):
|
||||
@@ -143,10 +162,46 @@ class Session(Base):
|
||||
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)
|
||||
# Passwort stimmt, der zweite Faktor fehlt noch. Solche Sitzungen gelten
|
||||
# nirgends als angemeldet (siehe services/sessions.validate) und laufen
|
||||
# nach wenigen Minuten ab.
|
||||
pending_2fa: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
|
||||
failed_2fa: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
||||
|
||||
user: Mapped[User] = relationship(back_populates="sessions")
|
||||
|
||||
|
||||
class RecoveryCode(Base):
|
||||
"""Einmal-Wiederherstellungscode für den Fall eines verlorenen Authenticators.
|
||||
|
||||
Gespeichert wird nur der SHA-256-Hash. Bewusst nicht Argon2: die Codes sind
|
||||
zufällig und hochentrop, und pro Prüfung stünden bis zu zehn Kandidaten an.
|
||||
"""
|
||||
|
||||
__tablename__ = "recovery_codes"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
||||
user_id: Mapped[int] = mapped_column(
|
||||
ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
|
||||
)
|
||||
code_hash: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
|
||||
used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
|
||||
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
|
||||
|
||||
user: Mapped[User] = relationship(back_populates="recovery_codes")
|
||||
|
||||
|
||||
class AppSetting(Base):
|
||||
"""Globale, vom Admin gepflegte Einstellungen – im Gegensatz zu
|
||||
``user_settings`` bewusst ohne ``owner_id``. Beherbergt heute die
|
||||
Passwortrichtlinie (siehe ``services/policy``)."""
|
||||
|
||||
__tablename__ = "app_settings"
|
||||
|
||||
key: Mapped[str] = mapped_column(String(64), primary_key=True)
|
||||
value: Mapped[str | None] = mapped_column(String(255), nullable=True)
|
||||
|
||||
|
||||
class UserSetting(Base):
|
||||
"""Freie Schlüssel-Wert-Einstellungen je Benutzer.
|
||||
|
||||
|
||||
+206
-14
@@ -1,4 +1,12 @@
|
||||
"""Anmeldung über server-seitige Sitzung im Cookie."""
|
||||
"""Anmeldung über server-seitige Sitzung im Cookie, inklusive Zwei-Faktor.
|
||||
|
||||
Ablauf mit aktivem TOTP: ``/login`` prüft das Passwort und legt eine Sitzung an,
|
||||
die als ``pending_2fa`` markiert ist – die gilt nirgends als angemeldet. Erst
|
||||
``/2fa/verify`` tauscht sie gegen eine echte Sitzung (neues Cookie, damit kein
|
||||
vorab untergeschobenes Token weiterlebt).
|
||||
"""
|
||||
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
|
||||
from fastapi.security import OAuth2PasswordRequestForm
|
||||
@@ -6,21 +14,32 @@ from sqlalchemy.orm import Session
|
||||
|
||||
from ..config import get_settings
|
||||
from ..database import get_db
|
||||
from ..deps import get_current_user
|
||||
from ..deps import get_current_user, user_out
|
||||
from ..models import User
|
||||
from ..schemas import PasswordChange, UserOut
|
||||
from ..security import hash_password, verify_password
|
||||
from ..services import sessions
|
||||
from ..schemas import (
|
||||
LoginOut,
|
||||
PasswordChange,
|
||||
RecoveryCodesOut,
|
||||
TwoFactorActivate,
|
||||
TwoFactorPassword,
|
||||
TwoFactorSetupOut,
|
||||
TwoFactorStatusOut,
|
||||
TwoFactorVerify,
|
||||
UserOut,
|
||||
)
|
||||
from ..security import hash_password, needs_rehash, verify_password
|
||||
from ..services import policy as policy_service
|
||||
from ..services import sessions, twofactor
|
||||
|
||||
router = APIRouter(prefix="/auth", tags=["auth"])
|
||||
|
||||
|
||||
def _set_session_cookie(response: Response, raw_token: str) -> None:
|
||||
def _set_session_cookie(response: Response, raw_token: str, max_age: int | None = None) -> None:
|
||||
settings = get_settings()
|
||||
response.set_cookie(
|
||||
key=settings.cookie_name,
|
||||
value=raw_token,
|
||||
max_age=settings.session_ttl_days * 86400,
|
||||
max_age=max_age if max_age is not None else settings.session_ttl_days * 86400,
|
||||
httponly=True,
|
||||
secure=settings.cookie_secure,
|
||||
samesite=settings.cookie_samesite,
|
||||
@@ -28,13 +47,13 @@ def _set_session_cookie(response: Response, raw_token: str) -> None:
|
||||
)
|
||||
|
||||
|
||||
@router.post("/login", response_model=UserOut)
|
||||
@router.post("/login", response_model=LoginOut)
|
||||
def login(
|
||||
response: Response,
|
||||
request: Request,
|
||||
form: OAuth2PasswordRequestForm = Depends(),
|
||||
db: Session = Depends(get_db),
|
||||
) -> User:
|
||||
) -> LoginOut:
|
||||
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(
|
||||
@@ -46,9 +65,27 @@ def login(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Dieses Konto ist deaktiviert",
|
||||
)
|
||||
|
||||
# Einziger Moment, in dem das Passwort im Klartext vorliegt: hier lässt sich
|
||||
# prüfen, ob es die (womöglich verschärfte) Richtlinie noch erfüllt. Über den
|
||||
# gespeicherten Hash ginge das nicht.
|
||||
if policy_service.violations(form.password, policy_service.load(db)):
|
||||
user.must_change_password = True
|
||||
user.must_change_password_reason = "policy"
|
||||
if needs_rehash(user.password_hash):
|
||||
user.password_hash = hash_password(form.password)
|
||||
db.commit()
|
||||
|
||||
if user.totp_enabled:
|
||||
raw = sessions.create_session(
|
||||
db, user, request.headers.get("user-agent"), pending_2fa=True
|
||||
)
|
||||
_set_session_cookie(response, raw, max_age=sessions.PENDING_TTL_MINUTES * 60)
|
||||
return LoginOut(status="2fa_required", user=None)
|
||||
|
||||
raw = sessions.create_session(db, user, request.headers.get("user-agent"))
|
||||
_set_session_cookie(response, raw)
|
||||
return user
|
||||
return LoginOut(status="ok", user=user_out(db, user))
|
||||
|
||||
|
||||
@router.post("/logout", status_code=status.HTTP_204_NO_CONTENT)
|
||||
@@ -60,23 +97,178 @@ def logout(request: Request, response: Response, db: Session = Depends(get_db))
|
||||
|
||||
|
||||
@router.get("/me", response_model=UserOut)
|
||||
def me(current_user: User = Depends(get_current_user)) -> User:
|
||||
return current_user
|
||||
def me(
|
||||
current_user: User = Depends(get_current_user), db: Session = Depends(get_db)
|
||||
) -> UserOut:
|
||||
return user_out(db, current_user)
|
||||
|
||||
|
||||
@router.post("/change-password", response_model=UserOut)
|
||||
def change_password(
|
||||
request: Request,
|
||||
response: Response,
|
||||
payload: PasswordChange,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db),
|
||||
) -> User:
|
||||
) -> UserOut:
|
||||
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",
|
||||
)
|
||||
if payload.new_password == payload.current_password:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Das neue Passwort muss sich vom bisherigen unterscheiden",
|
||||
)
|
||||
policy_service.assert_ok(db, payload.new_password)
|
||||
|
||||
current_user.password_hash = hash_password(payload.new_password)
|
||||
current_user.must_change_password = False
|
||||
current_user.must_change_password_reason = None
|
||||
db.commit()
|
||||
|
||||
# Ein Passwortwechsel soll alle übrigen Sitzungen beenden – wer das Passwort
|
||||
# kannte, soll danach draussen sein. Der aufrufende Client bekommt sofort
|
||||
# eine frische Sitzung, damit er nicht selbst herausfliegt.
|
||||
sessions.revoke_all(db, current_user)
|
||||
raw = sessions.create_session(db, current_user, request.headers.get("user-agent"))
|
||||
_set_session_cookie(response, raw)
|
||||
db.refresh(current_user)
|
||||
return current_user
|
||||
return user_out(db, current_user)
|
||||
|
||||
|
||||
# --- Zwei-Faktor ------------------------------------------------------------
|
||||
|
||||
|
||||
@router.post("/2fa/verify", response_model=LoginOut)
|
||||
def verify_two_factor(
|
||||
request: Request,
|
||||
response: Response,
|
||||
payload: TwoFactorVerify,
|
||||
db: Session = Depends(get_db),
|
||||
) -> LoginOut:
|
||||
"""Zweiter Schritt der Anmeldung: TOTP-Code oder Wiederherstellungscode."""
|
||||
token = request.cookies.get(get_settings().cookie_name, "")
|
||||
offen = sessions.validate_pending(db, token)
|
||||
if offen is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Anmeldung abgelaufen – bitte neu anmelden",
|
||||
)
|
||||
entry, user = offen
|
||||
|
||||
ok = (
|
||||
twofactor.consume_recovery_code(db, user, payload.code)
|
||||
if payload.recovery
|
||||
else twofactor.consume_code(db, user, payload.code)
|
||||
)
|
||||
if not ok:
|
||||
verbraucht = sessions.note_failed_2fa(db, entry)
|
||||
if verbraucht:
|
||||
response.delete_cookie(get_settings().cookie_name, path="/")
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Zu viele Fehlversuche – bitte neu anmelden",
|
||||
)
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED, detail="Code stimmt nicht"
|
||||
)
|
||||
|
||||
# Cookie tauschen statt weiterverwenden: die Pending-Sitzung verschwindet,
|
||||
# der Client bekommt ein frisches Token.
|
||||
db.delete(entry)
|
||||
db.commit()
|
||||
raw = sessions.create_session(db, user, request.headers.get("user-agent"))
|
||||
_set_session_cookie(response, raw)
|
||||
return LoginOut(status="ok", user=user_out(db, user))
|
||||
|
||||
|
||||
@router.get("/2fa/status", response_model=TwoFactorStatusOut)
|
||||
def two_factor_status(
|
||||
current_user: User = Depends(get_current_user), db: Session = Depends(get_db)
|
||||
) -> TwoFactorStatusOut:
|
||||
return TwoFactorStatusOut(
|
||||
enabled=current_user.totp_enabled,
|
||||
recovery_codes_left=twofactor.codes_left(db, current_user),
|
||||
)
|
||||
|
||||
|
||||
@router.post("/2fa/setup", response_model=TwoFactorSetupOut)
|
||||
def setup_two_factor(
|
||||
current_user: User = Depends(get_current_user), db: Session = Depends(get_db)
|
||||
) -> TwoFactorSetupOut:
|
||||
"""Legt ein neues, noch nicht aktives Secret an und liefert den QR-Code."""
|
||||
if current_user.totp_enabled:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Zwei-Faktor ist bereits aktiv",
|
||||
)
|
||||
secret = twofactor.new_secret()
|
||||
current_user.totp_secret = secret
|
||||
current_user.totp_last_step = None
|
||||
db.commit()
|
||||
uri = twofactor.otpauth_uri(current_user, secret)
|
||||
return TwoFactorSetupOut(secret=secret, otpauth_uri=uri, qr_png=twofactor.qr_data_uri(uri))
|
||||
|
||||
|
||||
@router.post("/2fa/activate", response_model=RecoveryCodesOut)
|
||||
def activate_two_factor(
|
||||
payload: TwoFactorActivate,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db),
|
||||
) -> RecoveryCodesOut:
|
||||
if current_user.totp_enabled:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Zwei-Faktor ist bereits aktiv",
|
||||
)
|
||||
if not current_user.totp_secret:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Zuerst die Einrichtung starten",
|
||||
)
|
||||
if not twofactor.consume_code(db, current_user, payload.code):
|
||||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Code stimmt nicht")
|
||||
|
||||
current_user.totp_enabled = True
|
||||
current_user.totp_confirmed_at = datetime.now(timezone.utc)
|
||||
db.commit()
|
||||
return RecoveryCodesOut(codes=twofactor.generate_recovery_codes(db, current_user))
|
||||
|
||||
|
||||
@router.post("/2fa/recovery-codes", response_model=RecoveryCodesOut)
|
||||
def new_recovery_codes(
|
||||
payload: TwoFactorPassword,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db),
|
||||
) -> RecoveryCodesOut:
|
||||
if not current_user.totp_enabled:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST, detail="Zwei-Faktor ist nicht aktiv"
|
||||
)
|
||||
if not verify_password(payload.password, current_user.password_hash):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST, detail="Passwort ist falsch"
|
||||
)
|
||||
return RecoveryCodesOut(codes=twofactor.generate_recovery_codes(db, current_user))
|
||||
|
||||
|
||||
@router.post("/2fa/disable", response_model=UserOut)
|
||||
def disable_two_factor(
|
||||
payload: TwoFactorPassword,
|
||||
current_user: User = Depends(get_current_user),
|
||||
db: Session = Depends(get_db),
|
||||
) -> UserOut:
|
||||
if not verify_password(payload.password, current_user.password_hash):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST, detail="Passwort ist falsch"
|
||||
)
|
||||
if policy_service.load(db).mfa_required:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Die Richtlinie verlangt Zwei-Faktor – Abschalten ist nicht möglich",
|
||||
)
|
||||
twofactor.disable(db, current_user)
|
||||
db.refresh(current_user)
|
||||
return user_out(db, current_user)
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
"""Backup & Export: kompletter DB-Download (Admin) und Datenexport je Nutzer."""
|
||||
"""Backup & Export: kompletter DB-Download (Admin), Datenexport je Nutzer und
|
||||
das wiederherstellbare Voll-Backup als JSON (siehe ``services/backup_data``)."""
|
||||
|
||||
import csv
|
||||
import io
|
||||
@@ -6,8 +7,9 @@ import os
|
||||
import sqlite3
|
||||
import tempfile
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Response
|
||||
from fastapi import APIRouter, Depends, File, Form, HTTPException, Response, UploadFile
|
||||
from fastapi.responses import FileResponse
|
||||
from sqlalchemy.orm import Session
|
||||
from starlette.background import BackgroundTask
|
||||
@@ -23,17 +25,49 @@ from ..models import (
|
||||
CategoryGroup,
|
||||
DashboardLayout,
|
||||
PlannedIncomeLine,
|
||||
Role,
|
||||
Transaction,
|
||||
User,
|
||||
)
|
||||
from ..services import backup_data, sessions
|
||||
|
||||
router = APIRouter(tags=["backup"])
|
||||
|
||||
# Ein Backup ist reiner Text; alles darüber ist mit Sicherheit keins mehr.
|
||||
MAX_BACKUP_BYTES = 64 * 1024 * 1024
|
||||
|
||||
|
||||
def _ts() -> str:
|
||||
return datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
|
||||
|
||||
|
||||
def _sqlite_path() -> str | None:
|
||||
url = get_settings().effective_database_url
|
||||
return url.replace("sqlite:///", "", 1) if url.startswith("sqlite:///") else None
|
||||
|
||||
|
||||
def _snapshot_before_restore() -> str | None:
|
||||
"""Legt vor einem ersetzenden Restore eine Kopie der Datenbank ab.
|
||||
|
||||
Dieselbe Absicherung, die ``deploy/update.sh`` vor einem Update macht: wer
|
||||
versehentlich das falsche Backup einspielt, hat noch einen Weg zurück."""
|
||||
db_path = _sqlite_path()
|
||||
if db_path is None or not os.path.exists(db_path):
|
||||
return None
|
||||
ziel_ordner = Path(get_settings().data_dir) / "backups"
|
||||
ziel_ordner.mkdir(parents=True, exist_ok=True)
|
||||
ziel = ziel_ordner / f"pre-restore-{_ts()}.db"
|
||||
src = sqlite3.connect(db_path)
|
||||
dst = sqlite3.connect(str(ziel))
|
||||
try:
|
||||
with dst:
|
||||
src.backup(dst)
|
||||
finally:
|
||||
dst.close()
|
||||
src.close()
|
||||
return str(ziel)
|
||||
|
||||
|
||||
def _money(cents: int | None) -> str:
|
||||
return f"{(cents or 0) / 100:.2f}"
|
||||
|
||||
@@ -43,10 +77,9 @@ def _money(cents: int | None) -> str:
|
||||
|
||||
@router.get("/backup/db")
|
||||
def backup_db(_: User = Depends(require_admin)):
|
||||
url = get_settings().effective_database_url
|
||||
if not url.startswith("sqlite:///"):
|
||||
db_path = _sqlite_path()
|
||||
if db_path is None:
|
||||
raise HTTPException(status_code=400, detail="DB-Backup ist nur für SQLite verfügbar.")
|
||||
db_path = url.replace("sqlite:///", "", 1)
|
||||
tmp = tempfile.NamedTemporaryFile(prefix="cf-backup-", suffix=".db", delete=False)
|
||||
tmp.close()
|
||||
# Online-Backup-API: konsistente Kopie auch bei laufendem Betrieb.
|
||||
@@ -198,3 +231,131 @@ def export_all_json(db: Session = Depends(get_db), user: User = Depends(get_curr
|
||||
media_type="application/json",
|
||||
headers={"Content-Disposition": f'attachment; filename="certain-finance-export-{_ts()}.json"'},
|
||||
)
|
||||
|
||||
|
||||
# --- Voll-Backup (wiederherstellbar) ---------------------------------------
|
||||
|
||||
|
||||
def _download(obj: dict, name: str) -> Response:
|
||||
return Response(
|
||||
content=backup_data.to_bytes(obj),
|
||||
media_type="application/json",
|
||||
headers={"Content-Disposition": f'attachment; filename="{name}-{_ts()}.json"'},
|
||||
)
|
||||
|
||||
|
||||
@router.get("/export/backup.json")
|
||||
def export_backup(db: Session = Depends(get_db), user: User = Depends(get_current_user)):
|
||||
"""Alle eigenen Daten – vollständig und wieder einspielbar."""
|
||||
return _download(backup_data.export_user(db, user), "certain-finance-backup")
|
||||
|
||||
|
||||
@router.get("/export/system-backup.json")
|
||||
def export_system_backup(db: Session = Depends(get_db), _: User = Depends(require_admin)):
|
||||
"""Die ganze Instanz inklusive Benutzerkonten. Enthält Passwort-Hashes und
|
||||
TOTP-Secrets – die Datei ist wie ein Passwort zu behandeln."""
|
||||
return _download(backup_data.export_system(db), "certain-finance-system-backup")
|
||||
|
||||
|
||||
async def _read_upload(file: UploadFile) -> bytes:
|
||||
roh = await file.read()
|
||||
if len(roh) > MAX_BACKUP_BYTES:
|
||||
raise HTTPException(status_code=400, detail="Die Datei ist zu gross für ein Backup.")
|
||||
return roh
|
||||
|
||||
|
||||
@router.post("/backup/restore/preview")
|
||||
async def restore_preview(
|
||||
file: UploadFile = File(...),
|
||||
db: Session = Depends(get_db),
|
||||
user: User = Depends(get_current_user),
|
||||
):
|
||||
"""Prüft die Datei und meldet, was darin steckt. Schreibt nichts."""
|
||||
try:
|
||||
data = backup_data.parse(await _read_upload(file))
|
||||
except backup_data.BackupError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||||
|
||||
scope = data["scope"]
|
||||
if scope == "system" and user.role != Role.admin:
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail="Ein systemweites Backup darf nur ein Administrator einspielen.",
|
||||
)
|
||||
hinweise: list[str] = []
|
||||
if scope == "system":
|
||||
hinweise.append(
|
||||
"Systemweite Wiederherstellung: alle Benutzer und deren Daten werden "
|
||||
"ersetzt. Danach sind alle Sitzungen beendet und du musst dich neu anmelden."
|
||||
)
|
||||
if not any(u.get("role") == "admin" for u in data.get("users") or []):
|
||||
hinweise.append("Achtung: im Backup ist kein Administrator enthalten.")
|
||||
return {
|
||||
"scope": scope,
|
||||
"exported_at": data.get("exported_at"),
|
||||
"app_version": data.get("app_version"),
|
||||
"username": data.get("username"),
|
||||
"user_count": len(data.get("users") or []) if scope == "system" else 1,
|
||||
"tables": backup_data.summarize(data),
|
||||
"notes": hinweise,
|
||||
}
|
||||
|
||||
|
||||
@router.post("/backup/restore")
|
||||
async def restore(
|
||||
file: UploadFile = File(...),
|
||||
mode: str = Form("replace"),
|
||||
db: Session = Depends(get_db),
|
||||
user: User = Depends(get_current_user),
|
||||
):
|
||||
"""Spielt ein Backup ein – ``replace`` (ersetzen) oder ``merge`` (ergänzen).
|
||||
|
||||
Alles läuft in einer Transaktion: entweder es geht komplett durch oder es
|
||||
wird gar nichts geschrieben."""
|
||||
if mode not in ("replace", "merge"):
|
||||
raise HTTPException(status_code=400, detail="Unbekannter Modus.")
|
||||
try:
|
||||
data = backup_data.parse(await _read_upload(file))
|
||||
except backup_data.BackupError as exc:
|
||||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||||
|
||||
scope = data["scope"]
|
||||
if scope == "system" and user.role != Role.admin:
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail="Ein systemweites Backup darf nur ein Administrator einspielen.",
|
||||
)
|
||||
if scope == "system" and mode != "replace":
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail="Ein systemweites Backup kann nur ersetzend eingespielt werden.",
|
||||
)
|
||||
|
||||
schnappschuss = _snapshot_before_restore() if mode == "replace" else None
|
||||
try:
|
||||
if scope == "system":
|
||||
bilanz = backup_data.restore_system(db, data)
|
||||
else:
|
||||
bilanz = backup_data.restore_user_data(
|
||||
db, user, data.get("data") or {}, merge=(mode == "merge")
|
||||
)
|
||||
db.commit()
|
||||
except backup_data.BackupError as exc:
|
||||
db.rollback()
|
||||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||||
except Exception as exc: # pragma: no cover - Schutznetz gegen halbe Zustände
|
||||
db.rollback()
|
||||
raise HTTPException(
|
||||
status_code=400, detail=f"Wiederherstellung fehlgeschlagen: {exc}"
|
||||
) from exc
|
||||
|
||||
if scope == "system":
|
||||
# Die eigene Sitzung wurde mitgelöscht; alle übrigen ebenfalls beenden.
|
||||
sessions.revoke_all_everywhere(db)
|
||||
return {
|
||||
"scope": scope,
|
||||
"mode": mode,
|
||||
"restored": bilanz,
|
||||
"snapshot": os.path.basename(schnappschuss) if schnappschuss else None,
|
||||
"logout_required": scope == "system",
|
||||
}
|
||||
|
||||
@@ -0,0 +1,33 @@
|
||||
"""Globale Passwortrichtlinie – lesen für jeden, ändern nur für Admins.
|
||||
|
||||
Lesbar muss sie für alle sein, weil die Anforderungs-Checkliste beim
|
||||
Passwortsetzen dieselben Werte braucht.
|
||||
"""
|
||||
|
||||
from fastapi import APIRouter, Depends
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from ..database import get_db
|
||||
from ..deps import get_current_user, require_admin
|
||||
from ..models import User
|
||||
from ..schemas import PolicyIn, PolicyOut
|
||||
from ..services import policy as policy_service
|
||||
|
||||
router = APIRouter(prefix="/policy", tags=["policy"])
|
||||
|
||||
|
||||
@router.get("", response_model=PolicyOut)
|
||||
def get_policy(
|
||||
db: Session = Depends(get_db), _: User = Depends(get_current_user)
|
||||
) -> PolicyOut:
|
||||
return PolicyOut(**policy_service.load(db).as_dict())
|
||||
|
||||
|
||||
@router.put("", response_model=PolicyOut)
|
||||
def set_policy(
|
||||
payload: PolicyIn,
|
||||
db: Session = Depends(get_db),
|
||||
_: User = Depends(require_admin),
|
||||
) -> PolicyOut:
|
||||
saved = policy_service.save(db, policy_service.Policy(**payload.model_dump()))
|
||||
return PolicyOut(**saved.as_dict())
|
||||
@@ -4,24 +4,28 @@ from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from ..database import get_db
|
||||
from ..deps import require_admin
|
||||
from ..deps import require_admin, user_out
|
||||
from ..models import Role, User
|
||||
from ..schemas import UserCreate, UserOut, UserUpdate
|
||||
from ..security import hash_password
|
||||
from ..seed import ensure_starter_categories
|
||||
from ..services import policy as policy_service
|
||||
from ..services import sessions, twofactor
|
||||
|
||||
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()
|
||||
def list_users(db: Session = Depends(get_db)) -> list[UserOut]:
|
||||
rows = db.query(User).order_by(User.username).all()
|
||||
return [user_out(db, u) for u in rows]
|
||||
|
||||
|
||||
@router.post("", response_model=UserOut, status_code=status.HTTP_201_CREATED)
|
||||
def create_user(payload: UserCreate, db: Session = Depends(get_db)) -> User:
|
||||
def create_user(payload: UserCreate, db: Session = Depends(get_db)) -> UserOut:
|
||||
if db.query(User).filter(User.username == payload.username).first():
|
||||
raise HTTPException(status.HTTP_409_CONFLICT, "Benutzername existiert bereits")
|
||||
policy_service.assert_ok(db, payload.password)
|
||||
user = User(
|
||||
username=payload.username,
|
||||
password_hash=hash_password(payload.password),
|
||||
@@ -30,24 +34,29 @@ def create_user(payload: UserCreate, db: Session = Depends(get_db)) -> User:
|
||||
is_active=True,
|
||||
# Vom Admin gesetztes Temporär-Passwort: beim ersten Login ändern lassen.
|
||||
must_change_password=True,
|
||||
must_change_password_reason="temp",
|
||||
)
|
||||
db.add(user)
|
||||
db.commit()
|
||||
db.refresh(user)
|
||||
# Neuen Nutzern die Startkategorien mitgeben.
|
||||
ensure_starter_categories(db, user)
|
||||
return user
|
||||
return user_out(db, user)
|
||||
|
||||
|
||||
@router.patch("/{user_id}", response_model=UserOut)
|
||||
def update_user(user_id: int, payload: UserUpdate, db: Session = Depends(get_db)) -> User:
|
||||
def update_user(user_id: int, payload: UserUpdate, db: Session = Depends(get_db)) -> UserOut:
|
||||
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:
|
||||
policy_service.assert_ok(db, payload.password)
|
||||
user.password_hash = hash_password(payload.password)
|
||||
# Vom Admin gesetztes Passwort gilt als temporär.
|
||||
user.must_change_password = True
|
||||
user.must_change_password_reason = "temp"
|
||||
# Ein zurückgesetztes Passwort soll bestehende Sitzungen beenden.
|
||||
sessions.revoke_all(db, user)
|
||||
if payload.role is not None:
|
||||
user.role = payload.role
|
||||
if payload.display_name is not None:
|
||||
@@ -56,7 +65,22 @@ def update_user(user_id: int, payload: UserUpdate, db: Session = Depends(get_db)
|
||||
user.is_active = payload.is_active
|
||||
db.commit()
|
||||
db.refresh(user)
|
||||
return user
|
||||
return user_out(db, user)
|
||||
|
||||
|
||||
@router.post("/{user_id}/2fa/reset", response_model=UserOut)
|
||||
def reset_two_factor(user_id: int, db: Session = Depends(get_db)) -> UserOut:
|
||||
"""Notausgang, wenn jemand seinen Authenticator und alle Codes verloren hat.
|
||||
|
||||
Verlangt die Richtlinie MFA, muss der Benutzer beim nächsten Anmelden sofort
|
||||
neu einrichten – dafür sorgt ``enforce_account_state``."""
|
||||
user = db.get(User, user_id)
|
||||
if user is None:
|
||||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Benutzer nicht gefunden")
|
||||
twofactor.disable(db, user)
|
||||
sessions.revoke_all(db, user)
|
||||
db.refresh(user)
|
||||
return user_out(db, user)
|
||||
|
||||
|
||||
@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
|
||||
@@ -34,7 +34,13 @@ class UserOut(BaseModel):
|
||||
display_name: str | None = None
|
||||
is_active: bool
|
||||
must_change_password: bool
|
||||
must_change_password_reason: str | None = None
|
||||
totp_enabled: bool = False
|
||||
created_at: datetime
|
||||
# Kein Datenbankfeld: „Richtlinie verlangt MFA und dieser Benutzer hat
|
||||
# keins". Wird beim Ausliefern gesetzt (siehe deps.user_out), damit das
|
||||
# Frontend ohne zweite Abfrage weiss, wohin es umleiten muss.
|
||||
mfa_setup_required: bool = False
|
||||
|
||||
|
||||
class UserCreate(BaseModel):
|
||||
@@ -61,6 +67,68 @@ class SettingOut(BaseModel):
|
||||
value: str | None = None
|
||||
|
||||
|
||||
# --- Passwortrichtlinie -----------------------------------------------------
|
||||
|
||||
|
||||
class PolicyOut(BaseModel):
|
||||
min_length: int
|
||||
require_upper: bool
|
||||
require_lower: bool
|
||||
require_digit: bool
|
||||
require_special: bool
|
||||
mfa_required: bool
|
||||
|
||||
|
||||
class PolicyIn(BaseModel):
|
||||
min_length: int = Field(ge=8, le=128)
|
||||
require_upper: bool = False
|
||||
require_lower: bool = False
|
||||
require_digit: bool = False
|
||||
require_special: bool = False
|
||||
mfa_required: bool = False
|
||||
|
||||
|
||||
# --- Anmeldung / Zwei-Faktor ------------------------------------------------
|
||||
|
||||
|
||||
class LoginOut(BaseModel):
|
||||
"""Antwort auf ``POST /auth/login``.
|
||||
|
||||
``status`` ist ``"ok"`` (angemeldet, ``user`` gesetzt) oder ``"2fa_required"``
|
||||
(Passwort stimmt, es fehlt noch der Code – ``user`` bleibt leer)."""
|
||||
|
||||
status: str = "ok"
|
||||
user: UserOut | None = None
|
||||
|
||||
|
||||
class TwoFactorVerify(BaseModel):
|
||||
code: str = Field(min_length=1, max_length=64)
|
||||
recovery: bool = False
|
||||
|
||||
|
||||
class TwoFactorSetupOut(BaseModel):
|
||||
secret: str
|
||||
otpauth_uri: str
|
||||
qr_png: str # data:-URI, direkt in <img src=...> verwendbar
|
||||
|
||||
|
||||
class TwoFactorActivate(BaseModel):
|
||||
code: str = Field(min_length=1, max_length=16)
|
||||
|
||||
|
||||
class TwoFactorPassword(BaseModel):
|
||||
password: str = Field(min_length=1, max_length=1024)
|
||||
|
||||
|
||||
class RecoveryCodesOut(BaseModel):
|
||||
codes: list[str]
|
||||
|
||||
|
||||
class TwoFactorStatusOut(BaseModel):
|
||||
enabled: bool
|
||||
recovery_codes_left: int
|
||||
|
||||
|
||||
# --- Konten ----------------------------------------------------------------
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,479 @@
|
||||
"""Voll-Backup als JSON: Export und Wiederherstellung.
|
||||
|
||||
Warum ein eigenes Format neben ``/export/all.json``: jener Export ist zum
|
||||
Ansehen gedacht und referenziert alles über Namen. Kategorienamen sind aber
|
||||
nicht eindeutig, mehrere Tabellen fehlen ganz, und eine Version steht auch nicht
|
||||
drin – zurückladen lässt sich das nicht verlustfrei.
|
||||
|
||||
Hier wird stattdessen tabellengetrieben gearbeitet: jede Zeile bringt ihre
|
||||
Quell-ID als ``local_id`` mit, Fremdschlüssel bleiben IDs. Beim Einspielen wird
|
||||
eine Abbildung ``local_id -> neue ID`` aufgebaut und jeder Verweis übersetzt.
|
||||
Das ist unabhängig von Namen und überlebt auch doppelte Bezeichnungen.
|
||||
|
||||
Zwei Reichweiten:
|
||||
|
||||
``user`` – die Daten eines Benutzers. Jeder kann sein eigenes Backup ziehen.
|
||||
``system`` – alle Benutzer inklusive Passwort-Hash und TOTP-Secret, dazu die
|
||||
globalen Tabellen. Nur für Admins; die Datei ist wie ein Passwort
|
||||
zu behandeln.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import date as date_t, datetime, time as time_t
|
||||
from decimal import Decimal
|
||||
from typing import Callable
|
||||
|
||||
from sqlalchemy import Boolean, Date, DateTime, Numeric, Time
|
||||
from sqlalchemy import Enum as SAEnum
|
||||
from sqlalchemy import inspect as sa_inspect
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from ..branding import VERSION
|
||||
from ..models import (
|
||||
Account,
|
||||
AccountIdentifier,
|
||||
AppSetting,
|
||||
BudgetPeriod,
|
||||
BudgetPeriodLine,
|
||||
BudgetTemplateLine,
|
||||
Category,
|
||||
CategoryGroup,
|
||||
CategorizationRule,
|
||||
DashboardLayout,
|
||||
ExchangeRate,
|
||||
ImportBatch,
|
||||
ImportTemplate,
|
||||
PlannedIncomeLine,
|
||||
RecoveryCode,
|
||||
Transaction,
|
||||
User,
|
||||
UserSetting,
|
||||
)
|
||||
|
||||
FORMAT_NAME = "certain-finance-backup"
|
||||
SCHEMA_VERSION = 2
|
||||
ALEMBIC_REVISION = "0008"
|
||||
|
||||
|
||||
class BackupError(ValueError):
|
||||
"""Fachlicher Fehler beim Lesen oder Einspielen eines Backups."""
|
||||
|
||||
|
||||
# --- Tabellenbeschreibung ---------------------------------------------------
|
||||
|
||||
|
||||
@dataclass
|
||||
class TableSpec:
|
||||
key: str # Name im JSON
|
||||
model: type # ORM-Klasse
|
||||
scope: str # owner | user | account | period | global
|
||||
refs: dict[str, str] = field(default_factory=dict) # Spalte -> Tabellen-key
|
||||
has_id: bool = True # False bei Tabellen mit natürlichem Schlüssel
|
||||
label: str = "" # Anzeigename für die Prüfansicht
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not self.label:
|
||||
self.label = self.key
|
||||
|
||||
|
||||
# Reihenfolge = Einfügereihenfolge. Verweise dürfen nur auf bereits
|
||||
# eingefügte Tabellen zeigen; gelöscht wird in umgekehrter Reihenfolge.
|
||||
USER_TABLES: list[TableSpec] = [
|
||||
TableSpec("accounts", Account, "owner", label="Konten"),
|
||||
TableSpec("account_identifiers", AccountIdentifier, "account",
|
||||
{"account_id": "accounts"}, label="Konto-Merkmale"),
|
||||
TableSpec("category_groups", CategoryGroup, "owner", label="Kategoriegruppen"),
|
||||
TableSpec("categories", Category, "owner",
|
||||
{"group_id": "category_groups"}, label="Kategorien"),
|
||||
TableSpec("import_templates", ImportTemplate, "owner", label="Import-Vorlagen"),
|
||||
TableSpec("import_batches", ImportBatch, "owner",
|
||||
{"account_id": "accounts", "template_id": "import_templates"},
|
||||
label="Import-Läufe"),
|
||||
TableSpec("transactions", Transaction, "owner", {
|
||||
"account_id": "accounts",
|
||||
"to_account_id": "accounts",
|
||||
"category_id": "categories",
|
||||
"budget_category_id": "categories",
|
||||
"import_batch_id": "import_batches",
|
||||
}, label="Buchungen"),
|
||||
TableSpec("budget_template_lines", BudgetTemplateLine, "owner",
|
||||
{"category_id": "categories"}, label="Budget-Vorlage"),
|
||||
TableSpec("planned_income_lines", PlannedIncomeLine, "owner",
|
||||
{"category_id": "categories", "target_account_id": "accounts"},
|
||||
label="Geplante Einnahmen"),
|
||||
TableSpec("budget_periods", BudgetPeriod, "owner", label="Budget-Monate"),
|
||||
TableSpec("budget_period_lines", BudgetPeriodLine, "period",
|
||||
{"budget_period_id": "budget_periods", "category_id": "categories"},
|
||||
label="Budget-Monatswerte"),
|
||||
TableSpec("dashboard_layouts", DashboardLayout, "owner", label="Dashboard-Layout"),
|
||||
TableSpec("categorization_rules", CategorizationRule, "owner",
|
||||
{"category_id": "categories"}, label="Schlagwortregeln"),
|
||||
TableSpec("user_settings", UserSetting, "user", has_id=False,
|
||||
label="Persönliche Einstellungen"),
|
||||
]
|
||||
|
||||
# Nur im systemweiten Backup, je Benutzer.
|
||||
SYSTEM_USER_TABLES: list[TableSpec] = [
|
||||
TableSpec("recovery_codes", RecoveryCode, "user", label="Wiederherstellungscodes"),
|
||||
]
|
||||
|
||||
# Globale Tabellen (nicht benutzergebunden), nur im systemweiten Backup.
|
||||
GLOBAL_TABLES: list[TableSpec] = [
|
||||
TableSpec("app_settings", AppSetting, "global", has_id=False, label="Globale Einstellungen"),
|
||||
TableSpec("exchange_rates", ExchangeRate, "global", has_id=False, label="Wechselkurse"),
|
||||
]
|
||||
|
||||
# Spalten, die nie exportiert werden: sie werden beim Einspielen neu gesetzt.
|
||||
_SKIP_COLUMNS = {"id", "owner_id", "user_id"}
|
||||
|
||||
|
||||
# --- Werte umwandeln --------------------------------------------------------
|
||||
|
||||
|
||||
def _dump_value(col, value):
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(col.type, SAEnum):
|
||||
return value.value if hasattr(value, "value") else str(value)
|
||||
if isinstance(col.type, DateTime) or isinstance(value, datetime):
|
||||
return value.isoformat()
|
||||
if isinstance(col.type, Date) or isinstance(value, date_t):
|
||||
return value.isoformat()
|
||||
if isinstance(col.type, Time) or isinstance(value, time_t):
|
||||
return value.isoformat()
|
||||
if isinstance(col.type, Numeric) or isinstance(value, Decimal):
|
||||
return str(value)
|
||||
if isinstance(col.type, Boolean):
|
||||
return bool(value)
|
||||
return value
|
||||
|
||||
|
||||
def _load_value(col, value):
|
||||
if value is None:
|
||||
return None
|
||||
try:
|
||||
if isinstance(col.type, SAEnum) and col.type.enum_class is not None:
|
||||
return col.type.enum_class(value)
|
||||
if isinstance(col.type, DateTime):
|
||||
return datetime.fromisoformat(value)
|
||||
if isinstance(col.type, Date):
|
||||
return date_t.fromisoformat(value)
|
||||
if isinstance(col.type, Time):
|
||||
return time_t.fromisoformat(value)
|
||||
if isinstance(col.type, Numeric):
|
||||
return Decimal(str(value))
|
||||
if isinstance(col.type, Boolean):
|
||||
return bool(value)
|
||||
except (ValueError, TypeError) as exc:
|
||||
raise BackupError(f"Unlesbarer Wert für '{col.key}': {value!r}") from exc
|
||||
return value
|
||||
|
||||
|
||||
def _columns(spec: TableSpec):
|
||||
"""Zu sichernde Spalten – ohne technische IDs, die neu vergeben werden."""
|
||||
return [c for c in sa_inspect(spec.model).columns if c.key not in _SKIP_COLUMNS]
|
||||
|
||||
|
||||
# --- Export -----------------------------------------------------------------
|
||||
|
||||
|
||||
def _query(db: Session, spec: TableSpec, user: User | None):
|
||||
q = db.query(spec.model)
|
||||
if spec.scope == "owner":
|
||||
q = q.filter(spec.model.owner_id == user.id)
|
||||
elif spec.scope == "user":
|
||||
q = q.filter(spec.model.user_id == user.id)
|
||||
elif spec.scope == "account":
|
||||
q = q.join(Account, spec.model.account_id == Account.id).filter(
|
||||
Account.owner_id == user.id
|
||||
)
|
||||
elif spec.scope == "period":
|
||||
q = q.join(BudgetPeriod, spec.model.budget_period_id == BudgetPeriod.id).filter(
|
||||
BudgetPeriod.owner_id == user.id
|
||||
)
|
||||
return q
|
||||
|
||||
|
||||
def _dump_table(db: Session, spec: TableSpec, user: User | None) -> list[dict]:
|
||||
cols = _columns(spec)
|
||||
rows = []
|
||||
for obj in _query(db, spec, user).all():
|
||||
row = {c.key: _dump_value(c, getattr(obj, c.key)) for c in cols}
|
||||
if spec.has_id:
|
||||
row["local_id"] = obj.id
|
||||
rows.append(row)
|
||||
return rows
|
||||
|
||||
|
||||
def _dump_user_data(db: Session, user: User, specs: list[TableSpec]) -> dict:
|
||||
return {spec.key: _dump_table(db, spec, user) for spec in specs}
|
||||
|
||||
|
||||
def _envelope(scope: str, payload: dict) -> dict:
|
||||
return {
|
||||
"format": FORMAT_NAME,
|
||||
"schema_version": SCHEMA_VERSION,
|
||||
"app_version": VERSION,
|
||||
"alembic_revision": ALEMBIC_REVISION,
|
||||
"exported_at": datetime.now().astimezone().isoformat(timespec="seconds"),
|
||||
"scope": scope,
|
||||
**payload,
|
||||
}
|
||||
|
||||
|
||||
def export_user(db: Session, user: User) -> dict:
|
||||
"""Backup der Daten eines Benutzers."""
|
||||
return _envelope("user", {
|
||||
"username": user.username,
|
||||
"data": _dump_user_data(db, user, USER_TABLES),
|
||||
})
|
||||
|
||||
|
||||
def export_system(db: Session) -> dict:
|
||||
"""Backup der ganzen Instanz inklusive Anmeldegeheimnissen (nur Admin)."""
|
||||
user_cols = _columns(TableSpec("users", User, "global"))
|
||||
benutzer = []
|
||||
for u in db.query(User).order_by(User.id).all():
|
||||
satz = {c.key: _dump_value(c, getattr(u, c.key)) for c in user_cols}
|
||||
satz["local_id"] = u.id
|
||||
satz["data"] = _dump_user_data(db, u, USER_TABLES + SYSTEM_USER_TABLES)
|
||||
benutzer.append(satz)
|
||||
return _envelope("system", {
|
||||
"users": benutzer,
|
||||
"global": {spec.key: _dump_table(db, spec, None) for spec in GLOBAL_TABLES},
|
||||
})
|
||||
|
||||
|
||||
def to_bytes(obj: dict) -> bytes:
|
||||
return json.dumps(obj, ensure_ascii=False, indent=2).encode("utf-8")
|
||||
|
||||
|
||||
# --- Umschlag prüfen --------------------------------------------------------
|
||||
|
||||
|
||||
def parse(raw: bytes) -> dict:
|
||||
"""Liest eine hochgeladene Datei und prüft den Umschlag."""
|
||||
try:
|
||||
data = json.loads(raw.decode("utf-8"))
|
||||
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
||||
raise BackupError("Die Datei ist kein lesbares JSON.") from exc
|
||||
if not isinstance(data, dict):
|
||||
raise BackupError("Unerwarteter Dateiinhalt – erwartet wird ein JSON-Objekt.")
|
||||
if data.get("format") != FORMAT_NAME:
|
||||
# Der alte, nur zum Ansehen gedachte Export – klar benennen statt
|
||||
# halbherzig zu raten.
|
||||
if "accounts" in data and "category_groups" in data:
|
||||
raise BackupError(
|
||||
"Das ist der Ansichts-Export (all.json). Er lässt sich nicht "
|
||||
"wiederherstellen – bitte ein Backup über „Voll-Backup“ erzeugen."
|
||||
)
|
||||
raise BackupError("Das ist kein Backup von Certain Finance.")
|
||||
version = data.get("schema_version")
|
||||
if version != SCHEMA_VERSION:
|
||||
raise BackupError(
|
||||
f"Backup-Version {version} wird nicht unterstützt "
|
||||
f"(diese Fassung liest Version {SCHEMA_VERSION})."
|
||||
)
|
||||
if data.get("scope") not in ("user", "system"):
|
||||
raise BackupError("Unbekannte Reichweite im Backup.")
|
||||
return data
|
||||
|
||||
|
||||
def summarize(data: dict) -> list[dict]:
|
||||
"""Zählt je Tabelle, was in der Datei steckt – für die Prüfansicht."""
|
||||
if data["scope"] == "user":
|
||||
haeufen = [data.get("data", {})]
|
||||
else:
|
||||
haeufen = [u.get("data", {}) for u in data.get("users", [])]
|
||||
|
||||
zeilen = []
|
||||
for spec in USER_TABLES + SYSTEM_USER_TABLES:
|
||||
anzahl = sum(len(h.get(spec.key) or []) for h in haeufen)
|
||||
if anzahl:
|
||||
zeilen.append({"key": spec.key, "label": spec.label, "count": anzahl})
|
||||
if data["scope"] == "system":
|
||||
zeilen.insert(0, {"key": "users", "label": "Benutzer", "count": len(data.get("users", []))})
|
||||
for spec in GLOBAL_TABLES:
|
||||
anzahl = len(data.get("global", {}).get(spec.key) or [])
|
||||
if anzahl:
|
||||
zeilen.append({"key": spec.key, "label": spec.label, "count": anzahl})
|
||||
return zeilen
|
||||
|
||||
|
||||
# --- Einspielen -------------------------------------------------------------
|
||||
|
||||
# Wie im Modus „Ergänzen" entschieden wird, ob eine Zeile schon da ist.
|
||||
# Rückgabe: Schlüssel, unter dem verglichen wird – oder None für „immer neu".
|
||||
_MERGE_KEYS: dict[str, Callable[[dict], tuple | None]] = {
|
||||
"accounts": lambda r: ("name", (r.get("name") or "").strip().lower()),
|
||||
# Ohne diesen Eintrag würde ein zweiter Ergänzen-Lauf die Eindeutigkeit
|
||||
# (account_id, kind, value_norm) verletzen.
|
||||
"account_identifiers": lambda r: ("ident", r.get("account_id"), r.get("kind"),
|
||||
r.get("value_norm")),
|
||||
"category_groups": lambda r: ("name", (r.get("name") or "").strip().lower()),
|
||||
"categories": lambda r: ("group", r.get("group_id"), (r.get("name") or "").strip().lower()),
|
||||
"import_templates": lambda r: ("name", (r.get("name") or "").strip().lower()),
|
||||
"import_batches": lambda r: ("batch", r.get("account_id"), r.get("filename"),
|
||||
r.get("created_at")),
|
||||
"transactions": lambda r: (
|
||||
("hash", r["import_hash"]) if r.get("import_hash")
|
||||
else ("nat", r.get("date"), r.get("account_id"), r.get("amount_cents"),
|
||||
(r.get("payee") or "").strip().lower())
|
||||
),
|
||||
}
|
||||
|
||||
# Tabellen, die im Modus „Ergänzen" nur dann übernommen werden, wenn der
|
||||
# Benutzer davon noch gar nichts hat. Zeilenweise mischen ergäbe hier Unsinn
|
||||
# (halbe Budgets, doppelte Layouts).
|
||||
_ALL_OR_NOTHING = {
|
||||
"budget_template_lines",
|
||||
"planned_income_lines",
|
||||
"budget_periods",
|
||||
"budget_period_lines",
|
||||
"dashboard_layouts",
|
||||
"categorization_rules",
|
||||
"user_settings",
|
||||
}
|
||||
|
||||
|
||||
def _existing_keys(db: Session, spec: TableSpec, user: User) -> dict[tuple, int]:
|
||||
"""Vergleichsschlüssel der schon vorhandenen Zeilen -> deren ID."""
|
||||
schluessel = _MERGE_KEYS.get(spec.key)
|
||||
if schluessel is None:
|
||||
return {}
|
||||
cols = _columns(spec)
|
||||
treffer: dict[tuple, int] = {}
|
||||
for obj in _query(db, spec, user).all():
|
||||
roh = {c.key: _dump_value(c, getattr(obj, c.key)) for c in cols}
|
||||
k = schluessel(roh)
|
||||
if k is not None:
|
||||
treffer.setdefault(k, obj.id)
|
||||
return treffer
|
||||
|
||||
|
||||
def _delete_user_data(db: Session, user: User, specs: list[TableSpec]) -> None:
|
||||
"""Löscht die Daten eines Benutzers in fremdschlüsselsicherer Reihenfolge."""
|
||||
for spec in reversed(specs):
|
||||
for obj in _query(db, spec, user).all():
|
||||
db.delete(obj)
|
||||
db.flush()
|
||||
|
||||
|
||||
def _insert_rows(
|
||||
db: Session, spec: TableSpec, user: User, rows: list[dict], maps: dict[str, dict[int, int]],
|
||||
merge: bool,
|
||||
) -> int:
|
||||
cols = {c.key: c for c in _columns(spec)}
|
||||
id_map = maps.setdefault(spec.key, {})
|
||||
vorhanden = _existing_keys(db, spec, user) if merge else {}
|
||||
schluessel = _MERGE_KEYS.get(spec.key)
|
||||
neu = 0
|
||||
|
||||
for row in rows:
|
||||
# Verweise zuerst übersetzen – sie fließen auch in den Vergleichsschlüssel ein.
|
||||
aufgeloest = dict(row)
|
||||
for spalte, ziel in spec.refs.items():
|
||||
alt = row.get(spalte)
|
||||
aufgeloest[spalte] = maps.get(ziel, {}).get(alt) if alt is not None else None
|
||||
|
||||
if merge and schluessel is not None:
|
||||
k = schluessel(aufgeloest)
|
||||
if k is not None and k in vorhanden:
|
||||
# Nicht anlegen, aber den Verweis auf die bestehende Zeile merken.
|
||||
if spec.has_id and row.get("local_id") is not None:
|
||||
id_map[row["local_id"]] = vorhanden[k]
|
||||
continue
|
||||
|
||||
werte = {}
|
||||
for name, col in cols.items():
|
||||
if name in spec.refs:
|
||||
werte[name] = aufgeloest[name]
|
||||
else:
|
||||
werte[name] = _load_value(col, row.get(name))
|
||||
if spec.scope in ("owner",):
|
||||
werte["owner_id"] = user.id
|
||||
elif spec.scope == "user":
|
||||
werte["user_id"] = user.id
|
||||
|
||||
obj = spec.model(**werte)
|
||||
db.add(obj)
|
||||
db.flush()
|
||||
if spec.has_id and row.get("local_id") is not None:
|
||||
id_map[row["local_id"]] = obj.id
|
||||
if merge and schluessel is not None:
|
||||
k = schluessel(aufgeloest)
|
||||
if k is not None:
|
||||
vorhanden[k] = obj.id
|
||||
neu += 1
|
||||
return neu
|
||||
|
||||
|
||||
def restore_user_data(
|
||||
db: Session, user: User, data: dict, merge: bool, specs: list[TableSpec] | None = None
|
||||
) -> dict[str, int]:
|
||||
"""Spielt den Datenblock eines Benutzers ein. Ruft **kein** commit."""
|
||||
specs = specs or USER_TABLES
|
||||
if not merge:
|
||||
_delete_user_data(db, user, specs)
|
||||
|
||||
maps: dict[str, dict[int, int]] = {}
|
||||
bilanz: dict[str, int] = {}
|
||||
for spec in specs:
|
||||
rows = data.get(spec.key) or []
|
||||
if merge and spec.key in _ALL_OR_NOTHING:
|
||||
# Nur übernehmen, wenn dazu noch nichts existiert.
|
||||
if _query(db, spec, user).first() is not None:
|
||||
bilanz[spec.key] = 0
|
||||
continue
|
||||
bilanz[spec.key] = _insert_rows(db, spec, user, rows, maps, merge)
|
||||
return bilanz
|
||||
|
||||
|
||||
def restore_system(db: Session, data: dict) -> dict[str, int]:
|
||||
"""Baut die ganze Instanz aus einem systemweiten Backup neu auf.
|
||||
|
||||
Immer ersetzend – ein teilweises Verschmelzen fremder Benutzerkonten wäre
|
||||
nicht sinnvoll deutbar. Ruft **kein** commit; das übernimmt der Aufrufer,
|
||||
damit im Fehlerfall nichts halb eingespielt zurückbleibt."""
|
||||
benutzer = data.get("users") or []
|
||||
if not any(u.get("role") == "admin" for u in benutzer):
|
||||
raise BackupError(
|
||||
"Das Backup enthält keinen Administrator – damit käme niemand mehr hinein."
|
||||
)
|
||||
|
||||
# Benutzer weg: alles Benutzergebundene hängt per ON DELETE CASCADE daran.
|
||||
for u in db.query(User).all():
|
||||
db.delete(u)
|
||||
db.flush()
|
||||
for spec in GLOBAL_TABLES:
|
||||
for obj in db.query(spec.model).all():
|
||||
db.delete(obj)
|
||||
db.flush()
|
||||
|
||||
user_cols = {c.key: c for c in _columns(TableSpec("users", User, "global"))}
|
||||
bilanz: dict[str, int] = {"users": 0}
|
||||
for satz in benutzer:
|
||||
werte = {k: _load_value(col, satz.get(k)) for k, col in user_cols.items()}
|
||||
neuer = User(**werte)
|
||||
db.add(neuer)
|
||||
db.flush()
|
||||
bilanz["users"] += 1
|
||||
teil = restore_user_data(
|
||||
db, neuer, satz.get("data") or {}, merge=False,
|
||||
specs=USER_TABLES + SYSTEM_USER_TABLES,
|
||||
)
|
||||
for key, anzahl in teil.items():
|
||||
bilanz[key] = bilanz.get(key, 0) + anzahl
|
||||
|
||||
global_data = data.get("global") or {}
|
||||
for spec in GLOBAL_TABLES:
|
||||
cols = {c.key: c for c in _columns(spec)}
|
||||
rows = global_data.get(spec.key) or []
|
||||
for row in rows:
|
||||
db.add(spec.model(**{k: _load_value(col, row.get(k)) for k, col in cols.items()}))
|
||||
bilanz[spec.key] = len(rows)
|
||||
db.flush()
|
||||
return bilanz
|
||||
@@ -0,0 +1,134 @@
|
||||
"""Passwortrichtlinie: global vom Admin einstellbar, serverseitig durchgesetzt.
|
||||
|
||||
Die Werte liegen in ``app_settings`` (nicht in ``user_settings`` – die Richtlinie
|
||||
gilt für die ganze Instanz). Das Frontend wertet dieselben Regeln nochmal lokal
|
||||
aus, um beim Tippen eine Checkliste abzuhaken; verbindlich ist ausschließlich
|
||||
diese Datei.
|
||||
|
||||
Zur wiederkehrenden Frage „kann man bestehende Passwörter gegen eine neue
|
||||
Richtlinie prüfen?": nicht über den Hash – der gibt das Passwort nicht her. Wohl
|
||||
aber beim Anmelden, denn dort liegt das Klartext-Passwort für einen Moment vor.
|
||||
Genau das macht ``routers/auth.login``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import asdict, dataclass
|
||||
|
||||
from fastapi import HTTPException, status
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from ..models import AppSetting
|
||||
|
||||
# Grenzen für die Mindestlänge – nach unten, damit die Richtlinie nicht sinnlos
|
||||
# wird, nach oben, damit sich niemand selbst aussperrt.
|
||||
MIN_LENGTH_FLOOR = 8
|
||||
MIN_LENGTH_CEILING = 128
|
||||
|
||||
|
||||
def is_special(c: str) -> bool:
|
||||
"""Sonderzeichen = alles, was weder Buchstabe/Ziffer noch Leerraum ist.
|
||||
|
||||
Bewusst nicht über eine feste ASCII-Liste: „§" oder „€" sollen zählen,
|
||||
Umlaute wie „ü" dagegen nicht (die sind Buchstaben)."""
|
||||
return not c.isalnum() and not c.isspace()
|
||||
|
||||
|
||||
@dataclass
|
||||
class Policy:
|
||||
min_length: int = 10
|
||||
require_upper: bool = True
|
||||
require_lower: bool = True
|
||||
require_digit: bool = True
|
||||
require_special: bool = False
|
||||
mfa_required: bool = False
|
||||
|
||||
def as_dict(self) -> dict:
|
||||
return asdict(self)
|
||||
|
||||
|
||||
# Schlüssel in app_settings <-> Feldname im Datenobjekt.
|
||||
_KEYS = {
|
||||
"password_min_length": "min_length",
|
||||
"password_require_upper": "require_upper",
|
||||
"password_require_lower": "require_lower",
|
||||
"password_require_digit": "require_digit",
|
||||
"password_require_special": "require_special",
|
||||
"mfa_required": "mfa_required",
|
||||
}
|
||||
|
||||
# Menschenlesbare Begründung je Regel – wird bei Verstößen zusammengesetzt.
|
||||
RULE_TEXTS = {
|
||||
"length": "mindestens {n} Zeichen",
|
||||
"upper": "mindestens einen Grossbuchstaben",
|
||||
"lower": "mindestens einen Kleinbuchstaben",
|
||||
"digit": "mindestens eine Ziffer",
|
||||
"special": "mindestens ein Sonderzeichen",
|
||||
}
|
||||
|
||||
|
||||
def _clamp_length(value: int) -> int:
|
||||
return max(MIN_LENGTH_FLOOR, min(MIN_LENGTH_CEILING, value))
|
||||
|
||||
|
||||
def load(db: Session) -> Policy:
|
||||
"""Liest die Richtlinie; fehlende Schlüssel fallen auf die Vorgaben zurück."""
|
||||
stored = {row.key: row.value for row in db.query(AppSetting).all()}
|
||||
pol = Policy()
|
||||
for key, field in _KEYS.items():
|
||||
raw = stored.get(key)
|
||||
if raw is None:
|
||||
continue
|
||||
if field == "min_length":
|
||||
try:
|
||||
pol.min_length = _clamp_length(int(raw))
|
||||
except ValueError:
|
||||
pass
|
||||
else:
|
||||
setattr(pol, field, raw == "true")
|
||||
return pol
|
||||
|
||||
|
||||
def save(db: Session, pol: Policy) -> Policy:
|
||||
"""Schreibt die Richtlinie; die Mindestlänge wird auf den erlaubten Bereich begrenzt."""
|
||||
pol.min_length = _clamp_length(pol.min_length)
|
||||
values = pol.as_dict()
|
||||
for key, field in _KEYS.items():
|
||||
value = values[field]
|
||||
text = str(value) if field == "min_length" else ("true" if value else "false")
|
||||
row = db.get(AppSetting, key)
|
||||
if row is None:
|
||||
db.add(AppSetting(key=key, value=text))
|
||||
else:
|
||||
row.value = text
|
||||
db.commit()
|
||||
return pol
|
||||
|
||||
|
||||
def violations(password: str, pol: Policy) -> list[str]:
|
||||
"""Gibt die Kennungen der *nicht* erfüllten Regeln zurück (leer = in Ordnung)."""
|
||||
offen: list[str] = []
|
||||
if len(password) < pol.min_length:
|
||||
offen.append("length")
|
||||
if pol.require_upper and not any(c.isupper() for c in password):
|
||||
offen.append("upper")
|
||||
if pol.require_lower and not any(c.islower() for c in password):
|
||||
offen.append("lower")
|
||||
if pol.require_digit and not any(c.isdigit() for c in password):
|
||||
offen.append("digit")
|
||||
if pol.require_special and not any(is_special(c) for c in password):
|
||||
offen.append("special")
|
||||
return offen
|
||||
|
||||
|
||||
def assert_ok(db: Session, password: str) -> None:
|
||||
"""Wirft 400 mit einer Klartext-Begründung, wenn das Passwort nicht genügt."""
|
||||
pol = load(db)
|
||||
offen = violations(password, pol)
|
||||
if not offen:
|
||||
return
|
||||
teile = [RULE_TEXTS[r].format(n=pol.min_length) for r in offen]
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Das Passwort erfüllt die Richtlinie nicht: " + ", ".join(teile) + ".",
|
||||
)
|
||||
@@ -23,24 +23,39 @@ def _aware(dt: datetime) -> datetime:
|
||||
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:
|
||||
# Wie lange der Zwischenzustand „Passwort ok, Code fehlt" gilt. Kurz, weil in
|
||||
# dieser Zeit nur noch der zweite Faktor eingegeben werden soll.
|
||||
PENDING_TTL_MINUTES = 10
|
||||
# So viele Fehlversuche verträgt eine Pending-Sitzung, dann ist sie verbraucht.
|
||||
MAX_2FA_ATTEMPTS = 5
|
||||
|
||||
|
||||
def create_session(
|
||||
db: DBSession, user: User, user_agent: str | None = None, pending_2fa: bool = False
|
||||
) -> str:
|
||||
"""Legt eine Sitzung an und gibt das Klartext-Token zurück (nur hier sichtbar)."""
|
||||
settings = get_settings()
|
||||
gueltig_bis = (
|
||||
_now() + timedelta(minutes=PENDING_TTL_MINUTES)
|
||||
if pending_2fa
|
||||
else _now() + timedelta(days=settings.session_ttl_days)
|
||||
)
|
||||
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),
|
||||
expires_at=gueltig_bis,
|
||||
last_seen_at=_now(),
|
||||
user_agent=(user_agent or "")[:255] or None,
|
||||
pending_2fa=pending_2fa,
|
||||
)
|
||||
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."""
|
||||
def _lookup(db: DBSession, raw_token: str) -> UserSession | None:
|
||||
"""Sucht die Sitzung und räumt sie weg, falls sie abgelaufen ist."""
|
||||
if not raw_token:
|
||||
return None
|
||||
entry = (
|
||||
@@ -54,6 +69,18 @@ def validate(db: DBSession, raw_token: str) -> User | None:
|
||||
db.delete(entry)
|
||||
db.commit()
|
||||
return None
|
||||
return entry
|
||||
|
||||
|
||||
def validate(db: DBSession, raw_token: str) -> User | None:
|
||||
"""Gibt den Benutzer zurück, wenn das Token gültig und nicht abgelaufen ist.
|
||||
|
||||
Sitzungen, bei denen der zweite Faktor noch aussteht, gelten hier bewusst
|
||||
als *nicht* angemeldet – sonst wäre das Zwischen-Cookie ein vollwertiger
|
||||
Zugang."""
|
||||
entry = _lookup(db, raw_token)
|
||||
if entry is None or entry.pending_2fa:
|
||||
return None
|
||||
user = db.get(User, entry.user_id)
|
||||
if user is None or not user.is_active:
|
||||
return None
|
||||
@@ -62,6 +89,27 @@ def validate(db: DBSession, raw_token: str) -> User | None:
|
||||
return user
|
||||
|
||||
|
||||
def validate_pending(db: DBSession, raw_token: str) -> tuple[UserSession, User] | None:
|
||||
"""Gegenstück zu ``validate`` für den Zwei-Faktor-Schritt."""
|
||||
entry = _lookup(db, raw_token)
|
||||
if entry is None or not entry.pending_2fa:
|
||||
return None
|
||||
user = db.get(User, entry.user_id)
|
||||
if user is None or not user.is_active:
|
||||
return None
|
||||
return entry, user
|
||||
|
||||
|
||||
def note_failed_2fa(db: DBSession, entry: UserSession) -> bool:
|
||||
"""Zählt einen Fehlversuch. Gibt ``True`` zurück, wenn die Sitzung verfällt."""
|
||||
entry.failed_2fa += 1
|
||||
verbraucht = entry.failed_2fa >= MAX_2FA_ATTEMPTS
|
||||
if verbraucht:
|
||||
db.delete(entry)
|
||||
db.commit()
|
||||
return verbraucht
|
||||
|
||||
|
||||
def revoke(db: DBSession, raw_token: str) -> None:
|
||||
entry = (
|
||||
db.query(UserSession)
|
||||
@@ -76,3 +124,10 @@ def revoke(db: DBSession, raw_token: str) -> None:
|
||||
def revoke_all(db: DBSession, user: User) -> None:
|
||||
db.query(UserSession).filter(UserSession.user_id == user.id).delete()
|
||||
db.commit()
|
||||
|
||||
|
||||
def revoke_all_everywhere(db: DBSession) -> None:
|
||||
"""Beendet sämtliche Sitzungen – nach einer systemweiten Wiederherstellung,
|
||||
bei der die Benutzertabelle komplett ausgetauscht wurde."""
|
||||
db.query(UserSession).delete()
|
||||
db.commit()
|
||||
|
||||
@@ -0,0 +1,144 @@
|
||||
"""Zwei-Faktor per TOTP (Authenticator-App) plus Einmal-Wiederherstellungscodes.
|
||||
|
||||
Der QR-Code wird hier serverseitig als PNG-Data-URI erzeugt. Das spart eine
|
||||
JavaScript-Abhängigkeit im Frontend und erspart es, fremdes SVG ins DOM zu
|
||||
schreiben – das Bild landet einfach in einem ``<img src=…>``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import io
|
||||
import secrets
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import pyotp
|
||||
import qrcode
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from ..branding import APP_NAME
|
||||
from ..models import RecoveryCode, User
|
||||
from ..security import hash_session_token
|
||||
|
||||
STEP_SECONDS = 30
|
||||
# Ein Schritt Toleranz nach vorn und hinten – deckt übliche Uhrendrift ab,
|
||||
# ohne das Zeitfenster unnötig aufzuweiten.
|
||||
VALID_WINDOW = 1
|
||||
|
||||
RECOVERY_CODE_COUNT = 10
|
||||
RECOVERY_CODE_LENGTH = 10
|
||||
# Ohne 0/O/1/I/L – abgeschriebene Codes sollen nicht an Verwechslungen scheitern.
|
||||
_ALPHABET = "ABCDEFGHJKMNPQRSTUVWXYZ23456789"
|
||||
|
||||
|
||||
def new_secret() -> str:
|
||||
return pyotp.random_base32()
|
||||
|
||||
|
||||
def otpauth_uri(user: User, secret: str) -> str:
|
||||
return pyotp.TOTP(secret).provisioning_uri(name=user.username, issuer_name=APP_NAME)
|
||||
|
||||
|
||||
def qr_data_uri(uri: str) -> str:
|
||||
"""QR-Code als ``data:image/png;base64,…``."""
|
||||
img = qrcode.make(uri)
|
||||
puffer = io.BytesIO()
|
||||
img.save(puffer, format="PNG")
|
||||
return "data:image/png;base64," + base64.b64encode(puffer.getvalue()).decode("ascii")
|
||||
|
||||
|
||||
def matching_step(secret: str, code: str) -> int | None:
|
||||
"""Zu welchem 30-Sekunden-Fenster gehört dieser Code? ``None`` = zu keinem.
|
||||
|
||||
Nicht ``TOTP.verify`` benutzt, weil das nur ja/nein sagt. Für den
|
||||
Wiederverwendungsschutz muss aber feststehen, *welches* Fenster verbraucht
|
||||
wurde – sonst liesse sich ein Code aus dem nächsten Fenster im nächsten
|
||||
Moment erneut einlösen."""
|
||||
code = (code or "").strip().replace(" ", "")
|
||||
if not code.isdigit():
|
||||
return None
|
||||
totp = pyotp.TOTP(secret)
|
||||
jetzt = int(time.time()) // STEP_SECONDS
|
||||
for versatz in range(-VALID_WINDOW, VALID_WINDOW + 1):
|
||||
schritt = jetzt + versatz
|
||||
if secrets.compare_digest(totp.at(schritt * STEP_SECONDS), code):
|
||||
return schritt
|
||||
return None
|
||||
|
||||
|
||||
def consume_code(db: Session, user: User, code: str) -> bool:
|
||||
"""Prüft den Code und verbraucht sein Zeitfenster.
|
||||
|
||||
Ein einmal benutzter Code bleibt zwar bis zu 30 Sekunden rechnerisch gültig,
|
||||
darf aber kein zweites Mal zählen – sonst könnte ein abgefangener Code sofort
|
||||
nachgespielt werden."""
|
||||
if not user.totp_secret:
|
||||
return False
|
||||
schritt = matching_step(user.totp_secret, code)
|
||||
if schritt is None:
|
||||
return False
|
||||
if user.totp_last_step is not None and schritt <= user.totp_last_step:
|
||||
return False
|
||||
user.totp_last_step = schritt
|
||||
db.commit()
|
||||
return True
|
||||
|
||||
|
||||
def _format_code(raw: str) -> str:
|
||||
"""Fünfergruppen mit Bindestrich – deutlich leichter abzuschreiben."""
|
||||
return f"{raw[:5]}-{raw[5:]}"
|
||||
|
||||
|
||||
def generate_recovery_codes(db: Session, user: User) -> list[str]:
|
||||
"""Erzeugt einen frischen Satz Codes; alte werden dabei ungültig.
|
||||
|
||||
Die Klartext-Codes gibt es nur dieses eine Mal – in der Datenbank steht
|
||||
ausschliesslich ihr Hash."""
|
||||
db.query(RecoveryCode).filter(RecoveryCode.user_id == user.id).delete()
|
||||
codes: list[str] = []
|
||||
for _ in range(RECOVERY_CODE_COUNT):
|
||||
roh = "".join(secrets.choice(_ALPHABET) for _ in range(RECOVERY_CODE_LENGTH))
|
||||
codes.append(_format_code(roh))
|
||||
db.add(RecoveryCode(user_id=user.id, code_hash=hash_session_token(roh)))
|
||||
db.commit()
|
||||
return codes
|
||||
|
||||
|
||||
def consume_recovery_code(db: Session, user: User, code: str) -> bool:
|
||||
"""Setzt einen Wiederherstellungscode ein. Jeder Code zählt genau einmal."""
|
||||
roh = (code or "").strip().upper().replace("-", "").replace(" ", "")
|
||||
if not roh:
|
||||
return False
|
||||
treffer = (
|
||||
db.query(RecoveryCode)
|
||||
.filter(
|
||||
RecoveryCode.user_id == user.id,
|
||||
RecoveryCode.code_hash == hash_session_token(roh),
|
||||
RecoveryCode.used_at.is_(None),
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if treffer is None:
|
||||
return False
|
||||
treffer.used_at = datetime.now(timezone.utc)
|
||||
db.commit()
|
||||
return True
|
||||
|
||||
|
||||
def codes_left(db: Session, user: User) -> int:
|
||||
return (
|
||||
db.query(RecoveryCode)
|
||||
.filter(RecoveryCode.user_id == user.id, RecoveryCode.used_at.is_(None))
|
||||
.count()
|
||||
)
|
||||
|
||||
|
||||
def disable(db: Session, user: User) -> None:
|
||||
"""Schaltet 2FA ab und räumt Secret sowie Wiederherstellungscodes weg."""
|
||||
db.query(RecoveryCode).filter(RecoveryCode.user_id == user.id).delete()
|
||||
user.totp_secret = None
|
||||
user.totp_enabled = False
|
||||
user.totp_confirmed_at = None
|
||||
user.totp_last_step = None
|
||||
db.commit()
|
||||
@@ -0,0 +1,78 @@
|
||||
"""Passwortrichtlinie und Zwei-Faktor-Authentisierung.
|
||||
|
||||
Bringt die globale Einstellungstabelle (``app_settings``), die TOTP-Felder am
|
||||
Benutzer, die Wiederherstellungscodes und den Zwischenzustand „Passwort ok,
|
||||
zweiter Faktor fehlt noch" an der Sitzung.
|
||||
|
||||
Revision ID: 0008
|
||||
Revises: 0007
|
||||
Create Date: 2026-08-20
|
||||
"""
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
revision = "0008"
|
||||
down_revision = "0007"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.create_table(
|
||||
"app_settings",
|
||||
sa.Column("key", sa.String(length=64), primary_key=True),
|
||||
sa.Column("value", sa.String(length=255), nullable=True),
|
||||
)
|
||||
|
||||
op.create_table(
|
||||
"recovery_codes",
|
||||
sa.Column("id", sa.Integer(), primary_key=True),
|
||||
sa.Column("user_id", sa.Integer(), nullable=False),
|
||||
sa.Column("code_hash", sa.String(length=64), nullable=False),
|
||||
sa.Column("used_at", sa.DateTime(timezone=True), nullable=True),
|
||||
sa.Column("created_at", sa.DateTime(timezone=True), nullable=True),
|
||||
sa.ForeignKeyConstraint(["user_id"], ["users.id"], ondelete="CASCADE"),
|
||||
)
|
||||
op.create_index("ix_recovery_codes_user_id", "recovery_codes", ["user_id"])
|
||||
op.create_index("ix_recovery_codes_code_hash", "recovery_codes", ["code_hash"])
|
||||
|
||||
with op.batch_alter_table("users") as batch:
|
||||
batch.add_column(sa.Column("must_change_password_reason", sa.String(length=16), nullable=True))
|
||||
batch.add_column(sa.Column("totp_secret", sa.String(length=64), nullable=True))
|
||||
batch.add_column(
|
||||
sa.Column("totp_enabled", sa.Boolean(), nullable=False, server_default=sa.text("0"))
|
||||
)
|
||||
batch.add_column(sa.Column("totp_confirmed_at", sa.DateTime(timezone=True), nullable=True))
|
||||
batch.add_column(sa.Column("totp_last_step", sa.Integer(), nullable=True))
|
||||
|
||||
with op.batch_alter_table("sessions") as batch:
|
||||
batch.add_column(
|
||||
sa.Column("pending_2fa", sa.Boolean(), nullable=False, server_default=sa.text("0"))
|
||||
)
|
||||
batch.add_column(
|
||||
sa.Column("failed_2fa", sa.Integer(), nullable=False, server_default=sa.text("0"))
|
||||
)
|
||||
|
||||
# Bestehende Nutzer, die bereits ein Temporär-Passwort haben, bekommen den
|
||||
# passenden Grund nachgetragen – sonst stünde dort der Richtlinien-Hinweis.
|
||||
op.execute(
|
||||
"UPDATE users SET must_change_password_reason = 'temp' WHERE must_change_password = 1"
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
with op.batch_alter_table("sessions") as batch:
|
||||
batch.drop_column("failed_2fa")
|
||||
batch.drop_column("pending_2fa")
|
||||
|
||||
with op.batch_alter_table("users") as batch:
|
||||
batch.drop_column("totp_last_step")
|
||||
batch.drop_column("totp_confirmed_at")
|
||||
batch.drop_column("totp_enabled")
|
||||
batch.drop_column("totp_secret")
|
||||
batch.drop_column("must_change_password_reason")
|
||||
|
||||
op.drop_index("ix_recovery_codes_code_hash", table_name="recovery_codes")
|
||||
op.drop_index("ix_recovery_codes_user_id", table_name="recovery_codes")
|
||||
op.drop_table("recovery_codes")
|
||||
op.drop_table("app_settings")
|
||||
@@ -5,6 +5,10 @@ alembic==1.14.0
|
||||
pydantic==2.10.4
|
||||
pydantic-settings==2.7.1
|
||||
argon2-cffi==23.1.0
|
||||
# Zwei-Faktor: TOTP-Codes (pyotp) und der QR-Code zur Einrichtung (qrcode).
|
||||
# qrcode nutzt Pillow, das unten ohnehin schon für die OCR dabei ist.
|
||||
pyotp==2.9.0
|
||||
qrcode==8.0
|
||||
python-multipart==0.0.20
|
||||
httpx==0.28.1
|
||||
python-dateutil==2.9.0.post0
|
||||
|
||||
@@ -8,12 +8,16 @@ import tempfile
|
||||
os.environ.setdefault("APP_DATA_DIR", tempfile.mkdtemp(prefix="cf-test-"))
|
||||
os.environ.setdefault("APP_ECB_FETCH_ON_START", "false")
|
||||
os.environ.setdefault("APP_ADMIN_USERNAME", "admin")
|
||||
os.environ.setdefault("APP_ADMIN_PASSWORD", "adminpass")
|
||||
# Muss die Standard-Passwortrichtlinie erfüllen (Länge 10, Gross/Klein/Ziffer),
|
||||
# sonst verlangt der erste Login sofort einen Wechsel und alle Tests scheitern.
|
||||
os.environ.setdefault("APP_ADMIN_PASSWORD", "AdminPass1")
|
||||
os.environ.setdefault("APP_SESSION_SECRET", "test-secret")
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
ADMIN_PASSWORD = "AdminPass1"
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def client():
|
||||
@@ -25,15 +29,67 @@ def client():
|
||||
|
||||
@pytest.fixture()
|
||||
def admin(client):
|
||||
r = client.post("/api/auth/login", data={"username": "admin", "password": "adminpass"})
|
||||
r = client.post("/api/auth/login", data={"username": "admin", "password": ADMIN_PASSWORD})
|
||||
assert r.status_code == 200
|
||||
return client
|
||||
|
||||
|
||||
def new_password() -> str:
|
||||
"""Zufälliges, richtlinienkonformes Testpasswort."""
|
||||
return "Pw1_" + secrets.token_hex(4)
|
||||
|
||||
|
||||
def new_user(admin_client) -> tuple[str, str]:
|
||||
"""Legt (als Admin) einen frischen Nutzer an und gibt (username, passwort) zurück."""
|
||||
username = "u_" + secrets.token_hex(4)
|
||||
password = "pw_" + secrets.token_hex(4)
|
||||
password = new_password()
|
||||
r = admin_client.post("/api/users", json={"username": username, "password": password, "role": "user"})
|
||||
assert r.status_code == 201, r.text
|
||||
return username, password
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def policy_admin(admin):
|
||||
"""Admin-Client für Tests, die die Passwortrichtlinie verändern.
|
||||
|
||||
Die Testdatenbank lebt über den ganzen Testlauf und die Richtlinie ist
|
||||
global – ohne Zurücksetzen würde eine Verschärfung alle folgenden Tests
|
||||
kippen.
|
||||
|
||||
Zurückgesetzt wird bewusst direkt über die Dienstschicht, nicht über die
|
||||
API: wer ``mfa_required`` einschaltet, sperrt sich damit selbst aus
|
||||
``PUT /policy`` aus, bis er einen zweiten Faktor eingerichtet hat. Über die
|
||||
API bliebe die Verschärfung also stehen."""
|
||||
from app.database import SessionLocal
|
||||
from app.services import policy as policy_service
|
||||
|
||||
vorher = admin.get("/api/policy").json()
|
||||
try:
|
||||
yield admin
|
||||
finally:
|
||||
db = SessionLocal()
|
||||
try:
|
||||
policy_service.save(db, policy_service.Policy(**vorher))
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def login_ready(username: str, password: str) -> tuple[TestClient, str]:
|
||||
"""Meldet einen frisch angelegten Nutzer an und erledigt den Pflichtwechsel.
|
||||
|
||||
Vom Admin angelegte Konten tragen ``must_change_password``; solange das
|
||||
offen ist, sperrt der Server alle fachlichen Endpunkte. Für Tests, die den
|
||||
Nutzer einfach benutzen wollen, wird der Wechsel hier gleich mit erledigt.
|
||||
Gibt den angemeldeten Client und das nun gültige Passwort zurück."""
|
||||
from app.main import app
|
||||
|
||||
c = TestClient(app)
|
||||
r = c.post("/api/auth/login", data={"username": username, "password": password})
|
||||
assert r.status_code == 200, r.text
|
||||
neu = new_password()
|
||||
r = c.post(
|
||||
"/api/auth/change-password",
|
||||
json={"current_password": password, "new_password": neu},
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
return c, neu
|
||||
|
||||
@@ -0,0 +1,162 @@
|
||||
"""Anmeldung, Passwortrichtlinie und die serverseitige Sperre offener Pflichten."""
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from conftest import ADMIN_PASSWORD, login_ready, new_password, new_user
|
||||
|
||||
|
||||
def _client() -> TestClient:
|
||||
from app.main import app
|
||||
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
def test_login_wrong_password(client):
|
||||
r = client.post("/api/auth/login", data={"username": "admin", "password": "falsch"})
|
||||
assert r.status_code == 401
|
||||
|
||||
|
||||
def test_login_unknown_user(client):
|
||||
r = client.post("/api/auth/login", data={"username": "gibtsnicht", "password": "egal"})
|
||||
assert r.status_code == 401
|
||||
|
||||
|
||||
def test_login_returns_status_ok(client):
|
||||
r = client.post("/api/auth/login", data={"username": "admin", "password": ADMIN_PASSWORD})
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body["status"] == "ok"
|
||||
assert body["user"]["username"] == "admin"
|
||||
|
||||
|
||||
def test_inactive_account_cannot_log_in(admin):
|
||||
username, password = new_user(admin)
|
||||
liste = admin.get("/api/users").json()
|
||||
u = next(x for x in liste if x["username"] == username)
|
||||
assert admin.patch(f"/api/users/{u['id']}", json={"is_active": False}).status_code == 200
|
||||
|
||||
r = _client().post("/api/auth/login", data={"username": username, "password": password})
|
||||
assert r.status_code == 403
|
||||
|
||||
|
||||
def test_fresh_user_is_blocked_until_password_changed(admin):
|
||||
"""Vom Admin angelegte Konten kommen nur an die Wechsel-Endpunkte."""
|
||||
username, password = new_user(admin)
|
||||
c = _client()
|
||||
assert c.post("/api/auth/login", data={"username": username, "password": password}).status_code == 200
|
||||
|
||||
# Eigene Daten und der Wechsel selbst bleiben erreichbar …
|
||||
me = c.get("/api/auth/me").json()
|
||||
assert me["must_change_password"] is True
|
||||
assert me["must_change_password_reason"] == "temp"
|
||||
assert c.get("/api/policy").status_code == 200
|
||||
|
||||
# … alles Fachliche nicht.
|
||||
r = c.get("/api/accounts")
|
||||
assert r.status_code == 403
|
||||
assert r.json()["detail"] == "password_change_required"
|
||||
|
||||
neu = new_password()
|
||||
assert c.post(
|
||||
"/api/auth/change-password",
|
||||
json={"current_password": password, "new_password": neu},
|
||||
).status_code == 200
|
||||
assert c.get("/api/accounts").status_code == 200
|
||||
|
||||
|
||||
def test_change_password_checks_current(admin):
|
||||
username, password = new_user(admin)
|
||||
c, _gueltig = login_ready(username, password)
|
||||
r = c.post(
|
||||
"/api/auth/change-password",
|
||||
json={"current_password": "stimmt-nicht", "new_password": new_password()},
|
||||
)
|
||||
assert r.status_code == 400
|
||||
|
||||
|
||||
def test_change_password_rejects_same_password(admin):
|
||||
username, password = new_user(admin)
|
||||
c, gueltig = login_ready(username, password)
|
||||
r = c.post(
|
||||
"/api/auth/change-password",
|
||||
json={"current_password": gueltig, "new_password": gueltig},
|
||||
)
|
||||
assert r.status_code == 400
|
||||
|
||||
|
||||
def test_change_password_enforces_policy(admin):
|
||||
username, password = new_user(admin)
|
||||
c, gueltig = login_ready(username, password)
|
||||
# Zu kurz, keine Grossbuchstaben, keine Ziffer.
|
||||
r = c.post(
|
||||
"/api/auth/change-password",
|
||||
json={"current_password": gueltig, "new_password": "kurz"},
|
||||
)
|
||||
assert r.status_code == 400
|
||||
assert "Richtlinie" in r.json()["detail"]
|
||||
|
||||
|
||||
def test_create_user_enforces_policy(admin):
|
||||
r = admin.post("/api/users", json={"username": "schwach_x", "password": "abc", "role": "user"})
|
||||
assert r.status_code == 400
|
||||
|
||||
|
||||
def test_policy_is_admin_only(admin):
|
||||
other, _ = login_ready(*new_user(admin))
|
||||
assert other.get("/api/policy").status_code == 200
|
||||
assert other.put("/api/policy", json={"min_length": 8}).status_code == 403
|
||||
|
||||
|
||||
def test_tightened_policy_forces_change_on_next_login(policy_admin):
|
||||
"""Kernpunkt: der Hash gibt nichts her, aber beim Anmelden liegt das
|
||||
Passwort im Klartext vor – dort wird die Richtlinie geprüft."""
|
||||
username, password = new_user(policy_admin)
|
||||
c, gueltig = login_ready(username, password)
|
||||
assert c.get("/api/auth/me").json()["must_change_password"] is False
|
||||
|
||||
# Richtlinie verschärfen: das bisherige Passwort ist nun zu kurz.
|
||||
assert policy_admin.put("/api/policy", json={
|
||||
"min_length": 40, "require_upper": True, "require_lower": True,
|
||||
"require_digit": True, "require_special": False, "mfa_required": False,
|
||||
}).status_code == 200
|
||||
|
||||
zweiter = _client()
|
||||
assert zweiter.post(
|
||||
"/api/auth/login", data={"username": username, "password": gueltig}
|
||||
).status_code == 200
|
||||
me = zweiter.get("/api/auth/me").json()
|
||||
assert me["must_change_password"] is True
|
||||
assert me["must_change_password_reason"] == "policy"
|
||||
assert zweiter.get("/api/accounts").status_code == 403
|
||||
|
||||
|
||||
def test_policy_min_length_unter_untergrenze_wird_abgelehnt(policy_admin):
|
||||
vorher = policy_admin.get("/api/policy").json()
|
||||
assert policy_admin.put("/api/policy", json={"min_length": 4}).status_code == 422
|
||||
# Die gespeicherte Richtlinie bleibt unverändert.
|
||||
assert policy_admin.get("/api/policy").json() == vorher
|
||||
|
||||
|
||||
def test_change_password_revokes_other_sessions(admin):
|
||||
username, password = new_user(admin)
|
||||
erste, gueltig = login_ready(username, password)
|
||||
zweite = _client()
|
||||
assert zweite.post(
|
||||
"/api/auth/login", data={"username": username, "password": gueltig}
|
||||
).status_code == 200
|
||||
|
||||
neu = new_password()
|
||||
assert erste.post(
|
||||
"/api/auth/change-password",
|
||||
json={"current_password": gueltig, "new_password": neu},
|
||||
).status_code == 200
|
||||
|
||||
# Die andere Sitzung ist beendet, die eigene läuft weiter.
|
||||
assert zweite.get("/api/auth/me").status_code == 401
|
||||
assert erste.get("/api/auth/me").status_code == 200
|
||||
|
||||
|
||||
def test_logout_ends_session(admin):
|
||||
c, _ = login_ready(*new_user(admin))
|
||||
assert c.post("/api/auth/logout").status_code == 204
|
||||
assert c.get("/api/auth/me").status_code == 401
|
||||
@@ -1,7 +1,8 @@
|
||||
import json
|
||||
import secrets
|
||||
from datetime import date
|
||||
|
||||
from conftest import login_ready, new_user
|
||||
|
||||
|
||||
def _setup(admin):
|
||||
acc = admin.post("/api/accounts", json={"name": "Exp", "currency": "EUR"}).json()
|
||||
@@ -41,19 +42,13 @@ def test_export_all_json(admin):
|
||||
|
||||
|
||||
def test_db_backup_admin_only(admin):
|
||||
from fastapi.testclient import TestClient
|
||||
from app.main import app
|
||||
|
||||
# Admin darf.
|
||||
r = admin.get("/api/backup/db")
|
||||
assert r.status_code == 200
|
||||
assert r.content[:16].startswith(b"SQLite format 3")
|
||||
|
||||
# Normaler Nutzer darf nicht.
|
||||
u, p = "u_" + secrets.token_hex(4), "pw"
|
||||
admin.post("/api/users", json={"username": u, "password": p, "role": "user"})
|
||||
other = TestClient(app)
|
||||
other.post("/api/auth/login", data={"username": u, "password": p})
|
||||
other, _ = login_ready(*new_user(admin))
|
||||
assert other.get("/api/backup/db").status_code == 403
|
||||
# ...aber seinen eigenen Export schon (leer).
|
||||
assert other.get("/api/export/all.json").status_code == 200
|
||||
|
||||
@@ -1,21 +1,15 @@
|
||||
import secrets
|
||||
from datetime import date
|
||||
|
||||
from conftest import login_ready, new_user
|
||||
|
||||
|
||||
def test_users_are_isolated(admin):
|
||||
from fastapi.testclient import TestClient
|
||||
from app.main import app
|
||||
|
||||
# Admin legt ein Konto an.
|
||||
acc = admin.post("/api/accounts", json={"name": "AdminKonto", "currency": "EUR"}).json()
|
||||
|
||||
# Frischen Nutzer anlegen und getrennt einloggen.
|
||||
username = "iso_" + secrets.token_hex(4)
|
||||
password = "pw_" + secrets.token_hex(4)
|
||||
assert admin.post("/api/users", json={"username": username, "password": password, "role": "user"}).status_code == 201
|
||||
|
||||
other = TestClient(app)
|
||||
assert other.post("/api/auth/login", data={"username": username, "password": password}).status_code == 200
|
||||
username, password = new_user(admin)
|
||||
other, _ = login_ready(username, password)
|
||||
|
||||
# Sieht keine fremden Konten/Buchungen.
|
||||
assert other.get("/api/accounts").json() == []
|
||||
|
||||
@@ -0,0 +1,189 @@
|
||||
"""Voll-Backup: Export, Prüfansicht und Wiederherstellung."""
|
||||
|
||||
import json
|
||||
from datetime import date
|
||||
|
||||
from conftest import login_ready, new_user
|
||||
|
||||
|
||||
def _datensatz(c):
|
||||
"""Legt ein Konto mit zwei Buchungen an und gibt das Konto zurück."""
|
||||
acc = c.post("/api/accounts", json={"name": "Restore-Konto", "currency": "EUR"}).json()
|
||||
for betrag, wer in ((1234, "Laden A"), (5678, "Laden B")):
|
||||
c.post("/api/transactions", json={
|
||||
"type": "expense", "date": str(date.today()), "account_id": acc["id"],
|
||||
"amount_cents": betrag, "payee": wer,
|
||||
})
|
||||
return acc
|
||||
|
||||
|
||||
def _backup(c) -> dict:
|
||||
r = c.get("/api/export/backup.json")
|
||||
assert r.status_code == 200, r.text
|
||||
return json.loads(r.content)
|
||||
|
||||
|
||||
def _hochladen(c, daten: dict, pfad: str, mode: str | None = None):
|
||||
dateien = {"file": ("backup.json", json.dumps(daten).encode("utf-8"), "application/json")}
|
||||
if mode is None:
|
||||
return c.post(pfad, files=dateien)
|
||||
return c.post(pfad, files=dateien, data={"mode": mode})
|
||||
|
||||
|
||||
def test_backup_hat_umschlag_und_alle_tabellen(admin):
|
||||
_datensatz(admin)
|
||||
daten = _backup(admin)
|
||||
assert daten["format"] == "certain-finance-backup"
|
||||
assert daten["schema_version"] == 2
|
||||
assert daten["scope"] == "user"
|
||||
assert daten["username"] == "admin"
|
||||
# Beziehungen laufen über lokale IDs, nicht über Namen.
|
||||
konto = daten["data"]["accounts"][0]
|
||||
assert "local_id" in konto
|
||||
buchung = daten["data"]["transactions"][0]
|
||||
assert buchung["account_id"] in {a["local_id"] for a in daten["data"]["accounts"]}
|
||||
# Felder, die der alte Ansichts-Export verlor.
|
||||
assert "import_hash" in buchung
|
||||
assert "fx_rate" in buchung
|
||||
assert "budget_treatment" in buchung
|
||||
|
||||
|
||||
def test_pruefansicht_schreibt_nichts(admin):
|
||||
c, _ = login_ready(*new_user(admin))
|
||||
_datensatz(c)
|
||||
daten = _backup(c)
|
||||
vorher = len(c.get("/api/accounts").json())
|
||||
|
||||
r = _hochladen(c, daten, "/api/backup/restore/preview")
|
||||
assert r.status_code == 200, r.text
|
||||
body = r.json()
|
||||
assert body["scope"] == "user"
|
||||
tabellen = {z["key"]: z["count"] for z in body["tables"]}
|
||||
assert tabellen["accounts"] >= 1
|
||||
assert tabellen["transactions"] == 2
|
||||
assert len(c.get("/api/accounts").json()) == vorher
|
||||
|
||||
|
||||
def test_replace_stellt_geloeschte_daten_wieder_her(admin):
|
||||
c, _ = login_ready(*new_user(admin))
|
||||
acc = _datensatz(c)
|
||||
daten = _backup(c)
|
||||
konten_vorher = c.get("/api/accounts").json()
|
||||
buchungen_vorher = c.get("/api/transactions").json()
|
||||
|
||||
# Alles wegwerfen …
|
||||
assert c.delete(f"/api/accounts/{acc['id']}").status_code in (200, 204)
|
||||
assert c.get("/api/accounts").json() == []
|
||||
|
||||
# … und aus dem Backup zurückholen.
|
||||
r = _hochladen(c, daten, "/api/backup/restore", "replace")
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json()["mode"] == "replace"
|
||||
|
||||
konten = c.get("/api/accounts").json()
|
||||
buchungen = c.get("/api/transactions").json()
|
||||
assert len(konten) == len(konten_vorher)
|
||||
assert konten[0]["name"] == "Restore-Konto"
|
||||
assert len(buchungen) == len(buchungen_vorher)
|
||||
assert {b["payee"] for b in buchungen} == {"Laden A", "Laden B"}
|
||||
# Die Buchungen hängen wieder am wiederhergestellten Konto.
|
||||
assert {b["account_id"] for b in buchungen} == {konten[0]["id"]}
|
||||
|
||||
|
||||
def test_replace_ersetzt_statt_zu_ergaenzen(admin):
|
||||
c, _ = login_ready(*new_user(admin))
|
||||
_datensatz(c)
|
||||
daten = _backup(c)
|
||||
|
||||
# Zwischendurch etwas anlegen, das im Backup nicht vorkommt.
|
||||
c.post("/api/accounts", json={"name": "Danach angelegt", "currency": "EUR"})
|
||||
assert len(c.get("/api/accounts").json()) == 2
|
||||
|
||||
assert _hochladen(c, daten, "/api/backup/restore", "replace").status_code == 200
|
||||
namen = [a["name"] for a in c.get("/api/accounts").json()]
|
||||
assert namen == ["Restore-Konto"]
|
||||
|
||||
|
||||
def test_merge_erzeugt_keine_duplikate(admin):
|
||||
c, _ = login_ready(*new_user(admin))
|
||||
_datensatz(c)
|
||||
daten = _backup(c)
|
||||
|
||||
assert _hochladen(c, daten, "/api/backup/restore", "merge").status_code == 200
|
||||
assert len(c.get("/api/accounts").json()) == 1
|
||||
assert len(c.get("/api/transactions").json()) == 2
|
||||
|
||||
# Zweiter Durchlauf ändert weiterhin nichts.
|
||||
assert _hochladen(c, daten, "/api/backup/restore", "merge").status_code == 200
|
||||
assert len(c.get("/api/accounts").json()) == 1
|
||||
assert len(c.get("/api/transactions").json()) == 2
|
||||
|
||||
|
||||
def test_merge_ergaenzt_fehlendes(admin):
|
||||
geber, _ = login_ready(*new_user(admin))
|
||||
_datensatz(geber)
|
||||
daten = _backup(geber)
|
||||
|
||||
nehmer, _ = login_ready(*new_user(admin))
|
||||
nehmer.post("/api/accounts", json={"name": "Eigenes Konto", "currency": "EUR"})
|
||||
|
||||
assert _hochladen(nehmer, daten, "/api/backup/restore", "merge").status_code == 200
|
||||
namen = sorted(a["name"] for a in nehmer.get("/api/accounts").json())
|
||||
assert namen == ["Eigenes Konto", "Restore-Konto"]
|
||||
assert len(nehmer.get("/api/transactions").json()) == 2
|
||||
|
||||
|
||||
def test_fremde_version_wird_abgelehnt(admin):
|
||||
daten = _backup(admin)
|
||||
daten["schema_version"] = 99
|
||||
r = _hochladen(admin, daten, "/api/backup/restore/preview")
|
||||
assert r.status_code == 400
|
||||
assert "99" in r.json()["detail"]
|
||||
|
||||
|
||||
def test_ansichts_export_wird_erkannt(admin):
|
||||
_datensatz(admin)
|
||||
alt = json.loads(admin.get("/api/export/all.json").content)
|
||||
r = _hochladen(admin, alt, "/api/backup/restore/preview")
|
||||
assert r.status_code == 400
|
||||
assert "all.json" in r.json()["detail"]
|
||||
|
||||
|
||||
def test_fremde_datei_wird_abgelehnt(admin):
|
||||
r = _hochladen(admin, {"irgendwas": 1}, "/api/backup/restore/preview")
|
||||
assert r.status_code == 400
|
||||
|
||||
|
||||
def test_systembackup_nur_fuer_admin(admin):
|
||||
assert admin.get("/api/export/system-backup.json").status_code == 200
|
||||
other, _ = login_ready(*new_user(admin))
|
||||
assert other.get("/api/export/system-backup.json").status_code == 403
|
||||
|
||||
|
||||
def test_systembackup_enthaelt_alle_benutzer(admin):
|
||||
daten = json.loads(admin.get("/api/export/system-backup.json").content)
|
||||
assert daten["scope"] == "system"
|
||||
namen = {u["username"] for u in daten["users"]}
|
||||
assert "admin" in namen
|
||||
admin_satz = next(u for u in daten["users"] if u["username"] == "admin")
|
||||
# Für einen echten Wiederaufbau muss der Hash mit – deshalb der Hinweis
|
||||
# in der Oberfläche, die Datei wie ein Passwort zu behandeln.
|
||||
assert admin_satz["password_hash"].startswith("$argon2")
|
||||
assert "data" in admin_satz
|
||||
|
||||
|
||||
def test_systembackup_darf_kein_normaler_nutzer_einspielen(admin):
|
||||
daten = json.loads(admin.get("/api/export/system-backup.json").content)
|
||||
other, _ = login_ready(*new_user(admin))
|
||||
r = _hochladen(other, daten, "/api/backup/restore/preview")
|
||||
assert r.status_code == 403
|
||||
|
||||
|
||||
def test_systembackup_ohne_admin_wird_abgelehnt(admin):
|
||||
daten = json.loads(admin.get("/api/export/system-backup.json").content)
|
||||
daten["users"] = [u for u in daten["users"] if u["role"] != "admin"]
|
||||
r = _hochladen(admin, daten, "/api/backup/restore", "replace")
|
||||
assert r.status_code == 400
|
||||
assert "Administrator" in r.json()["detail"]
|
||||
# Nichts halb eingespielt: der Admin ist noch da.
|
||||
assert admin.get("/api/auth/me").status_code == 200
|
||||
@@ -0,0 +1,213 @@
|
||||
"""Zwei-Faktor: Einrichtung, zweistufige Anmeldung, Wiederherstellungscodes."""
|
||||
|
||||
import time
|
||||
|
||||
import pyotp
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from conftest import login_ready, new_user
|
||||
|
||||
|
||||
def _client() -> TestClient:
|
||||
from app.main import app
|
||||
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
def _code(secret: str, versatz: int = 0) -> str:
|
||||
"""Code für das aktuelle (oder ein verschobenes) Zeitfenster.
|
||||
|
||||
Jeder Code verbraucht sein Fenster dauerhaft, deshalb muss ein Test, der
|
||||
mehrfach anmeldet, jeweils ein späteres Fenster nehmen."""
|
||||
schritt = int(time.time()) // 30 + versatz
|
||||
return pyotp.TOTP(secret).at(schritt * 30)
|
||||
|
||||
|
||||
def _mit_2fa(admin) -> tuple[str, str, str]:
|
||||
"""Legt einen Nutzer an, richtet 2FA ein und gibt (name, passwort, secret) zurück.
|
||||
|
||||
Aktiviert wird mit dem Code des aktuellen Fensters – spätere Anmeldungen im
|
||||
Test brauchen deshalb ``versatz >= 1``."""
|
||||
username, password = new_user(admin)
|
||||
c, gueltig = login_ready(username, password)
|
||||
setup = c.post("/api/auth/2fa/setup").json()
|
||||
r = c.post("/api/auth/2fa/activate", json={"code": _code(setup["secret"])})
|
||||
assert r.status_code == 200, r.text
|
||||
return username, gueltig, setup["secret"]
|
||||
|
||||
|
||||
def test_setup_liefert_qr_und_secret(admin):
|
||||
c, _ = login_ready(*new_user(admin))
|
||||
r = c.post("/api/auth/2fa/setup")
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body["qr_png"].startswith("data:image/png;base64,")
|
||||
assert body["otpauth_uri"].startswith("otpauth://totp/")
|
||||
assert len(body["secret"]) >= 16
|
||||
# Noch nicht aktiv, solange kein Code bestätigt wurde.
|
||||
assert c.get("/api/auth/2fa/status").json()["enabled"] is False
|
||||
|
||||
|
||||
def test_activate_liefert_wiederherstellungscodes(admin):
|
||||
c, _ = login_ready(*new_user(admin))
|
||||
setup = c.post("/api/auth/2fa/setup").json()
|
||||
r = c.post("/api/auth/2fa/activate", json={"code": _code(setup["secret"])})
|
||||
assert r.status_code == 200
|
||||
codes = r.json()["codes"]
|
||||
assert len(codes) == 10
|
||||
assert len(set(codes)) == 10
|
||||
status = c.get("/api/auth/2fa/status").json()
|
||||
assert status["enabled"] is True
|
||||
assert status["recovery_codes_left"] == 10
|
||||
|
||||
|
||||
def test_activate_mit_falschem_code(admin):
|
||||
c, _ = login_ready(*new_user(admin))
|
||||
c.post("/api/auth/2fa/setup")
|
||||
assert c.post("/api/auth/2fa/activate", json={"code": "000000"}).status_code == 400
|
||||
|
||||
|
||||
def test_zweistufige_anmeldung(admin):
|
||||
username, password, secret = _mit_2fa(admin)
|
||||
|
||||
c = _client()
|
||||
r = c.post("/api/auth/login", data={"username": username, "password": password})
|
||||
assert r.status_code == 200
|
||||
assert r.json()["status"] == "2fa_required"
|
||||
assert r.json()["user"] is None
|
||||
|
||||
# Das Zwischen-Cookie ist kein Zugang.
|
||||
assert c.get("/api/auth/me").status_code == 401
|
||||
assert c.get("/api/accounts").status_code == 401
|
||||
|
||||
r = c.post("/api/auth/2fa/verify", json={"code": _code(secret, 1)})
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json()["status"] == "ok"
|
||||
assert r.json()["user"]["username"] == username
|
||||
assert c.get("/api/accounts").status_code == 200
|
||||
|
||||
|
||||
def test_code_gilt_nur_einmal(admin):
|
||||
username, password, secret = _mit_2fa(admin)
|
||||
code = _code(secret, 1)
|
||||
|
||||
c = _client()
|
||||
c.post("/api/auth/login", data={"username": username, "password": password})
|
||||
assert c.post("/api/auth/2fa/verify", json={"code": code}).status_code == 200
|
||||
|
||||
# Derselbe Code ein zweites Mal: das Fenster ist verbraucht.
|
||||
zweiter = _client()
|
||||
zweiter.post("/api/auth/login", data={"username": username, "password": password})
|
||||
assert zweiter.post("/api/auth/2fa/verify", json={"code": code}).status_code == 401
|
||||
|
||||
|
||||
def test_bereits_verbrauchtes_fenster_wird_abgelehnt(admin):
|
||||
"""Der Code, mit dem aktiviert wurde, taugt nicht mehr zum Anmelden."""
|
||||
username, password = new_user(admin)
|
||||
c, gueltig = login_ready(username, password)
|
||||
setup = c.post("/api/auth/2fa/setup").json()
|
||||
aktivierungscode = _code(setup["secret"])
|
||||
assert c.post("/api/auth/2fa/activate", json={"code": aktivierungscode}).status_code == 200
|
||||
|
||||
zweiter = _client()
|
||||
zweiter.post("/api/auth/login", data={"username": username, "password": gueltig})
|
||||
assert zweiter.post(
|
||||
"/api/auth/2fa/verify", json={"code": aktivierungscode}
|
||||
).status_code == 401
|
||||
|
||||
|
||||
def test_falscher_code_wird_abgelehnt(admin):
|
||||
username, password, _ = _mit_2fa(admin)
|
||||
c = _client()
|
||||
c.post("/api/auth/login", data={"username": username, "password": password})
|
||||
assert c.post("/api/auth/2fa/verify", json={"code": "000000"}).status_code == 401
|
||||
assert c.get("/api/auth/me").status_code == 401
|
||||
|
||||
|
||||
def test_zu_viele_fehlversuche_beenden_den_versuch(admin):
|
||||
username, password, secret = _mit_2fa(admin)
|
||||
c = _client()
|
||||
c.post("/api/auth/login", data={"username": username, "password": password})
|
||||
for _ in range(5):
|
||||
c.post("/api/auth/2fa/verify", json={"code": "000000"})
|
||||
# Die Pending-Sitzung ist verbraucht – auch ein richtiger Code hilft nicht mehr.
|
||||
r = c.post("/api/auth/2fa/verify", json={"code": _code(secret, 1)})
|
||||
assert r.status_code == 401
|
||||
|
||||
|
||||
def test_wiederherstellungscode_funktioniert_einmal(admin):
|
||||
username, password = new_user(admin)
|
||||
c, gueltig = login_ready(username, password)
|
||||
setup = c.post("/api/auth/2fa/setup").json()
|
||||
codes = c.post("/api/auth/2fa/activate", json={"code": _code(setup["secret"])}).json()["codes"]
|
||||
|
||||
erster = _client()
|
||||
erster.post("/api/auth/login", data={"username": username, "password": gueltig})
|
||||
r = erster.post("/api/auth/2fa/verify", json={"code": codes[0], "recovery": True})
|
||||
assert r.status_code == 200, r.text
|
||||
assert erster.get("/api/auth/2fa/status").json()["recovery_codes_left"] == 9
|
||||
|
||||
zweiter = _client()
|
||||
zweiter.post("/api/auth/login", data={"username": username, "password": gueltig})
|
||||
r = zweiter.post("/api/auth/2fa/verify", json={"code": codes[0], "recovery": True})
|
||||
assert r.status_code == 401
|
||||
|
||||
|
||||
def test_abschalten_verlangt_das_richtige_passwort(admin):
|
||||
username, password = new_user(admin)
|
||||
c, gueltig = login_ready(username, password)
|
||||
setup = c.post("/api/auth/2fa/setup").json()
|
||||
assert c.post("/api/auth/2fa/activate", json={"code": _code(setup["secret"])}).status_code == 200
|
||||
|
||||
assert c.post("/api/auth/2fa/disable", json={"password": "falsch"}).status_code == 400
|
||||
r = c.post("/api/auth/2fa/disable", json={"password": gueltig})
|
||||
assert r.status_code == 200
|
||||
assert r.json()["totp_enabled"] is False
|
||||
assert c.get("/api/auth/2fa/status").json()["recovery_codes_left"] == 0
|
||||
|
||||
|
||||
def test_admin_kann_zwei_faktor_zuruecksetzen(admin):
|
||||
username, password, secret = _mit_2fa(admin)
|
||||
u = next(x for x in admin.get("/api/users").json() if x["username"] == username)
|
||||
assert u["totp_enabled"] is True
|
||||
|
||||
r = admin.post(f"/api/users/{u['id']}/2fa/reset")
|
||||
assert r.status_code == 200
|
||||
assert r.json()["totp_enabled"] is False
|
||||
|
||||
# Danach genügt wieder das Passwort allein.
|
||||
c = _client()
|
||||
r = c.post("/api/auth/login", data={"username": username, "password": password})
|
||||
assert r.json()["status"] == "ok"
|
||||
assert secret # das alte Secret ist damit wertlos
|
||||
|
||||
|
||||
def test_mfa_pflicht_sperrt_bis_zur_einrichtung(policy_admin):
|
||||
username, password = new_user(policy_admin)
|
||||
c, gueltig = login_ready(username, password)
|
||||
|
||||
assert policy_admin.put("/api/policy", json={
|
||||
"min_length": 10, "require_upper": True, "require_lower": True,
|
||||
"require_digit": True, "require_special": False, "mfa_required": True,
|
||||
}).status_code == 200
|
||||
|
||||
zweiter = _client()
|
||||
assert zweiter.post(
|
||||
"/api/auth/login", data={"username": username, "password": gueltig}
|
||||
).status_code == 200
|
||||
me = zweiter.get("/api/auth/me").json()
|
||||
assert me["mfa_setup_required"] is True
|
||||
|
||||
r = zweiter.get("/api/accounts")
|
||||
assert r.status_code == 403
|
||||
assert r.json()["detail"] == "mfa_setup_required"
|
||||
|
||||
# Einrichten ist erlaubt und hebt die Sperre auf.
|
||||
setup = zweiter.post("/api/auth/2fa/setup").json()
|
||||
assert zweiter.post(
|
||||
"/api/auth/2fa/activate", json={"code": _code(setup["secret"])}
|
||||
).status_code == 200
|
||||
assert zweiter.get("/api/accounts").status_code == 200
|
||||
|
||||
# Abschalten bleibt verwehrt, solange die Pflicht gilt.
|
||||
assert zweiter.post("/api/auth/2fa/disable", json={"password": gueltig}).status_code == 400
|
||||
Reference in New Issue
Block a user