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

1from datetime import datetime, timedelta 

2from typing import Optional 

3from uuid import UUID 

4 

5import jwt 

6from jwt.api_jwk import PyJWK 

7from opal_common.authentication.types import ( 

8 JWTAlgorithm, 

9 JWTClaims, 

10 PrivateKey, 

11 PublicKey, 

12) 

13from opal_common.authentication.verifier import JWTVerifier 

14from opal_common.logger import logger 

15 

16 

17class InvalidJWTCryptoKeysException(Exception): 

18 """Raised when JWT signer provided with invalid crypto keys.""" 

19 

20 pass 

21 

22 

23class JWTSigner(JWTVerifier): 

24 """Given cryptographic keys, signs and verifies jwt tokens.""" 

25 

26 def __init__( 

27 self, 

28 private_key: Optional[PrivateKey], 

29 public_key: Optional[PublicKey], 

30 algorithm: JWTAlgorithm, 

31 audience: str, 

32 issuer: str, 

33 ): 

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

35 generate together are are valid. otherwise will throw. 

36 

37 JWT signer can be initialized with empty keys (None), 

38 in which case signer.enabled == False. 

39 

40 This allows opal to run both in secure mode (which keys, requires jwt authentication) 

41 and in insecure mode (good for development and running locally). 

42 

43 Args: 

44 private_key (PrivateKey): a valid private key or None 

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

46 algorithm (JWTAlgorithm): the jwt algorithm to use 

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

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

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

50 """ 

51 super().__init__( 

52 public_key=public_key, algorithm=algorithm, audience=audience, issuer=issuer 

53 ) 

54 self._private_key = private_key 

55 self._verify_crypto_keys() 

56 

57 def _verify_crypto_keys(self): 

58 """Verifies whether or not valid crypto keys were provided to the 

59 signer. if both keys are valid, encodes and decodes a JWT to make sure 

60 the keys match. 

61 

62 if both private and public keys are valid and are matching => 

63 signer is enabled if both private and public keys are None => 

64 signer is disabled (self.enabled == False) if only one key is 

65 valid/not-None => throws ValueError any other case => throws 

66 ValueError 

67 """ 

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

69 # both keys provided, let's make sure these keys were generated correctly 

70 token = jwt.encode( 

71 {"some": "payload"}, self._private_key, algorithm=self._algorithm 

72 ) 

73 try: 

74 jwt.decode(token, self._public_key, algorithms=[self._algorithm]) 

75 except jwt.PyJWTError as exc: 

76 logger.info( 

77 "JWT Signer key verification failed with error: {err}", 

78 err=repr(exc), 

79 ) 

80 raise InvalidJWTCryptoKeysException( 

81 "private key and public key do not match!" 

82 ) from exc 

83 # save jwk 

84 self._jwk: PyJWK = PyJWK.from_json( 

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

86 ) 

87 elif self._private_key is None and self._public_key is not None: 

88 raise ValueError( 

89 "JWT Signer not valid, you provided a public key without a private key!" 

90 ) 

91 elif self._private_key is not None and self._public_key is None: 

92 raise ValueError( 

93 "JWT Signer not valid, you provided a private key without a public key!" 

94 ) 

95 elif self._private_key is None and self._public_key is None: 

96 # valid situation, running in dev mode and api security is off 

97 self._disable() 

98 else: 

99 raise ValueError("Invalid JWT Signer input!") 

100 

101 def sign( 

102 self, sub: UUID, token_lifetime: timedelta, custom_claims: dict = {} 

103 ) -> str: 

104 payload = {} 

105 issued_at = datetime.utcnow() 

106 expire_at = issued_at + token_lifetime 

107 payload = { 

108 "iat": issued_at, 

109 "exp": expire_at, 

110 "aud": self._audience, 

111 "iss": self._issuer, 

112 "sub": sub.hex, 

113 } 

114 if custom_claims: 114 ↛ 117line 114 didn't jump to line 117 because the condition on line 114 was always true

115 payload.update(custom_claims) 

116 

117 headers = {} 

118 if self._jwk.key_id is not None: 118 ↛ 119line 118 didn't jump to line 119 because the condition on line 118 was never true

119 headers = {"kid": self._jwk.key_id} 

120 return jwt.encode( 

121 payload, self._private_key, algorithm=self._algorithm, headers=headers 

122 )