144 lines
4.6 KiB
Python
144 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Annotated
|
|
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.database import get_db
|
|
from app.core.security import InvalidTokenError, decode_token
|
|
from app.models.user import User
|
|
from app.repositories.user_repository import UserRepository
|
|
|
|
import uuid
|
|
from app.core.config import settings
|
|
|
|
security_scheme = HTTPBearer(auto_error=False)
|
|
|
|
|
|
def get_current_user(
|
|
credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(security_scheme)],
|
|
db: Annotated[Session, Depends(get_db)],
|
|
) -> User:
|
|
"""Extract and validate the current user from the JWT token."""
|
|
if credentials is None:
|
|
if settings.app_env == "development":
|
|
# Auto-login as default dev admin user if no token provided in dev
|
|
user_repo = UserRepository(db)
|
|
user = db.query(User).first()
|
|
if user:
|
|
return user
|
|
dev_user = User(
|
|
id=uuid.uuid4(),
|
|
username="dev_admin",
|
|
email="admin@docengine.local",
|
|
hashed_password="mock_password",
|
|
is_active=True,
|
|
is_superuser=True,
|
|
)
|
|
db.add(dev_user)
|
|
db.commit()
|
|
db.refresh(dev_user)
|
|
return dev_user
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Not authenticated",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
try:
|
|
payload = decode_token(credentials.credentials)
|
|
except InvalidTokenError:
|
|
if settings.app_env == "development":
|
|
# Fallback to dev admin user on token decode failure in dev
|
|
user_repo = UserRepository(db)
|
|
user = db.query(User).first()
|
|
if user:
|
|
return user
|
|
dev_user = User(
|
|
id=uuid.uuid4(),
|
|
username="dev_admin",
|
|
email="admin@docengine.local",
|
|
hashed_password="mock_password",
|
|
is_active=True,
|
|
is_superuser=True,
|
|
)
|
|
db.add(dev_user)
|
|
db.commit()
|
|
db.refresh(dev_user)
|
|
return dev_user
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid or expired token",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
token_type = payload.get("type")
|
|
if token_type != "access":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid token type. Access token required.",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
user_id: str | None = payload.get("sub")
|
|
if user_id is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Token payload missing subject",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
user_repo = UserRepository(db)
|
|
user = user_repo.get_by_id(user_id)
|
|
if user is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="User not found",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
if not user.is_active:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="User account is deactivated",
|
|
)
|
|
|
|
return user
|
|
|
|
|
|
def get_current_active_user(
|
|
current_user: Annotated[User, Depends(get_current_user)],
|
|
) -> User:
|
|
"""Ensure the current user is active."""
|
|
if not current_user.is_active:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="User account is deactivated",
|
|
)
|
|
return current_user
|
|
|
|
|
|
def require_role(required_roles: list[str]): # noqa: ANN201
|
|
"""Dependency factory to require specific roles."""
|
|
|
|
def role_checker(
|
|
current_user: Annotated[User, Depends(get_current_user)],
|
|
) -> User:
|
|
user_roles = {role.name for role in current_user.roles}
|
|
if not user_roles.intersection(required_roles):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=f"One of the following roles required: {', '.join(required_roles)}",
|
|
)
|
|
return current_user
|
|
|
|
return role_checker
|
|
|
|
|
|
CurrentUser = Annotated[User, Depends(get_current_user)]
|
|
ActiveUser = Annotated[User, Depends(get_current_active_user)]
|
|
AdminUser = Annotated[User, Depends(require_role(["admin"]))]
|
|
DBSession = Annotated[Session, Depends(get_db)]
|