add encrypt/decrypt utility functions with appropriate exceptions and tests

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={},
+    )