pythonauthenticationfastapidecoratorrbac

role_required decorator for FastAPI route


Disclaimer and sorry words.. It's been quite a long time since I do not ask questions here and also I am a complete novice in FastAPI, so.. please do not judge too strong

I am playing with FastAPI authorization and wondering how can I protected my routes from user who are authenticated but do not have permission for that specific routes.

I have done some code that solves this.

Here is my routes:

@router.get('/student_route_only')
@role_required(allowed_roles=[UserRole(name='student')])
async def student_route_only(
    token: Annotated[str, Depends(oauth2_scheme)],
    auth_service: AuthService = Depends(get_auth_service),
    user_data: UserAuthResponse = None,
):
    return UserAuthResponse(
        user_id=user_data.user_id,
        role=user_data.role,
        name=user_data.name,
    )


@router.get('/routes_for_student_and_admin')
@role_required(allowed_roles=[UserRole(name='student'), UserRole(name='admin')])
async def routes_for_student_and_admin(
    token: Annotated[str, Depends(oauth2_scheme)],
    auth_service: AuthService = Depends(get_auth_service),
    user_data: UserAuthResponse = None,
):
    return UserAuthResponse(
        user_id=user_data.user_id,
        role=user_data.role,
        name=user_data.name,
    )

This are my pydantic model:

class UserAuthResponse(BaseModel):
    user_id: int
    role: str
    name: str

class UserRole(BaseModel):
    name: str

    @validator('name')
    def name_must_be_valid(cls, value):
        allowed_roles = ['admin', 'student', 'teacher']
        if value.lower() not in allowed_roles:
            raise ValueError(
                f"Invalid role. Allowed roles are: {', '.join(allowed_roles)}"
            )
        return value

and this are my decorator that does the job:

def role_required(allowed_roles: list[UserRole]):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            auth_service = kwargs.get('auth_service')
            token = kwargs.get('token')
            user_data = await auth_service.get_current_user(token)
            if not user_data or user_data.role not in [role.name for role in allowed_roles]:
                raise HTTPException(
                    status_code=status.HTTP_403_FORBIDDEN,
                    detail='This operation is forbidden for you',
                )
            kwargs['user_data'] = user_data  # pushing gotten user_data back
            return await func(*args, **kwargs)

        return wrapper

    return decorator

I have searched for the relatively simple decision for role route protecting in FastAPI but did not find something that are not hard to implement.

So I would like to ask you, is it okay to use such kind of code in production?

Because I think it is quite simple to use, for instance if I have to protect some route and make it available let's say only for admin I can simply do this:

@router.get('/admin_route_only')
@role_required(allowed_roles=[UserRole(name='admin')])  # and that's it. The decorator does the rest of the job
async def admin_route_only(
    token: Annotated[str, Depends(oauth2_scheme)],
    auth_service: AuthService = Depends(get_auth_service),
    user_data: UserAuthResponse = None,
):
    return UserAuthResponse(
        user_id=user_data.user_id,
        role=user_data.role,
        name=user_data.name,
    )

On the other hand my PyCharm does not like that auth_service and token in function are not used (but they are needed in the decorator). Would it be okay for other developers? And for linters?

And also is it okay to delegate authorization to decorator like this and then push the user data back through kwargs.. ?

Thank you very much in advance!


Solution

  • It is possible and convenient to use Dependencies for this purpose.

    You can implement dependency as class.

    class Authorization:
    
        def __init__(self, allowed_roles: list[UserRole]):
            self.allowed_roles = allowed_roles
    
        def __call__(
            self,
            token: : Annotated[str, Depends(oauth2_scheme),
            auth_service: AuthService = Depends(get_auth_service),
        ):
            user_data = await auth_service.get_current_user(token)
            if not user_data or user_data.role not in [role.name for role in self.allowed_roles]:
                raise HTTPException(
                    status_code=status.HTTP_403_FORBIDDEN,
                    detail='This operation is forbidden for you',
                )
    
            return user_data
    

    And then your routes:

    @router.get('/routes_for_student_and_admin')
    async def routes_for_student_and_admin(
        user_data: Annotated[UserAuthResponse, Depends(Authorization(allowed_roles=[UserRole(name='student')]))],
    ):
        return user_data
    
    

    Implement other endpoints in the same way with another list of UserRoles. Probably a bit edits are required to run this code. But I hope you got approach!