42 lines
1.8 KiB
Python
42 lines
1.8 KiB
Python
from fastapi import Depends, HTTPException, Request, status
|
|
from jose import JWTError, jwt
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.config import settings
|
|
from app.db.session import get_db
|
|
from app.models.user import User
|
|
|
|
|
|
def get_current_user(request: Request, db: Session = Depends(get_db)) -> User:
|
|
token = request.cookies.get("access_token")
|
|
if not token:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Nicht angemeldet.")
|
|
try:
|
|
payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
|
|
user_id = int(payload.get("sub"))
|
|
except (JWTError, TypeError, ValueError) as exc:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Ungültiger Token.") from exc
|
|
|
|
user = db.query(User).filter(User.id == user_id, User.is_active.is_(True)).first()
|
|
if not user:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Benutzer nicht gefunden.")
|
|
return user
|
|
|
|
|
|
def get_admin_user(current_user: User = Depends(get_current_user)) -> User:
|
|
if current_user.role != "admin":
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin-Rechte erforderlich.")
|
|
return current_user
|
|
|
|
|
|
def require_editor_or_admin(current_user: User = Depends(get_current_user)) -> User:
|
|
if current_user.role not in {"admin", "editor"}:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Editor- oder Admin-Rechte erforderlich.")
|
|
return current_user
|
|
|
|
|
|
def require_reader_or_higher(current_user: User = Depends(get_current_user)) -> User:
|
|
if current_user.role not in {"admin", "editor", "reader"}:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin-Rechte erforderlich.")
|
|
return current_user
|