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

1"""Signer implementation for AWS Key Management Service""" 

2 

3from __future__ import annotations 

4 

5import hashlib 

6import logging 

7from urllib import parse 

8 

9from securesystemslib.exceptions import ( 

10 UnsupportedAlgorithmError, 

11 UnsupportedLibraryError, 

12) 

13from securesystemslib.signer._constants import ( 

14 ECDSA_SHA2_NISTP256, 

15 ECDSA_SHA2_NISTP384, 

16 KEY_TYPE_ECDSA, 

17 KEY_TYPE_RSA, 

18 RSA_PKCS1V15_SHA256, 

19 RSA_PKCS1V15_SHA384, 

20 RSA_PKCS1V15_SHA512, 

21 RSASSA_PSS_SHA256, 

22 RSASSA_PSS_SHA384, 

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 

28 

29logger = logging.getLogger(__name__) 

30 

31AWS_IMPORT_ERROR = None 

32try: 

33 import boto3 

34 from botocore.exceptions import BotoCoreError, ClientError 

35 from cryptography.hazmat.primitives import serialization 

36except ImportError: 

37 AWS_IMPORT_ERROR = "Signing with AWS KMS requires aws-kms and cryptography." 

38 

39 

40class AWSSigner(Signer): 

41 """AWS Key Management Service Signer. 

42 

43 This Signer uses AWS KMS to sign, supporting RSA and EC keys. 

44 The signer computes hash digests locally and sends only the digest to AWS KMS. 

45 

46 The private key URI scheme is: ``awskms:<AWS_KEY_ID>``, where ``<AWS_KEY_ID>`` 

47 can be a key ID, key ARN, alias name, or alias ARN. 

48 

49 Authentication uses ambient credentials (typically environment variables such as 

50 ``AWS_ACCESS_KEY_ID``, ``AWS_SECRET_ACCESS_KEY``, and ``AWS_SESSION_TOKEN``) 

51 recognized by the boto3 SDK. 

52 

53 For more details on AWS authentication, refer to the `AWS Command Line 

54 Interface User Guide 

55 <https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-files.html>`_. 

56 

57 The specific IAM permissions that AWSSigner needs are: 

58 

59 * ``kms:GetPublicKey`` for ``AWSSigner.import_()`` 

60 * ``kms:Sign`` for ``Signer.sign()`` 

61 

62 Raises: 

63 UnsupportedLibraryError: If boto3 or cryptography are not installed. 

64 """ 

65 

66 SCHEME = "awskms" 

67 

68 # Ordered dict of securesystemslib schemes to aws signing algorithms 

69 # NOTE: the order matters when choosing a default (see _get_default_scheme) 

70 aws_algos = { 

71 ECDSA_SHA2_NISTP256: "ECDSA_SHA_256", 

72 ECDSA_SHA2_NISTP384: "ECDSA_SHA_384", 

73 # "ecdsa-sha2-nistp521": "ECDSA_SHA_512", # FIXME: needs SSlibKey support 

74 RSASSA_PSS_SHA256: "RSASSA_PSS_SHA_256", 

75 RSASSA_PSS_SHA384: "RSASSA_PSS_SHA_384", 

76 RSASSA_PSS_SHA512: "RSASSA_PSS_SHA_512", 

77 RSA_PKCS1V15_SHA256: "RSASSA_PKCS1_V1_5_SHA_256", 

78 RSA_PKCS1V15_SHA384: "RSASSA_PKCS1_V1_5_SHA_384", 

79 RSA_PKCS1V15_SHA512: "RSASSA_PKCS1_V1_5_SHA_512", 

80 } 

81 

82 def __init__(self, aws_key_id: str, public_key: SSlibKey): 

83 if AWS_IMPORT_ERROR: 

84 raise UnsupportedLibraryError(AWS_IMPORT_ERROR) 

85 

86 self.aws_key_id = aws_key_id 

87 self._public_key = public_key 

88 self.client = boto3.client("kms") 

89 self.aws_algo = self.aws_algos[self.public_key.scheme] 

90 

91 @property 

92 def public_key(self) -> SSlibKey: 

93 return self._public_key 

94 

95 @classmethod 

96 def from_priv_key_uri( 

97 cls, 

98 priv_key_uri: str, 

99 public_key: Key, 

100 secrets_handler: SecretsHandler | None = None, 

101 ) -> AWSSigner: 

102 if not isinstance(public_key, SSlibKey): 

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

104 

105 uri = parse.urlparse(priv_key_uri) 

106 

107 if uri.scheme != cls.SCHEME: 

108 raise ValueError(f"AWSSigner does not support {priv_key_uri}") 

109 

110 return cls(uri.path, public_key) 

111 

112 @classmethod 

113 def _get_default_scheme(cls, supported_by_key: list[str]) -> str | None: 

114 # Iterate over supported AWS algorithms, pick the **first** that is also 

115 # supported by the key, and return the related securesystemslib scheme. 

116 for scheme, algo in cls.aws_algos.items(): 

117 if algo in supported_by_key: 

118 return scheme 

119 return None 

120 

121 @staticmethod 

122 def _get_keytype_for_scheme(scheme: str) -> str: 

123 if scheme.startswith(KEY_TYPE_ECDSA): 

124 return KEY_TYPE_ECDSA 

125 if scheme.startswith(KEY_TYPE_RSA): 

126 return KEY_TYPE_RSA 

127 raise RuntimeError 

128 

129 @classmethod 

130 def import_( 

131 cls, aws_key_id: str, local_scheme: str | None = None 

132 ) -> tuple[str, SSlibKey]: 

133 """Loads a key and signer details from AWS KMS. 

134 

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

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

137 

138 Arguments: 

139 aws_key_id: AWS KMS key ID. 

140 local_scheme: securesystemslib key scheme. 

141 Defaults to 'rsassa-pss-sha256' if not provided. 

142 

143 Raises: 

144 UnsupportedAlgorithmError: If the AWS KMS signing algorithm is 

145 unsupported. 

146 BotoCoreError: Errors from the botocore library. 

147 ClientError: Errors related to AWS KMS client. 

148 

149 Returns: 

150 A tuple of private key URI string and public key. 

151 """ 

152 if AWS_IMPORT_ERROR: 

153 raise UnsupportedLibraryError(AWS_IMPORT_ERROR) 

154 

155 if local_scheme: 

156 if local_scheme not in cls.aws_algos: 

157 raise ValueError(f"Unsupported scheme '{local_scheme}'") 

158 

159 client = boto3.client("kms") 

160 request = client.get_public_key(KeyId=aws_key_id) 

161 key_algos = request["SigningAlgorithms"] 

162 

163 if local_scheme: 

164 if cls.aws_algos[local_scheme] not in key_algos: 

165 raise UnsupportedAlgorithmError( 

166 f"Unsupported scheme '{local_scheme}' for AWS key" 

167 ) 

168 

169 else: 

170 local_scheme = cls._get_default_scheme(key_algos) 

171 if not local_scheme: 

172 raise UnsupportedAlgorithmError( 

173 f"Unsupported AWS key algorithms: {key_algos}" 

174 ) 

175 

176 keytype = cls._get_keytype_for_scheme(local_scheme) 

177 

178 kms_pubkey = serialization.load_der_public_key(request["PublicKey"]) 

179 

180 public_key_pem = kms_pubkey.public_bytes( 

181 encoding=serialization.Encoding.PEM, 

182 format=serialization.PublicFormat.SubjectPublicKeyInfo, 

183 ).decode("utf-8") 

184 

185 keyval = {"public": public_key_pem} 

186 keyid = compute_default_keyid(keytype, local_scheme, keyval) 

187 public_key = SSlibKey(keyid, keytype, local_scheme, keyval) 

188 return f"{cls.SCHEME}:{aws_key_id}", public_key 

189 

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

191 """Sign the payload with the AWS KMS key 

192 

193 This method computes the hash of the payload locally and sends only the 

194 digest to AWS KMS for signing. 

195 

196 Arguments: 

197 payload (bytes): The payload to be signed. 

198 

199 Raises: 

200 BotoCoreError, ClientError: If an error occurs during the signing process. 

201 

202 Returns: 

203 Signature: A signature object containing the key ID and the signature. 

204 """ 

205 try: 

206 hash_algorithm = self.public_key.get_hash_algorithm_name() 

207 hasher = hashlib.new(hash_algorithm) 

208 hasher.update(payload) 

209 digest = hasher.digest() 

210 

211 sign_request = self.client.sign( 

212 KeyId=self.aws_key_id, 

213 Message=digest, 

214 MessageType="DIGEST", 

215 SigningAlgorithm=self.aws_algo, 

216 ) 

217 

218 logger.debug("Signing response: %s", sign_request) 

219 response = sign_request["Signature"] 

220 logger.debug("Signature response: %s", response) 

221 

222 return Signature(self.public_key.keyid, response.hex()) 

223 except (BotoCoreError, ClientError) as e: 

224 logger.error( 

225 "Failed to sign using AWS KMS key ID %s: %s", 

226 self.aws_key_id, 

227 str(e), 

228 ) 

229 raise e