Coverage for auth/auth_jwt.py: 45%
101 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:56 +0000
1import datetime
2import logging
3from typing import Optional
5from decouple import config
6from fastapi import Request
7from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
8from starlette import status
9from starlette.exceptions import HTTPException
11import schemas
12from chalicelib.core import authorizers, users, spot
14logger = logging.getLogger(__name__)
17def _get_jwt_leeway() -> datetime.timedelta:
18 days = config("JWT_LEEWAY_DAYS", default=None)
19 if days is not None:
20 return datetime.timedelta(days=int(days))
21 return datetime.timedelta(seconds=config("JWT_LEEWAY_S", cast=int, default=300))
24def _get_current_auth_context(request: Request, jwt_payload: dict) -> schemas.CurrentContext:
25 user = users.get_user(user_id=jwt_payload.get("userId", -1), tenant_id=jwt_payload.get("tenantId", -1))
26 if user is None: 26 ↛ 27line 26 didn't jump to line 27 because the condition on line 26 was never true
27 logger.warning("User not found.")
28 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="User not found.")
29 request.state.authorizer_identity = "jwt"
30 request.state.currentContext = schemas.CurrentContext(tenantId=jwt_payload.get("tenantId", -1),
31 userId=jwt_payload.get("userId", -1),
32 email=user["email"],
33 role=user["role"])
34 return request.state.currentContext
37class JWTAuth(HTTPBearer):
38 def __init__(self, auto_error: bool = True):
39 super(JWTAuth, self).__init__(auto_error=auto_error)
41 async def __call__(self, request: Request) -> Optional[schemas.CurrentContext]:
42 if request.url.path in ["/refresh", "/api/refresh"]: 42 ↛ 43line 42 didn't jump to line 43 because the condition on line 42 was never true
43 return await self.__process_refresh_call(request)
45 elif request.url.path in ["/spot/refresh", "/api/spot/refresh"]:
46 return await self.__process_spot_refresh_call(request)
48 else:
49 credentials: HTTPAuthorizationCredentials = await super(JWTAuth, self).__call__(request)
50 if credentials: 50 ↛ 81line 50 didn't jump to line 81 because the condition on line 50 was always true
51 if not credentials.scheme == "Bearer": 51 ↛ 52line 51 didn't jump to line 52 because the condition on line 51 was never true
52 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
53 detail="Invalid authentication scheme.")
54 jwt_payload = authorizers.jwt_authorizer(scheme=credentials.scheme, token=credentials.credentials)
55 auth_exists = jwt_payload is not None and users.auth_exists(user_id=jwt_payload.get("userId", -1),
56 jwt_iat=jwt_payload.get("iat", 100))
57 if jwt_payload is None \
58 or jwt_payload.get("iat") is None or jwt_payload.get("aud") is None \
59 or not auth_exists:
60 if jwt_payload is not None:
61 logger.debug(jwt_payload)
62 if jwt_payload.get("iat") is None: 62 ↛ 63line 62 didn't jump to line 63 because the condition on line 62 was never true
63 logger.debug("iat is None")
64 if jwt_payload.get("aud") is None: 64 ↛ 65line 64 didn't jump to line 65 because the condition on line 64 was never true
65 logger.debug("aud is None")
66 if not auth_exists: 66 ↛ 69line 66 didn't jump to line 69 because the condition on line 66 was always true
67 logger.warning("not users.auth_exists")
69 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Invalid token or expired token.")
71 if jwt_payload.get("aud", "").startswith("spot") and not request.url.path.startswith("/spot"): 71 ↛ 73line 71 didn't jump to line 73 because the condition on line 71 was never true
72 # Allow access to spot endpoints only
73 raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
74 detail="Unauthorized access (spot).")
75 elif jwt_payload.get("aud", "").startswith("front") and request.url.path.startswith("/spot"):
76 raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
77 detail="Unauthorized access endpoint reserved for Spot only.")
79 return _get_current_auth_context(request=request, jwt_payload=jwt_payload)
81 logger.warning("Invalid authorization code.")
82 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid authorization code.")
84 async def __process_refresh_call(self, request: Request) -> schemas.CurrentContext:
85 if "refreshToken" not in request.cookies:
86 logger.warning("Missing refreshToken cookie.")
87 jwt_payload = None
88 else:
89 jwt_payload = authorizers.jwt_refresh_authorizer(scheme="Bearer", token=request.cookies["refreshToken"])
91 if jwt_payload is None or jwt_payload.get("jti") is None:
92 logger.warning("Null refreshToken's payload, or null JTI.")
93 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN,
94 detail="Invalid refresh-token or expired refresh-token.")
95 auth_exists = users.refresh_auth_exists(user_id=jwt_payload.get("userId", -1),
96 jwt_jti=jwt_payload["jti"])
97 if not auth_exists:
98 logger.warning("refreshToken's user not found.")
99 logger.warning(jwt_payload)
100 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN,
101 detail="Invalid refresh-token or expired refresh-token.")
103 credentials: HTTPAuthorizationCredentials = await super(JWTAuth, self).__call__(request)
104 if credentials:
105 if not credentials.scheme == "Bearer":
106 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
107 detail="Invalid authentication scheme.")
108 old_jwt_payload = authorizers.jwt_authorizer(scheme=credentials.scheme, token=credentials.credentials,
109 leeway=_get_jwt_leeway())
110 if old_jwt_payload is None \
111 or old_jwt_payload.get("userId") is None \
112 or old_jwt_payload.get("userId") != jwt_payload.get("userId"):
113 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Invalid token or expired token.")
115 return _get_current_auth_context(request=request, jwt_payload=jwt_payload)
117 logger.warning("Invalid authorization code (refresh logic).")
118 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid authorization code for refresh.")
120 async def __process_spot_refresh_call(self, request: Request) -> schemas.CurrentContext:
121 if "spotRefreshToken" not in request.cookies: 121 ↛ 125line 121 didn't jump to line 125 because the condition on line 121 was always true
122 logger.warning("Missing spotRefreshToken cookie.")
123 jwt_payload = None
124 else:
125 jwt_payload = authorizers.jwt_refresh_authorizer(scheme="Bearer", token=request.cookies["spotRefreshToken"])
127 if jwt_payload is None or jwt_payload.get("jti") is None: 127 ↛ 131line 127 didn't jump to line 131 because the condition on line 127 was always true
128 logger.warning("Null spotRefreshToken's payload, or null JTI.")
129 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN,
130 detail="Invalid spotRefreshToken or expired refresh-token.")
131 auth_exists = spot.refresh_auth_exists(user_id=jwt_payload.get("userId", -1),
132 jwt_jti=jwt_payload["jti"])
133 if not auth_exists:
134 logger.warning("spotRefreshToken's user not found.")
135 logger.warning(jwt_payload)
136 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN,
137 detail="Invalid spotRefreshToken or expired refresh-token.")
139 credentials: HTTPAuthorizationCredentials = await super(JWTAuth, self).__call__(request)
140 if credentials:
141 if not credentials.scheme == "Bearer":
142 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
143 detail="Invalid spot-authentication scheme.")
144 old_jwt_payload = authorizers.jwt_authorizer(scheme=credentials.scheme, token=credentials.credentials,
145 leeway=_get_jwt_leeway())
146 if old_jwt_payload is None \
147 or old_jwt_payload.get("userId") is None \
148 or old_jwt_payload.get("userId") != jwt_payload.get("userId"):
149 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN,
150 detail="Invalid spot-token or expired token.")
152 return _get_current_auth_context(request=request, jwt_payload=jwt_payload)
154 logger.warning("Invalid authorization code (spot-refresh logic).")
155 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
156 detail="Invalid authorization code for spot-refresh.")