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

1import logging 

2import os 

3from typing import Optional 

4 

5from cryptography.hazmat.backends import default_backend 

6from cryptography.hazmat.primitives import serialization 

7from opal_common.authentication.types import EncryptionKeyFormat, PrivateKey, PublicKey 

8 

9logger = logging.getLogger("opal.authentication") 

10 

11 

12def to_bytes(key: str, encoding: str = "utf-8"): 

13 """Crypto lib expect 'bytes' keys, convert 'str' keys to 'bytes'.""" 

14 return key.encode(encoding) 

15 

16 

17def maybe_decode_multiline_key(key: str) -> bytes: 

18 """If key contents are passed via env var, we allow to encode multiline 

19 keys with a simple replace of each newline (\n) char with underscore (_). 

20 

21 this method detects if the provided key contains such encoding, and 

22 if so reverses it. 

23 """ 

24 if "\n" not in key: 24 ↛ 25line 24 didn't jump to line 25 because the condition on line 24 was never true

25 key = key.replace("_", "\n") 

26 if not key.endswith("\n"): 

27 key = key + "\n" 

28 return to_bytes(key) 

29 

30 

31def cast_private_key( 

32 value: str, key_format: EncryptionKeyFormat, passphrase: Optional[str] = None 

33) -> Optional[PrivateKey]: 

34 """Parse a string into a valid cryptographic private key. 

35 

36 the string can represent a file path in which the key exists, or the 

37 actual key contents. 

38 """ 

39 if value is None: 39 ↛ 40line 39 didn't jump to line 40 because the condition on line 39 was never true

40 return None 

41 

42 if isinstance(value, PrivateKey.__args__): 42 ↛ 43line 42 didn't jump to line 43 because the condition on line 42 was never true

43 return value 

44 

45 if passphrase is None: 45 ↛ 48line 45 didn't jump to line 48 because the condition on line 45 was always true

46 password = None 

47 else: 

48 password = passphrase.encode("utf-8") 

49 

50 key_path = os.path.expanduser(value) 

51 if os.path.isfile(key_path): 51 ↛ 52line 51 didn't jump to line 52 because the condition on line 51 was never true

52 with open(key_path, "rb") as file: 

53 raw_key = file.read() 

54 else: 

55 raw_key = maybe_decode_multiline_key(value) 

56 

57 if key_format == EncryptionKeyFormat.pem: 

58 return serialization.load_pem_private_key( 

59 raw_key, password=password, backend=default_backend() 

60 ) 

61 

62 if key_format == EncryptionKeyFormat.ssh: 62 ↛ 67line 62 didn't jump to line 67 because the condition on line 62 was always true

63 return serialization.load_ssh_private_key( 

64 raw_key, password=password, backend=default_backend() 

65 ) 

66 

67 if key_format == EncryptionKeyFormat.der: 

68 return serialization.load_der_private_key( 

69 raw_key, password=password, backend=default_backend() 

70 ) 

71 

72 

73def cast_public_key(value: str, key_format: EncryptionKeyFormat) -> Optional[PublicKey]: 

74 """Parse a string into a valid cryptographic public key. 

75 

76 the string can represent a file path in which the key exists, or the 

77 actual key contents. 

78 """ 

79 if value is None: 79 ↛ 80line 79 didn't jump to line 80 because the condition on line 79 was never true

80 return None 

81 

82 if isinstance(value, PublicKey.__args__): 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true

83 return value 

84 

85 key_path = os.path.expanduser(value) 

86 if os.path.isfile(key_path): 86 ↛ 87line 86 didn't jump to line 87 because the condition on line 86 was never true

87 with open(key_path, "rb") as file: 

88 raw_key = file.read() 

89 elif key_format == EncryptionKeyFormat.ssh: # ssh key format is one line 89 ↛ 92line 89 didn't jump to line 92 because the condition on line 89 was always true

90 raw_key = to_bytes(value) 

91 else: 

92 raw_key = maybe_decode_multiline_key(value) 

93 

94 if key_format == EncryptionKeyFormat.pem: 94 ↛ 95line 94 didn't jump to line 95 because the condition on line 94 was never true

95 return serialization.load_pem_public_key(raw_key, backend=default_backend()) 

96 

97 if key_format == EncryptionKeyFormat.ssh: 97 ↛ 100line 97 didn't jump to line 100 because the condition on line 97 was always true

98 return serialization.load_ssh_public_key(raw_key, backend=default_backend()) 

99 

100 if key_format == EncryptionKeyFormat.der: 

101 return serialization.load_der_public_key(raw_key, backend=default_backend())