a1167df4f1
Backend: - User-Modell (sqlmodel) mit bcrypt-Passwort-Hash und is_admin-Flag - Job.owner_id (FK auf user.id), per Migration nachgerüstet (SQLite ALTER) - Starlette SessionMiddleware mit signiertem Cookie (SESSION_SECRET) - /api/auth: login / logout / me / needs-setup / setup (Initial-Admin) - /api/admin: User-CRUD + Passwort-Reset (Admin-only); letzter Admin / eigener Account vor Löschung geschützt - /api/jobs: Owner-Filter (Admin sieht alle inkl. owner_username), 404 statt 403 bei fremden Jobs (kein Existenz-Leak) Frontend: - Auth-Overlay mit Login bzw. Erst-Setup-Modus (auto-Detection) - Header zeigt User + Logout, Admin-Badge bei Admin-Konten - Admin-Karte: User anlegen, Passwort zurücksetzen, Löschen (mit Confirm) - 401 → Login-Overlay (auch bei XHR-Upload) Deploy: - install.sh generiert SESSION_SECRET einmalig per python-secrets - .env wird auf chmod 600 gesetzt - Erstaufruf zeigt automatisch das Setup-Formular; bestehende Jobs werden dem ersten Admin zugeordnet
46 lines
1.5 KiB
Python
46 lines
1.5 KiB
Python
"""Auth-Helfer: Passwort-Hashing, current_user-Dependency, Admin-Check."""
|
|
from __future__ import annotations
|
|
|
|
import bcrypt
|
|
from fastapi import Depends, HTTPException, Request, status
|
|
from sqlmodel import Session, select
|
|
|
|
from app.core.db import get_session
|
|
from app.models.user import User
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt(rounds=12)).decode("utf-8")
|
|
|
|
|
|
def verify_password(password: str, password_hash: str) -> bool:
|
|
try:
|
|
return bcrypt.checkpw(password.encode("utf-8"), password_hash.encode("utf-8"))
|
|
except (ValueError, TypeError):
|
|
return False
|
|
|
|
|
|
def current_user(request: Request, session: Session = Depends(get_session)) -> User:
|
|
user_id = request.session.get("user_id")
|
|
if not user_id:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Nicht angemeldet")
|
|
user = session.get(User, user_id)
|
|
if not user:
|
|
request.session.clear()
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Sitzung ungültig")
|
|
return user
|
|
|
|
|
|
def admin_required(user: User = Depends(current_user)) -> User:
|
|
if not user.is_admin:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin-Rechte erforderlich")
|
|
return user
|
|
|
|
|
|
def get_user_by_username(session: Session, username: str) -> User | None:
|
|
return session.exec(select(User).where(User.username == username)).first()
|
|
|
|
|
|
def has_any_user(session: Session) -> bool:
|
|
return session.exec(select(User).limit(1)).first() is not None
|