Coverage for chalicelib/core/authorizers.py: 49%

1import logging 

2 

3import jwt 

4from decouple import config 

5 

6from chalicelib.core import tenants 

7from chalicelib.core import users, spot 

8from chalicelib.utils.TimeUTC import TimeUTC 

9from chalicelib.utils.log import sanitize 

10 

11logger = logging.getLogger(__name__) 

12 

13 

14def get_supported_audience(): 

15 return [users.AUDIENCE, spot.AUDIENCE] 

16 

17 

18def is_spot_token(token: str) -> bool: 

19 try: 

20 if len(token) < 5 or "." not in token: 20 ↛ 21line 20 didn't jump to line 21 because the condition on line 20 was never true

21 return False 

22 decoded_token = jwt.decode(token, options={"verify_signature": False, "verify_exp": False}) 

23 audience = decoded_token.get("aud") 

24 return audience == spot.AUDIENCE 

25 except jwt.InvalidTokenError: 

26 logger.error(f"Invalid token for is_spot_token: {sanitize(token, max_length=16)}...") 

27 raise 

28 

29 

30def jwt_authorizer(scheme: str, token: str, leeway=0) -> dict | None: 

31 if scheme.lower() != "bearer" or len(token) < 5 or "." not in token: 

32 return None 

33 if not token or token.count(".") != 2: 33 ↛ 34line 33 didn't jump to line 34 because the condition on line 33 was never true

34 logger.debug("! JWT Malformed token") 

35 return None 

36 try: 

37 payload = jwt.decode(jwt=token, 

38 key=config("JWT_SECRET") if not is_spot_token(token) else config("JWT_SPOT_SECRET"), 

39 algorithms=config("JWT_ALGORITHM"), 

40 audience=get_supported_audience(), 

41 leeway=leeway) 

42 except jwt.ExpiredSignatureError: 

43 logger.debug("! JWT Expired signature") 

44 return None 

45 except jwt.exceptions.InvalidSignatureError: 

46 logger.warning("! JWT Signature verification failed") 

47 return None 

48 except BaseException as e: 

49 logger.warning("! JWT Base Exception", exc_info=e) 

50 return None 

51 return payload 

52 

53 

54def jwt_refresh_authorizer(scheme: str, token: str): 

55 if scheme.lower() != "bearer" or len(token) < 5 or "." not in token: 

56 return None 

57 if not token or token.count(".") != 2: 

58 logger.debug("! JWT-refresh Malformed token") 

59 logger.debug(token) 

60 return None 

61 try: 

62 payload = jwt.decode(jwt=token, 

63 key=config("JWT_REFRESH_SECRET") if not is_spot_token(token) \ 

64 else config("JWT_SPOT_REFRESH_SECRET"), 

65 algorithms=config("JWT_ALGORITHM"), 

66 audience=get_supported_audience()) 

67 except jwt.ExpiredSignatureError: 

68 logger.debug("! JWT-refresh Expired signature") 

69 return None 

70 except jwt.exceptions.InvalidSignatureError: 

71 logger.warning("! JWT-refresh Signature verification failed") 

72 return None 

73 except BaseException as e: 

74 logger.error("! JWT-refresh Base Exception", exc_info=e) 

75 return None 

76 return payload 

77 

78 

79def generate_jwt(user_id, tenant_id, iat, aud, for_spot=False): 

80 token = jwt.encode( 

81 payload={ 

82 "userId": user_id, 

83 "tenantId": tenant_id, 

84 "exp": iat + (config("JWT_EXPIRATION", cast=int) if not for_spot 

85 else config("JWT_SPOT_EXPIRATION", cast=int)), 

86 "iss": config("JWT_ISSUER"), 

87 "iat": iat, 

88 "aud": aud 

89 }, 

90 key=config("JWT_SECRET") if not for_spot else config("JWT_SPOT_SECRET"), 

91 algorithm=config("JWT_ALGORITHM") 

92 ) 

93 return token 

94 

95 

96def generate_jwt_refresh(user_id, tenant_id, iat, aud, jwt_jti, for_spot=False): 

97 token = jwt.encode( 

98 payload={ 

99 "userId": user_id, 

100 "tenantId": tenant_id, 

101 "exp": iat + (config("JWT_REFRESH_EXPIRATION", cast=int) if not for_spot 

102 else config("JWT_SPOT_REFRESH_EXPIRATION", cast=int)), 

103 "iss": config("JWT_ISSUER"), 

104 "iat": iat, 

105 "aud": aud, 

106 "jti": jwt_jti 

107 }, 

108 key=config("JWT_REFRESH_SECRET") if not for_spot else config("JWT_SPOT_REFRESH_SECRET"), 

109 algorithm=config("JWT_ALGORITHM") 

110 ) 

111 return token 

112 

113 

114def api_key_authorizer(token): 

115 t = tenants.get_by_api_key(token) 

116 if t is not None: 116 ↛ 117line 116 didn't jump to line 117 because the condition on line 116 was never true

117 t["createdAt"] = TimeUTC.datetime_to_timestamp(t["createdAt"]) 

118 return t