7788c74cfe
- Updated the .env.example file to reflect a development environment setup with enhanced secret key requirements and local host settings. - Modified the Docker Compose configuration to enhance security with read-only settings and no-new-privileges options. - Updated requirements.txt to pin package versions for better dependency management. - Enhanced the FastAPI application to include dynamic OpenAPI and documentation URLs based on the environment. - Implemented session versioning in JWT tokens to improve security and user session management. - Added new validation for user roles and password strength in schemas. - Improved email sending logic to handle recipient lists more robustly and added logging for SMTP operations. - Updated dashboard and profile templates to reflect new features and improve user experience.
55 lines
1.9 KiB
Python
55 lines
1.9 KiB
Python
from fastapi import Depends, HTTPException, Request, status
|
|
from jose import JWTError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.security import decode_access_token
|
|
from app.db.session import get_db
|
|
from app.models.user import User
|
|
|
|
|
|
def _user_from_token(token: str, db: Session) -> User | None:
|
|
try:
|
|
payload = decode_access_token(token)
|
|
user_id = int(payload.get("sub"))
|
|
token_sv = int(payload.get("sv", 0))
|
|
except (JWTError, TypeError, ValueError):
|
|
return None
|
|
|
|
user = db.query(User).filter(User.id == user_id, User.is_active.is_(True)).first()
|
|
if not user or (user.session_version or 0) != token_sv:
|
|
return None
|
|
return user
|
|
|
|
|
|
def get_optional_user(request: Request, db: Session = Depends(get_db)) -> User | None:
|
|
token = request.cookies.get("access_token")
|
|
if not token:
|
|
return None
|
|
return _user_from_token(token, db)
|
|
|
|
|
|
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.")
|
|
user = _user_from_token(token, db)
|
|
if not user:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Ungültiger oder abgelaufener Token.")
|
|
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 can_view_sensitive_logs(user: User) -> bool:
|
|
return user.role in {"admin", "editor"}
|