# Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"). You
# may not use this file except in compliance with the License. A copy of
# the License is located at
#
# http://aws.amazon.com/apache2.0/
#
# or in the "license" file accompanying this file. This file is
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.
"""
AWS Encryption SDK Decrypt Message Generation manifest handler.

Described in AWS Crypto Tools Test Vector Framework feature #0006 AWS Encryption SDK Decrypt Message Generation.
"""
import json
import os
import uuid
from copy import copy

import attr
import six
from aws_encryption_sdk.caches.local import LocalCryptoMaterialsCache
from aws_encryption_sdk.materials_managers.base import CryptoMaterialsManager
from aws_encryption_sdk.materials_managers.caching import CachingCryptoMaterialsManager
from aws_encryption_sdk.materials_managers.default import DefaultCryptoMaterialsManager

from awses_test_vectors.internal.defaults import ENCODING
from awses_test_vectors.internal.util import (
    dictionary_validator,
    file_reader,
    file_writer,
    iterable_validator,
    membership_validator,
    validate_manifest_type,
)
from awses_test_vectors.manifests.full_message.decrypt import (
    DecryptionMethod,
    MessageDecryptionManifest,
    MessageDecryptionTestResult,
    MessageDecryptionTestScenario,
)
from awses_test_vectors.manifests.full_message.encrypt import MessageEncryptionTestScenario
from awses_test_vectors.manifests.keys import KeysManifest

try:
    from aws_encryption_sdk.identifiers import AlgorithmSuite
except ImportError:
    from aws_encryption_sdk.identifiers import Algorithm as AlgorithmSuite

from awses_test_vectors.manifests.master_key import MasterKeySpec, master_key_provider_from_master_key_specs

try:  # Python 3.5.0 and 3.5.1 have incompatible typing modules
    from typing import IO, Callable, Dict, Iterable, Optional  # noqa pylint: disable=unused-import

    from awses_test_vectors.internal.mypy_types import (  # noqa pylint: disable=unused-import
        ENCRYPT_SCENARIO_SPEC,
        PLAINTEXTS_SPEC,
    )
except ImportError:  # pragma: no cover
    # We only actually need these imports when running the mypy checks
    pass

SUPPORTED_VERSIONS = (2,)


class TamperingMethod:
    """Base class for all tampering methods."""

    @classmethod
    def from_tampering_spec(cls, spec):
        """Load from a tampering specification"""
        if spec is None:
            return TamperingMethod()
        if spec == "truncate":
            return TruncateTamperingMethod()
        if spec == "mutate":
            return MutateTamperingMethod()
        if spec == "half-sign":
            return HalfSigningTamperingMethod()
        ((tampering_tag, tampering_values_spec),) = spec.items()
        if tampering_tag == "change-edk-provider-info":
            return ChangeEDKProviderInfoTamperingMethod.from_values_spec(tampering_values_spec)
        raise ValueError("Unrecognized tampering method tag: " + tampering_tag)

    # pylint: disable=R0201
    def run_scenario_with_tampering(self, ciphertext_writer, generation_scenario, plaintext_uri):
        """
        Run a given scenario, tampering with the input or the result.

        return: a list of (ciphertext, result) pairs
        """
        materials_manager = DefaultCryptoMaterialsManager(
            generation_scenario.encryption_scenario.master_key_provider_fn()
        )
        ciphertext_to_decrypt = generation_scenario.encryption_scenario.run(materials_manager)
        if generation_scenario.result:
            expected_result = generation_scenario.result
        else:
            expected_result = MessageDecryptionTestResult.expect_output(
                plaintext_uri=plaintext_uri, plaintext=generation_scenario.encryption_scenario.plaintext
            )
        return [
            generation_scenario.decryption_test_scenario_pair(ciphertext_writer, ciphertext_to_decrypt, expected_result)
        ]


class ChangeEDKProviderInfoTamperingMethod(TamperingMethod):
    """Tampering method that changes the provider info on all EDKs."""

    new_provider_infos = attr.ib(validator=iterable_validator(list, six.string_types))

    def __init__(self, new_provider_infos):
        """Create a new instance for a given new provider info value."""
        self.new_provider_infos = new_provider_infos

    @classmethod
    def from_values_spec(cls, values_spec):
        """Load from a tampering parameters specification"""
        return ChangeEDKProviderInfoTamperingMethod(values_spec)

    # pylint: disable=R0201
    def run_scenario_with_tampering(self, ciphertext_writer, generation_scenario, _plaintext_uri):
        """
        Run a given scenario, tampering with the input or the result.

        return: a list of (ciphertext, result) pairs.
        """
        master_key_provider = generation_scenario.encryption_scenario.master_key_provider_fn()

        # Use a caching CMM to avoid generating a new data key every time.
        cache = LocalCryptoMaterialsCache(10)
        caching_cmm = CachingCryptoMaterialsManager(
            master_key_provider=master_key_provider,
            cache=cache,
            max_age=60.0,
            max_messages_encrypted=100,
        )
        return [
            self.run_scenario_with_new_provider_info(
                ciphertext_writer, generation_scenario, caching_cmm, new_provider_info
            )
            for new_provider_info in self.new_provider_infos
        ]

    def run_scenario_with_new_provider_info(
        self, ciphertext_writer, generation_scenario, materials_manager, new_provider_info
    ):
        """Run with tampering for a specific new provider info value"""
        tampering_materials_manager = ProviderInfoChangingCryptoMaterialsManager(materials_manager, new_provider_info)
        ciphertext_to_decrypt = generation_scenario.encryption_scenario.run(tampering_materials_manager)
        expected_result = MessageDecryptionTestResult.expect_error(
            "Incorrect encrypted data key provider info: " + new_provider_info
        )
        return generation_scenario.decryption_test_scenario_pair(
            ciphertext_writer, ciphertext_to_decrypt, expected_result
        )


class ProviderInfoChangingCryptoMaterialsManager(CryptoMaterialsManager):
    """
    Custom CMM that modifies the provider info field on EDKS.

    THIS IS ONLY USED TO CREATE INVALID MESSAGES and should never be used in
    production!
    """

    wrapped_cmm = attr.ib(validator=attr.validators.instance_of(CryptoMaterialsManager))
    new_provider_info = attr.ib(validator=attr.validators.instance_of(six.string_types))

    def __init__(self, materials_manager, new_provider_info):
        """Create a new CMM that wraps a the given CMM."""
        self.wrapped_cmm = materials_manager
        self.new_provider_info = new_provider_info

    def get_encryption_materials(self, request):
        """
        Request materials from the wrapped CMM, and then change the provider info
        on each EDK.
        """
        result = self.wrapped_cmm.get_encryption_materials(request)
        for encrypted_data_key in result.encrypted_data_keys:
            encrypted_data_key.key_provider.key_info = self.new_provider_info
        return result

    def decrypt_materials(self, request):
        """Thunks to the wrapped CMM"""
        return self.wrapped_cmm.decrypt_materials(request)


BITS_PER_BYTE = 8


class TruncateTamperingMethod(TamperingMethod):
    """Tampering method that truncates a good message at every byte (except zero)."""

    # pylint: disable=R0201
    def run_scenario_with_tampering(self, ciphertext_writer, generation_scenario, _plaintext_uri):
        """
        Run a given scenario, tampering with the input or the result.

        return: a list of (ciphertext, result) pairs.
        """
        ciphertext_to_decrypt = generation_scenario.encryption_scenario.run()
        return [
            generation_scenario.decryption_test_scenario_pair(
                ciphertext_writer,
                TruncateTamperingMethod.flip_bit(ciphertext_to_decrypt, bit),
                MessageDecryptionTestResult.expect_error("Bit {} flipped".format(bit)),
            )
            for bit in range(0, len(ciphertext_to_decrypt) * BITS_PER_BYTE)
        ]

    @classmethod
    def flip_bit(cls, ciphertext, bit):
        """Flip only the given bit in the given ciphertext"""
        byte_index, bit_index = divmod(bit, BITS_PER_BYTE)
        result = bytearray(ciphertext)
        result[byte_index] ^= 1 << (BITS_PER_BYTE - bit_index - 1)
        return bytes(result)


class MutateTamperingMethod(TamperingMethod):
    """Tampering method that produces a message with a single bit flipped, for every possible bit."""

    # pylint: disable=R0201
    def run_scenario_with_tampering(self, ciphertext_writer, generation_scenario, _plaintext_uri):
        """
        Run a given scenario, tampering with the input or the result.

        return: a list of (ciphertext, result) pairs.
        """
        ciphertext_to_decrypt = generation_scenario.encryption_scenario.run()
        return [
            generation_scenario.decryption_test_scenario_pair(
                ciphertext_writer,
                ciphertext_to_decrypt[0:length],
                MessageDecryptionTestResult.expect_error("Truncated at byte {}".format(length)),
            )
            for length in range(1, len(ciphertext_to_decrypt))
        ]


class HalfSigningTamperingMethod(TamperingMethod):
    """Tampering method that changes the provider info on all EDKs."""

    # pylint: disable=R0201
    def run_scenario_with_tampering(self, ciphertext_writer, generation_scenario, _plaintext_uri):
        """
        Run a given scenario, tampering with the input or the result.

        return: a list of (ciphertext, result) pairs.
        """
        tampering_materials_manager = HalfSigningCryptoMaterialsManager(
            generation_scenario.encryption_scenario.master_key_provider_fn()
        )
        ciphertext_to_decrypt = generation_scenario.encryption_scenario.run(tampering_materials_manager)
        expected_result = MessageDecryptionTestResult.expect_error(
            "Unsigned message using a data key with a public key"
        )
        return [
            generation_scenario.decryption_test_scenario_pair(ciphertext_writer, ciphertext_to_decrypt, expected_result)
        ]


class HalfSigningCryptoMaterialsManager(CryptoMaterialsManager):
    """
    Custom CMM that generates materials for an unsigned algorithm suite
    that includes the "aws-crypto-public-key" encryption context.

    THIS IS ONLY USED TO CREATE INVALID MESSAGES and should never be used in
    production! It is imitating what a malicious decryptor without encryption
    permissions might do, to attempt to forge an unsigned message from a decrypted
    signed message, and therefore this is an important case for ESDKs to reject.
    """

    wrapped_default_cmm = attr.ib(validator=attr.validators.instance_of(CryptoMaterialsManager))

    def __init__(self, master_key_provider):
        """
        Create a new CMM that wraps a new DefaultCryptoMaterialsManager
        based on the given master key provider.
        """
        self.wrapped_default_cmm = DefaultCryptoMaterialsManager(master_key_provider)

    def get_encryption_materials(self, request):
        """
        Generate half-signing materials by requesting signing materials
        from the wrapped default CMM, and then changing the algorithm suite
        and removing the signing key from teh result.
        """
        if request.algorithm == AlgorithmSuite.AES_256_GCM_HKDF_SHA512_COMMIT_KEY:
            signing_request = copy(request)
            signing_request.algorithm = AlgorithmSuite.AES_256_GCM_HKDF_SHA512_COMMIT_KEY_ECDSA_P384

            result = self.wrapped_default_cmm.get_encryption_materials(signing_request)
            result.algorithm = request.algorithm
            result.signing_key = None

            return result

        raise NotImplementedError(
            "The half-sign tampering method is only supported on the "
            "AES_256_GCM_HKDF_SHA512_COMMIT_KEY algorithm suite."
        )

    def decrypt_materials(self, request):
        """Thunks to the wrapped default CMM"""
        return self.wrapped_default_cmm.decrypt_materials(request)


@attr.s
class MessageDecryptionTestScenarioGenerator(object):
    # pylint: disable=too-many-instance-attributes
    """Data class for a single full message decrypt test scenario.

    Handles serialization and deserialization to and from manifest specs.

    :param MessageEncryptionTestScenario encryption_scenario: Encryption parameters
    :param tampering_method: Optional method used to tamper with the ciphertext
    :type tampering_method: :class:`TamperingMethod`
    :param decryption_method:
    :param decryption_master_key_specs: Iterable of master key specifications
    :type decryption_master_key_specs: iterable of :class:`MasterKeySpec`
    :param Callable decryption_master_key_provider_fn:
    :param result:
    """

    encryption_scenario = attr.ib(validator=attr.validators.instance_of(MessageEncryptionTestScenario))
    tampering_method = attr.ib(validator=attr.validators.optional(attr.validators.instance_of(TamperingMethod)))
    decryption_method = attr.ib(validator=attr.validators.optional(attr.validators.instance_of(DecryptionMethod)))
    decryption_master_key_specs = attr.ib(validator=iterable_validator(list, MasterKeySpec))
    decryption_master_key_provider_fn = attr.ib(validator=attr.validators.is_callable())
    result = attr.ib(validator=attr.validators.optional(attr.validators.instance_of(MessageDecryptionTestResult)))

    @classmethod
    def from_scenario(cls, scenario, keys, plaintexts):
        """Load from a scenario specification.

        :param dict scenario: Scenario specification JSON
        :param KeysManifest keys: Loaded keys
        :param dict plaintexts: Mapping of plaintext names to plaintext values
        :return: Loaded test scenario
        :rtype: MessageDecryptionTestScenarioGenerator
        """
        encryption_scenario_spec = scenario["encryption-scenario"]
        encryption_scenario = MessageEncryptionTestScenario.from_scenario(encryption_scenario_spec, keys, plaintexts)
        tampering = scenario.get("tampering")
        tampering_method = TamperingMethod.from_tampering_spec(tampering)
        decryption_method_spec = scenario.get("decryption-method")
        decryption_method = DecryptionMethod(decryption_method_spec) if decryption_method_spec else None
        if "decryption-master-keys" in scenario:
            decryption_master_key_specs = [
                MasterKeySpec.from_scenario(spec) for spec in scenario["decryption-master-keys"]
            ]

            def decryption_master_key_provider_fn():
                return master_key_provider_from_master_key_specs(keys, decryption_master_key_specs)

        else:
            decryption_master_key_specs = encryption_scenario.master_key_specs
            decryption_master_key_provider_fn = encryption_scenario.master_key_provider_fn
        result_spec = scenario.get("result")
        result = MessageDecryptionTestResult.from_result_spec(result_spec, None) if result_spec else None

        return cls(
            encryption_scenario=encryption_scenario,
            tampering_method=tampering_method,
            decryption_method=decryption_method,
            decryption_master_key_specs=decryption_master_key_specs,
            decryption_master_key_provider_fn=decryption_master_key_provider_fn,
            result=result,
        )

    def run(self, ciphertext_writer, plaintext_uri):
        """Run this scenario, writing the resulting ciphertext with ``ciphertext_writer`` and returning
        a :class:`MessageDecryptionTestScenario` that describes the matching decrypt scenario.

        :param callable ciphertext_writer: Callable that will write the requested named ciphertext and
            return a URI locating the written data
        :param str plaintext_uri: URI locating the written plaintext data for this scenario
        :return: Decrypt test scenario that describes the generated scenario
        :rtype: MessageDecryptionTestScenario
        """
        return dict(self.tampering_method.run_scenario_with_tampering(ciphertext_writer, self, plaintext_uri))

    def decryption_test_scenario_pair(self, ciphertext_writer, ciphertext_to_decrypt, expected_result):
        """Create a new (name, decryption scenario) pair"""
        ciphertext_name = str(uuid.uuid4())
        ciphertext_uri = ciphertext_writer(ciphertext_name, ciphertext_to_decrypt)

        return (
            ciphertext_name,
            MessageDecryptionTestScenario(
                ciphertext_uri=ciphertext_uri,
                ciphertext=ciphertext_to_decrypt,
                master_key_specs=self.decryption_master_key_specs,
                master_key_provider_fn=self.decryption_master_key_provider_fn,
                decryption_method=self.decryption_method,
                result=expected_result,
            ),
        )


@attr.s
class MessageDecryptionGenerationManifest(object):
    """AWS Encryption SDK Decryption Message Generation manifest handler.

    Described in AWS Crypto Tools Test Vector Framework feature #0006 AWS Encryption SDK Decrypt Message Generation.

    :param int version: Version of this manifest
    :param KeysManifest keys: Loaded keys
    :param dict plaintexts: Mapping of plaintext names to plaintext values
    :param dict tests: Mapping of test scenario names to :class:`MessageDecryptionGenerationManifest`s
    """

    version = attr.ib(validator=membership_validator(SUPPORTED_VERSIONS))
    keys = attr.ib(validator=attr.validators.instance_of(KeysManifest))
    plaintexts = attr.ib(validator=dictionary_validator(six.string_types, six.binary_type))
    tests = attr.ib(validator=dictionary_validator(six.string_types, MessageDecryptionTestScenarioGenerator))
    type_name = "awses-decrypt-generate"

    @staticmethod
    def _generate_plaintexts(plaintexts_specs):
        # type: (PLAINTEXTS_SPEC) -> Dict[str, bytes]
        """Generate required plaintext values.

        :param dict plaintexts_specs: Mapping of plaintext name to size in bytes
        :return: Mapping of plaintext name to randomly generated bytes
        :rtype: dict
        """
        return {name: os.urandom(size) for name, size in plaintexts_specs.items()}

    @classmethod
    def from_file(cls, input_file):
        # type: (IO) -> MessageDecryptionGenerationManifest
        """Load from a file containing a full message encrypt manifest.

        :param file input_file: File object for file containing JSON manifest
        :return: Loaded manifest
        :rtype: MessageEncryptionManifest
        """
        raw_manifest = json.load(input_file)
        validate_manifest_type(
            type_name=cls.type_name, manifest_version=raw_manifest["manifest"], supported_versions=SUPPORTED_VERSIONS
        )

        parent_dir = os.path.abspath(os.path.dirname(input_file.name))
        reader = file_reader(parent_dir)
        raw_keys_manifest = json.loads(reader(raw_manifest["keys"]).decode(ENCODING))
        keys = KeysManifest.from_manifest_spec(raw_keys_manifest)
        plaintexts = cls._generate_plaintexts(raw_manifest["plaintexts"])
        tests = {}
        for name, scenario in raw_manifest["tests"].items():
            try:
                tests[name] = MessageDecryptionTestScenarioGenerator.from_scenario(
                    scenario=scenario, keys=keys, plaintexts=plaintexts
                )
            except NotImplementedError:
                continue
        return cls(version=raw_manifest["manifest"]["version"], keys=keys, plaintexts=plaintexts, tests=tests)

    def run_and_write_to_dir(self, target_directory, json_indent=None):
        # type: (str, Optional[int]) -> None
        """Process all known encrypt test scenarios and write the resulting data and manifests to disk.

        :param str target_directory: Directory in which to write all output
        :param int json_indent: Number of spaces to indent JSON files (optional: default is to write minified)
        """
        root_dir = os.path.abspath(target_directory)
        root_writer = file_writer(root_dir)

        root_writer("keys.json", json.dumps(self.keys.manifest_spec, indent=json_indent).encode(ENCODING))

        plaintext_writer = file_writer(os.path.join(root_dir, "plaintexts"))
        plaintext_uris = {name: plaintext_writer(name, plaintext) for name, plaintext in self.plaintexts.items()}

        ciphertext_writer = file_writer(os.path.join(root_dir, "ciphertexts"))

        test_scenarios = {
            decrypt_scenario_name: decrypt_scenario
            for name, scenario in self.tests.items()
            for decrypt_scenario_name, decrypt_scenario in scenario.run(
                ciphertext_writer, plaintext_uris[scenario.encryption_scenario.plaintext_name]
            ).items()
        }

        decrypt_manifest = MessageDecryptionManifest(
            keys_uri="file://keys.json", keys=self.keys, test_scenarios=test_scenarios
        )

        root_writer("manifest.json", json.dumps(decrypt_manifest.manifest_spec, indent=json_indent).encode(ENCODING))

[((134, 16, 134, 45), 'aws_encryption_sdk.caches.local.LocalCryptoMaterialsCache', 'LocalCryptoMaterialsCache', ({(134, 42, 134, 44): '10'}, {}), '(10)', False, 'from aws_encryption_sdk.caches.local import LocalCryptoMaterialsCache\n'), ((135, 22, 140, 9), 'aws_encryption_sdk.materials_managers.caching.CachingCryptoMaterialsManager', 'CachingCryptoMaterialsManager', (), '', False, 'from aws_encryption_sdk.materials_managers.caching import CachingCryptoMaterialsManager\n'), ((154, 26, 156, 9), 'awses_test_vectors.manifests.full_message.decrypt.MessageDecryptionTestResult.expect_error', 'MessageDecryptionTestResult.expect_error', ({(155, 12, 155, 78): "'Incorrect encrypted data key provider info: ' + new_provider_info"}, {}), "(\n    'Incorrect encrypted data key provider info: ' + new_provider_info)", False, 'from awses_test_vectors.manifests.full_message.decrypt import DecryptionMethod, MessageDecryptionManifest, MessageDecryptionTestResult, MessageDecryptionTestScenario\n'), ((260, 26, 262, 9), 'awses_test_vectors.manifests.full_message.decrypt.MessageDecryptionTestResult.expect_error', 'MessageDecryptionTestResult.expect_error', ({(261, 12, 261, 65): '"""Unsigned message using a data key with a public key"""'}, {}), "(\n    'Unsigned message using a data key with a public key')", False, 'from awses_test_vectors.manifests.full_message.decrypt import DecryptionMethod, MessageDecryptionManifest, MessageDecryptionTestResult, MessageDecryptionTestScenario\n'), ((286, 35, 286, 85), 'aws_encryption_sdk.materials_managers.default.DefaultCryptoMaterialsManager', 'DefaultCryptoMaterialsManager', ({(286, 65, 286, 84): 'master_key_provider'}, {}), '(master_key_provider)', False, 'from aws_encryption_sdk.materials_managers.default import DefaultCryptoMaterialsManager\n'), ((349, 30, 349, 117), 'awses_test_vectors.manifests.full_message.encrypt.MessageEncryptionTestScenario.from_scenario', 'MessageEncryptionTestScenario.from_scenario', ({(349, 74, 349, 98): 'encryption_scenario_spec', (349, 100, 349, 104): 'keys', (349, 106, 349, 116): 'plaintexts'}, {}), '(encryption_scenario_spec, keys,\n    plaintexts)', False, 'from awses_test_vectors.manifests.full_message.encrypt import MessageEncryptionTestScenario\n'), ((445, 23, 445, 44), 'json.load', 'json.load', ({(445, 33, 445, 43): 'input_file'}, {}), '(input_file)', False, 'import json\n'), ((446, 8, 448, 9), 'awses_test_vectors.internal.util.validate_manifest_type', 'validate_manifest_type', (), '', False, 'from awses_test_vectors.internal.util import dictionary_validator, file_reader, file_writer, iterable_validator, membership_validator, validate_manifest_type\n'), ((451, 17, 451, 40), 'awses_test_vectors.internal.util.file_reader', 'file_reader', ({(451, 29, 451, 39): 'parent_dir'}, {}), '(parent_dir)', False, 'from awses_test_vectors.internal.util import dictionary_validator, file_reader, file_writer, iterable_validator, membership_validator, validate_manifest_type\n'), ((453, 15, 453, 65), 'awses_test_vectors.manifests.keys.KeysManifest.from_manifest_spec', 'KeysManifest.from_manifest_spec', ({(453, 47, 453, 64): 'raw_keys_manifest'}, {}), '(raw_keys_manifest)', False, 'from awses_test_vectors.manifests.keys import KeysManifest\n'), ((472, 19, 472, 52), 'os.path.abspath', 'os.path.abspath', ({(472, 35, 472, 51): 'target_directory'}, {}), '(target_directory)', False, 'import os\n'), ((473, 22, 473, 43), 'awses_test_vectors.internal.util.file_writer', 'file_writer', ({(473, 34, 473, 42): 'root_dir'}, {}), '(root_dir)', False, 'from awses_test_vectors.internal.util import dictionary_validator, file_reader, file_writer, iterable_validator, membership_validator, validate_manifest_type\n'), ((490, 27, 492, 9), 'awses_test_vectors.manifests.full_message.decrypt.MessageDecryptionManifest', 'MessageDecryptionManifest', (), '', False, 'from awses_test_vectors.manifests.full_message.decrypt import DecryptionMethod, MessageDecryptionManifest, MessageDecryptionTestResult, MessageDecryptionTestScenario\n'), ((102, 30, 104, 13), 'awses_test_vectors.manifests.full_message.decrypt.MessageDecryptionTestResult.expect_output', 'MessageDecryptionTestResult.expect_output', (), '', False, 'from awses_test_vectors.manifests.full_message.decrypt import DecryptionMethod, MessageDecryptionManifest, MessageDecryptionTestResult, MessageDecryptionTestScenario\n'), ((113, 43, 113, 85), 'awses_test_vectors.internal.util.iterable_validator', 'iterable_validator', ({(113, 62, 113, 66): 'list', (113, 68, 113, 84): 'six.string_types'}, {}), '(list, six.string_types)', False, 'from awses_test_vectors.internal.util import dictionary_validator, file_reader, file_writer, iterable_validator, membership_validator, validate_manifest_type\n'), ((170, 36, 170, 87), 'attr.validators.instance_of', 'attr.validators.instance_of', ({(170, 64, 170, 86): 'CryptoMaterialsManager'}, {}), '(CryptoMaterialsManager)', False, 'import attr\n'), ((171, 42, 171, 87), 'attr.validators.instance_of', 'attr.validators.instance_of', ({(171, 70, 171, 86): 'six.string_types'}, {}), '(six.string_types)', False, 'import attr\n'), ((279, 44, 279, 95), 'attr.validators.instance_of', 'attr.validators.instance_of', ({(279, 72, 279, 94): 'CryptoMaterialsManager'}, {}), '(CryptoMaterialsManager)', False, 'import attr\n'), ((295, 30, 295, 43), 'copy.copy', 'copy', ({(295, 35, 295, 42): 'request'}, {}), '(request)', False, 'from copy import copy\n'), ((331, 44, 331, 102), 'attr.validators.instance_of', 'attr.validators.instance_of', ({(331, 72, 331, 101): 'MessageEncryptionTestScenario'}, {}), '(MessageEncryptionTestScenario)', False, 'import attr\n'), ((334, 52, 334, 91), 'awses_test_vectors.internal.util.iterable_validator', 'iterable_validator', ({(334, 71, 334, 75): 'list', (334, 77, 334, 90): 'MasterKeySpec'}, {}), '(list, MasterKeySpec)', False, 'from awses_test_vectors.internal.util import dictionary_validator, file_reader, file_writer, iterable_validator, membership_validator, validate_manifest_type\n'), ((335, 58, 335, 87), 'attr.validators.is_callable', 'attr.validators.is_callable', ({}, {}), '()', False, 'import attr\n'), ((353, 28, 353, 68), 'awses_test_vectors.manifests.full_message.decrypt.DecryptionMethod', 'DecryptionMethod', ({(353, 45, 353, 67): 'decryption_method_spec'}, {}), '(decryption_method_spec)', False, 'from awses_test_vectors.manifests.full_message.decrypt import DecryptionMethod, MessageDecryptionManifest, MessageDecryptionTestResult, MessageDecryptionTestScenario\n'), ((366, 17, 366, 80), 'awses_test_vectors.manifests.full_message.decrypt.MessageDecryptionTestResult.from_result_spec', 'MessageDecryptionTestResult.from_result_spec', ({(366, 62, 366, 73): 'result_spec', (366, 75, 366, 79): 'None'}, {}), '(result_spec, None)', False, 'from awses_test_vectors.manifests.full_message.decrypt import DecryptionMethod, MessageDecryptionManifest, MessageDecryptionTestResult, MessageDecryptionTestScenario\n'), ((391, 30, 391, 42), 'uuid.uuid4', 'uuid.uuid4', ({}, {}), '()', False, 'import uuid\n'), ((396, 12, 403, 13), 'awses_test_vectors.manifests.full_message.decrypt.MessageDecryptionTestScenario', 'MessageDecryptionTestScenario', (), '', False, 'from awses_test_vectors.manifests.full_message.decrypt import DecryptionMethod, MessageDecryptionManifest, MessageDecryptionTestResult, MessageDecryptionTestScenario\n'), ((419, 32, 419, 72), 'awses_test_vectors.internal.util.membership_validator', 'membership_validator', ({(419, 53, 419, 71): 'SUPPORTED_VERSIONS'}, {}), '(SUPPORTED_VERSIONS)', False, 'from awses_test_vectors.internal.util import dictionary_validator, file_reader, file_writer, iterable_validator, membership_validator, validate_manifest_type\n'), ((420, 29, 420, 70), 'attr.validators.instance_of', 'attr.validators.instance_of', ({(420, 57, 420, 69): 'KeysManifest'}, {}), '(KeysManifest)', False, 'import attr\n'), ((421, 35, 421, 90), 'awses_test_vectors.internal.util.dictionary_validator', 'dictionary_validator', ({(421, 56, 421, 72): 'six.string_types', (421, 74, 421, 89): 'six.binary_type'}, {}), '(six.string_types, six.binary_type)', False, 'from awses_test_vectors.internal.util import dictionary_validator, file_reader, file_writer, iterable_validator, membership_validator, validate_manifest_type\n'), ((422, 30, 422, 108), 'awses_test_vectors.internal.util.dictionary_validator', 'dictionary_validator', ({(422, 51, 422, 67): 'six.string_types', (422, 69, 422, 107): 'MessageDecryptionTestScenarioGenerator'}, {}), '(six.string_types, MessageDecryptionTestScenarioGenerator)', False, 'from awses_test_vectors.internal.util import dictionary_validator, file_reader, file_writer, iterable_validator, membership_validator, validate_manifest_type\n'), ((434, 22, 434, 38), 'os.urandom', 'os.urandom', ({(434, 33, 434, 37): 'size'}, {}), '(size)', False, 'import os\n'), ((450, 37, 450, 69), 'os.path.dirname', 'os.path.dirname', ({(450, 53, 450, 68): 'input_file.name'}, {}), '(input_file.name)', False, 'import os\n'), ((477, 39, 477, 75), 'os.path.join', 'os.path.join', ({(477, 52, 477, 60): 'root_dir', (477, 62, 477, 74): '"""plaintexts"""'}, {}), "(root_dir, 'plaintexts')", False, 'import os\n'), ((480, 40, 480, 77), 'os.path.join', 'os.path.join', ({(480, 53, 480, 61): 'root_dir', (480, 63, 480, 76): '"""ciphertexts"""'}, {}), "(root_dir, 'ciphertexts')", False, 'import os\n'), ((332, 66, 332, 110), 'attr.validators.instance_of', 'attr.validators.instance_of', ({(332, 94, 332, 109): 'TamperingMethod'}, {}), '(TamperingMethod)', False, 'import attr\n'), ((333, 67, 333, 112), 'attr.validators.instance_of', 'attr.validators.instance_of', ({(333, 95, 333, 111): 'DecryptionMethod'}, {}), '(DecryptionMethod)', False, 'import attr\n'), ((336, 56, 336, 112), 'attr.validators.instance_of', 'attr.validators.instance_of', ({(336, 84, 336, 111): 'MessageDecryptionTestResult'}, {}), '(MessageDecryptionTestResult)', False, 'import attr\n'), ((356, 16, 356, 49), 'awses_test_vectors.manifests.master_key.MasterKeySpec.from_scenario', 'MasterKeySpec.from_scenario', ({(356, 44, 356, 48): 'spec'}, {}), '(spec)', False, 'from awses_test_vectors.manifests.master_key import MasterKeySpec, master_key_provider_from_master_key_specs\n'), ((360, 23, 360, 99), 'awses_test_vectors.manifests.master_key.master_key_provider_from_master_key_specs', 'master_key_provider_from_master_key_specs', ({(360, 65, 360, 69): 'keys', (360, 71, 360, 98): 'decryption_master_key_specs'}, {}), '(keys, decryption_master_key_specs)', False, 'from awses_test_vectors.manifests.master_key import MasterKeySpec, master_key_provider_from_master_key_specs\n'), ((475, 33, 475, 88), 'json.dumps', 'json.dumps', (), '', False, 'import json\n'), ((494, 37, 494, 99), 'json.dumps', 'json.dumps', (), '', False, 'import json\n')]