Coverage for auth/auth_project.py: 94%

1import logging 

2 

3from fastapi import Request 

4from starlette import status 

5from starlette.exceptions import HTTPException 

6 

7import schemas 

8from chalicelib.core import projects 

9from chalicelib.utils.log import sanitize 

10from or_dependencies import OR_context 

11 

12logger = logging.getLogger(__name__) 

13 

14 

15class ProjectAuthorizer: 

16 def __init__(self, project_identifier): 

17 self.project_identifier: str = project_identifier 

18 

19 async def __call__(self, request: Request) -> None: 

20 if len(request.path_params.keys()) == 0 or request.path_params.get(self.project_identifier) is None: 

21 return 

22 current_user: schemas.CurrentContext = await OR_context(request) 

23 value = request.path_params[self.project_identifier] 

24 current_project = None 

25 if self.project_identifier == "projectId" \ 

26 and (isinstance(value, int) or isinstance(value, str) and value.isnumeric()): 

27 current_project = projects.get_project(project_id=value, tenant_id=current_user.tenant_id) 

28 elif self.project_identifier == "projectKey": 28 ↛ 29line 28 didn't jump to line 29 because the condition on line 28 was never true

29 current_project = projects.get_by_project_key(project_key=value) 

30 

31 if current_project is None: 

32 logger.debug(f"unauthorized project {self.project_identifier}:{sanitize(value)}") 

33 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="project not found.") 

34 else: 

35 current_project = schemas.ProjectContext(projectId=current_project["projectId"], 

36 projectKey=current_project["projectKey"], 

37 platform=current_project["platform"], 

38 name=current_project["name"]) 

39 request.state.currentContext.project = current_project