107 lines
3.3 KiB
Python
107 lines
3.3 KiB
Python
"""Shared FastAPI dependencies: DB session, current user, RBAC enforcement."""
|
|
import jwt
|
|
from fastapi import Depends, HTTPException, Request, status
|
|
from fastapi.security import OAuth2PasswordBearer
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.config import settings
|
|
from app.core.database import get_db
|
|
from app.core.security import ACCESS_TOKEN_TYPE, decode_token
|
|
from app.crud import user as user_crud
|
|
from app.models.user import User
|
|
|
|
# auto_error=False so we can fall back to the cookie used by the HTMX UI.
|
|
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login", auto_error=False)
|
|
|
|
|
|
class NotAuthenticated(Exception):
|
|
"""Raised by web dependencies; handled by redirecting to the login page."""
|
|
|
|
|
|
def _token_from_request(request: Request, bearer: str | None) -> str | None:
|
|
if bearer:
|
|
return bearer
|
|
return request.cookies.get(settings.cookie_name)
|
|
|
|
|
|
async def _user_from_token(token: str | None, db: AsyncSession) -> User | None:
|
|
if not token:
|
|
return None
|
|
try:
|
|
payload = decode_token(token)
|
|
except jwt.PyJWTError:
|
|
return None
|
|
if payload.get("type") != ACCESS_TOKEN_TYPE:
|
|
return None
|
|
sub = payload.get("sub")
|
|
if sub is None:
|
|
return None
|
|
user = await user_crud.get(db, int(sub))
|
|
if user is None or not user.is_active:
|
|
return None
|
|
return user
|
|
|
|
|
|
# --- JSON API auth -----------------------------------------------------------
|
|
|
|
async def get_current_user(
|
|
request: Request,
|
|
bearer: str | None = Depends(oauth2_scheme),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> User:
|
|
user = await _user_from_token(_token_from_request(request, bearer), db)
|
|
if user is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Not authenticated",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
return user
|
|
|
|
|
|
def require_permission(*required: str):
|
|
"""Dependency factory enforcing that the user holds all given permissions."""
|
|
|
|
async def checker(user: User = Depends(get_current_user)) -> User:
|
|
if user.is_superuser:
|
|
return user
|
|
missing = set(required) - user.permissions
|
|
if missing:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=f"Missing permission(s): {', '.join(sorted(missing))}",
|
|
)
|
|
return user
|
|
|
|
return checker
|
|
|
|
|
|
def require_superuser(user: User = Depends(get_current_user)) -> User:
|
|
if not user.is_superuser:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN, detail="Superuser required"
|
|
)
|
|
return user
|
|
|
|
|
|
# --- Web (HTMX) auth ---------------------------------------------------------
|
|
|
|
async def get_current_web_user(
|
|
request: Request,
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> User:
|
|
"""Like get_current_user but reads the cookie and redirects on failure."""
|
|
token = request.cookies.get(settings.cookie_name)
|
|
user = await _user_from_token(token, db)
|
|
if user is None:
|
|
raise NotAuthenticated()
|
|
return user
|
|
|
|
|
|
async def get_optional_web_user(
|
|
request: Request,
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> User | None:
|
|
token = request.cookies.get(settings.cookie_name)
|
|
return await _user_from_token(token, db)
|