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

1"""Signer implementation for Azure Key Vault""" 

2 

3from __future__ import annotations 

4 

5import hashlib 

6import logging 

7from urllib import parse 

8 

9from securesystemslib.exceptions import UnsupportedLibraryError 

10from securesystemslib.signer._constants import ( 

11 ECDSA_SHA2_NISTP256, 

12 ECDSA_SHA2_NISTP384, 

13 ECDSA_SHA2_NISTP521, 

14 KEY_TYPE_ECDSA, 

15) 

16from securesystemslib.signer._key import Key, SSlibKey 

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

18from securesystemslib.signer._utils import compute_default_keyid 

19 

20AZURE_IMPORT_ERROR = None 

21try: 

22 from azure.core.exceptions import HttpResponseError 

23 from azure.identity import DefaultAzureCredential 

24 from azure.keyvault.keys import KeyClient, KeyCurveName, KeyVaultKey 

25 from azure.keyvault.keys.crypto import ( 

26 CryptographyClient, 

27 SignatureAlgorithm, 

28 ) 

29 from cryptography.hazmat.primitives.asymmetric import ec 

30 from cryptography.hazmat.primitives.asymmetric.utils import ( 

31 encode_dss_signature, 

32 ) 

33 from cryptography.hazmat.primitives.serialization import ( 

34 Encoding, 

35 PublicFormat, 

36 ) 

37 

38 KEYTYPES_AND_SCHEMES = { 

39 KeyCurveName.p_256: (KEY_TYPE_ECDSA, ECDSA_SHA2_NISTP256), 

40 KeyCurveName.p_384: (KEY_TYPE_ECDSA, ECDSA_SHA2_NISTP384), 

41 KeyCurveName.p_521: (KEY_TYPE_ECDSA, ECDSA_SHA2_NISTP521), 

42 } 

43 

44 SIGNATURE_ALGORITHMS = { 

45 ECDSA_SHA2_NISTP256: SignatureAlgorithm.es256, 

46 ECDSA_SHA2_NISTP384: SignatureAlgorithm.es384, 

47 ECDSA_SHA2_NISTP521: SignatureAlgorithm.es512, 

48 } 

49 

50 

51except ImportError: 

52 AZURE_IMPORT_ERROR = ( 

53 "Signing with Azure Key Vault requires azure-identity, " 

54 "azure-keyvault-keys and cryptography." 

55 ) 

56 

57logger = logging.getLogger(__name__) 

58 

59 

60class UnsupportedKeyType(Exception): # noqa: N818 

61 pass 

62 

63 

64class AzureSigner(Signer): 

65 """Azure Key Vault Signer. 

66 

67 This Signer uses Azure Key Vault to sign. 

68 Currently this signer supports signing with EC keys (NIST curves P-256, P-384, 

69 and P-521). 

70 

71 The private key URI scheme is: 

72 ``azurekms://<vault-name>.vault.azure.net/keys/<key-name>/<version>`` 

73 

74 Authentication uses ambient credentials via 

75 ``azure.identity.DefaultAzureCredential`` (such as environment variables 

76 ``AZURE_CLIENT_ID``, ``AZURE_CLIENT_SECRET``, ``AZURE_TENANT_ID``, managed 

77 identities, or Azure CLI login). 

78 

79 The specific permissions that AzureSigner needs are: 

80 

81 * "Key Vault Crypto User" (or equivalent custom RBAC role) for 

82 ``AzureSigner.import_()`` and ``Signer.sign()`` 

83 

84 See `Azure Key Vault RBAC guide 

85 <https://learn.microsoft.com/en-us/azure/key-vault/general/rbac-guide?tabs=azure-cli>`_ 

86 for a list of all built-in Azure Key Vault roles. 

87 

88 Raises: 

89 UnsupportedLibraryError: If azure-identity, azure-keyvault-keys, or 

90 cryptography are not installed. 

91 """ 

92 

93 SCHEME = "azurekms" 

94 

95 def __init__(self, az_key_uri: str, public_key: SSlibKey): 

96 if AZURE_IMPORT_ERROR: 

97 raise UnsupportedLibraryError(AZURE_IMPORT_ERROR) 

98 

99 if (public_key.keytype, public_key.scheme) not in KEYTYPES_AND_SCHEMES.values(): 

100 logger.info("only EC keys are supported for now") 

101 raise UnsupportedKeyType( 

102 "Supplied key must be an EC key on curve " 

103 "nistp256, nistp384, or nistp521" 

104 ) 

105 

106 cred = DefaultAzureCredential() 

107 self.crypto_client = CryptographyClient( 

108 az_key_uri, 

109 credential=cred, 

110 ) 

111 self.signature_algorithm = SIGNATURE_ALGORITHMS[public_key.scheme] 

112 self.hash_algorithm = public_key.get_hash_algorithm_name() 

113 self._public_key = public_key 

114 

115 @property 

116 def public_key(self) -> SSlibKey: 

117 return self._public_key 

118 

119 @staticmethod 

120 def _get_key_vault_key( 

121 cred: DefaultAzureCredential, 

122 vault_name: str, 

123 key_name: str, 

124 ) -> KeyVaultKey: 

125 """Return KeyVaultKey created from the Vault name and key name""" 

126 vault_url = f"https://{vault_name}.vault.azure.net/" 

127 

128 try: 

129 key_client = KeyClient(vault_url=vault_url, credential=cred) 

130 return key_client.get_key(key_name) 

131 except (HttpResponseError,) as e: 

132 logger.info( 

133 "Key %s/%s failed to create key client from credentials, " 

134 "key ID, and Vault URL: %s", 

135 vault_name, 

136 key_name, 

137 str(e), 

138 ) 

139 raise e 

140 

141 @staticmethod 

142 def _create_crypto_client( 

143 cred: DefaultAzureCredential, 

144 kv_key: KeyVaultKey, 

145 ) -> CryptographyClient: 

146 """Return CryptographyClient created Azure credentials and a KeyVaultKey""" 

147 try: 

148 return CryptographyClient(kv_key, credential=cred) 

149 except (HttpResponseError,) as e: 

150 logger.info( 

151 "Key %s failed to create crypto client from " 

152 "credentials and KeyVaultKey: %s", 

153 kv_key, 

154 str(e), 

155 ) 

156 raise e 

157 

158 @staticmethod 

159 def _get_keytype_and_scheme(crv: str) -> tuple[str, str]: 

160 try: 

161 return KEYTYPES_AND_SCHEMES[crv] 

162 except KeyError: 

163 raise UnsupportedKeyType("Unsupported curve supplied by key") 

164 

165 @classmethod 

166 def from_priv_key_uri( 

167 cls, 

168 priv_key_uri: str, 

169 public_key: Key, 

170 secrets_handler: SecretsHandler | None = None, 

171 ) -> AzureSigner: 

172 if not isinstance(public_key, SSlibKey): 

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

174 

175 uri = parse.urlparse(priv_key_uri) 

176 

177 if uri.scheme != cls.SCHEME: 

178 raise ValueError(f"AzureSigner does not support {priv_key_uri}") 

179 

180 az_key_uri = priv_key_uri.replace("azurekms:", "https:") 

181 return cls(az_key_uri, public_key) 

182 

183 @classmethod 

184 def import_(cls, az_vault_name: str, az_key_name: str) -> tuple[str, SSlibKey]: 

185 """Load key and signer details from KMS 

186 

187 Returns the private key uri and the public key. This method should only 

188 be called once per key: the uri and Key should be stored for later use. 

189 """ 

190 if AZURE_IMPORT_ERROR: 

191 raise UnsupportedLibraryError(AZURE_IMPORT_ERROR) 

192 

193 credential = DefaultAzureCredential() 

194 key_vault_key = cls._get_key_vault_key(credential, az_vault_name, az_key_name) 

195 

196 if not key_vault_key.key.kty.startswith("EC"): 

197 raise UnsupportedKeyType(f"Unsupported key type {key_vault_key.key.kty}") 

198 

199 if key_vault_key.key.crv == KeyCurveName.p_256: 

200 crv: ec.EllipticCurve = ec.SECP256R1() 

201 elif key_vault_key.key.crv == KeyCurveName.p_384: 

202 crv = ec.SECP384R1() 

203 elif key_vault_key.key.crv == KeyCurveName.p_521: 

204 crv = ec.SECP521R1() 

205 else: 

206 raise UnsupportedKeyType(f"Unsupported curve type {key_vault_key.key.crv}") 

207 

208 # Key is in JWK format, create a curve from it with the parameters 

209 x = int.from_bytes(key_vault_key.key.x, byteorder="big") 

210 y = int.from_bytes(key_vault_key.key.y, byteorder="big") 

211 

212 cpub = ec.EllipticCurvePublicNumbers(x, y, crv) 

213 pub_key = cpub.public_key() 

214 pem = pub_key.public_bytes(Encoding.PEM, PublicFormat.SubjectPublicKeyInfo) 

215 

216 keytype, scheme = cls._get_keytype_and_scheme(key_vault_key.key.crv) 

217 keyval = {"public": pem.decode("utf-8")} 

218 keyid = compute_default_keyid(keytype, scheme, keyval) 

219 public_key = SSlibKey(keyid, keytype, scheme, keyval) 

220 priv_key_uri = key_vault_key.key.kid.replace("https:", "azurekms:") 

221 

222 return priv_key_uri, public_key 

223 

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

225 """Signs payload with Azure Key Vault. 

226 

227 Arguments: 

228 payload: bytes to be signed. 

229 

230 Raises: 

231 Various errors from azure.keyvault.keys. 

232 

233 Returns: 

234 Signature. 

235 """ 

236 

237 hasher = hashlib.new(self.hash_algorithm) 

238 hasher.update(payload) 

239 digest = hasher.digest() 

240 response = self.crypto_client.sign(self.signature_algorithm, digest) 

241 

242 # This code is copied from: 

243 # https://github.com/secure-systems-lab/securesystemslib/blob/135567fa04f10d0c6a4cd32eb45ce736e1f50a93/securesystemslib/signer/_hsm_signer.py#L379 

244 # 

245 # The PKCS11 signature octets correspond to the concatenation of the 

246 # ECDSA values r and s, both represented as an octet string of equal 

247 # length of at most nLen with the most significant byte first (i.e. 

248 # big endian) 

249 # https://docs.oasis-open.org/pkcs11/pkcs11-curr/v3.0/cs01/pkcs11-curr-v3.0-cs01.html#_Toc30061178 

250 r_s_len = int(len(response.signature) / 2) 

251 r = int.from_bytes(response.signature[:r_s_len], byteorder="big") 

252 s = int.from_bytes(response.signature[r_s_len:], byteorder="big") 

253 

254 # Create an ASN.1 encoded Dss-Sig-Value to be used with 

255 # pyca/cryptography 

256 dss_sig_value = encode_dss_signature(r, s).hex() 

257 

258 return Signature(self.public_key.keyid, dss_sig_value)