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

1"""Signer implementation for Google Cloud KMS""" 

2 

3from __future__ import annotations 

4 

5import hashlib 

6import logging 

7from typing import Any 

8from urllib import parse 

9 

10from securesystemslib import exceptions 

11from securesystemslib.signer._constants import ( 

12 ECDSA_SHA2_NISTP256, 

13 ECDSA_SHA2_NISTP384, 

14 KEY_TYPE_ECDSA, 

15 KEY_TYPE_MLDSA, 

16 KEY_TYPE_RSA, 

17 MLDSA_44_1, 

18 MLDSA_65_1, 

19 MLDSA_87_1, 

20 RSA_PKCS1V15_SHA256, 

21 RSA_PKCS1V15_SHA512, 

22 RSASSA_PSS_SHA256, 

23 RSASSA_PSS_SHA512, 

24) 

25from securesystemslib.signer._key import Key, SSlibKey 

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

27from securesystemslib.signer._utils import compute_default_keyid, get_mldsa_payload 

28 

29logger = logging.getLogger(__name__) 

30 

31GCP_IMPORT_ERROR = None 

32try: 

33 from google.cloud import kms 

34 from google.cloud.kms_v1.types import CryptoKeyVersion 

35 

36 KEYTYPES_AND_SCHEMES = { 

37 CryptoKeyVersion.CryptoKeyVersionAlgorithm.EC_SIGN_P256_SHA256: ( 

38 KEY_TYPE_ECDSA, 

39 ECDSA_SHA2_NISTP256, 

40 ), 

41 CryptoKeyVersion.CryptoKeyVersionAlgorithm.EC_SIGN_P384_SHA384: ( 

42 KEY_TYPE_ECDSA, 

43 ECDSA_SHA2_NISTP384, 

44 ), 

45 CryptoKeyVersion.CryptoKeyVersionAlgorithm.RSA_SIGN_PSS_2048_SHA256: ( 

46 KEY_TYPE_RSA, 

47 RSASSA_PSS_SHA256, 

48 ), 

49 CryptoKeyVersion.CryptoKeyVersionAlgorithm.RSA_SIGN_PSS_3072_SHA256: ( 

50 KEY_TYPE_RSA, 

51 RSASSA_PSS_SHA256, 

52 ), 

53 CryptoKeyVersion.CryptoKeyVersionAlgorithm.RSA_SIGN_PSS_4096_SHA256: ( 

54 KEY_TYPE_RSA, 

55 RSASSA_PSS_SHA256, 

56 ), 

57 CryptoKeyVersion.CryptoKeyVersionAlgorithm.RSA_SIGN_PSS_4096_SHA512: ( 

58 KEY_TYPE_RSA, 

59 RSASSA_PSS_SHA512, 

60 ), 

61 CryptoKeyVersion.CryptoKeyVersionAlgorithm.RSA_SIGN_PKCS1_2048_SHA256: ( 

62 KEY_TYPE_RSA, 

63 RSA_PKCS1V15_SHA256, 

64 ), 

65 CryptoKeyVersion.CryptoKeyVersionAlgorithm.RSA_SIGN_PKCS1_3072_SHA256: ( 

66 KEY_TYPE_RSA, 

67 RSA_PKCS1V15_SHA256, 

68 ), 

69 CryptoKeyVersion.CryptoKeyVersionAlgorithm.RSA_SIGN_PKCS1_4096_SHA256: ( 

70 KEY_TYPE_RSA, 

71 RSA_PKCS1V15_SHA256, 

72 ), 

73 CryptoKeyVersion.CryptoKeyVersionAlgorithm.RSA_SIGN_PKCS1_4096_SHA512: ( 

74 KEY_TYPE_RSA, 

75 RSA_PKCS1V15_SHA512, 

76 ), 

77 CryptoKeyVersion.CryptoKeyVersionAlgorithm.PQ_SIGN_ML_DSA_44: ( 

78 KEY_TYPE_MLDSA, 

79 MLDSA_44_1, 

80 ), 

81 CryptoKeyVersion.CryptoKeyVersionAlgorithm.PQ_SIGN_ML_DSA_65: ( 

82 KEY_TYPE_MLDSA, 

83 MLDSA_65_1, 

84 ), 

85 CryptoKeyVersion.CryptoKeyVersionAlgorithm.PQ_SIGN_ML_DSA_87: ( 

86 KEY_TYPE_MLDSA, 

87 MLDSA_87_1, 

88 ), 

89 } 

90except ImportError: 

91 GCP_IMPORT_ERROR = ( 

92 "google-cloud-kms library required to sign with Google Cloud keys." 

93 ) 

94 

95 

96class GCPSigner(Signer): 

97 """Google Cloud KMS Signer. 

98 

99 This Signer uses Google Cloud KMS to sign. The payload is hashed locally, 

100 but the signature is created on the KMS. 

101 

102 The private key URI scheme is: ``gcpkms:<gcp_keyid>``, where ``<gcp_keyid>`` 

103 is the fully qualified GCP KMS key name: 

104 ``projects/<project>/locations/<location>/keyRings/<keyRing>/cryptoKeys/<key>/cryptoKeyVersions/<version>``. 

105 

106 Authentication uses ambient credentials (typically the environment variable 

107 ``GOOGLE_APPLICATION_CREDENTIALS`` pointing to a service account key file, 

108 or Google Application Default Credentials): 

109 https://cloud.google.com/docs/authentication/getting-started. 

110 

111 The specific Google Cloud IAM roles that GCPSigner needs are: 

112 

113 * ``roles/cloudkms.publicKeyViewer`` for ``GCPSigner.import_()`` 

114 * ``roles/cloudkms.signer`` for ``Signer.sign()`` 

115 """ 

116 

117 SCHEME = "gcpkms" 

118 

119 def __init__(self, gcp_keyid: str, public_key: SSlibKey): 

120 if GCP_IMPORT_ERROR: 

121 raise exceptions.UnsupportedLibraryError(GCP_IMPORT_ERROR) 

122 

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

124 raise exceptions.UnsupportedAlgorithmError( 

125 f"Unsupported key ({public_key.keytype}/{public_key.scheme}) " 

126 f"in key {public_key.keyid}" 

127 ) 

128 

129 self.hash_algorithm = public_key.get_hash_algorithm_name() 

130 self.gcp_keyid = gcp_keyid 

131 self._public_key = public_key 

132 self.client = kms.KeyManagementServiceClient() 

133 

134 @property 

135 def public_key(self) -> SSlibKey: 

136 return self._public_key 

137 

138 @classmethod 

139 def from_priv_key_uri( 

140 cls, 

141 priv_key_uri: str, 

142 public_key: Key, 

143 secrets_handler: SecretsHandler | None = None, 

144 ) -> GCPSigner: 

145 if not isinstance(public_key, SSlibKey): 

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

147 

148 uri = parse.urlparse(priv_key_uri) 

149 

150 if uri.scheme != cls.SCHEME: 

151 raise ValueError(f"GCPSigner does not support {priv_key_uri}") 

152 

153 return cls(uri.path, public_key) 

154 

155 @classmethod 

156 def import_(cls, gcp_keyid: str) -> tuple[str, SSlibKey]: 

157 """Load key and signer details from KMS 

158 

159 This method should only be called once per key: the uri and 

160 Key should be stored for later use. 

161 

162 Requires ``roles/cloudkms.publicKeyViewer`` role on Google Cloud. 

163 

164 Args: 

165 gcp_id: Fully qualified GCP KMS key name 

166 (``projects/<P>/locations/<L>/keyRings/<R>/cryptoKeys/<K>/cryptoKeyVersions/<V>``). 

167 Returns: 

168 Tuple with private key URI and the public key 

169 """ 

170 if GCP_IMPORT_ERROR: 

171 raise exceptions.UnsupportedLibraryError(GCP_IMPORT_ERROR) 

172 

173 client = kms.KeyManagementServiceClient() 

174 request = { 

175 "name": gcp_keyid, 

176 "public_key_format": kms.PublicKey.PublicKeyFormat.PEM, 

177 } 

178 kms_pubkey = client.get_public_key(request) 

179 try: 

180 keytype, scheme = KEYTYPES_AND_SCHEMES[kms_pubkey.algorithm] 

181 except KeyError as e: 

182 raise exceptions.UnsupportedAlgorithmError( 

183 f"{kms_pubkey.algorithm} is not a supported signing algorithm" 

184 ) from e 

185 

186 keyval = {"public": kms_pubkey.public_key.data.decode("utf-8")} 

187 keyid = compute_default_keyid(keytype, scheme, keyval) 

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

189 

190 return f"{cls.SCHEME}:{gcp_keyid}", public_key 

191 

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

193 # NOTE: request and response can contain CRC32C of the digest/sig: 

194 # Verifying could be useful but would require another dependency... 

195 

196 request: dict[str, Any] = {"name": self.gcp_keyid} 

197 if self.public_key.keytype == KEY_TYPE_MLDSA: 

198 request["data"] = get_mldsa_payload(payload, 1) 

199 else: 

200 hasher = hashlib.new(self.hash_algorithm) 

201 hasher.update(payload) 

202 request["digest"] = {self.hash_algorithm: hasher.digest()} 

203 

204 logger.debug("signing request %s", request) 

205 response = self.client.asymmetric_sign(request) 

206 logger.debug("signing response %s", response) 

207 

208 return Signature(self.public_key.keyid, response.signature.hex())