Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/securesystemslib/signer/_vault_signer.py: 44%

1"""Signer implementation for HashiCorp Vault (Transit secrets engine)""" 

2 

3from __future__ import annotations 

4 

5from base64 import b64decode, b64encode 

6from urllib import parse 

7 

8from securesystemslib.exceptions import UnsupportedLibraryError 

9from securesystemslib.signer._key import Key, SSlibKey 

10from securesystemslib.signer._signer import SecretsHandler, Signature, Signer 

11 

12VAULT_IMPORT_ERROR = None 

13try: 

14 import hvac 

15 from cryptography.hazmat.primitives.asymmetric.ed25519 import ( 

16 Ed25519PublicKey, 

17 ) 

18 

19except ImportError: 

20 VAULT_IMPORT_ERROR = "Signing with HashiCorp Vault requires hvac and cryptography." 

21 

22 

23class VaultSigner(Signer): 

24 """Signer for HashiCorp Vault Transit secrets engine. 

25 

26 The signer uses ambient credentials to connect to Vault, most notably 

27 the environment variables ``VAULT_ADDR`` and ``VAULT_TOKEN`` must be set: 

28 https://developer.hashicorp.com/vault/docs/commands#environment-variables 

29 

30 The private key URI scheme is: ``hv:<KEY NAME>/<KEY VERSION>``. 

31 

32 Raises: 

33 UnsupportedLibraryError: If hvac or cryptography are not installed. 

34 """ 

35 

36 SCHEME = "hv" 

37 

38 def __init__(self, hv_key_name: str, public_key: SSlibKey, hv_key_version: int): 

39 if VAULT_IMPORT_ERROR: 

40 raise UnsupportedLibraryError(VAULT_IMPORT_ERROR) 

41 

42 self.hv_key_name = hv_key_name 

43 self._public_key = public_key 

44 self.hv_key_version = hv_key_version 

45 

46 # Client caches ambient settings in __init__. This means settings are 

47 # stable for subsequent calls to sign, also if the environment changes. 

48 self._client = hvac.Client() 

49 

50 def sign(self, payload: bytes) -> Signature: 

51 """Signs payload with HashiCorp Vault Transit secrets engine. 

52 

53 Arguments: 

54 payload: bytes to be signed. 

55 

56 Raises: 

57 Various errors from hvac. 

58 

59 Returns: 

60 Signature. 

61 """ 

62 resp = self._client.secrets.transit.sign_data( 

63 self.hv_key_name, 

64 hash_input=b64encode(payload).decode(), 

65 key_version=self.hv_key_version, 

66 ) 

67 

68 sig_b64 = resp["data"]["signature"].split(":")[2] 

69 sig = b64decode(sig_b64).hex() 

70 

71 return Signature(self.public_key.keyid, sig) 

72 

73 @property 

74 def public_key(self) -> SSlibKey: 

75 return self._public_key 

76 

77 @classmethod 

78 def from_priv_key_uri( 

79 cls, 

80 priv_key_uri: str, 

81 public_key: Key, 

82 secrets_handler: SecretsHandler | None = None, 

83 ) -> VaultSigner: 

84 if not isinstance(public_key, SSlibKey): 

85 raise ValueError(f"Expected SSlibKey for {priv_key_uri}") 

86 

87 uri = parse.urlparse(priv_key_uri) 

88 

89 if uri.scheme != cls.SCHEME: 

90 raise ValueError(f"VaultSigner does not support {priv_key_uri}") 

91 

92 name, version = uri.path.split("/") 

93 

94 return cls(name, public_key, int(version)) 

95 

96 @classmethod 

97 def import_(cls, hv_key_name: str) -> tuple[str, SSlibKey]: 

98 """Load key and signer details from HashiCorp Vault. 

99 

100 If multiple keys exist in the vault under the passed name, only the 

101 newest key is returned. Supported key type is: ed25519 

102 

103 See class documentation for details about settings and uri format. 

104 

105 Arguments: 

106 hv_key_name: Name of vault key to import. 

107 

108 Raises: 

109 UnsupportedLibraryError: hvac or cryptography are not installed. 

110 Various errors from hvac. 

111 

112 Returns: 

113 Private key uri and public key. 

114 

115 """ 

116 if VAULT_IMPORT_ERROR: 

117 raise UnsupportedLibraryError(VAULT_IMPORT_ERROR) 

118 

119 client = hvac.Client() 

120 resp = client.secrets.transit.read_key(hv_key_name) 

121 

122 # Pick key with highest version number 

123 version, key_info = sorted(resp["data"]["keys"].items())[-1] 

124 

125 crypto_key = Ed25519PublicKey.from_public_bytes( 

126 b64decode(key_info["public_key"]) 

127 ) 

128 

129 key = SSlibKey.from_crypto(crypto_key) 

130 uri = f"{VaultSigner.SCHEME}:{hv_key_name}/{version}" 

131 

132 return uri, key