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

1from typing import Optional 

2 

3import jwt 

4from fastapi import HTTPException, status 

5from jwt.algorithms import Algorithm, get_default_algorithms 

6from jwt.api_jwk import PyJWK 

7from opal_common.authentication.types import JWTAlgorithm, JWTClaims, PublicKey 

8from opal_common.logger import logger 

9 

10 

11class Unauthorized(HTTPException): 

12 """HTTP 401 Unauthorized exception.""" 

13 

14 def __init__(self, description="Bearer token is not valid!", **kwargs): 

15 super().__init__( 

16 status_code=status.HTTP_401_UNAUTHORIZED, 

17 detail={"error": description, **kwargs}, 

18 headers={"WWW-Authenticate": "Bearer"}, 

19 ) 

20 

21 

22class JWTVerifier: 

23 """Given a cryptographic public key, can verify jwt tokens.""" 

24 

25 def __init__( 

26 self, 

27 public_key: Optional[PublicKey], 

28 algorithm: JWTAlgorithm, 

29 audience: str, 

30 issuer: str, 

31 ): 

32 """Inits the signer if and only if the keys provided to __init__ were 

33 generate together are are valid. otherwise will throw. 

34 

35 JWT verifier can be initialized without a public key (None) 

36 in which case verifier.enabled == False and jwt verification is turned off. 

37 

38 This allows opal to run both in secure mode (with jwt-based authentication) 

39 and in insecure mode (intended for development environments and running locally). 

40 

41 Args: 

42 public_key (PublicKey): a valid public key or None 

43 algorithm (JWTAlgorithm): the jwt algorithm to use 

44 (possible values: https://pyjwt.readthedocs.io/en/stable/algorithms.html) 

45 audience (string): the value for the aud claim: https://tools.ietf.org/html/rfc7519#section-4.1.3 

46 issuer (string): the value for the iss claim: https://tools.ietf.org/html/rfc7519#section-4.1.1 

47 """ 

48 self._public_key = public_key 

49 self._algorithm: str = algorithm.value 

50 self._audience = audience 

51 self._issuer = issuer 

52 self._enabled = True 

53 self._verify_public_key() 

54 

55 def _verify_public_key(self): 

56 """Verifies whether or not the public key is a valid crypto key 

57 (according to the JWT algorithm).""" 

58 if self._public_key is not None: 58 ↛ 68line 58 didn't jump to line 68 because the condition on line 58 was always true

59 # save jwk 

60 try: 

61 self._jwk: PyJWK = PyJWK.from_json( 

62 self.get_jwk(), algorithm=self._algorithm 

63 ) 

64 except jwt.exceptions.InvalidKeyError as e: 

65 logger.error(f"Invalid public key for jwt verification, error: {e}!") 

66 self._disable() 

67 else: 

68 self._disable() 

69 

70 def get_jwk(self) -> str: 

71 """Returns the jwk json contents.""" 

72 algorithm: Optional[Algorithm] = get_default_algorithms().get(self._algorithm) 

73 if algorithm is None: 73 ↛ 74line 73 didn't jump to line 74 because the condition on line 73 was never true

74 raise ValueError(f"invalid jwt algorithm: {self._algorithm}") 

75 return algorithm.to_jwk(self._public_key) 

76 

77 def _disable(self): 

78 self._enabled = False 

79 

80 @property 

81 def enabled(self): 

82 """Whether or not the verifier has valid cryptographic keys.""" 

83 return self._enabled 

84 

85 def verify(self, token: str) -> JWTClaims: 

86 """Verifies a JWT token is valid. 

87 

88 if valid returns dict with jwt claims, otherwise throws. 

89 """ 

90 try: 

91 return jwt.decode( 

92 token, 

93 self._public_key, 

94 algorithms=[self._algorithm], 

95 audience=self._audience, 

96 issuer=self._issuer, 

97 ) 

98 except jwt.ExpiredSignatureError: 

99 raise Unauthorized(token=token, description="Access token is expired") 

100 except jwt.InvalidAudienceError: 

101 raise Unauthorized( 

102 token=token, description="Invalid access token: invalid audience claim" 

103 ) 

104 except jwt.InvalidIssuerError: 

105 raise Unauthorized( 

106 token=token, description="Invalid access token: invalid issuer claim" 

107 ) 

108 except jwt.DecodeError: 

109 raise Unauthorized(token=token, description="Could not decode access token") 

110 except Exception: 

111 raise Unauthorized(token=token, description="Unknown JWT error")