From 7eeead8a37768956db50a81ef7073f4b9fde1c18 Mon Sep 17 00:00:00 2001 From: mattsb42-aws Date: Mon, 26 Aug 2019 23:24:31 -0700 Subject: [PATCH] add encrypt/decrypt utility functions with appropriate exceptions and tests --- moto/kms/exceptions.py | 20 +++++ moto/kms/models.py | 3 +- moto/kms/utils.py | 135 ++++++++++++++++++++++++++++++++ tests/test_kms/test_utils.py | 146 +++++++++++++++++++++++++++++++++++ 4 files changed, 303 insertions(+), 1 deletion(-) create mode 100644 tests/test_kms/test_utils.py diff --git a/moto/kms/exceptions.py b/moto/kms/exceptions.py index 70edd3dcd..c9094e8f8 100644 --- a/moto/kms/exceptions.py +++ b/moto/kms/exceptions.py @@ -34,3 +34,23 @@ class NotAuthorizedException(JsonRESTError): "NotAuthorizedException", None) self.description = '{"__type":"NotAuthorizedException"}' + + +class AccessDeniedException(JsonRESTError): + code = 400 + + def __init__(self, message): + super(AccessDeniedException, self).__init__( + "AccessDeniedException", message) + + self.description = '{"__type":"AccessDeniedException"}' + + +class InvalidCiphertextException(JsonRESTError): + code = 400 + + def __init__(self): + super(InvalidCiphertextException, self).__init__( + "InvalidCiphertextException", None) + + self.description = '{"__type":"InvalidCiphertextException"}' diff --git a/moto/kms/models.py b/moto/kms/models.py index 577840b06..d1b61d86c 100644 --- a/moto/kms/models.py +++ b/moto/kms/models.py @@ -4,7 +4,7 @@ import os import boto.kms from moto.core import BaseBackend, BaseModel from moto.core.utils import iso_8601_datetime_without_milliseconds -from .utils import generate_key_id +from .utils import generate_key_id, generate_master_key from collections import defaultdict from datetime import datetime, timedelta @@ -23,6 +23,7 @@ class Key(BaseModel): self.key_rotation_status = False self.deletion_date = None self.tags = tags or {} + self.key_material = generate_master_key() @property def physical_resource_id(self): diff --git a/moto/kms/utils.py b/moto/kms/utils.py index fad38150f..96d3f25cc 100644 --- a/moto/kms/utils.py +++ b/moto/kms/utils.py @@ -1,7 +1,142 @@ from __future__ import unicode_literals +from collections import namedtuple +import io +import os +import struct import uuid +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives.ciphers import algorithms, Cipher, modes + +from .exceptions import InvalidCiphertextException, AccessDeniedException, NotFoundException + + +MASTER_KEY_LEN = 32 +KEY_ID_LEN = 36 +IV_LEN = 12 +TAG_LEN = 16 +HEADER_LEN = KEY_ID_LEN + IV_LEN + TAG_LEN +# NOTE: This is just a simple binary format. It is not what KMS actually does. +CIPHERTEXT_HEADER_FORMAT = ">{key_id_len}s{iv_len}s{tag_len}s".format( + key_id_len=KEY_ID_LEN, iv_len=IV_LEN, tag_len=TAG_LEN +) +Ciphertext = namedtuple("Ciphertext", ("key_id", "iv", "ciphertext", "tag")) + def generate_key_id(): return str(uuid.uuid4()) + + +def generate_data_key(number_of_bytes): + """Generate a data key.""" + return os.urandom(number_of_bytes) + + +def generate_master_key(): + """Generate a master key.""" + return generate_data_key(MASTER_KEY_LEN) + + +def _serialize_ciphertext_blob(ciphertext): + """Serialize Ciphertext object into a ciphertext blob. + + NOTE: This is just a simple binary format. It is not what KMS actually does. + """ + header = struct.pack(CIPHERTEXT_HEADER_FORMAT, ciphertext.key_id.encode("utf-8"), ciphertext.iv, ciphertext.tag) + return header + ciphertext.ciphertext + + +def _deserialize_ciphertext_blob(ciphertext_blob): + """Deserialize ciphertext blob into a Ciphertext object. + + NOTE: This is just a simple binary format. It is not what KMS actually does. + """ + header = ciphertext_blob[:HEADER_LEN] + ciphertext = ciphertext_blob[HEADER_LEN:] + key_id, iv, tag = struct.unpack(CIPHERTEXT_HEADER_FORMAT, header) + return Ciphertext(key_id=key_id.decode("utf-8"), iv=iv, ciphertext=ciphertext, tag=tag) + + +def _serialize_encryption_context(encryption_context): + """Serialize encryption context for use a AAD. + + NOTE: This is not necessarily what KMS does, but it retains the same properties. + """ + aad = io.BytesIO() + for key, value in sorted(encryption_context.items(), key=lambda x: x[0]): + aad.write(key.encode("utf-8")) + aad.write(value.encode("utf-8")) + return aad.getvalue() + + +def encrypt(master_keys, key_id, plaintext, encryption_context): + """Encrypt data using a master key material. + + NOTE: This is not necessarily what KMS does, but it retains the same properties. + + NOTE: This function is NOT compatible with KMS APIs. + :param dict master_keys: Mapping of a KmsBackend's known master keys + :param str key_id: Key ID of moto master key + :param bytes plaintext: Plaintext data to encrypt + :param dict[str, str] encryption_context: KMS-style encryption context + :returns: Moto-structured ciphertext blob encrypted under a moto master key in master_keys + :rtype: bytes + """ + try: + key = master_keys[key_id] + except KeyError: + is_alias = key_id.startswith("alias/") or ":alias/" in key_id + raise NotFoundException( + "{id_type} {key_id} is not found.".format(id_type="Alias" if is_alias else "keyId", key_id=key_id) + ) + + iv = os.urandom(IV_LEN) + aad = _serialize_encryption_context(encryption_context=encryption_context) + + encryptor = Cipher(algorithms.AES(key.key_material), modes.GCM(iv), backend=default_backend()).encryptor() + encryptor.authenticate_additional_data(aad) + ciphertext = encryptor.update(plaintext) + encryptor.finalize() + return _serialize_ciphertext_blob( + ciphertext=Ciphertext(key_id=key_id, iv=iv, ciphertext=ciphertext, tag=encryptor.tag) + ) + + +def decrypt(master_keys, ciphertext_blob, encryption_context): + """Decrypt a ciphertext blob using a master key material. + + NOTE: This is not necessarily what KMS does, but it retains the same properties. + + NOTE: This function is NOT compatible with KMS APIs. + + :param dict master_keys: Mapping of a KmsBackend's known master keys + :param bytes ciphertext_blob: moto-structured ciphertext blob encrypted under a moto master key in master_keys + :param dict[str, str] encryption_context: KMS-style encryption context + :returns: plaintext bytes and moto key ID + :rtype: bytes and str + """ + try: + ciphertext = _deserialize_ciphertext_blob(ciphertext_blob=ciphertext_blob) + except Exception: + raise InvalidCiphertextException() + + aad = _serialize_encryption_context(encryption_context=encryption_context) + + try: + key = master_keys[ciphertext.key_id] + except KeyError: + raise AccessDeniedException( + "The ciphertext refers to a customer master key that does not exist, " + "does not exist in this region, or you are not allowed to access." + ) + + try: + decryptor = Cipher( + algorithms.AES(key.key_material), modes.GCM(ciphertext.iv, ciphertext.tag), backend=default_backend() + ).decryptor() + decryptor.authenticate_additional_data(aad) + plaintext = decryptor.update(ciphertext.ciphertext) + decryptor.finalize() + except Exception: + raise InvalidCiphertextException() + + return plaintext, ciphertext.key_id diff --git a/tests/test_kms/test_utils.py b/tests/test_kms/test_utils.py new file mode 100644 index 000000000..9d540d50a --- /dev/null +++ b/tests/test_kms/test_utils.py @@ -0,0 +1,146 @@ +from __future__ import unicode_literals + +from nose.tools import assert_raises +from parameterized import parameterized + +from moto.kms.exceptions import AccessDeniedException, InvalidCiphertextException, NotFoundException +from moto.kms.models import Key +from moto.kms.utils import ( + generate_data_key, + generate_master_key, + _serialize_ciphertext_blob, + _deserialize_ciphertext_blob, + _serialize_encryption_context, + encrypt, + decrypt, + Ciphertext, +) + +ENCRYPTION_CONTEXT_VECTORS = ( + ({"this": "is", "an": "encryption", "context": "example"}, b"an" b"encryption" b"context" b"example" b"this" b"is"), + ({"a_this": "one", "b_is": "actually", "c_in": "order"}, b"a_this" b"one" b"b_is" b"actually" b"c_in" b"order"), +) +CIPHERTEXT_BLOB_VECTORS = ( + ( + Ciphertext( + key_id="d25652e4-d2d2-49f7-929a-671ccda580c6", + iv=b"123456789012", + ciphertext=b"some ciphertext", + tag=b"1234567890123456", + ), + b"d25652e4-d2d2-49f7-929a-671ccda580c6" b"123456789012" b"1234567890123456" b"some ciphertext", + ), + ( + Ciphertext( + key_id="d25652e4-d2d2-49f7-929a-671ccda580c6", + iv=b"123456789012", + ciphertext=b"some ciphertext that is much longer now", + tag=b"1234567890123456", + ), + b"d25652e4-d2d2-49f7-929a-671ccda580c6" + b"123456789012" + b"1234567890123456" + b"some ciphertext that is much longer now", + ), +) + + +@parameterized(ENCRYPTION_CONTEXT_VECTORS) +def test_serialize_encryption_context(raw, serialized): + test = _serialize_encryption_context(raw) + test.should.equal(serialized) + + +@parameterized(CIPHERTEXT_BLOB_VECTORS) +def test_cycle_ciphertext_blob(raw, _serialized): + test_serialized = _serialize_ciphertext_blob(raw) + test_deserialized = _deserialize_ciphertext_blob(test_serialized) + test_deserialized.should.equal(raw) + + +@parameterized(CIPHERTEXT_BLOB_VECTORS) +def test_serialize_ciphertext_blob(raw, serialized): + test = _serialize_ciphertext_blob(raw) + test.should.equal(serialized) + + +@parameterized(CIPHERTEXT_BLOB_VECTORS) +def test_deserialize_ciphertext_blob(raw, serialized): + test = _deserialize_ciphertext_blob(serialized) + test.should.equal(raw) + + +@parameterized(((ec[0],) for ec in ENCRYPTION_CONTEXT_VECTORS)) +def test_encrypt_decrypt_cycle(encryption_context): + plaintext = b"some secret plaintext" + master_key = Key("nop", "nop", "nop", [], "nop") + master_key_map = {master_key.id: master_key} + + ciphertext_blob = encrypt( + master_keys=master_key_map, key_id=master_key.id, plaintext=plaintext, encryption_context=encryption_context + ) + ciphertext_blob.should_not.equal(plaintext) + + decrypted, decrypting_key_id = decrypt( + master_keys=master_key_map, ciphertext_blob=ciphertext_blob, encryption_context=encryption_context + ) + decrypted.should.equal(plaintext) + decrypting_key_id.should.equal(master_key.id) + + +def test_encrypt_unknown_key_id(): + assert_raises( + NotFoundException, encrypt, master_keys={}, key_id="anything", plaintext=b"secrets", encryption_context={} + ) + + +def test_decrypt_invalid_ciphertext_format(): + master_key = Key("nop", "nop", "nop", [], "nop") + master_key_map = {master_key.id: master_key} + + assert_raises( + InvalidCiphertextException, decrypt, master_keys=master_key_map, ciphertext_blob=b"", encryption_context={} + ) + + +def test_decrypt_unknwown_key_id(): + ciphertext_blob = b"d25652e4-d2d2-49f7-929a-671ccda580c6" b"123456789012" b"1234567890123456" b"some ciphertext" + + assert_raises( + AccessDeniedException, decrypt, master_keys={}, ciphertext_blob=ciphertext_blob, encryption_context={} + ) + + +def test_decrypt_invalid_ciphertext(): + master_key = Key("nop", "nop", "nop", [], "nop") + master_key_map = {master_key.id: master_key} + ciphertext_blob = master_key.id.encode("utf-8") + b"123456789012" b"1234567890123456" b"some ciphertext" + + assert_raises( + InvalidCiphertextException, + decrypt, + master_keys=master_key_map, + ciphertext_blob=ciphertext_blob, + encryption_context={}, + ) + + +def test_decrypt_invalid_encryption_context(): + plaintext = b"some secret plaintext" + master_key = Key("nop", "nop", "nop", [], "nop") + master_key_map = {master_key.id: master_key} + + ciphertext_blob = encrypt( + master_keys=master_key_map, + key_id=master_key.id, + plaintext=plaintext, + encryption_context={"some": "encryption", "context": "here"}, + ) + + assert_raises( + InvalidCiphertextException, + decrypt, + master_keys=master_key_map, + ciphertext_blob=ciphertext_blob, + encryption_context={}, + )