# Copyright 2012-2014 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.

"""Builtin event handlers.

This module contains builtin handlers for events emitted by botocore.
"""

import base64
import logging
import xml.etree.cElementTree
import copy
import re
import warnings
import uuid

from botocore.compat import unquote, json, six, unquote_str, \
    ensure_bytes, get_md5, MD5_AVAILABLE
from botocore.docs.utils import AutoPopulatedParam
from botocore.docs.utils import HideParamFromOperations
from botocore.docs.utils import AppendParamDocumentation
from botocore.signers import add_generate_presigned_url
from botocore.signers import add_generate_presigned_post
from botocore.exceptions import ParamValidationError
from botocore.exceptions import AliasConflictParameterError
from botocore.exceptions import UnsupportedTLSVersionWarning
from botocore.utils import percent_encode, SAFE_CHARS
from botocore.utils import switch_host_with_param

from botocore import retryhandler
from botocore import utils
from botocore import translate
import botocore
import botocore.auth


logger = logging.getLogger(__name__)

REGISTER_FIRST = object()
REGISTER_LAST = object()
# From the S3 docs:
# The rules for bucket names in the US Standard region allow bucket names
# to be as long as 255 characters, and bucket names can contain any
# combination of uppercase letters, lowercase letters, numbers, periods
# (.), hyphens (-), and underscores (_).
VALID_BUCKET = re.compile('^[a-zA-Z0-9.\-_]{1,255}$')
VERSION_ID_SUFFIX = re.compile(r'\?versionId=[^\s]+$')


def check_for_200_error(response, **kwargs):
    # From: http://docs.aws.amazon.com/AmazonS3/latest/API/RESTObjectCOPY.html
    # There are two opportunities for a copy request to return an error. One
    # can occur when Amazon S3 receives the copy request and the other can
    # occur while Amazon S3 is copying the files. If the error occurs before
    # the copy operation starts, you receive a standard Amazon S3 error. If the
    # error occurs during the copy operation, the error response is embedded in
    # the 200 OK response. This means that a 200 OK response can contain either
    # a success or an error. Make sure to design your application to parse the
    # contents of the response and handle it appropriately.
    #
    # So this handler checks for this case.  Even though the server sends a
    # 200 response, conceptually this should be handled exactly like a
    # 500 response (with respect to raising exceptions, retries, etc.)
    # We're connected *before* all the other retry logic handlers, so as long
    # as we switch the error code to 500, we'll retry the error as expected.
    if response is None:
        # A None response can happen if an exception is raised while
        # trying to retrieve the response.  See Endpoint._get_response().
        return
    http_response, parsed = response
    if _looks_like_special_case_error(http_response):
        logger.debug("Error found for response with 200 status code, "
                     "errors: %s, changing status code to "
                     "500.", parsed)
        http_response.status_code = 500


def _looks_like_special_case_error(http_response):
    if http_response.status_code == 200:
        parser = xml.etree.cElementTree.XMLParser(
            target=xml.etree.cElementTree.TreeBuilder(),
            encoding='utf-8')
        parser.feed(http_response.content)
        root = parser.close()
        if root.tag == 'Error':
            return True
    return False


def decode_console_output(parsed, **kwargs):
    if 'Output' in parsed:
        try:
            # We're using 'replace' for errors because it is
            # possible that console output contains non string
            # chars we can't utf-8 decode.
            value = base64.b64decode(six.b(parsed['Output'])).decode(
                'utf-8', 'replace')
            parsed['Output'] = value
        except (ValueError, TypeError, AttributeError):
            logger.debug('Error decoding base64', exc_info=True)


def generate_idempotent_uuid(params, model, **kwargs):
    for name in model.idempotent_members:
        if name not in params:
            params[name] = str(uuid.uuid4())
            logger.debug("injecting idempotency token (%s) into param '%s'." %
                         (params[name], name))


def decode_quoted_jsondoc(value):
    try:
        value = json.loads(unquote(value))
    except (ValueError, TypeError):
        logger.debug('Error loading quoted JSON', exc_info=True)
    return value


def json_decode_template_body(parsed, **kwargs):
    if 'TemplateBody' in parsed:
        try:
            value = json.loads(parsed['TemplateBody'])
            parsed['TemplateBody'] = value
        except (ValueError, TypeError):
            logger.debug('error loading JSON', exc_info=True)


def calculate_md5(params, **kwargs):
    request_dict = params
    if request_dict['body'] and 'Content-MD5' not in params['headers']:
        body = request_dict['body']
        if isinstance(body, (bytes, bytearray)):
            binary_md5 = _calculate_md5_from_bytes(body)
        else:
            binary_md5 = _calculate_md5_from_file(body)
        base64_md5 = base64.b64encode(binary_md5).decode('ascii')
        params['headers']['Content-MD5'] = base64_md5


def _calculate_md5_from_bytes(body_bytes):
    md5 = get_md5(body_bytes)
    return md5.digest()


def _calculate_md5_from_file(fileobj):
    start_position = fileobj.tell()
    md5 = get_md5()
    for chunk in iter(lambda: fileobj.read(1024 * 1024), b''):
        md5.update(chunk)
    fileobj.seek(start_position)
    return md5.digest()


def conditionally_calculate_md5(params, context, request_signer, **kwargs):
    """Only add a Content-MD5 if the system supports it."""
    if MD5_AVAILABLE:
        calculate_md5(params, **kwargs)


def validate_bucket_name(params, **kwargs):
    if 'Bucket' not in params:
        return
    bucket = params['Bucket']
    if VALID_BUCKET.search(bucket) is None:
        error_msg = (
            'Invalid bucket name "%s": Bucket name must match '
            'the regex "%s"' % (bucket, VALID_BUCKET.pattern))
        raise ParamValidationError(report=error_msg)


def sse_md5(params, **kwargs):
    """
    S3 server-side encryption requires the encryption key to be sent to the
    server base64 encoded, as well as a base64-encoded MD5 hash of the
    encryption key. This handler does both if the MD5 has not been set by
    the caller.
    """
    _sse_md5(params, 'SSECustomer')


def copy_source_sse_md5(params, **kwargs):
    """
    S3 server-side encryption requires the encryption key to be sent to the
    server base64 encoded, as well as a base64-encoded MD5 hash of the
    encryption key. This handler does both if the MD5 has not been set by
    the caller specifically if the parameter is for the copy-source sse-c key.
    """
    _sse_md5(params, 'CopySourceSSECustomer')


def _sse_md5(params, sse_member_prefix='SSECustomer'):
    if not _needs_s3_sse_customization(params, sse_member_prefix):
        return

    sse_key_member = sse_member_prefix + 'Key'
    sse_md5_member = sse_member_prefix + 'KeyMD5'
    key_as_bytes = params[sse_key_member]
    if isinstance(key_as_bytes, six.text_type):
        key_as_bytes = key_as_bytes.encode('utf-8')
    key_md5_str = base64.b64encode(
        get_md5(key_as_bytes).digest()).decode('utf-8')
    key_b64_encoded = base64.b64encode(key_as_bytes).decode('utf-8')
    params[sse_key_member] = key_b64_encoded
    params[sse_md5_member] = key_md5_str


def _needs_s3_sse_customization(params, sse_member_prefix):
    return (params.get(sse_member_prefix + 'Key') is not None and
            sse_member_prefix + 'KeyMD5' not in params)


def register_retries_for_service(service_data, session,
                                 service_name, **kwargs):
    loader = session.get_component('data_loader')
    endpoint_prefix = service_data.get('metadata', {}).get('endpointPrefix')
    if endpoint_prefix is None:
        logger.debug("Not registering retry handlers, could not endpoint "
                     "prefix from model for service %s", service_name)
        return
    config = _load_retry_config(loader, endpoint_prefix)
    if not config:
        return
    logger.debug("Registering retry handlers for service: %s", service_name)
    handler = retryhandler.create_retry_handler(
        config, endpoint_prefix)
    unique_id = 'retry-config-%s' % endpoint_prefix
    session.register('needs-retry.%s' % endpoint_prefix,
                     handler, unique_id=unique_id)
    _register_for_operations(config, session,
                             service_name=endpoint_prefix)


def _load_retry_config(loader, endpoint_prefix):
    original_config = loader.load_data('_retry')
    retry_config = translate.build_retry_config(
        endpoint_prefix, original_config['retry'],
        original_config.get('definitions', {}))
    return retry_config


def _register_for_operations(config, session, service_name):
    # There's certainly a tradeoff for registering the retry config
    # for the operations when the service is created.  In practice,
    # there aren't a whole lot of per operation retry configs so
    # this is ok for now.
    for key in config:
        if key == '__default__':
            continue
        handler = retryhandler.create_retry_handler(config, key)
        unique_id = 'retry-config-%s-%s' % (service_name, key)
        session.register('needs-retry.%s.%s' % (service_name, key),
                         handler, unique_id=unique_id)


def disable_signing(**kwargs):
    """
    This handler disables request signing by setting the signer
    name to a special sentinel value.
    """
    return botocore.UNSIGNED


def add_expect_header(model, params, **kwargs):
    if model.http.get('method', '') not in ['PUT', 'POST']:
        return
    if 'body' in params:
        body = params['body']
        if hasattr(body, 'read'):
            # Any file like object will use an expect 100-continue
            # header regardless of size.
            logger.debug("Adding expect 100 continue header to request.")
            params['headers']['Expect'] = '100-continue'


def document_copy_source_form(section, event_name, **kwargs):
    if 'request-example' in event_name:
        parent = section.get_section('structure-value')
        param_line = parent.get_section('CopySource')
        value_portion = param_line.get_section('member-value')
        value_portion.clear_text()
        value_portion.write("'string' or {'Bucket': 'string', "
                            "'Key': 'string', 'VersionId': 'string'}")
    elif 'request-params' in event_name:
        param_section = section.get_section('CopySource')
        type_section = param_section.get_section('param-type')
        type_section.clear_text()
        type_section.write(':type CopySource: str or dict')
        doc_section = param_section.get_section('param-documentation')
        doc_section.clear_text()
        doc_section.write(
            "The name of the source bucket, key name of the source object, "
            "and optional version ID of the source object.  You can either "
            "provide this value as a string or a dictionary.  The "
            "string form is {bucket}/{key} or "
            "{bucket}/{key}?versionId={versionId} if you want to copy a "
            "specific version.  You can also provide this value as a "
            "dictionary.  The dictionary format is recommended over "
            "the string format because it is more explicit.  The dictionary "
            "format is: {'Bucket': 'bucket', 'Key': 'key', 'VersionId': 'id'}."
            "  Note that the VersionId key is optional and may be omitted."
        )


def handle_copy_source_param(params, **kwargs):
    """Convert CopySource param for CopyObject/UploadPartCopy.

    This handler will deal with two cases:

        * CopySource provided as a string.  We'll make a best effort
          to URL encode the key name as required.  This will require
          parsing the bucket and version id from the CopySource value
          and only encoding the key.
        * CopySource provided as a dict.  In this case we're
          explicitly given the Bucket, Key, and VersionId so we're
          able to encode the key and ensure this value is serialized
          and correctly sent to S3.

    """
    source = params.get('CopySource')
    if source is None:
        # The call will eventually fail but we'll let the
        # param validator take care of this.  It will
        # give a better error message.
        return
    if isinstance(source, six.string_types):
        params['CopySource'] = _quote_source_header(source)
    elif isinstance(source, dict):
        params['CopySource'] = _quote_source_header_from_dict(source)


def _quote_source_header_from_dict(source_dict):
    try:
        bucket = source_dict['Bucket']
        key = percent_encode(source_dict['Key'], safe=SAFE_CHARS + '/')
        version_id = source_dict.get('VersionId')