Refactor configuration and security features for improved development experience
- 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.
This commit is contained in:
@@ -1,42 +1,45 @@
|
||||
from fastapi import Depends, HTTPException, Request, status
|
||||
from jose import JWTError, jwt
|
||||
from jose import JWTError
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core.config import settings
|
||||
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
|
||||
try:
|
||||
payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
|
||||
user_id = int(payload.get("sub"))
|
||||
except (JWTError, TypeError, ValueError):
|
||||
return None
|
||||
return db.query(User).filter(User.id == user_id, User.is_active.is_(True)).first()
|
||||
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.")
|
||||
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()
|
||||
user = _user_from_token(token, db)
|
||||
if not user:
|
||||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Benutzer nicht gefunden.")
|
||||
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" and not current_user.is_admin:
|
||||
if current_user.role != "admin":
|
||||
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin-Rechte erforderlich.")
|
||||
return current_user
|
||||
|
||||
@@ -47,7 +50,5 @@ def require_editor_or_admin(current_user: User = Depends(get_current_user)) -> U
|
||||
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
|
||||
def can_view_sensitive_logs(user: User) -> bool:
|
||||
return user.role in {"admin", "editor"}
|
||||
|
||||
Reference in New Issue
Block a user