Add sudo-style elevation for superadmin actions

CurrentSuperuser now requires a short-lived (5 min) elevated JWT
claim in addition to the is_superadmin flag, obtained via the new
POST /auth/elevate (no re-authentication — the user is already the
only superadmin in practice; this is a deliberate-confirmation guard
against accidentally triggering a destructive action, not a defense
against a stolen session). Regular login tokens keep working unchanged
for all non-superadmin routes; DELETE /participants/{id} from task 5
now needs a fresh /auth/elevate call, no other code changes required
since CurrentSuperuser's meaning changed underneath it.

create_access_token gained an extra_claims param (backward compatible)
to carry the elevated flag.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Carsten Gram 2026-09-28 13:19:35 +02:00
parent ded23efc6a
commit 0ee6ef2e68
5 changed files with 36 additions and 3 deletions

View file

@ -2,3 +2,4 @@ DATABASE_URL=postgresql+psycopg://user:password@localhost:5432/vinindkoeb
ENVIRONMENT=development ENVIRONMENT=development
SECRET_KEY=change-me # generér med: openssl rand -hex 32 SECRET_KEY=change-me # generér med: openssl rand -hex 32
ACCESS_TOKEN_EXPIRE_MINUTES=120 ACCESS_TOKEN_EXPIRE_MINUTES=120
ELEVATION_EXPIRE_MINUTES=5

View file

@ -8,6 +8,7 @@ class Settings(BaseSettings):
environment: str = "development" environment: str = "development"
secret_key: str secret_key: str
access_token_expire_minutes: int = 120 access_token_expire_minutes: int = 120
elevation_expire_minutes: int = 5
model_config = SettingsConfigDict( model_config = SettingsConfigDict(
env_file=".env", env_file=".env",

View file

@ -17,12 +17,19 @@ def verify_password(plain: str, hashed: str) -> bool:
return _password_hash.verify(plain, hashed) 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() settings = get_settings()
expire = datetime.now(timezone.utc) + ( expire = datetime.now(timezone.utc) + (
expires_delta or timedelta(minutes=settings.access_token_expire_minutes) 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: def decode_access_token(token: str) -> dict | None:

View file

@ -31,9 +31,18 @@ def get_current_user(session: SessionDep, token: Annotated[str, Depends(oauth2_s
CurrentUser = Annotated[User, Depends(get_current_user)] 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: if not current_user.is_superadmin:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough privileges") 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 return current_user

View file

@ -1,9 +1,11 @@
from datetime import timedelta
from typing import Annotated from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, status from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm from fastapi.security import OAuth2PasswordRequestForm
from sqlmodel import SQLModel, select from sqlmodel import SQLModel, select
from app.core.config import get_settings
from app.core.security import create_access_token, verify_password from app.core.security import create_access_token, verify_password
from app.db import SessionDep from app.db import SessionDep
from app.dependencies import CurrentUser from app.dependencies import CurrentUser
@ -35,3 +37,16 @@ def login(
@router.get("/me", response_model=UserPublic) @router.get("/me", response_model=UserPublic)
def me(current_user: CurrentUser) -> User: def me(current_user: CurrentUser) -> User:
return current_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)