from __future__ import unicode_literals import base64 import json import os import re import six from moto.core.responses import BaseResponse from .models import kms_backends from .exceptions import NotFoundException, ValidationException, AlreadyExistsException, NotAuthorizedException ACCOUNT_ID = "012345678912" reserved_aliases = [ 'alias/aws/ebs', 'alias/aws/s3', 'alias/aws/redshift', 'alias/aws/rds', ] class KmsResponse(BaseResponse): @property def parameters(self): 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): return kms_backends[self.region] def _display_arn(self, key_id): if key_id.startswith("arn:"): return key_id if key_id.startswith("alias/"): id_type = "" else: id_type = "key/" return "arn:aws:kms:{region}:{account}:{id_type}{key_id}".format( region=self.region, account=ACCOUNT_ID, id_type=id_type, key_id=key_id ) def _validate_cmk_id(self, key_id): """Determine whether a CMK ID exists. - raw key ID - key ARN """ is_arn = key_id.startswith("arn:") and ":key/" in key_id is_raw_key_id = 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) if not is_arn and not is_raw_key_id: raise NotFoundException("Invalid keyId {key_id}".format(key_id=key_id)) cmk_id = self.kms_backend.get_key_id(key_id) if cmk_id not in self.kms_backend.keys: raise NotFoundException("Key '{key_id}' does not exist".format(key_id=self._display_arn(key_id))) def _validate_alias(self, key_id): """Determine whether an alias exists. - alias name - alias ARN """ error = NotFoundException("Alias {key_id} is not found.".format(key_id=self._display_arn(key_id))) is_arn = key_id.startswith("arn:") and ":alias/" in key_id is_name = key_id.startswith("alias/") if not is_arn and not is_name: raise error alias_name = self.kms_backend.get_alias_name(key_id) cmk_id = self.kms_backend.get_key_id_from_alias(alias_name) if cmk_id is None: raise error def _validate_key_id(self, key_id): """Determine whether or not a key ID exists. - raw key ID - key ARN - alias name - alias ARN """ is_alias_arn = key_id.startswith("arn:") and ":alias/" in key_id is_alias_name = key_id.startswith("alias/") if is_alias_arn or is_alias_name: self._validate_alias(key_id) return self._validate_cmk_id(key_id) def create_key(self): """https://docs.aws.amazon.com/kms/latest/APIReference/API_CreateKey.html""" policy = self.parameters.get('Policy') key_usage = self.parameters.get('KeyUsage') description = self.parameters.get('Description') tags = self.parameters.get('Tags') key = self.kms_backend.create_key( policy, key_usage, description, tags, self.region) return json.dumps(key.to_dict()) def update_key_description(self): """https://docs.aws.amazon.com/kms/latest/APIReference/API_UpdateKeyDescription.html""" key_id = self.parameters.get('KeyId') description = self.parameters.get('Description') self._validate_cmk_id(key_id) self.kms_backend.update_key_description(key_id, description) return json.dumps(None) def tag_resource(self): """https://docs.aws.amazon.com/kms/latest/APIReference/API_TagResource.html""" key_id = self.parameters.get('KeyId') tags = self.parameters.get('Tags') self._validate_cmk_id(key_id) self.kms_backend.tag_resource(key_id, tags) return json.dumps({}) def list_resource_tags(self): """https://docs.aws.amazon.com/kms/latest/APIReference/API_ListResourceTags.html""" key_id = self.parameters.get('KeyId') self._validate_cmk_id(key_id) tags = self.kms_backend.list_resource_tags(key_id) return json.dumps({ "Tags": tags, "NextMarker": None, "Truncated": False, }) def describe_key(self): """https://docs.aws.amazon.com/kms/latest/APIReference/API_DescribeKey.html""" key_id = self.parameters.get('KeyId') self._validate_key_id(key_id) key = self.kms_backend.describe_key( self.kms_backend.get_key_id(key_id) ) return json.dumps(key.to_dict()) def list_keys(self): """https://docs.aws.amazon.com/kms/latest/APIReference/API_ListKeys.html""" keys = self.kms_backend.list_keys() return json.dumps({ "Keys": [ { "KeyArn": key.arn, "KeyId": key.id, } for key in keys ], "NextMarker": None, "Truncated": False, }) def create_alias(self): """https://docs.aws.amazon.com/kms/latest/APIReference/API_CreateAlias.html""" alias_name = self.parameters['AliasName'] target_key_id = self.parameters['TargetKeyId'] if not alias_name.startswith('alias/'): raise ValidationException('Invalid identifier') if alias_name in reserved_aliases: raise NotAuthorizedException() if ':' in alias_name: raise ValidationException('{alias_name} contains invalid characters for an alias'.format(alias_name=alias_name)) if not re.match(r'^[a-zA-Z0-9:/_-]+$', alias_name): raise ValidationException("1 validation error detected: Value '{alias_name}' at 'aliasName' " "failed to satisfy constraint: Member must satisfy regular " "expression pattern: ^[a-zA-Z0-9:/_-]+$" .format(alias_name=alias_name)) if self.kms_backend.alias_exists(target_key_id): raise ValidationException('Aliases must refer to keys. Not aliases') if self.kms_backend.alias_exists(alias_name): raise AlreadyExistsException('An alias with the name arn:aws:kms:{region}:012345678912:{alias_name} ' 'already exists'.format(region=self.region, alias_name=alias_name)) self._validate_cmk_id(target_key_id) self.kms_backend.add_alias(target_key_id, alias_name) return json.dumps(None) def delete_alias(self): """https://docs.aws.amazon.com/kms/latest/APIReference/API_DeleteAlias.html""" alias_name = self.parameters['AliasName'] if not alias_name.startswith('alias/'): raise ValidationException('Invalid identifier') self._validate_alias(alias_name) self.kms_backend.delete_alias(alias_name) return json.dumps(None) def list_aliases(self): """https://docs.aws.amazon.com/kms/latest/APIReference/API_ListAliases.html""" region = self.region # TODO: The actual API can filter on KeyId. response_aliases = [ { 'AliasArn': u'arn:aws:kms:{region}:012345678912:{reserved_alias}'.format(region=region, reserved_alias=reserved_alias), 'AliasName': reserved_alias } for reserved_alias in reserved_aliases ] backend_aliases = self.kms_backend.get_all_aliases() for target_key_id, aliases in backend_aliases.items(): for alias_name in aliases: response_aliases.append({ 'AliasArn': u'arn:aws:kms:{region}:012345678912:{alias_name}'.format(region=region, alias_name=alias_name), 'AliasName': alias_name, 'TargetKeyId': target_key_id, }) return json.dumps({ 'Truncated': False, 'Aliases': response_aliases, }) def enable_key_rotation(self): """https://docs.aws.amazon.com/kms/latest/APIReference/API_EnableKeyRotation.html""" key_id = self.parameters.get('KeyId') self._validate_cmk_id(key_id) self.kms_backend.enable_key_rotation(key_id) return json.dumps(None) def disable_key_rotation(self): """https://docs.aws.amazon.com/kms/latest/APIReference/API_EnableKeyRotation.html""" key_id = self.parameters.get('KeyId') self._validate_cmk_id(key_id) self.kms_backend.disable_key_rotation(key_id) return json.dumps(None) def get_key_rotation_status(self): """https://docs.aws.amazon.com/kms/latest/APIReference/API_GetKeyRotationStatus.html""" key_id = self.parameters.get('KeyId') self._validate_cmk_id(key_id) rotation_enabled = self.kms_backend.get_key_rotation_status(key_id) return json.dumps({'KeyRotationEnabled': rotation_enabled}) def put_key_policy(self): """https://docs.aws.amazon.com/kms/latest/APIReference/API_PutKeyPolicy.html""" key_id = self.parameters.get('KeyId') policy_name = self.parameters.get('PolicyName') policy = self.parameters.get('Policy') _assert_default_policy(policy_name) self._validate_cmk_id(key_id) self.kms_backend.put_key_policy(key_id, policy) return json.dumps(None) def get_key_policy(self): """https://docs.aws.amazon.com/kms/latest/APIReference/API_GetKeyPolicy.html""" key_id = self.parameters.get('KeyId') policy_name = self.parameters.get('PolicyName') _assert_default_policy(policy_name) self._validate_cmk_id(key_id) return json.dumps({'Policy': self.kms_backend.get_key_policy(key_id)}) def list_key_policies(self): """https://docs.aws.amazon.com/kms/latest/APIReference/API_ListKeyPolicies.html""" key_id = self.parameters.get('KeyId') self._validate_cmk_id(key_id) self.kms_backend.describe_key(key_id) return json.dumps({'Truncated': False, 'PolicyNames': ['default']}) def encrypt(self): """https://docs.aws.amazon.com/kms/latest/APIReference/API_Encrypt.html""" key_id = self.parameters.get("KeyId") encryption_context = self.parameters.get('EncryptionContext', {}) plaintext = self.parameters.get("Plaintext") self._validate_key_id(key_id) 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): """https://docs.aws.amazon.com/kms/latest/APIReference/API_Decrypt.html""" ciphertext_blob = self.parameters.get("CiphertextBlob") encryption_context = self.parameters.get('EncryptionContext', {}) 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): """https://docs.aws.amazon.com/kms/latest/APIReference/API_ReEncrypt.html""" 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", {}) self._validate_cmk_id(destination_key_id) 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): """https://docs.aws.amazon.com/kms/latest/APIReference/API_DisableKey.html""" key_id = self.parameters.get('KeyId') self._validate_cmk_id(key_id) self.kms_backend.disable_key(key_id) return json.dumps(None) def enable_key(self): """https://docs.aws.amazon.com/kms/latest/APIReference/API_EnableKey.html""" key_id = self.parameters.get('KeyId') self._validate_cmk_id(key_id) self.kms_backend.enable_key(key_id) return json.dumps(None) def cancel_key_deletion(self): """https://docs.aws.amazon.com/kms/latest/APIReference/API_CancelKeyDeletion.html""" key_id = self.parameters.get('KeyId') self._validate_cmk_id(key_id) self.kms_backend.cancel_key_deletion(key_id) return json.dumps({'KeyId': key_id}) def schedule_key_deletion(self): """https://docs.aws.amazon.com/kms/latest/APIReference/API_ScheduleKeyDeletion.html""" key_id = self.parameters.get('KeyId') if self.parameters.get('PendingWindowInDays') is None: pending_window_in_days = 30 else: pending_window_in_days = self.parameters.get('PendingWindowInDays') self._validate_cmk_id(key_id) return json.dumps({ 'KeyId': key_id, 'DeletionDate': self.kms_backend.schedule_key_deletion(key_id, pending_window_in_days) }) def generate_data_key(self): """https://docs.aws.amazon.com/kms/latest/APIReference/API_GenerateDataKey.html""" key_id = self.parameters.get('KeyId') 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') # Param validation self._validate_key_id(key_id) if number_of_bytes and (number_of_bytes > 1024 or number_of_bytes < 1): 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 '{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, 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_response = base64.b64encode(plaintext).decode("utf-8") ciphertext_blob_response = base64.b64encode(ciphertext_blob).decode("utf-8") return json.dumps({ 'CiphertextBlob': ciphertext_blob_response, 'Plaintext': plaintext_response, 'KeyId': key_arn # not alias }) def generate_data_key_without_plaintext(self): """https://docs.aws.amazon.com/kms/latest/APIReference/API_GenerateDataKeyWithoutPlaintext.html""" result = json.loads(self.generate_data_key()) del result['Plaintext'] return json.dumps(result) def generate_random(self): """https://docs.aws.amazon.com/kms/latest/APIReference/API_GenerateRandom.html""" number_of_bytes = self.parameters.get("NumberOfBytes") if number_of_bytes and (number_of_bytes > 1024 or number_of_bytes < 1): 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)) entropy = os.urandom(number_of_bytes) response_entropy = base64.b64encode(entropy).decode("utf-8") return json.dumps({"Plaintext": response_entropy}) def _assert_default_policy(policy_name): if policy_name != 'default': raise NotFoundException("No such policy exists")