Coverage for /usr/local/lib/python3.10/site-packages/opal_common-0.0.0-py3.10.egg/opal_common/authentication/deps.py: 58%

1from typing import Optional 

2from uuid import UUID 

3 

4from fastapi import Header 

5from fastapi.exceptions import HTTPException 

6from fastapi.security.utils import get_authorization_scheme_param 

7from opal_common.authentication.types import JWTClaims 

8from opal_common.authentication.verifier import JWTVerifier, Unauthorized 

9from opal_common.logger import logger 

10 

11 

12def get_token_from_header(authorization_header: str) -> Optional[str]: 

13 """Extracts a bearer token from an HTTP Authorization header. 

14 

15 when provided bearer token via websocket, we cannot use the fastapi 

16 built-in: oauth2_scheme. 

17 """ 

18 if not authorization_header: 18 ↛ 19line 18 didn't jump to line 19 because the condition on line 18 was never true

19 return None 

20 

21 scheme, token = get_authorization_scheme_param(authorization_header) 

22 if not token or scheme.lower() != "bearer": 22 ↛ 23line 22 didn't jump to line 23 because the condition on line 22 was never true

23 return None 

24 

25 return token 

26 

27 

28def verify_logged_in(verifier: JWTVerifier, token: Optional[str]) -> JWTClaims: 

29 """Forces bearer token authentication with valid JWT or throws 401.""" 

30 try: 

31 if not verifier.enabled: 31 ↛ 32line 31 didn't jump to line 32 because the condition on line 31 was never true

32 logger.debug("JWT verification disabled, cannot verify requests!") 

33 return {} 

34 if token is None: 34 ↛ 35line 34 didn't jump to line 35 because the condition on line 34 was never true

35 raise Unauthorized(description="access token was not provided") 

36 claims: JWTClaims = verifier.verify(token) 

37 subject = claims.get("sub", "") 

38 

39 invalid = Unauthorized(description="invalid sub claim") 

40 if not subject: 40 ↛ 41line 40 didn't jump to line 41 because the condition on line 40 was never true

41 raise invalid 

42 try: 

43 _ = UUID(subject) 

44 except ValueError: 

45 raise invalid 

46 

47 # returns the entire claims dict so we can do more checks on it if needed 

48 return claims or {} 

49 

50 except (Unauthorized, HTTPException) as err: 

51 # err.details is sometimes string and sometimes dict 

52 details: dict = {} 

53 if isinstance(err.detail, dict): 

54 details = err.detail.copy() 

55 elif isinstance(err.detail, str): 

56 details = {"msg": err.detail} 

57 else: 

58 details = {"msg": repr(err.detail)} 

59 

60 # pop the token before logging - tokens should not appear in logs 

61 details.pop("token", None) 

62 

63 # logs the error and reraises 

64 logger.error( 

65 f"Authentication failed with {err.status_code} due to error: {details}" 

66 ) 

67 raise 

68 

69 

70class _JWTAuthenticator: 

71 def __init__(self, verifier: JWTVerifier): 

72 self._verifier = verifier 

73 

74 @property 

75 def verifier(self) -> JWTVerifier: 

76 return self._verifier 

77 

78 @property 

79 def enabled(self) -> JWTVerifier: 

80 return self._verifier.enabled 

81 

82 

83class JWTAuthenticator(_JWTAuthenticator): 

84 """Bearer token authentication for http(s) api endpoints. 

85 

86 throws 401 if a valid jwt is not provided. 

87 """ 

88 

89 def __call__(self, authorization: Optional[str] = Header(None)) -> JWTClaims: 

90 token = get_token_from_header(authorization) 

91 return verify_logged_in(self._verifier, token) 

92 

93 

94class WebsocketJWTAuthenticator(_JWTAuthenticator): 

95 """Bearer token authentication for websocket endpoints. 

96 

97 with fastapi ws endpoint, we cannot throw http exceptions inside dependencies, 

98 because no matter the http status code, uvicorn will treat it as http 500. 

99 see: https://github.com/encode/uvicorn/blob/master/uvicorn/protocols/websockets/websockets_impl.py#L168 

100 

101 Instead we return the claims or None to the endpoint, in order for it to gracefully 

102 close the connection in case authentication was unsuccessful. 

103 

104 In this case uvicorn's hardcoded behavior suits us: 

105 - if websocket.accept() was called, http 200 will be sent 

106 - if websocket.close() was called instead, http 403 will be sent 

107 no other status code are supported. 

108 see: https://github.com/encode/uvicorn/blob/master/uvicorn/protocols/websockets/websockets_impl.py#L189-L207 

109 

110 thus we return a the claims or None and the endpoint can use it to potentially call websocket.close() 

111 """ 

112 

113 def __call__(self, authorization: Optional[str] = Header(None)) -> bool: 

114 token = get_token_from_header(authorization) 

115 try: 

116 return verify_logged_in(self._verifier, token) 

117 except (Unauthorized, HTTPException): 

118 return None 

119 

120 

121class StaticBearerAuthenticator: 

122 """Bearer token authentication for http(s) api endpoints. 

123 

124 throws 401 if token does not match a preconfigured value. 

125 """ 

126 

127 def __init__(self, preconfigured_token: Optional[str]): 

128 self._preconfigured_token = preconfigured_token 

129 

130 def __call__(self, authorization: Optional[str] = Header(None)): 

131 if self._preconfigured_token is None: 131 ↛ 133line 131 didn't jump to line 133 because the condition on line 131 was never true

132 # always allow 

133 return 

134 

135 if authorization is None: 135 ↛ 136line 135 didn't jump to line 136 because the condition on line 135 was never true

136 raise Unauthorized(description="Authorization header is required!") 

137 

138 token = get_token_from_header(authorization) 

139 if token is None or token != self._preconfigured_token: 139 ↛ 140line 139 didn't jump to line 140 because the condition on line 139 was never true

140 raise Unauthorized( 

141 token=token, description="unauthorized to access this endpoint!" 

142 )