diff --git a/.env.example b/.env.example index fd9f9cf..3083dc4 100644 --- a/.env.example +++ b/.env.example @@ -2,3 +2,4 @@ DATABASE_URL=postgresql+psycopg://user:password@localhost:5432/vinindkoeb ENVIRONMENT=development SECRET_KEY=change-me # generér med: openssl rand -hex 32 ACCESS_TOKEN_EXPIRE_MINUTES=120 +ELEVATION_EXPIRE_MINUTES=5 diff --git a/app/core/config.py b/app/core/config.py index 2ec5fa7..7670015 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -8,6 +8,7 @@ class Settings(BaseSettings): environment: str = "development" secret_key: str access_token_expire_minutes: int = 120 + elevation_expire_minutes: int = 5 model_config = SettingsConfigDict( env_file=".env", diff --git a/app/core/security.py b/app/core/security.py index ebb4c7b..736f7fa 100644 --- a/app/core/security.py +++ b/app/core/security.py @@ -17,12 +17,19 @@ def verify_password(plain: str, hashed: str) -> bool: return _password_hash.verify(plain, hashed) -def create_access_token(subject: str, expires_delta: timedelta | None = None) -> str: +def create_access_token( + subject: str, + expires_delta: timedelta | None = None, + extra_claims: dict | None = None, +) -> str: settings = get_settings() expire = datetime.now(timezone.utc) + ( expires_delta or timedelta(minutes=settings.access_token_expire_minutes) ) - return jwt.encode({"sub": subject, "exp": expire}, settings.secret_key, algorithm=ALGORITHM) + payload = {"sub": subject, "exp": expire} + if extra_claims: + payload.update(extra_claims) + return jwt.encode(payload, settings.secret_key, algorithm=ALGORITHM) def decode_access_token(token: str) -> dict | None: diff --git a/app/dependencies.py b/app/dependencies.py index e910e63..c8bdfc4 100644 --- a/app/dependencies.py +++ b/app/dependencies.py @@ -31,9 +31,18 @@ def get_current_user(session: SessionDep, token: Annotated[str, Depends(oauth2_s CurrentUser = Annotated[User, Depends(get_current_user)] -def get_current_active_superuser(current_user: CurrentUser) -> User: +def get_current_active_superuser( + current_user: CurrentUser, + token: Annotated[str, Depends(oauth2_scheme)], +) -> User: if not current_user.is_superadmin: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough privileges") + payload = decode_access_token(token) + if not payload or not payload.get("elevated"): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="This action requires an elevated session — call POST /auth/elevate first", + ) return current_user diff --git a/app/routers/auth.py b/app/routers/auth.py index 0ba6a57..dc55eb2 100644 --- a/app/routers/auth.py +++ b/app/routers/auth.py @@ -1,9 +1,11 @@ +from datetime import timedelta from typing import Annotated from fastapi import APIRouter, Depends, HTTPException, status from fastapi.security import OAuth2PasswordRequestForm from sqlmodel import SQLModel, select +from app.core.config import get_settings from app.core.security import create_access_token, verify_password from app.db import SessionDep from app.dependencies import CurrentUser @@ -35,3 +37,16 @@ def login( @router.get("/me", response_model=UserPublic) def me(current_user: CurrentUser) -> User: return current_user + + +@router.post("/elevate") +def elevate(current_user: CurrentUser) -> Token: + if not current_user.is_superadmin: + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="User is not a superadmin") + settings = get_settings() + token = create_access_token( + subject=str(current_user.id), + expires_delta=timedelta(minutes=settings.elevation_expire_minutes), + extra_claims={"elevated": True}, + ) + return Token(access_token=token)