KMS: add get_public_key (#6222)

This commit is contained in:
sercan tor 2023-04-17 20:59:33 +01:00 committed by GitHub
parent 94db0c6fa4
commit ae11c02593
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 42 additions and 2 deletions

View File

@ -4056,7 +4056,7 @@
- [X] get_key_policy - [X] get_key_policy
- [X] get_key_rotation_status - [X] get_key_rotation_status
- [ ] get_parameters_for_import - [ ] get_parameters_for_import
- [ ] get_public_key - [X] get_public_key
- [ ] import_key_material - [ ] import_key_material
- [ ] list_aliases - [ ] list_aliases
- [X] list_grants - [X] list_grants

View File

@ -4,7 +4,7 @@ from collections import defaultdict
from copy import copy from copy import copy
from datetime import datetime, timedelta from datetime import datetime, timedelta
from cryptography.exceptions import InvalidSignature from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding from cryptography.hazmat.primitives.asymmetric import padding
from typing import Any, Dict, List, Tuple, Optional, Iterable, Set from typing import Any, Dict, List, Tuple, Optional, Iterable, Set
@ -682,5 +682,14 @@ class KmsBackend(BaseBackend):
except InvalidSignature: except InvalidSignature:
return key.arn, False, signing_algorithm return key.arn, False, signing_algorithm
def get_public_key(self, key_id: str) -> Tuple[Key, bytes]:
key = self.describe_key(key_id)
public_key = key.private_key.public_key().public_bytes(
encoding=serialization.Encoding.DER,
format=serialization.PublicFormat.SubjectPublicKeyInfo,
)
return key, public_key
kms_backends = BackendDict(KmsBackend, "kms") kms_backends = BackendDict(KmsBackend, "kms")

View File

@ -721,6 +721,23 @@ class KmsResponse(BaseResponse):
} }
) )
def get_public_key(self) -> str:
key_id = self._get_param("KeyId")
self._validate_key_id(key_id)
self._validate_cmk_id(key_id)
key, public_key = self.kms_backend.get_public_key(key_id)
return json.dumps(
{
"CustomerMasterKeySpec": key.key_spec,
"EncryptionAlgorithms": key.encryption_algorithms,
"KeyId": key.id,
"KeyUsage": key.key_usage,
"PublicKey": base64.b64encode(public_key).decode("UTF-8"),
"SigningAlgorithms": key.signing_algorithms,
}
)
def _assert_default_policy(policy_name: str) -> None: def _assert_default_policy(policy_name: str) -> None:
if policy_name != "default": if policy_name != "default":

View File

@ -1335,6 +1335,20 @@ def test_verify_empty_signature():
) )
@mock_kms
def test_get_public_key():
client = boto3.client("kms", region_name="us-east-1")
key = client.create_key(KeyUsage="SIGN_VERIFY", KeySpec="RSA_2048")
key_id = key["KeyMetadata"]["KeyId"]
public_key_response = client.get_public_key(KeyId=key_id)
public_key_response.should.contain("PublicKey")
public_key_response["SigningAlgorithms"].should.equal(
key["KeyMetadata"]["SigningAlgorithms"]
)
public_key_response.shouldnt.contain("EncryptionAlgorithms")
def create_simple_key(client, id_or_arn="KeyId", description=None, policy=None): def create_simple_key(client, id_or_arn="KeyId", description=None, policy=None):
with mock.patch.object(rsa, "generate_private_key", return_value=""): with mock.patch.object(rsa, "generate_private_key", return_value=""):
params = {} params = {}