Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #2397 from mattsb42-aws/kms

Browse Source 
Making kms:Encrypt/Decrypt/GenerateDataKey more real
This commit is contained in:
Steve Pulec 2019-09-11 22:13:33 -05:00 committed by GitHub
commit b8a79611d6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 981 additions and 550 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
IMPLEMENTATION_COVERAGE.md
View File

@ -3801,14 +3801,14 @@
- [ ] update_stream
## kms
41% implemented
54% implemented
- [X] cancel_key_deletion
- [ ] connect_custom_key_store
- [ ] create_alias
- [ ] create_custom_key_store
- [ ] create_grant
- [X] create_key
- [ ] decrypt
- [X] decrypt
- [X] delete_alias
- [ ] delete_custom_key_store
- [ ] delete_imported_key_material
@ -3819,10 +3819,10 @@
- [ ] disconnect_custom_key_store
- [X] enable_key
- [X] enable_key_rotation
- [ ] encrypt
- [X] encrypt
- [X] generate_data_key
- [ ] generate_data_key_without_plaintext
- [ ] generate_random
- [X] generate_data_key_without_plaintext
- [X] generate_random
- [X] get_key_policy
- [X] get_key_rotation_status
- [ ] get_parameters_for_import
@ -3834,7 +3834,7 @@
- [X] list_resource_tags
- [ ] list_retirable_grants
- [X] put_key_policy
- [ ] re_encrypt
- [X] re_encrypt
- [ ] retire_grant
- [ ] revoke_grant
- [X] schedule_key_deletion

20
moto/kms/exceptions.py
View File

@ -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"}'

105
moto/kms/models.py
View File

@ -4,13 +4,12 @@ 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 decrypt, encrypt, generate_key_id, generate_master_key
from collections import defaultdict
from datetime import datetime, timedelta
class Key(BaseModel):
def __init__(self, policy, key_usage, description, tags, region):
self.id = generate_key_id()
self.policy = policy
@ -23,6 +22,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):
@ -45,8 +45,8 @@ class Key(BaseModel):
"KeyState": self.key_state,
}
}
if self.key_state == 'PendingDeletion':
key_dict['KeyMetadata']['DeletionDate'] = iso_8601_datetime_without_milliseconds(self.deletion_date)
if self.key_state == "PendingDeletion":
key_dict["KeyMetadata"]["DeletionDate"] = iso_8601_datetime_without_milliseconds(self.deletion_date)
return key_dict
def delete(self, region_name):
@ -55,28 +55,28 @@ class Key(BaseModel):
@classmethod
def create_from_cloudformation_json(self, resource_name, cloudformation_json, region_name):
kms_backend = kms_backends[region_name]
properties = cloudformation_json['Properties']
properties = cloudformation_json["Properties"]
key = kms_backend.create_key(
policy=properties['KeyPolicy'],
key_usage='ENCRYPT_DECRYPT',
description=properties['Description'],
tags=properties.get('Tags'),
policy=properties["KeyPolicy"],
key_usage="ENCRYPT_DECRYPT",
description=properties["Description"],
tags=properties.get("Tags"),
region=region_name,
)
key.key_rotation_status = properties['EnableKeyRotation']
key.enabled = properties['Enabled']
key.key_rotation_status = properties["EnableKeyRotation"]
key.enabled = properties["Enabled"]
return key
def get_cfn_attribute(self, attribute_name):
from moto.cloudformation.exceptions import UnformattedGetAttTemplateException
if attribute_name == 'Arn':
if attribute_name == "Arn":
return self.arn
raise UnformattedGetAttTemplateException()
class KmsBackend(BaseBackend):
def __init__(self):
self.keys = {}
self.key_to_aliases = defaultdict(set)
@ -109,8 +109,8 @@ class KmsBackend(BaseBackend):
# allow the different methods (alias, ARN :key/, keyId, ARN alias) to
# describe key not just KeyId
key_id = self.get_key_id(key_id)
if r'alias/' in str(key_id).lower():
key_id = self.get_key_id_from_alias(key_id.split('alias/')[1])
if r"alias/" in str(key_id).lower():
key_id = self.get_key_id_from_alias(key_id.split("alias/")[1])
return self.keys[self.get_key_id(key_id)]
def list_keys(self):
@ -118,7 +118,26 @@ class KmsBackend(BaseBackend):
def get_key_id(self, key_id):
# Allow use of ARN as well as pure KeyId
return str(key_id).split(r':key/')[1] if r':key/' in str(key_id).lower() else key_id
return str(key_id).split(r":key/")[1] if r":key/" in str(key_id).lower() else key_id
def get_alias_name(self, alias_name):
# Allow use of ARN as well as alias name
return str(alias_name).split(r":alias/")[1] if r":alias/" in str(alias_name).lower() else alias_name
def any_id_to_key_id(self, key_id):
"""Go from any valid key ID to the raw key ID.
Acceptable inputs:
- raw key ID
- key ARN
- alias name
- alias ARN
"""
key_id = self.get_alias_name(key_id)
key_id = self.get_key_id(key_id)
if key_id.startswith("alias/"):
key_id = self.get_key_id_from_alias(key_id)
return key_id
def alias_exists(self, alias_name):
for aliases in self.key_to_aliases.values():
@ -162,37 +181,69 @@ class KmsBackend(BaseBackend):
def disable_key(self, key_id):
self.keys[key_id].enabled = False
self.keys[key_id].key_state = 'Disabled'
self.keys[key_id].key_state = "Disabled"
def enable_key(self, key_id):
self.keys[key_id].enabled = True
self.keys[key_id].key_state = 'Enabled'
self.keys[key_id].key_state = "Enabled"
def cancel_key_deletion(self, key_id):
self.keys[key_id].key_state = 'Disabled'
self.keys[key_id].key_state = "Disabled"
self.keys[key_id].deletion_date = None
def schedule_key_deletion(self, key_id, pending_window_in_days):
if 7 <= pending_window_in_days <= 30:
self.keys[key_id].enabled = False
self.keys[key_id].key_state = 'PendingDeletion'
self.keys[key_id].key_state = "PendingDeletion"
self.keys[key_id].deletion_date = datetime.now() + timedelta(days=pending_window_in_days)
return iso_8601_datetime_without_milliseconds(self.keys[key_id].deletion_date)
def encrypt(self, key_id, plaintext, encryption_context):
key_id = self.any_id_to_key_id(key_id)
ciphertext_blob = encrypt(
master_keys=self.keys, key_id=key_id, plaintext=plaintext, encryption_context=encryption_context
)
arn = self.keys[key_id].arn
return ciphertext_blob, arn
def decrypt(self, ciphertext_blob, encryption_context):
plaintext, key_id = decrypt(
master_keys=self.keys, ciphertext_blob=ciphertext_blob, encryption_context=encryption_context
)
arn = self.keys[key_id].arn
return plaintext, arn
def re_encrypt(
self, ciphertext_blob, source_encryption_context, destination_key_id, destination_encryption_context
):
destination_key_id = self.any_id_to_key_id(destination_key_id)
plaintext, decrypting_arn = self.decrypt(
ciphertext_blob=ciphertext_blob, encryption_context=source_encryption_context
)
new_ciphertext_blob, encrypting_arn = self.encrypt(
key_id=destination_key_id, plaintext=plaintext, encryption_context=destination_encryption_context
)
return new_ciphertext_blob, decrypting_arn, encrypting_arn
def generate_data_key(self, key_id, encryption_context, number_of_bytes, key_spec, grant_tokens):
key = self.keys[self.get_key_id(key_id)]
key_id = self.any_id_to_key_id(key_id)
if key_spec:
if key_spec == 'AES_128':
bytes = 16
# Note: Actual validation of key_spec is done in kms.responses
if key_spec == "AES_128":
plaintext_len = 16
else:
bytes = 32
plaintext_len = 32
else:
bytes = number_of_bytes
plaintext_len = number_of_bytes
plaintext = os.urandom(bytes)
plaintext = os.urandom(plaintext_len)
return plaintext, key.arn
ciphertext_blob, arn = self.encrypt(key_id=key_id, plaintext=plaintext, encryption_context=encryption_context)
return plaintext, ciphertext_blob, arn
kms_backends = {}

112
moto/kms/responses.py
View File

@ -2,7 +2,9 @@ from __future__ import unicode_literals
import base64
import json
import os
import re
import six
from moto.core.responses import BaseResponse
@ -21,7 +23,13 @@ class KmsResponse(BaseResponse):
@property
def parameters(self):
return json.loads(self.body)
params = json.loads(self.body)
for key in ("Plaintext", "CiphertextBlob"):
if key in params:
params[key] = base64.b64decode(params[key].encode("utf-8"))
return params
@property
def kms_backend(self):
@ -224,24 +232,53 @@ class KmsResponse(BaseResponse):
return json.dumps({'Truncated': False, 'PolicyNames': ['default']})
def encrypt(self):
"""
We perform no encryption, we just encode the value as base64 and then
decode it in decrypt().
"""
value = self.parameters.get("Plaintext")
if isinstance(value, six.text_type):
value = value.encode('utf-8')
return json.dumps({"CiphertextBlob": base64.b64encode(value).decode("utf-8"), 'KeyId': 'key_id'})
key_id = self.parameters.get("KeyId")
encryption_context = self.parameters.get('EncryptionContext', {})
plaintext = self.parameters.get("Plaintext")
if isinstance(plaintext, six.text_type):
plaintext = plaintext.encode('utf-8')
ciphertext_blob, arn = self.kms_backend.encrypt(
key_id=key_id,
plaintext=plaintext,
encryption_context=encryption_context,
)
ciphertext_blob_response = base64.b64encode(ciphertext_blob).decode("utf-8")
return json.dumps({"CiphertextBlob": ciphertext_blob_response, "KeyId": arn})
def decrypt(self):
# TODO refuse decode if EncryptionContext is not the same as when it was encrypted / generated
ciphertext_blob = self.parameters.get("CiphertextBlob")
encryption_context = self.parameters.get('EncryptionContext', {})
value = self.parameters.get("CiphertextBlob")
try:
return json.dumps({"Plaintext": base64.b64decode(value).decode("utf-8"), 'KeyId': 'key_id'})
except UnicodeDecodeError:
# Generate data key will produce random bytes which when decrypted is still returned as base64
return json.dumps({"Plaintext": value})
plaintext, arn = self.kms_backend.decrypt(
ciphertext_blob=ciphertext_blob,
encryption_context=encryption_context,
)
plaintext_response = base64.b64encode(plaintext).decode("utf-8")
return json.dumps({"Plaintext": plaintext_response, 'KeyId': arn})
def re_encrypt(self):
ciphertext_blob = self.parameters.get("CiphertextBlob")
source_encryption_context = self.parameters.get("SourceEncryptionContext", {})
destination_key_id = self.parameters.get("DestinationKeyId")
destination_encryption_context = self.parameters.get("DestinationEncryptionContext", {})
new_ciphertext_blob, decrypting_arn, encrypting_arn = self.kms_backend.re_encrypt(
ciphertext_blob=ciphertext_blob,
source_encryption_context=source_encryption_context,
destination_key_id=destination_key_id,
destination_encryption_context=destination_encryption_context,
)
response_ciphertext_blob = base64.b64encode(new_ciphertext_blob).decode("utf-8")
return json.dumps(
{"CiphertextBlob": response_ciphertext_blob, "KeyId": encrypting_arn, "SourceKeyId": decrypting_arn}
)
def disable_key(self):
key_id = self.parameters.get('KeyId')
@ -291,7 +328,7 @@ class KmsResponse(BaseResponse):
def generate_data_key(self):
key_id = self.parameters.get('KeyId')
encryption_context = self.parameters.get('EncryptionContext')
encryption_context = self.parameters.get('EncryptionContext', {})
number_of_bytes = self.parameters.get('NumberOfBytes')
key_spec = self.parameters.get('KeySpec')
grant_tokens = self.parameters.get('GrantTokens')
@ -306,27 +343,37 @@ class KmsResponse(BaseResponse):
raise NotFoundException('Invalid keyId')
if number_of_bytes and (number_of_bytes > 1024 or number_of_bytes < 0):
raise ValidationException("1 validation error detected: Value '2048' at 'numberOfBytes' failed "
"to satisfy constraint: Member must have value less than or "
"equal to 1024")
raise ValidationException((
"1 validation error detected: Value '{number_of_bytes:d}' at 'numberOfBytes' failed "
"to satisfy constraint: Member must have value less than or "
"equal to 1024"
).format(number_of_bytes=number_of_bytes))
if key_spec and key_spec not in ('AES_256', 'AES_128'):
raise ValidationException("1 validation error detected: Value 'AES_257' at 'keySpec' failed "
"to satisfy constraint: Member must satisfy enum value set: "
"[AES_256, AES_128]")
raise ValidationException((
"1 validation error detected: Value '{key_spec}' at 'keySpec' failed "
"to satisfy constraint: Member must satisfy enum value set: "
"[AES_256, AES_128]"
).format(key_spec=key_spec))
if not key_spec and not number_of_bytes:
raise ValidationException("Please specify either number of bytes or key spec.")
if key_spec and number_of_bytes:
raise ValidationException("Please specify either number of bytes or key spec.")
plaintext, key_arn = self.kms_backend.generate_data_key(key_id, encryption_context,
number_of_bytes, key_spec, grant_tokens)
plaintext, ciphertext_blob, key_arn = self.kms_backend.generate_data_key(
key_id=key_id,
encryption_context=encryption_context,
number_of_bytes=number_of_bytes,
key_spec=key_spec,
grant_tokens=grant_tokens
)
plaintext = base64.b64encode(plaintext).decode()
plaintext_response = base64.b64encode(plaintext).decode("utf-8")
ciphertext_blob_response = base64.b64encode(ciphertext_blob).decode("utf-8")
return json.dumps({
'CiphertextBlob': plaintext,
'Plaintext': plaintext,
'CiphertextBlob': ciphertext_blob_response,
'Plaintext': plaintext_response,
'KeyId': key_arn # not alias
})
@ -336,6 +383,15 @@ class KmsResponse(BaseResponse):
return json.dumps(result)
def generate_random(self):
number_of_bytes = self.parameters.get("NumberOfBytes")
entropy = os.urandom(number_of_bytes)
response_entropy = base64.b64encode(entropy).decode("utf-8")
return json.dumps({"Plaintext": response_entropy})
def _assert_valid_key_id(key_id):
if not re.match(r'^[A-F0-9]{8}-[A-F0-9]{4}-[A-F0-9]{4}-[A-F0-9]{4}-[A-F0-9]{12}$', key_id, re.IGNORECASE):

135
moto/kms/utils.py
View File

@ -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

1
requirements-dev.txt
View File

@ -10,6 +10,7 @@ boto>=2.45.0
boto3>=1.4.4
botocore>=1.12.13
six>=1.9
parameterized>=0.7.0
prompt-toolkit==1.0.14
click==6.7
inflection==0.3.1

989
tests/test_kms/test_kms.py
View File

File diff suppressed because it is too large Load Diff

157
tests/test_kms/test_utils.py Normal file
View File

@ -0,0 +1,157 @@
from __future__ import unicode_literals
import sure # noqa
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 (
_deserialize_ciphertext_blob,
_serialize_ciphertext_blob,
_serialize_encryption_context,
generate_data_key,
generate_master_key,
MASTER_KEY_LEN,
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",
),
)
def test_generate_data_key():
test = generate_data_key(123)
test.should.be.a(bytes)
len(test).should.equal(123)
def test_generate_master_key():
test = generate_master_key()
test.should.be.a(bytes)
len(test).should.equal(MASTER_KEY_LEN)
@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():
with 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}
with 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"
with 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"
with 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"},
)
with assert_raises(InvalidCiphertextException):
decrypt(
master_keys=master_key_map,
ciphertext_blob=ciphertext_blob,
encryption_context={},
)