catalin/moto
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Add KMS Support to EBS Encrypted Volumes (#3383)

Browse source 
* Properly coerce `Encrypted` attribute to bool on request/response.
* Create and use a default AWS managed CMK for EBS when clients request
  an encrypted volume without specifying a KmsKeyId.

NOTE: A client-provided KmsKeyId is simply stored as-is, and is not validated
against the KMS backend. This is in keeping with other moto backends (RDS, Redshift)
that currently also accept unvalidated customer master key (CMK) parameters, but could
be an area for future improvement.

Closes #3248
This commit is contained in:
Brian Pandola 2020-10-14 07:18:50 -07:00 • committed by GitHub
commit ccda76898a
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 124 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

63
tests/test_ec2/test_elastic_block_store.py
View file

@ -13,6 +13,7 @@ import sure # noqa
from moto import mock_ec2_deprecated, mock_ec2
from moto.ec2.models import OWNER_ID
from moto.kms import mock_kms
@mock_ec2_deprecated
@ -915,3 +916,65 @@ def test_search_for_many_snapshots():
snapshots_response = ec2_client.describe_snapshots(SnapshotIds=snapshot_ids)
assert len(snapshots_response["Snapshots"]) == len(snapshot_ids)
@mock_ec2
def test_create_unencrypted_volume_with_kms_key_fails():
resource = boto3.resource("ec2", region_name="us-east-1")
with assert_raises(ClientError) as ex:
resource.create_volume(
AvailabilityZone="us-east-1a", Encrypted=False, KmsKeyId="key", Size=10
)
ex.exception.response["Error"]["Code"].should.equal("InvalidParameterDependency")
ex.exception.response["Error"]["Message"].should.contain("KmsKeyId")
@mock_kms
@mock_ec2
def test_create_encrypted_volume_without_kms_key_should_use_default_key():
kms = boto3.client("kms", region_name="us-east-1")
# Default master key for EBS does not exist until needed.
with assert_raises(ClientError) as ex:
kms.describe_key(KeyId="alias/aws/ebs")
ex.exception.response["Error"]["Code"].should.equal("NotFoundException")
# Creating an encrypted volume should create (and use) the default key.
resource = boto3.resource("ec2", region_name="us-east-1")
volume = resource.create_volume(
AvailabilityZone="us-east-1a", Encrypted=True, Size=10
)
default_ebs_key_arn = kms.describe_key(KeyId="alias/aws/ebs")["KeyMetadata"]["Arn"]
volume.kms_key_id.should.equal(default_ebs_key_arn)
volume.encrypted.should.be.true
# Subsequent encrypted volumes should use the now-created default key.
volume = resource.create_volume(
AvailabilityZone="us-east-1a", Encrypted=True, Size=10
)
volume.kms_key_id.should.equal(default_ebs_key_arn)
volume.encrypted.should.be.true
@mock_ec2
def test_create_volume_with_kms_key():
resource = boto3.resource("ec2", region_name="us-east-1")
volume = resource.create_volume(
AvailabilityZone="us-east-1a", Encrypted=True, KmsKeyId="key", Size=10
)
volume.kms_key_id.should.equal("key")
volume.encrypted.should.be.true
@mock_ec2
def test_kms_key_id_property_hidden_when_volume_not_encrypted():
client = boto3.client("ec2", region_name="us-east-1")
resp = client.create_volume(AvailabilityZone="us-east-1a", Encrypted=False, Size=10)
resp["Encrypted"].should.be.false
resp.should_not.have.key("KmsKeyId")
resp = client.describe_volumes(VolumeIds=[resp["VolumeId"]])
resp["Volumes"][0]["Encrypted"].should.be.false
resp["Volumes"][0].should_not.have.key("KmsKeyId")
resource = boto3.resource("ec2", region_name="us-east-1")
volume = resource.create_volume(
AvailabilityZone="us-east-1a", Encrypted=False, Size=10
)
volume.encrypted.should.be.false
volume.kms_key_id.should.be.none