# Copyright 2015 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.
import datetime
import re

import pytest
from dateutil.tz import tzutc

import botocore.session
from botocore import UNSIGNED
from botocore.compat import parse_qs, urlsplit
from botocore.config import Config
from botocore.exceptions import (
    ClientError,
    InvalidS3UsEast1RegionalEndpointConfigError,
    ParamValidationError,
    UnsupportedS3AccesspointConfigurationError,
    UnsupportedS3ConfigurationError,
)
from tests import (
    BaseSessionTest,
    ClientHTTPStubber,
    FreezeTime,
    create_session,
    get_checksum_cls,
    mock,
    requires_crt,
    temporary_file,
    unittest,
)

DATE = datetime.datetime(2021, 8, 27, 0, 0, 0)


class TestS3BucketValidation(unittest.TestCase):
    def test_invalid_bucket_name_raises_error(self):
        session = botocore.session.get_session()
        s3 = session.create_client("s3")
        with self.assertRaises(ParamValidationError):
            s3.put_object(
                Bucket="adfgasdfadfs/bucket/name", Key="foo", Body=b"asdf"
            )


class BaseS3OperationTest(BaseSessionTest):
    def setUp(self):
        super().setUp()
        self.region = "us-west-2"
        self.client = self.session.create_client("s3", self.region)
        self.http_stubber = ClientHTTPStubber(self.client)


class BaseS3ClientConfigurationTest(BaseSessionTest):
    _V4_AUTH_REGEX = re.compile(
        r"AWS4-HMAC-SHA256 "
        r"Credential=\w+/\d+/"
        r"(?P<signing_region>[a-z0-9-]+)/"
        r"(?P<signing_name>[a-z0-9-]+)/"
    )

    _V4A_AUTH_REGEX = re.compile(
        r"AWS4-ECDSA-P256-SHA256 "
        r"Credential=\w+/\d+/"
        r"(?P<signing_name>[a-z0-9-]+)/"
    )

    def setUp(self):
        super().setUp()
        self.region = "us-west-2"

    def _get_auth_regex(self, auth_header):
        if auth_header.startswith("AWS4-ECDSA"):
            return self._V4A_AUTH_REGEX
        return self._V4_AUTH_REGEX

    def assert_signing_region(self, request, expected_region):
        auth_header = request.headers["Authorization"].decode("utf-8")
        actual_region = None
        auth_regex = self._get_auth_regex(auth_header)
        match = auth_regex.match(auth_header)
        if match and auth_regex is self._V4_AUTH_REGEX:
            actual_region = match.group("signing_region")
            self.assertEqual(expected_region, actual_region)
        else:
            # SigV4a does not sign with a specific region
            region_set = request.headers.get('X-Amz-Region-Set')
            self.assertEqual(region_set, b'*')

    def assert_signing_name(self, request, expected_name):
        auth_header = request.headers["Authorization"].decode("utf-8")
        actual_name = None
        auth_regex = self._get_auth_regex(auth_header)
        match = auth_regex.match(auth_header)
        if match:
            actual_name = match.group("signing_name")
        self.assertEqual(expected_name, actual_name)

    def assert_signing_region_in_url(self, url, expected_region):
        qs_components = parse_qs(urlsplit(url).query)
        self.assertIn(expected_region, qs_components["X-Amz-Credential"][0])

    def assert_endpoint(self, request, expected_endpoint):
        actual_endpoint = urlsplit(request.url).netloc
        self.assertEqual(actual_endpoint, expected_endpoint)

    def create_s3_client(self, **kwargs):
        client_kwargs = {"region_name": self.region}
        client_kwargs.update(kwargs)
        return self.session.create_client("s3", **client_kwargs)

    def set_config_file(self, fileobj, contents):
        fileobj.write(contents)
        fileobj.flush()
        self.environ["AWS_CONFIG_FILE"] = fileobj.name


class TestS3ClientConfigResolution(BaseS3ClientConfigurationTest):
    def test_no_s3_config(self):
        client = self.create_s3_client()
        self.assertIsNone(client.meta.config.s3)

    def test_client_s3_dualstack_handles_uppercase_true(self):
        with temporary_file("w") as f:
            self.set_config_file(
                f, "[default]\n" "s3 = \n" "    use_dualstack_endpoint = True"
            )
            client = self.create_s3_client()
            self.assertEqual(
                client.meta.config.s3["use_dualstack_endpoint"], True
            )

    def test_client_s3_dualstack_handles_lowercase_true(self):
        with temporary_file("w") as f:
            self.set_config_file(
                f, "[default]\n" "s3 = \n" "    use_dualstack_endpoint = true"
            )
            client = self.create_s3_client()
            self.assertEqual(
                client.meta.config.s3["use_dualstack_endpoint"], True
            )

    def test_client_s3_accelerate_handles_uppercase_true(self):
        with temporary_file("w") as f:
            self.set_config_file(
                f, "[default]\n" "s3 = \n" "    use_accelerate_endpoint = True"
            )
            client = self.create_s3_client()
            self.assertEqual(
                client.meta.config.s3["use_accelerate_endpoint"], True
            )

    def test_client_s3_accelerate_handles_lowercase_true(self):
        with temporary_file("w") as f:
            self.set_config_file(
                f, "[default]\n" "s3 = \n" "    use_accelerate_endpoint = true"
            )
            client = self.create_s3_client()
            self.assertEqual(
                client.meta.config.s3["use_accelerate_endpoint"], True
            )

    def test_client_payload_signing_enabled_handles_uppercase_true(self):
        with temporary_file("w") as f:
            self.set_config_file(
                f, "[default]\n" "s3 = \n" "    payload_signing_enabled = True"
            )
            client = self.create_s3_client()
            self.assertEqual(
                client.meta.config.s3["payload_signing_enabled"], True
            )

    def test_client_payload_signing_enabled_handles_lowercase_true(self):
        with temporary_file("w") as f:
            self.set_config_file(
                f, "[default]\n" "s3 = \n" "    payload_signing_enabled = true"
            )
            client = self.create_s3_client()
            self.assertEqual(
                client.meta.config.s3["payload_signing_enabled"], True
            )

    def test_includes_unmodeled_s3_config_vars(self):
        with temporary_file("w") as f:
            self.set_config_file(
                f, "[default]\n" "s3 = \n" "    unmodeled = unmodeled_val"
            )
            client = self.create_s3_client()
            self.assertEqual(
                client.meta.config.s3["unmodeled"], "unmodeled_val"
            )

    def test_mixed_modeled_and_unmodeled_config_vars(self):
        with temporary_file("w") as f:
            self.set_config_file(
                f,
                "[default]\n"
                "s3 = \n"
                "    payload_signing_enabled = true\n"
                "    unmodeled = unmodeled_val",
            )
            client = self.create_s3_client()
            self.assertEqual(
                client.meta.config.s3,
                {
                    "payload_signing_enabled": True,
                    "unmodeled": "unmodeled_val",
                },
            )

    def test_use_arn_region(self):
        self.environ["AWS_S3_USE_ARN_REGION"] = "true"
        client = self.create_s3_client()
        self.assertEqual(
            client.meta.config.s3,
            {
                "use_arn_region": True,
            },
        )

    def test_use_arn_region_config_var(self):
        with temporary_file("w") as f:
            self.set_config_file(f, "[default]\n" "s3_use_arn_region = true")
            client = self.create_s3_client()
            self.assertEqual(
                client.meta.config.s3,
                {
                    "use_arn_region": True,
                },
            )

    def test_use_arn_region_nested_config_var(self):
        with temporary_file("w") as f:
            self.set_config_file(
                f, "[default]\n" "s3 = \n" "    use_arn_region = true"
            )
            client = self.create_s3_client()
            self.assertEqual(
                client.meta.config.s3,
                {
                    "use_arn_region": True,
                },
            )

    def test_use_arn_region_is_case_insensitive(self):
        self.environ["AWS_S3_USE_ARN_REGION"] = "True"
        client = self.create_s3_client()
        self.assertEqual(
            client.meta.config.s3,
            {
                "use_arn_region": True,
            },
        )

    def test_use_arn_region_env_var_overrides_config_var(self):
        self.environ["AWS_S3_USE_ARN_REGION"] = "false"
        with temporary_file("w") as f:
            self.set_config_file(
                f, "[default]\n" "s3 = \n" "    use_arn_region = true"
            )
            client = self.create_s3_client()
        self.assertEqual(
            client.meta.config.s3,
            {
                "use_arn_region": False,
            },
        )

    def test_client_config_use_arn_region_overrides_env_var(self):
        self.environ["AWS_S3_USE_ARN_REGION"] = "true"
        client = self.create_s3_client(
            config=Config(s3={"use_arn_region": False})
        )
        self.assertEqual(
            client.meta.config.s3,
            {
                "use_arn_region": False,
            },
        )

    def test_client_config_use_arn_region_overrides_config_var(self):
        with temporary_file("w") as f:
            self.set_config_file(
                f, "[default]\n" "s3 = \n" "    use_arn_region = true"
            )
            client = self.create_s3_client(
                config=Config(s3={"use_arn_region": False})
            )
        self.assertEqual(
            client.meta.config.s3,
            {
                "use_arn_region": False,
            },
        )

    def test_us_east_1_regional_env_var(self):
        self.environ["AWS_S3_US_EAST_1_REGIONAL_ENDPOINT"] = "regional"
        client = self.create_s3_client()
        self.assertEqual(
            client.meta.config.s3,
            {
                "us_east_1_regional_endpoint": "regional",
            },
        )

    def test_us_east_1_regional_config_var(self):
        with temporary_file("w") as f:
            self.set_config_file(
                f, "[default]\n" "s3_us_east_1_regional_endpoint = regional"
            )
            client = self.create_s3_client()
            self.assertEqual(
                client.meta.config.s3,
                {
                    "us_east_1_regional_endpoint": "regional",
                },
            )

    def test_us_east_1_regional_nested_config_var(self):
        with temporary_file("w") as f:
            self.set_config_file(
                f,
                "[default]\n"
                "s3 = \n"
                "    us_east_1_regional_endpoint = regional",
            )
            client = self.create_s3_client()
            self.assertEqual(
                client.meta.config.s3,
                {
                    "us_east_1_regional_endpoint": "regional",
                },
            )

    def test_us_east_1_regional_env_var_overrides_config_var(self):
        self.environ["AWS_S3_US_EAST_1_REGIONAL_ENDPOINT"] = "regional"
        with temporary_file("w") as f:
            self.set_config_file(
                f,
                "[default]\n"
                "s3 = \n"
                "    us_east_1_regional_endpoint = legacy",
            )
            client = self.create_s3_client()
        self.assertEqual(
            client.meta.config.s3,
            {
                "us_east_1_regional_endpoint": "regional",
            },
        )

    def test_client_config_us_east_1_regional_overrides_env_var(self):
        self.environ["AWS_S3_US_EAST_1_REGIONAL_ENDPOINT"] = "regional"
        client = self.create_s3_client(
            config=Config(s3={"us_east_1_regional_endpoint": "legacy"})
        )
        self.assertEqual(
            client.meta.config.s3,
            {
                "us_east_1_regional_endpoint": "legacy",
            },
        )

    def test_client_config_us_east_1_regional_overrides_config_var(self):
        with temporary_file("w") as f:
            self.set_config_file(
                f,
                "[default]\n"
                "s3 = \n"
                "    us_east_1_regional_endpoint = legacy",
            )
            client = self.create_s3_client(
                config=Config(s3={"us_east_1_regional_endpoint": "regional"})
            )
        self.assertEqual(
            client.meta.config.s3,
            {
                "us_east_1_regional_endpoint": "regional",
            },
        )

    def test_client_validates_us_east_1_regional(self):
        with self.assertRaises(InvalidS3UsEast1RegionalEndpointConfigError):
            self.create_s3_client(
                config=Config(s3={"us_east_1_regional_endpoint": "not-valid"})
            )

    def test_client_region_defaults_to_us_east_1(self):
        client = self.create_s3_client(region_name=None)
        self.assertEqual(client.meta.region_name, "us-east-1")

    def test_client_region_remains_us_east_1(self):
        client = self.create_s3_client(region_name="us-east-1")
        self.assertEqual(client.meta.region_name, "us-east-1")

    def test_client_region_remains_aws_global(self):
        client = self.create_s3_client(region_name="aws-global")
        self.assertEqual(client.meta.region_name, "aws-global")

    def test_client_region_defaults_to_aws_global_for_regional(self):
        self.environ["AWS_S3_US_EAST_1_REGIONAL_ENDPOINT"] = "regional"
        client = self.create_s3_client(region_name=None)
        self.assertEqual(client.meta.region_name, "aws-global")

    def test_client_region_remains_us_east_1_for_regional(self):
        self.environ["AWS_S3_US_EAST_1_REGIONAL_ENDPOINT"] = "regional"
        client = self.create_s3_client(region_name="us-east-1")
        self.assertEqual(client.meta.region_name, "us-east-1")

    def test_client_region_remains_aws_global_for_regional(self):
        self.environ["AWS_S3_US_EAST_1_REGIONAL_ENDPOINT"] = "regional"
        client = self.create_s3_client(region_name="aws-global")
        self.assertEqual(client.meta.region_name, "aws-global")


class TestS3Copy(BaseS3OperationTest):
    def create_s3_client(self, **kwargs):
        client_kwargs = {"region_name": self.region}
        client_kwargs.update(kwargs)
        return self.session.create_client("s3", **client_kwargs)

    def create_stubbed_s3_client(self, **kwargs):
        client = self.create_s3_client(**kwargs)
        http_stubber = ClientHTTPStubber(client)
        http_stubber.start()
        return client, http_stubber

    def test_s3_copy_object_with_incomplete_response(self):
        self.client, self.http_stubber = self.create_stubbed_s3_client(
            region_name="us-east-1"
        )

        incomplete_body = b'<?xml version="1.0" encoding="UTF-8"?>\n\n\n'
        complete_body = (
            b'<?xml version="1.0" encoding="UTF-8"?>\n\n'
            b"<CopyObjectResult "
            b'xmlns="http://s3.amazonaws.com/doc/2006-03-01/">'
            b"<LastModified>2020-04-21T21:03:31.000Z</LastModified>"
            b"<ETag>&quot;s0mEcH3cK5uM&quot;</ETag></CopyObjectResult>"
        )
        self.http_stubber.add_response(status=200, body=incomplete_body)
        self.http_stubber.add_response(status=200, body=complete_body)
        response = self.client.copy_object(
            Bucket="bucket",
            CopySource="other-bucket/test.txt",
            Key="test.txt",
        )

        # Validate we retried and got second body
        self.assertEqual(len(self.http_stubber.requests), 2)
        self.assertEqual(response["ResponseMetadata"]["HTTPStatusCode"], 200)
        self.assertTrue("CopyObjectResult" in response)


class TestS3200ErrorResponse(BaseS3OperationTest):
    def create_s3_client(self, **kwargs):
        client_kwargs = {"region_name": self.region}
        client_kwargs.update(kwargs)
        return self.session.create_client("s3", **client_kwargs)

    def create_stubbed_s3_client(self, **kwargs):
        client = self.create_s3_client(**kwargs)
        http_stubber = ClientHTTPStubber(client)
        http_stubber.start()
        return client, http_stubber

    def test_s3_200_with_error_response(self):
        self.client, self.http_stubber = self.create_stubbed_s3_client(
            region_name="us-east-1"
        )
        error_body = (
            b"<Error>"
            b"<Code>SlowDown</Code>"
            b"<Message>Please reduce your request rate.</Message>"
            b"</Error>"
        )
        # Populate 5 attempts for SlowDown to validate
        # we reached four max retries and raised an exception.
        for i in range(5):
            self.http_stubber.add_response(status=200, body=error_body)
        with self.assertRaises(botocore.exceptions.ClientError) as context:
            self.client.copy_object(
                Bucket="bucket",
                CopySource="other-bucket/test.txt",
                Key="test.txt",
            )
        self.assertEqual(len(self.http_stubber.requests), 5)
        self.assertEqual(
            context.exception.response["ResponseMetadata"]["HTTPStatusCode"],
            500,
        )
        self.assertEqual(
            context.exception.response["Error"]["Code"], "SlowDown"
        )

    def test_s3_200_with_no_error_response(self):
        self.client, self.http_stubber = self.create_stubbed_s3_client(
            region_name="us-east-1"
        )
        self.http_stubber.add_response(status=200, body=b"<NotAnError/>")

        response = self.client.copy_object(
            Bucket="bucket",
            CopySource="other-bucket/test.txt",
            Key="test.txt",
        )

        # Validate that the status code remains 200.
        self.assertEqual(len(self.http_stubber.requests), 1)
        self.assertEqual(response["ResponseMetadata"]["HTTPStatusCode"], 200)

    def test_s3_200_with_error_response_on_streaming_operation(self):
        self.client, self.http_stubber = self.create_stubbed_s3_client(
            region_name="us-east-1"
        )
        self.http_stubber.add_response(status=200, body=b"<Error/>")
        response = self.client.get_object(Bucket="bucket", Key="test.txt")

        # Validate that the status code remains 200 because we don't
        # process 200-with-error responses on streaming operations.
        self.assertEqual(len(self.http_stubber.requests), 1)
        self.assertEqual(response["ResponseMetadata"]["HTTPStatusCode"], 200)

    def test_s3_200_response_with_no_body(self):
        self.client, self.http_stubber = self.create_stubbed_s3_client(
            region_name="us-east-1"
        )
        self.http_stubber.add_response(status=200)
        response = self.client.head_object(Bucket="bucket", Key="test.txt")

        # Validate that the status code remains 200 on operations without a body.
        self.assertEqual(len(self.http_stubber.requests), 1)
        self.assertEqual(response["ResponseMetadata"]["HTTPStatusCode"], 200)


class TestAccesspointArn(BaseS3ClientConfigurationTest):
    def setUp(self):
        super().setUp()
        self.client, self.http_stubber = self.create_stubbed_s3_client()

    def create_stubbed_s3_client(self, **kwargs):
        client = self.create_s3_client(**kwargs)
        http_stubber = ClientHTTPStubber(client)
        http_stubber.start()
        return client, http_stubber

    def assert_expected_copy_source_header(
        self, http_stubber, expected_copy_source
    ):
        request = self.http_stubber.requests[0]
        self.assertIn("x-amz-copy-source", request.headers)
        self.assertEqual(
            request.headers["x-amz-copy-source"], expected_copy_source
        )

    def add_copy_object_response(self, http_stubber):
        http_stubber.add_response(
            body=b"<CopyObjectResult></CopyObjectResult>"
        )

    def assert_endpoint(self, request, expected_endpoint):
        actual_endpoint = urlsplit(request.url).netloc
        self.assertEqual(actual_endpoint, expected_endpoint)

    def assert_header_matches(self, request, header_key, expected_value):
        self.assertEqual(request.headers.get(header_key), expected_value)

    def test_missing_account_id_in_arn(self):
        accesspoint_arn = "arn:aws:s3:us-west-2::accesspoint:myendpoint"
        with self.assertRaises(botocore.exceptions.ParamValidationError):
            self.client.list_objects(Bucket=accesspoint_arn)

    def test_missing_accesspoint_name_in_arn(self):
        accesspoint_arn = "arn:aws:s3:us-west-2:123456789012:accesspoint"
        with self.assertRaises(botocore.exceptions.ParamValidationError):
            self.client.list_objects(Bucket=accesspoint_arn)

    def test_accesspoint_includes_asterisk(self):
        accesspoint_arn = "arn:aws:s3:us-west-2:123456789012:accesspoint:*"
        with self.assertRaises(botocore.exceptions.ParamValidationError):
            self.client.list_objects(Bucket=accesspoint_arn)

    def test_accesspoint_arn_contains_subresources(self):
        accesspoint_arn = (
            "arn:aws:s3:us-west-2:123456789012:accesspoint:myendpoint:object"
        )
        with self.assertRaises(botocore.exceptions.ParamValidationError):
            self.client.list_objects(Bucket=accesspoint_arn)

    def test_accesspoint_arn_with_custom_endpoint(self):
        accesspoint_arn = (
            "arn:aws:s3:us-west-2:123456789012:accesspoint:myendpoint"
        )
        self.client, http_stubber = self.create_stubbed_s3_client(
            endpoint_url="https://custom.com"
        )
        http_stubber.add_response()
        self.client.list_objects(Bucket=accesspoint_arn)
        expected_endpoint = "myendpoint-123456789012.custom.com"
        self.assert_endpoint(http_stubber.requests[0], expected_endpoint)

    def test_accesspoint_arn_with_custom_endpoint_and_dualstack(self):
        accesspoint_arn = (
            "arn:aws:s3:us-west-2:123456789012:accesspoint:myendpoint"
        )
        self.client, http_stubber = self.create_stubbed_s3_client(
            endpoint_url="https://custom.com",
            config=Config(s3={"use_dualstack_endpoint": True}),
        )
        http_stubber.add_response()
        self.client.list_objects(Bucket=accesspoint_arn)
        expected_endpoint = "myendpoint-123456789012.custom.com"
        self.assert_endpoint(http_stubber.requests[0], expected_endpoint)

    def test_accesspoint_arn_with_s3_accelerate(self):
        accesspoint_arn = (
            "arn:aws:s3:us-west-2:123456789012:accesspoint:myendpoint"
        )
        self.client, _ = self.create_stubbed_s3_client(
            config=Config(s3={"use_accelerate_endpoint": True})
        )
        with self.assertRaises(
            botocore.exceptions.UnsupportedS3AccesspointConfigurationError
        ):
            self.client.list_objects(Bucket=accesspoint_arn)

    def test_accesspoint_arn_cross_partition(self):
        accesspoint_arn = (
            "arn:aws:s3:us-west-2:123456789012:accesspoint:myendpoint"
        )
        self.client, _ = self.create_stubbed_s3_client(
            region_name="cn-north-1"
        )
        with self.assertRaises(
            botocore.exceptions.UnsupportedS3AccesspointConfigurationError
        ):
            self.client.list_objects(Bucket=accesspoint_arn)

    def test_accesspoint_arn_cross_partition_use_client_region(self):
        accesspoint_arn = (
            "arn:aws:s3:us-west-2:123456789012:accesspoint:myendpoint"
        )
        self.client, _ = self.create_stubbed_s3_client(
            region_name="cn-north-1",
            config=Config(s3={"use_accelerate_endpoint": True}),
        )
        with self.assertRaises(
            botocore.exceptions.UnsupportedS3AccesspointConfigurationError
        ):
            self.client.list_objects(Bucket=accesspoint_arn)

    def test_signs_with_arn_region(self):
        accesspoint_arn = (
            "arn:aws:s3:us-west-2:123456789012:accesspoint:myendpoint"
        )
        self.client, self.http_stubber = self.create_stubbed_s3_client(
            region_name="us-east-1"
        )
        self.http_stubber.add_response()
        self.client.list_objects(Bucket=accesspoint_arn)
        self.assert_signing_region(self.http_stubber.requests[0], "us-west-2")

    def test_signs_with_client_region_when_use_arn_region_false(self):
        accesspoint_arn = (
            "arn:aws:s3:us-west-2:123456789012:accesspoint:myendpoint"
        )
        self.client, self.http_stubber = self.create_stubbed_s3_client(
            region_name="us-east-1",
            config=Config(s3={"use_arn_region": False}),
        )
        self.http_stubber.add_response()
        with self.assertRaises(UnsupportedS3AccesspointConfigurationError):
            self.client.list_objects(Bucket=accesspoint_arn)

    def test_presign_signs_with_arn_region(self):
        accesspoint_arn = (
            "arn:aws:s3:us-west-2:123456789012:accesspoint:myendpoint"
        )
        self.client, _ = self.create_stubbed_s3_client(
            region_name="us-east-1", config=Config(signature_version="s3v4")
        )
        url = self.client.generate_presigned_url(
            "get_object", {"Bucket": accesspoint_arn, "Key": "mykey"}
        )
        self.assert_signing_region_in_url(url, "us-west-2")

    def test_presign_signs_with_client_region_when_use_arn_region_false(self):
        accesspoint_arn = (
            "arn:aws:s3:us-west-2:123456789012:accesspoint:myendpoint"
        )
        self.client, _ = self.create_stubbed_s3_client(
            region_name="us-east-1",
            config=Config(
                signature_version="s3v4", s3={"use_arn_region": False}
            ),
        )
        with self.assertRaises(UnsupportedS3AccesspointConfigurationError):
            self.client.generate_presigned_url(
                "get_object", {"Bucket": accesspoint_arn, "Key": "mykey"}
            )

    def test_copy_source_str_with_accesspoint_arn(self):
        copy_source = (
            "arn:aws:s3:us-west-2:123456789012:accesspoint:myendpoint/"
            "object/myprefix/myobject"
        )
        self.client, self.http_stubber = self.create_stubbed_s3_client()
        self.add_copy_object_response(self.http_stubber)
        self.client.copy_object(
            Bucket="mybucket", Key="mykey", CopySource=copy_source
        )
        self.assert_expected_copy_source_header(
            self.http_stubber,
            expected_copy_source=(
                b"arn%3Aaws%3As3%3Aus-west-2%3A123456789012%3Aaccesspoint%3A"
                b"myendpoint/object/myprefix/myobject"
            ),
        )

    def test_copy_source_str_with_accesspoint_arn_and_version_id(self):
        copy_source = (
            "arn:aws:s3:us-west-2:123456789012:accesspoint:myendpoint/"
            "object/myprefix/myobject?versionId=myversionid"
        )
        self.client, self.http_stubber = self.create_stubbed_s3_client()
        self.add_copy_object_response(self.http_stubber)
        self.client.copy_object(
            Bucket="mybucket", Key="mykey", CopySource=copy_source
        )
        self.assert_expected_copy_source_header(
            self.http_stubber,
            expected_copy_source=(
                b"arn%3Aaws%3As3%3Aus-west-2%3A123456789012%3Aaccesspoint%3A"
                b"myendpoint/object/myprefix/myobject?versionId=myversionid"
            ),
        )

    def test_copy_source_dict_with_accesspoint_arn(self):
        copy_source = {
            "Bucket": "arn:aws:s3:us-west-2:123456789012:accesspoint:myendpoint",
            "Key": "myprefix/myobject",
        }
        self.client, self.http_stubber = self.create_stubbed_s3_client()
        self.add_copy_object_response(self.http_stubber)
        self.client.copy_object(
            Bucket="mybucket", Key="mykey", CopySource=copy_source
        )
        self.assert_expected_copy_source_header(
            self.http_stubber,
            expected_copy_source=(
                b"arn%3Aaws%3As3%3Aus-west-2%3A123456789012%3Aaccesspoint%3A"
                b"myendpoint/object/myprefix/myobject"
            ),
        )

    def test_copy_source_dict_with_accesspoint_arn_and_version_id(self):
        copy_source = {
            "Bucket": "arn:aws:s3:us-west-2:123456789012:accesspoint:myendpoint",
            "Key": "myprefix/myobject",
            "VersionId": "myversionid",
        }
        self.client, self.http_stubber = self.create_stubbed_s3_client()
        self.add_copy_object_response(self.http_stubber)
        self.client.copy_object(
            Bucket="mybucket", Key="mykey", CopySource=copy_source
        )
        self.assert_expected_copy_source_header(
            self.http_stubber,
            expected_copy_source=(
                b"arn%3Aaws%3As3%3Aus-west-2%3A123456789012%3Aaccesspoint%3A"
                b"myendpoint/object/myprefix/myobject?versionId=myversionid"
            ),
        )

    def test_basic_outpost_arn(self):
        outpost_arn = (
            "arn:aws:s3-outposts:us-west-2:123456789012:outpost:"
            "op-01234567890123456:accesspoint:myaccesspoint"
        )
        self.client, self.http_stubber = self.create_stubbed_s3_client(
            region_name="us-east-1"
        )
        self.http_stubber.add_response()
        self.client.list_objects(Bucket=outpost_arn)
        request = self.http_stubber.requests[0]
        self.assert_signing_name(request, "s3-outposts")
        self.assert_signing_region(request, "us-west-2")
        expected_endpoint = (
            "myaccesspoint-123456789012.op-01234567890123456."
            "s3-outposts.us-west-2.amazonaws.com"
        )
        self.assert_endpoint(request, expected_endpoint)
        sha_header = request.headers.get("x-amz-content-sha256")
        self.assertIsNotNone(sha_header)
        self.assertNotEqual(sha_header, b"UNSIGNED-PAYLOAD")

    def test_basic_outpost_arn_custom_endpoint(self):
        outpost_arn = (
            "arn:aws:s3-outposts:us-west-2:123456789012:outpost:"
            "op-01234567890123456:accesspoint:myaccesspoint"
        )
        self.client, self.http_stubber = self.create_stubbed_s3_client(
            endpoint_url="https://custom.com", region_name="us-east-1"
        )
        self.http_stubber.add_response()
        self.client.list_objects(Bucket=outpost_arn)
        request = self.http_stubber.requests[0]
        self.assert_signing_name(request, "s3-outposts")
        self.assert_signing_region(request, "us-west-2")
        expected_endpoint = (
            "myaccesspoint-123456789012.op-01234567890123456.custom.com"
        )
        self.assert_endpoint(request, expected_endpoint)

    def test_outpost_arn_presigned_url(self):
        outpost_arn = (
            'arn:aws:s3-outposts:us-west-2:123456789012:outpost/'
            'op-01234567890123456/accesspoint/myaccesspoint'
        )
        expected_url = (
            'myaccesspoint-123456789012.op-01234567890123456.'
            's3-outposts.us-west-2.amazonaws.com'
        )
        expected_credentials = (
            '20210827%2Fus-west-2%2Fs3-outposts%2Faws4_request'
        )
        expected_signature = (
            'a944fbe2bfbae429f922746546d1c6f890649c88ba7826bd1d258ac13f327e09'
        )
        config = Config(signature_version='s3v4')
        presigned_url = self._get_presigned_url(
            outpost_arn, 'us-west-2', config=config
        )
        self._assert_presigned_url(
            presigned_url,
            expected_url,
            expected_signature,
            expected_credentials,
        )

    def test_outpost_arn_presigned_url_with_use_arn_region(self):
        outpost_arn = (
            'arn:aws:s3-outposts:us-west-2:123456789012:outpost/'
            'op-01234567890123456/accesspoint/myaccesspoint'
        )
        expected_url = (
            'myaccesspoint-123456789012.op-01234567890123456.'
            's3-outposts.us-west-2.amazonaws.com'
        )
        expected_credentials = (
            '20210827%2Fus-west-2%2Fs3-outposts%2Faws4_request'
        )
        expected_signature = (
            'a944fbe2bfbae429f922746546d1c6f890649c88ba7826bd1d258ac13f327e09'
        )
        config = Config(
            signature_version='s3v4',
            s3={
                'use_arn_region': True,
            },
        )
        presigned_url = self._get_presigned_url(
            outpost_arn, 'us-west-2', config=config
        )
        self._assert_presigned_url(
            presigned_url,
            expected_url,
            expected_signature,
            expected_credentials,
        )

    def test_outpost_arn_presigned_url_cross_region_arn(self):
        outpost_arn = (
            'arn:aws:s3-outposts:us-east-1:123456789012:outpost/'
            'op-01234567890123456/accesspoint/myaccesspoint'
        )
        expected_url = (
            'myaccesspoint-123456789012.op-01234567890123456.'
            's3-outposts.us-east-1.amazonaws.com'
        )
        expected_credentials = (
            '20210827%2Fus-east-1%2Fs3-outposts%2Faws4_request'
        )
        expected_signature = (
            '7f93df0b81f80e590d95442d579bd6cf749a35ff4bbdc6373fa669b89c7fce4e'
        )
        config = Config(
            signature_version='s3v4',
            s3={
                'use_arn_region': True,
            },
        )
        presigned_url = self._get_presigned_url(
            outpost_arn, 'us-west-2', config=config
        )
        self._assert_presigned_url(
            presigned_url,
            expected_url,
            expected_signature,
            expected_credentials,
        )

    def test_outpost_arn_with_s3_accelerate(self):
        outpost_arn = (
            "arn:aws:s3-outposts:us-west-2:123456789012:outpost:"
            "op-01234567890123456:accesspoint:myaccesspoint"
        )
        self.client, _ = self.create_stubbed_s3_client(
            config=Config(s3={"use_accelerate_endpoint": True})
        )
        with self.assertRaises(UnsupportedS3AccesspointConfigurationError):
            self.client.list_objects(Bucket=outpost_arn)

    def test_outpost_arn_with_s3_dualstack(self):
        outpost_arn = (
            "arn:aws:s3-outposts:us-west-2:123456789012:outpost:"
            "op-01234567890123456:accesspoint:myaccesspoint"
        )
        self.client, _ = self.create_stubbed_s3_client(
            config=Config(s3={"use_dualstack_endpoint": True})
        )
        with self.assertRaises(UnsupportedS3AccesspointConfigurationError):
            self.client.list_objects(Bucket=outpost_arn)

    def test_incorrect_outpost_format(self):
        outpost_arn = "arn:aws:s3-outposts:us-west-2:123456789012:outpost"
        with self.assertRaises(botocore.exceptions.ParamValidationError):
            self.client.list_objects(Bucket=outpost_arn)

    def test_incorrect_outpost_no_accesspoint(self):
        outpost_arn = (
            "arn:aws:s3-outposts:us-west-2:123456789012:outpost:"
            "op-01234567890123456"
        )
        with self.assertRaises(botocore.exceptions.ParamValidationError):
            self.client.list_objects(Bucket=outpost_arn)

    def test_incorrect_outpost_resource_format(self):
        outpost_arn = (
            "arn:aws:s3-outposts:us-west-2:123456789012:outpost:myaccesspoint"
        )
        with self.assertRaises(botocore.exceptions.ParamValidationError):
            self.client.list_objects(Bucket=outpost_arn)

    def test_incorrect_outpost_sub_resources(self):
        outpost_arn = (
            "arn:aws:s3-outposts:us-west-2:123456789012:outpost:"
            "op-01234567890123456:accesspoint:mybucket:object:foo"
        )
        with self.assertRaises(botocore.exceptions.ParamValidationError):
            self.client.list_objects(Bucket=outpost_arn)

    def test_incorrect_outpost_invalid_character(self):
        outpost_arn = (
            "arn:aws:s3-outposts:us-west-2:123456789012:outpost:"
            "op-0123456.890123456:accesspoint:myaccesspoint"
        )
        with self.assertRaises(botocore.exceptions.ParamValidationError):
            self.client.list_objects(Bucket=outpost_arn)

    def test_s3_object_lambda_arn_with_s3_dualstack(self):
        s3_object_lambda_arn = (
            "arn:aws:s3-object-lambda:us-west-2:123456789012:"
            "accesspoint/myBanner"
        )
        self.client, _ = self.create_stubbed_s3_client(
            config=Config(s3={"use_dualstack_endpoint": True})
        )
        with self.assertRaises(UnsupportedS3AccesspointConfigurationError):
            self.client.list_objects(Bucket=s3_object_lambda_arn)

    def test_s3_object_lambda_fips_raise_for_cross_region(self):
        s3_object_lambda_arn = (
            "arn:aws-us-gov:s3-object-lambda:us-gov-east-1:123456789012:"
            "accesspoint/mybanner"
        )
        self.client, _ = self.create_stubbed_s3_client(
            region_name="fips-us-gov-west-1",
            config=Config(s3={"use_arn_region": False}),
        )
        expected_exception = UnsupportedS3AccesspointConfigurationError
        with self.assertRaises(expected_exception):
            self.client.list_objects(Bucket=s3_object_lambda_arn)

    def test_s3_object_lambda_with_global_regions(self):
        s3_object_lambda_arn = (
            "arn:aws:s3-object-lambda:us-east-1:123456789012:"
            "accesspoint/mybanner"
        )
        expected_exception = UnsupportedS3AccesspointConfigurationError
        for region in ("aws-global", "s3-external-1"):
            self.client, _ = self.create_stubbed_s3_client(
                region_name=region, config=Config(s3={"use_arn_region": False})
            )
            with self.assertRaises(expected_exception):
                self.client.list_objects(Bucket=s3_object_lambda_arn)

    def test_s3_object_lambda_arn_with_us_east_1(self):
        # test that us-east-1 region is not resolved
        # into s3 global endpoint
        s3_object_lambda_arn = (
            "arn:aws:s3-object-lambda:us-east-1:123456789012:"
            "accesspoint/myBanner"
        )
        self.client, self.http_stubber = self.create_stubbed_s3_client(
            region_name="us-east-1",
            config=Config(s3={"use_arn_region": False}),
        )
        self.http_stubber.add_response()
        self.client.list_objects(Bucket=s3_object_lambda_arn)
        request = self.http_stubber.requests[0]
        self.assert_signing_name(request, "s3-object-lambda")
        self.assert_signing_region(request, "us-east-1")
        expected_endpoint = (
            "myBanner-123456789012.s3-object-lambda.us-east-1.amazonaws.com"
        )
        self.assert_endpoint(request, expected_endpoint)

    def test_basic_s3_object_lambda_arn(self):
        s3_object_lambda_arn = (
            "arn:aws:s3-object-lambda:us-west-2:123456789012:"
            "accesspoint/myBanner"
        )
        self.client, self.http_stubber = self.create_stubbed_s3_client(
            region_name="us-east-1"
        )
        self.http_stubber.add_response()
        self.client.list_objects(Bucket=s3_object_lambda_arn)
        request = self.http_stubber.requests[0]
        self.assert_signing_name(request, "s3-object-lambda")
        self.assert_signing_region(request, "us-west-2")
        expected_endpoint = (
            "myBanner-123456789012.s3-object-lambda.us-west-2.amazonaws.com"
        )
        self.assert_endpoint(request, expected_endpoint)
        sha_header = request.headers.get("x-amz-content-sha256")
        self.assertIsNotNone(sha_header)
        self.assertNotEqual(sha_header, b"UNSIGNED-PAYLOAD")

    def test_outposts_raise_exception_if_fips_region(self):
        outpost_arn = (
            "arn:aws:s3-outposts:us-gov-east-1:123456789012:outpost:"
            "op-01234567890123456:accesspoint:myaccesspoint"
        )
        self.client, _ = self.create_stubbed_s3_client(
            region_name="fips-east-1"
        )
        expected_exception = UnsupportedS3AccesspointConfigurationError
        with self.assertRaises(expected_exception):
            self.client.list_objects(Bucket=outpost_arn)

    def test_accesspoint_fips_raise_for_cross_region(self):
        s3_accesspoint_arn = (
            "arn:aws-us-gov:s3:us-gov-east-1:123456789012:"
            "accesspoint:myendpoint"
        )
        self.client, _ = self.create_stubbed_s3_client(
            region_name="fips-us-gov-west-1",
            config=Config(s3={"use_arn_region": False}),
        )
        expected_exception = UnsupportedS3AccesspointConfigurationError
        with self.assertRaises(expected_exception):
            self.client.list_objects(Bucket=s3_accesspoint_arn)

    def test_accesspoint_fips_raise_if_fips_in_arn(self):
        s3_accesspoint_arn = (
            "arn:aws-us-gov:s3:fips-us-gov-west-1:123456789012:"
            "accesspoint:myendpoint"
        )
        self.client, _ = self.create_stubbed_s3_client(
            region_name="fips-us-gov-west-1",
        )
        expected_exception = UnsupportedS3AccesspointConfigurationError
        with self.assertRaises(expected_exception):
            self.client.list_objects(Bucket=s3_accesspoint_arn)

    def test_accesspoint_with_global_regions(self):
        s3_accesspoint_arn = (
            "arn:aws:s3:us-east-1:123456789012:accesspoint:myendpoint"
        )
        self.client, _ = self.create_stubbed_s3_client(
            region_name="aws-global",
            config=Config(s3={"use_arn_region": False}),
        )
        expected_exception = UnsupportedS3AccesspointConfigurationError
        with self.assertRaises(expected_exception):
            self.client.list_objects(Bucket=s3_accesspoint_arn)

        # It shouldn't raise if use_arn_region is True
        self.client, self.http_stubber = self.create_stubbed_s3_client(
            region_name="s3-external-1",
            config=Config(s3={"use_arn_region": True}),
        )

        self.http_stubber.add_response()
        self.client.list_objects(Bucket=s3_accesspoint_arn)
        request = self.http_stubber.requests[0]
        expected_endpoint = (
            "myendpoint-123456789012.s3-accesspoint." "us-east-1.amazonaws.com"
        )
        self.assert_endpoint(request, expected_endpoint)

        # It shouldn't raise if no use_arn_region is specified since
        # use_arn_region defaults to True
        self.client, self.http_stubber = self.create_stubbed_s3_client(
            region_name="s3-external-1",
        )

        self.http_stubber.add_response()
        self.client.list_objects(Bucket=s3_accesspoint_arn)
        request = self.http_stubber.requests[0]
        expected_endpoint = (
            "myendpoint-123456789012.s3-accesspoint." "us-east-1.amazonaws.com"
        )
        self.assert_endpoint(request, expected_endpoint)

    @requires_crt()
    def test_mrap_arn_with_client_regions(self):
        mrap_arn = "arn:aws:s3::123456789012:accesspoint:mfzwi23gnjvgw.mrap"
        region_tests = [
            (
                "us-east-1",
                "mfzwi23gnjvgw.mrap.accesspoint.s3-global.amazonaws.com",
            ),
            (
                "us-west-2",
                "mfzwi23gnjvgw.mrap.accesspoint.s3-global.amazonaws.com",
            ),
            (
                "aws-global",
                "mfzwi23gnjvgw.mrap.accesspoint.s3-global.amazonaws.com",
            ),
        ]
        for region, expected in region_tests:
            self._assert_mrap_endpoint(mrap_arn, region, expected)

    @requires_crt()
    def test_mrap_arn_with_other_partition(self):
        mrap_arn = "arn:aws-cn:s3::123456789012:accesspoint:mfzwi23gnjvgw.mrap"
        expected = "mfzwi23gnjvgw.mrap.accesspoint.s3-global.amazonaws.com.cn"
        self._assert_mrap_endpoint(mrap_arn, "cn-north-1", expected)

    @requires_crt()
    def test_mrap_arn_with_invalid_s3_configs(self):
        mrap_arn = "arn:aws:s3::123456789012:accesspoint:mfzwi23gnjvgw.mrap"
        config_tests = [
            ("us-west-2", Config(s3={"use_dualstack_endpoint": True})),
            ("us-west-2", Config(s3={"use_accelerate_endpoint": True})),
        ]
        for region, config in config_tests:
            self._assert_mrap_config_failure(mrap_arn, region, config=config)

    @requires_crt()
    def test_mrap_arn_with_disable_config_enabled(self):
        mrap_arn = "arn:aws:s3::123456789012:accesspoint:mfzwi23gnjvgw.mrap"
        config = Config(s3={"s3_disable_multiregion_access_points": True})
        for region in ("us-west-2", "aws-global"):
            self._assert_mrap_config_failure(mrap_arn, region, config)

    @requires_crt()
    def test_mrap_arn_with_disable_config_enabled_custom_endpoint(self):
        mrap_arn = "arn:aws:s3::123456789012:accesspoint:myendpoint"
        config = Config(s3={"s3_disable_multiregion_access_points": True})
        self._assert_mrap_config_failure(mrap_arn, "us-west-2", config)

    @requires_crt()
    def test_mrap_arn_with_disable_config_disabled(self):
        mrap_arn = "arn:aws:s3::123456789012:accesspoint:mfzwi23gnjvgw.mrap"
        config = Config(s3={"s3_disable_multiregion_access_points": False})
        expected = "mfzwi23gnjvgw.mrap.accesspoint.s3-global.amazonaws.com"
        self._assert_mrap_endpoint(
            mrap_arn, "us-west-2", expected, config=config
        )

    @requires_crt()
    def test_global_arn_without_mrap_suffix(self):
        global_arn_tests = [
            (
                "arn:aws:s3::123456789012:accesspoint:myendpoint",
                "myendpoint.accesspoint.s3-global.amazonaws.com",
            ),
            (
                "arn:aws:s3::123456789012:accesspoint:my.bucket",
                "my.bucket.accesspoint.s3-global.amazonaws.com",
            ),
        ]
        for arn, expected in global_arn_tests:
            self._assert_mrap_endpoint(arn, "us-west-2", expected)

    @requires_crt()
    def test_mrap_signing_algorithm_is_sigv4a(self):
        s3_accesspoint_arn = (
            "arn:aws:s3::123456789012:accesspoint:mfzwi23gnjvgw.mrap"
        )
        self.client, self.http_stubber = self.create_stubbed_s3_client(
            region_name="us-west-2"
        )
        self.http_stubber.add_response()
        self.client.list_objects(Bucket=s3_accesspoint_arn)
        request = self.http_stubber.requests[0]
        self._assert_sigv4a_used(request.headers)

    @requires_crt()
    def test_mrap_presigned_url(self):
        mrap_arn = "arn:aws:s3::123456789012:accesspoint:mfzwi23gnjvgw.mrap"
        config = Config(s3={"s3_disable_multiregion_access_points": False})
        expected_url = "mfzwi23gnjvgw.mrap.accesspoint.s3-global.amazonaws.com"
        self._assert_mrap_presigned_url(
            mrap_arn, "us-west-2", expected_url, config=config
        )

    @requires_crt()
    def test_mrap_presigned_url_disabled(self):
        mrap_arn = "arn:aws:s3::123456789012:accesspoint:mfzwi23gnjvgw.mrap"
        config = Config(s3={"s3_disable_multiregion_access_points": True})
        self._assert_mrap_config_presigned_failure(
            mrap_arn, "us-west-2", config
        )

    def _assert_mrap_config_failure(self, arn, region, config):
        self.client, self.http_stubber = self.create_stubbed_s3_client(
            region_name=region, config=config
        )
        with self.assertRaises(
            botocore.exceptions.UnsupportedS3AccesspointConfigurationError
        ):
            self.client.list_objects(Bucket=arn)

    @FreezeTime(botocore.auth.datetime, date=DATE)
    def _get_presigned_url(self, arn, region, config=None, endpoint_url=None):
        self.client, self.http_stubber = self.create_stubbed_s3_client(
            region_name=region,
            endpoint_url=endpoint_url,
            config=config,
            aws_access_key_id='ACCESS_KEY_ID',
            aws_secret_access_key='SECRET_ACCESS_KEY',
        )
        presigned_url = self.client.generate_presigned_url(
            'get_object', Params={'Bucket': arn, 'Key': 'obj'}, ExpiresIn=900
        )
        return presigned_url

    def _assert_presigned_url(
        self,
        presigned_url,
        expected_url,
        expected_signature,
        expected_credentials,
    ):
        url_parts = urlsplit(presigned_url)
        assert url_parts.netloc == expected_url
        query_strs = url_parts.query.split('&')
        query_parts = dict(part.split('=') for part in query_strs)
        assert expected_signature == query_parts['X-Amz-Signature']
        assert expected_credentials in query_parts['X-Amz-Credential']

    def _assert_mrap_presigned_url(
        self, arn, region, expected, endpoint_url=None, config=None
    ):
        presigned_url = self._get_presigned_url(
            arn, region, endpoint_url=endpoint_url, config=config
        )
        url_parts = urlsplit(presigned_url)
        self.assertEqual(expected, url_parts.hostname)
        # X-Amz-Region-Set header MUST be * (percent-encoded as %2A) for MRAPs
        self.assertIn("X-Amz-Region-Set=%2A", url_parts.query)

    def _assert_mrap_config_presigned_failure(self, arn, region, config):
        self.client, self.http_stubber = self.create_stubbed_s3_client(
            region_name=region, config=config
        )
        with self.assertRaises(
            botocore.exceptions.UnsupportedS3AccesspointConfigurationError
        ):
            self.client.generate_presigned_url(
                "get_object", Params={"Bucket": arn, "Key": "test_object"}
            )

    def _assert_mrap_endpoint(
        self, arn, region, expected, endpoint_url=None, config=None
    ):
        self.client, self.http_stubber = self.create_stubbed_s3_client(
            region_name=region, endpoint_url=endpoint_url, config=config
        )
        self.http_stubber.add_response()
        self.client.list_objects(Bucket=arn)
        request = self.http_stubber.requests[0]
        self.assert_endpoint(request, expected)
        # MRAP requests MUST include a global signing region stored in the
        # X-Amz-Region-Set header as *.
        self.assert_header_matches(request, "X-Amz-Region-Set", b"*")

    def _assert_sigv4a_used(self, headers):
        self.assertIn(
            b"AWS4-ECDSA-P256-SHA256", headers.get("Authorization", "")
        )


class TestOnlyAsciiCharsAllowed(BaseS3OperationTest):
    def test_validates_non_ascii_chars_trigger_validation_error(self):
        self.http_stubber.add_response()
        with self.http_stubber:
            with self.assertRaises(ParamValidationError):
                self.client.put_object(
                    Bucket="foo",
                    Key="bar",
                    Metadata={"goodkey": "good", "non-ascii": "\u2713"},
                )


class TestS3GetBucketLifecycle(BaseS3OperationTest):
    def test_multiple_transitions_returns_one(self):
        response_body = (
            b'<?xml version="1.0" ?>'
            b'<LifecycleConfiguration xmlns="http://s3.amazonaws.'
            b'com/doc/2006-03-01/">'
            b"	<Rule>"
            b"		<ID>transitionRule</ID>"
            b"		<Prefix>foo</Prefix>"
            b"		<Status>Enabled</Status>"
            b"		<Transition>"
            b"			<Days>40</Days>"
            b"			<StorageClass>STANDARD_IA</StorageClass>"
            b"		</Transition>"
            b"		<Transition>"
            b"			<Days>70</Days>"
            b"			<StorageClass>GLACIER</StorageClass>"
            b"		</Transition>"
            b"	</Rule>"
            b"	<Rule>"
            b"		<ID>noncurrentVersionRule</ID>"
            b"		<Prefix>bar</Prefix>"
            b"		<Status>Enabled</Status>"
            b"		<NoncurrentVersionTransition>"
            b"			<NoncurrentDays>40</NoncurrentDays>"
            b"			<StorageClass>STANDARD_IA</StorageClass>"
            b"		</NoncurrentVersionTransition>"
            b"		<NoncurrentVersionTransition>"
            b"			<NoncurrentDays>70</NoncurrentDays>"
            b"			<StorageClass>GLACIER</StorageClass>"
            b"		</NoncurrentVersionTransition>"
            b"	</Rule>"
            b"</LifecycleConfiguration>"
        )
        s3 = self.session.create_client("s3")
        with ClientHTTPStubber(s3) as http_stubber:
            http_stubber.add_response(body=response_body)
            response = s3.get_bucket_lifecycle(Bucket="mybucket")
        # Each Transition member should have at least one of the
        # transitions provided.
        self.assertEqual(
            response["Rules"][0]["Transition"],
            {"Days": 40, "StorageClass": "STANDARD_IA"},
        )
        self.assertEqual(
            response["Rules"][1]["NoncurrentVersionTransition"],
            {"NoncurrentDays": 40, "StorageClass": "STANDARD_IA"},
        )


class TestS3PutObject(BaseS3OperationTest):
    def test_500_error_with_non_xml_body(self):
        # Note: This exact tesdict may not be applicable from
        # an integration standpoint if the issue is fixed in the future.
        #
        # The issue is that:
        # S3 returns a 200 response but the received response from urllib3 has
        # a 500 status code and the headers are in the body of the
        # the response. Botocore will try to parse out the error body as xml,
        # but the body is invalid xml because it is full of headers.
        # So instead of blowing up on an XML parsing error, we
        # should at least use the 500 status code because that can be
        # retried.
        #
        # We are unsure of what exactly causes the response to be mangled
        # but we expect it to be how 100 continues are handled.
        non_xml_content = (
            b"x-amz-id-2: foo\r\n"
            b"x-amz-request-id: bar\n"
            b"Date: Tue, 06 Oct 2015 03:20:38 GMT\r\n"
            b'ETag: "a6d856bc171fc6aa1b236680856094e2"\r\n'
            b"Content-Length: 0\r\n"
            b"Server: AmazonS3\r\n"
        )
        s3 = self.session.create_client("s3")
        with ClientHTTPStubber(s3) as http_stubber:
            http_stubber.add_response(status=500, body=non_xml_content)
            http_stubber.add_response()
            response = s3.put_object(
                Bucket="mybucket", Key="mykey", Body=b"foo"
            )
            # The first response should have been retried even though the xml is
            # invalid and eventually return the 200 response.
            self.assertEqual(
                response["ResponseMetadata"]["HTTPStatusCode"], 200
            )
            self.assertEqual(len(http_stubber.requests), 2)


class TestS3ExpiresHeaderResponse(BaseS3OperationTest):
    def test_valid_expires_value_in_response(self):
        expires_value = "Thu, 01 Jan 1970 00:00:00 GMT"
        mock_headers = {'expires': expires_value}
        s3 = self.session.create_client("s3")
        with ClientHTTPStubber(s3) as http_stubber:
            http_stubber.add_response(headers=mock_headers)
            response = s3.get_object(Bucket='mybucket', Key='mykey')
            self.assertEqual(
                response.get('Expires'),
                datetime.datetime(1970, 1, 1, tzinfo=tzutc()),
            )
            self.assertEqual(response.get('ExpiresString'), expires_value)

    def test_invalid_expires_value_in_response(self):
        expires_value = "Invalid Date"
        mock_headers = {'expires': expires_value}
        warning_msg = 'Failed to parse the "Expires" member as a timestamp'
        s3 = self.session.create_client("s3")
        with self.assertLogs('botocore.handlers', level='WARNING') as log:
            with ClientHTTPStubber(s3) as http_stubber:
                http_stubber.add_response(headers=mock_headers)
                response = s3.get_object(Bucket='mybucket', Key='mykey')
                self.assertNotIn(
                    'expires',
                    response.get('ResponseMetadata').get('HTTPHeaders'),
                )
                self.assertNotIn('Expires', response)
                self.assertEqual(response.get('ExpiresString'), expires_value)
                self.assertTrue(
                    any(warning_msg in entry for entry in log.output),
                    f'Expected warning message not found in logs. Logs: {log.output}',
                )


class TestWriteGetObjectResponse(BaseS3ClientConfigurationTest):
    def create_stubbed_s3_client(self, **kwargs):
        client = self.create_s3_client(**kwargs)
        http_stubber = ClientHTTPStubber(client)
        http_stubber.start()
        return client, http_stubber

    def test_endpoint_redirection(self):
        regions = ["us-west-2", "us-east-1"]
        for region in regions:
            self.client, self.http_stubber = self.create_stubbed_s3_client(
                region_name=region
            )
            self.http_stubber.add_response()
            self.client.write_get_object_response(
                RequestRoute="endpoint-io.a1c1d5c7",
                RequestToken="SecretToken",
            )
            request = self.http_stubber.requests[0]
            self.assert_signing_name(request, "s3-object-lambda")
            self.assert_signing_region(request, region)
            expected_endpoint = (
                "endpoint-io.a1c1d5c7.s3-object-lambda."
                f"{region}.amazonaws.com"
            )
            self.assert_endpoint(request, expected_endpoint)

    def test_endpoint_redirection_fails_with_custom_endpoint(self):
        self.client, self.http_stubber = self.create_stubbed_s3_client(
            region_name="us-west-2", endpoint_url="https://example.com"
        )
        self.http_stubber.add_response()
        self.client.write_get_object_response(
            RequestRoute="endpoint-io.a1c1d5c7",
            RequestToken="SecretToken",
        )
        request = self.http_stubber.requests[0]
        self.assert_signing_name(request, "s3-object-lambda")
        self.assert_signing_region(request, "us-west-2")
        self.assert_endpoint(request, "endpoint-io.a1c1d5c7.example.com")

    def test_endpoint_redirection_fails_with_accelerate_endpoint(self):
        config = Config(s3={"use_accelerate_endpoint": True})
        self.client, self.http_stubber = self.create_stubbed_s3_client(
            region_name="us-west-2",
            config=config,
        )
        self.http_stubber.add_response()
        with self.assertRaises(UnsupportedS3ConfigurationError):
            self.client.write_get_object_response(
                RequestRoute="endpoint-io.a1c1d5c7",
                RequestToken="SecretToken",
            )

    def test_invalid_request_route_raises(self):
        self.client, self.http_stubber = self.create_stubbed_s3_client(
            region_name="us-west-2"
        )
        self.http_stubber.add_response()
        with self.assertRaises(ParamValidationError):
            self.client.write_get_object_response(
                RequestRoute="my-route/",
                RequestToken="SecretToken",
            )


class TestS3SigV4(BaseS3OperationTest):
    def setUp(self):
        super().setUp()
        self.client = self.session.create_client(
            "s3", self.region, config=Config(signature_version="s3v4")
        )
        self.http_stubber = ClientHTTPStubber(self.client)
        self.http_stubber.add_response()

    def get_sent_headers(self):
        return self.http_stubber.requests[0].headers

    def test_trailing_checksum_set(self):
        with self.http_stubber:
            self.client.put_object(Bucket="foo", Key="bar", Body="baz")
        sent_headers = self.get_sent_headers()
        self.assertEqual(sent_headers["Content-Encoding"], b"aws-chunked")
        self.assertEqual(sent_headers["Transfer-Encoding"], b"chunked")
        self.assertEqual(
            sent_headers["X-Amz-Trailer"], b"x-amz-checksum-crc32"
        )
        self.assertEqual(sent_headers["X-Amz-Decoded-Content-Length"], b"3")
        self.assertEqual(
            sent_headers["x-amz-content-sha256"],
            b"STREAMING-UNSIGNED-PAYLOAD-TRAILER",
        )
        body = self.http_stubber.requests[0].body.read()
        self.assertIn(b"x-amz-checksum-crc32:eCQEmA==", body)

    def test_trailing_checksum_set_with_content_length_removes_header(self):
        with self.http_stubber:
            self.client.put_object(
                Bucket="foo", Key="bar", Body="baz", ContentLength=123
            )
        sent_headers = self.get_sent_headers()
        self.assertEqual(sent_headers["Content-Encoding"], b"aws-chunked")
        self.assertEqual(sent_headers["Transfer-Encoding"], b"chunked")
        self.assertEqual(
            sent_headers["X-Amz-Trailer"], b"x-amz-checksum-crc32"
        )
        self.assertEqual(sent_headers["X-Amz-Decoded-Content-Length"], b"3")
        self.assertEqual(
            sent_headers["x-amz-content-sha256"],
            b"STREAMING-UNSIGNED-PAYLOAD-TRAILER",
        )
        body = self.http_stubber.requests[0].body.read()
        self.assertIn(b"x-amz-checksum-crc32:eCQEmA==", body)
        self.assertNotIn("Content-Length", sent_headers)

    def test_trailing_checksum_set_empty_body(self):
        with self.http_stubber:
            self.client.put_object(Bucket="foo", Key="bar", Body="")
        sent_headers = self.get_sent_headers()
        self.assertEqual(sent_headers["Content-Encoding"], b"aws-chunked")
        self.assertEqual(sent_headers["Transfer-Encoding"], b"chunked")
        self.assertEqual(
            sent_headers["X-Amz-Trailer"], b"x-amz-checksum-crc32"
        )
        self.assertEqual(sent_headers["X-Amz-Decoded-Content-Length"], b"0")
        self.assertEqual(
            sent_headers["x-amz-content-sha256"],
            b"STREAMING-UNSIGNED-PAYLOAD-TRAILER",
        )
        body = self.http_stubber.requests[0].body.read()
        self.assertIn(b"x-amz-checksum-crc32:AAAAAA==", body)

    def test_trailing_checksum_set_empty_file(self):
        with self.http_stubber:
            with temporary_file("rb") as f:
                assert f.read() == b""
                self.client.put_object(Bucket="foo", Key="bar", Body=f)
                body = self.http_stubber.requests[0].body.read()
        sent_headers = self.get_sent_headers()
        self.assertEqual(sent_headers["Content-Encoding"], b"aws-chunked")
        self.assertEqual(sent_headers["Transfer-Encoding"], b"chunked")
        self.assertEqual(
            sent_headers["X-Amz-Trailer"], b"x-amz-checksum-crc32"
        )
        self.assertEqual(sent_headers["X-Amz-Decoded-Content-Length"], b"0")
        self.assertEqual(
            sent_headers["x-amz-content-sha256"],
            b"STREAMING-UNSIGNED-PAYLOAD-TRAILER",
        )
        self.assertIn(b"x-amz-checksum-crc32:AAAAAA==", body)

    def test_content_sha256_set_for_unsigned_request(self):
        config = Config(
            signature_version=UNSIGNED,
        )
        self.client = self.session.create_client(
            "s3", self.region, config=config
        )
        self.http_stubber = ClientHTTPStubber(self.client)
        self.http_stubber.add_response()
        with self.http_stubber:
            self.client.put_object(
                Bucket="foo", Key="bar", Body="baz", ChecksumAlgorithm="CRC32"
            )
        sent_headers = self.get_sent_headers()
        sha_header = sent_headers.get("x-amz-content-sha256")
        self.assertEqual(sha_header, b"STREAMING-UNSIGNED-PAYLOAD-TRAILER")

    def test_content_sha256_not_set_if_config_value_is_true(self):
        # By default, put_object() provides a trailing checksum and includes the
        # x-amz-content-sha256 header with a value of "STREAMING-UNSIGNED-PAYLOAD-TRAILER".
        # We do not support payload signing for streaming so the `payload_signing_enabled`
        # config has no effect here.
        config = Config(
            signature_version="s3v4",
            s3={"payload_signing_enabled": True},
        )
        self.client = self.session.create_client(
            "s3", self.region, config=config
        )
        self.http_stubber = ClientHTTPStubber(self.client)
        self.http_stubber.add_response()
        with self.http_stubber:
            self.client.put_object(Bucket="foo", Key="bar", Body="baz")
        sent_headers = self.get_sent_headers()
        sha_header = sent_headers.get("x-amz-content-sha256")
        self.assertEqual(sha_header, b"STREAMING-UNSIGNED-PAYLOAD-TRAILER")

    def test_content_sha256_not_set_if_config_value_is_false(self):
        # By default, put_object() provides a trailing checksum and includes the
        # x-amz-content-sha256 header with a value of "STREAMING-UNSIGNED-PAYLOAD-TRAILER".
        # We do not support payload signing for streaming so the `payload_signing_enabled`
        # config has no effect here.
        config = Config(
            signature_version="s3v4",
            s3={"payload_signing_enabled": False},
        )
        self.client = self.session.create_client(
            "s3", self.region, config=config
        )
        self.http_stubber = ClientHTTPStubber(self.client)
        self.http_stubber.add_response()
        with self.http_stubber:
            self.client.put_object(Bucket="foo", Key="bar", Body="baz")
        sent_headers = self.get_sent_headers()
        sha_header = sent_headers.get("x-amz-content-sha256")
        self.assertEqual(sha_header, b"STREAMING-UNSIGNED-PAYLOAD-TRAILER")

    def test_content_sha256_not_set_if_config_value_not_set_put_object(self):
        # By default, put_object() provides a trailing checksum and includes the
        # x-amz-content-sha256 header with a value of "STREAMING-UNSIGNED-PAYLOAD-TRAILER".
        # We do not support payload signing for streaming so the `payload_signing_enabled`
        # config has no effect here.
        config = Config(signature_version="s3v4")
        self.client = self.session.create_client(
            "s3",
            self.region,
            config=config,
        )
        self.http_stubber = ClientHTTPStubber(self.client)
        self.http_stubber.add_response()
        with self.http_stubber:
            self.client.put_object(Bucket="foo", Key="bar", Body="baz")
        sent_headers = self.get_sent_headers()
        sha_header = sent_headers.get("x-amz-content-sha256")
        self.assertEqual(sha_header, b"STREAMING-UNSIGNED-PAYLOAD-TRAILER")

    def test_content_sha256_set_if_config_value_not_set_list_objects(self):
        # The default behavior matches payload_signing_enabled=False. For
        # operations where the `Content-MD5` is not present, this means that
        # `x-amz-content-sha256` is present and set.
        config = Config(signature_version="s3v4")
        self.client = self.session.create_client(
            "s3", self.region, config=config
        )
        self.http_stubber = ClientHTTPStubber(self.client)
        self.http_stubber.add_response()
        with self.http_stubber:
            self.client.list_objects(Bucket="foo")
        sent_headers = self.get_sent_headers()
        sha_header = sent_headers.get("x-amz-content-sha256")
        self.assertIsNotNone(sha_header)
        self.assertNotEqual(sha_header, b"UNSIGNED-PAYLOAD")

    def test_content_sha256_set_s3_on_outpost(self):
        # S3 on Outpost bucket names should behave the same way.
        config = Config(signature_version="s3v4")
        bucket = (
            'test-accessp-e0000075431d83bebde8xz5w8ijx1qzlbp3i3kuse10--op-s3'
        )
        self.client = self.session.create_client(
            "s3", self.region, config=config
        )
        self.http_stubber = ClientHTTPStubber(self.client)
        self.http_stubber.add_response()
        with self.http_stubber:
            self.client.list_objects(Bucket=bucket)
        sent_headers = self.get_sent_headers()
        sha_header = sent_headers.get("x-amz-content-sha256")
        self.assertNotEqual(sha_header, b"UNSIGNED-PAYLOAD")


class TestCanSendIntegerHeaders(BaseSessionTest):
    def test_int_values_with_sigv4(self):
        s3 = self.session.create_client(
            "s3",
            config=Config(
                signature_version="s3v4",
                request_checksum_calculation="when_required",
            ),
        )
        with ClientHTTPStubber(s3) as http_stubber:
            http_stubber.add_response()
            s3.upload_part(
                Bucket="foo",
                Key="bar",
                Body=b"foo",
                UploadId="bar",
                PartNumber=1,
                ContentLength=3,
            )
            headers = http_stubber.requests[0].headers
            # Verify that the request integer value of 3 has been converted to
            # string '3'.  This also means we've made it pass the signer which
            # expects string values in order to sign properly.
            self.assertEqual(headers["Content-Length"], b"3")


class TestRegionRedirect(BaseS3OperationTest):
    def setUp(self):
        super().setUp()
        self.client = self.session.create_client(
            "s3",
            "us-west-2",
            config=Config(
                signature_version="s3v4",
                s3={"addressing_style": "path"},
            ),
        )
        self.http_stubber = ClientHTTPStubber(self.client)

        self.redirect_response = {
            "status": 301,
            "headers": {"x-amz-bucket-region": "eu-central-1"},
            "body": (
                b'<?xml version="1.0" encoding="UTF-8"?>\n'
                b"<Error>"
                b"    <Code>PermanentRedirect</Code>"
                b"    <Message>The bucket you are attempting to access must be"
                b"        addressed using the specified endpoint. Please send "
                b"        all future requests to this endpoint."
                b"    </Message>"
                b"    <Bucket>foo</Bucket>"
                b"    <Endpoint>foo.s3.eu-central-1.amazonaws.com</Endpoint>"
                b"</Error>"
            ),
        }
        self.bad_signing_region_response = {
            "status": 400,
            "headers": {"x-amz-bucket-region": "eu-central-1"},
            "body": (
                b'<?xml version="1.0" encoding="UTF-8"?>'
                b"<Error>"
                b"  <Code>AuthorizationHeaderMalformed</Code>"
                b"  <Message>the region us-west-2 is wrong; "
                b"expecting eu-central-1</Message>"
                b"  <Region>eu-central-1</Region>"
                b"  <RequestId>BD9AA1730D454E39</RequestId>"
                b"  <HostId></HostId>"
                b"</Error>"
            ),
        }
        self.success_response = {
            "status": 200,
            "headers": {},
            "body": (
                b'<?xml version="1.0" encoding="UTF-8"?>\n'
                b"<ListBucketResult>"
                b"    <Name>foo</Name>"
                b"    <Prefix></Prefix>"
                b"    <Marker></Marker>"
                b"    <MaxKeys>1000</MaxKeys>"
                b"    <EncodingType>url</EncodingType>"
                b"    <IsTruncated>false</IsTruncated>"
                b"</ListBucketResult>"
            ),
        }

    def test_region_redirect(self):
        self.http_stubber.add_response(**self.redirect_response)
        self.http_stubber.add_response(**self.success_response)
        with self.http_stubber:
            response = self.client.list_objects(Bucket="foo")
        self.assertEqual(response["ResponseMetadata"]["HTTPStatusCode"], 200)
        self.assertEqual(len(self.http_stubber.requests), 2)

        initial_url = (
            "https://s3.us-west-2.amazonaws.com/foo" "?encoding-type=url"
        )
        self.assertEqual(self.http_stubber.requests[0].url, initial_url)

        fixed_url = (
            "https://s3.eu-central-1.amazonaws.com/foo" "?encoding-type=url"
        )
        self.assertEqual(self.http_stubber.requests[1].url, fixed_url)

    def test_region_redirect_cache(self):
        self.http_stubber.add_response(**self.redirect_response)
        self.http_stubber.add_response(**self.success_response)
        self.http_stubber.add_response(**self.success_response)

        with self.http_stubber:
            first_response = self.client.list_objects(Bucket="foo")
            second_response = self.client.list_objects(Bucket="foo")

        self.assertEqual(
            first_response["ResponseMetadata"]["HTTPStatusCode"], 200
        )
        self.assertEqual(
            second_response["ResponseMetadata"]["HTTPStatusCode"], 200
        )

        self.assertEqual(len(self.http_stubber.requests), 3)
        initial_url = (
            "https://s3.us-west-2.amazonaws.com/foo" "?encoding-type=url"
        )
        self.assertEqual(self.http_stubber.requests[0].url, initial_url)

        fixed_url = (
            "https://s3.eu-central-1.amazonaws.com/foo" "?encoding-type=url"
        )
        self.assertEqual(self.http_stubber.requests[1].url, fixed_url)
        self.assertEqual(self.http_stubber.requests[2].url, fixed_url)

    def test_resign_request_with_region_when_needed(self):
        # Create a client with no explicit configuration so we can
        # verify the default behavior.
        client = self.session.create_client("s3", "us-west-2")
        with ClientHTTPStubber(client) as http_stubber:
            http_stubber.add_response(**self.bad_signing_region_response)
            http_stubber.add_response(**self.success_response)
            first_response = client.list_objects(Bucket="foo")
            self.assertEqual(
                first_response["ResponseMetadata"]["HTTPStatusCode"], 200
            )

            self.assertEqual(len(http_stubber.requests), 2)
            initial_url = (
                "https://foo.s3.us-west-2.amazonaws.com/" "?encoding-type=url"
            )
            self.assertEqual(http_stubber.requests[0].url, initial_url)

            fixed_url = (
                "https://foo.s3.eu-central-1.amazonaws.com/"
                "?encoding-type=url"
            )
            self.assertEqual(http_stubber.requests[1].url, fixed_url)

    def test_resign_request_in_us_east_1(self):
        region_headers = {"x-amz-bucket-region": "eu-central-1"}

        # Verify that the default behavior in us-east-1 will redirect
        client = self.session.create_client("s3", "us-east-1")
        with ClientHTTPStubber(client) as http_stubber:
            http_stubber.add_response(status=400)
            http_stubber.add_response(status=400, headers=region_headers)
            http_stubber.add_response(headers=region_headers)
            http_stubber.add_response()
            response = client.head_object(Bucket="foo", Key="bar")
            self.assertEqual(
                response["ResponseMetadata"]["HTTPStatusCode"], 200
            )

            self.assertEqual(len(http_stubber.requests), 4)
            initial_url = "https://foo.s3.amazonaws.com/bar"
            self.assertEqual(http_stubber.requests[0].url, initial_url)

            fixed_url = "https://foo.s3.eu-central-1.amazonaws.com/bar"
            self.assertEqual(http_stubber.requests[-1].url, fixed_url)

    def test_resign_request_in_us_east_1_fails(self):
        region_headers = {"x-amz-bucket-region": "eu-central-1"}

        # Verify that the final 400 response is propagated
        # back to the user.
        client = self.session.create_client("s3", "us-east-1")
        with ClientHTTPStubber(client) as http_stubber:
            http_stubber.add_response(status=400)
            http_stubber.add_response(status=400, headers=region_headers)
            http_stubber.add_response(headers=region_headers)
            # The final request still fails with a 400.
            http_stubber.add_response(status=400)
            with self.assertRaises(ClientError):
                client.head_object(Bucket="foo", Key="bar")
            self.assertEqual(len(http_stubber.requests), 4)

    def test_no_region_redirect_for_accesspoint(self):
        self.http_stubber.add_response(**self.redirect_response)
        accesspoint_arn = (
            "arn:aws:s3:us-west-2:123456789012:accesspoint:myendpoint"
        )
        with self.http_stubber:
            try:
                self.client.list_objects(Bucket=accesspoint_arn)
            except self.client.exceptions.ClientError as e:
                self.assertEqual(
                    e.response["Error"]["Code"], "PermanentRedirect"
                )
            else:
                self.fail("PermanentRedirect error should have been raised")


class TestFipsRegionRedirect(BaseS3OperationTest):
    def setUp(self):
        super().setUp()
        self.client = self.session.create_client(
            "s3",
            "fips-us-west-2",
            config=Config(signature_version="s3v4"),
        )
        self.http_stubber = ClientHTTPStubber(self.client)

        self.redirect_response = {
            "status": 301,
            "headers": {"x-amz-bucket-region": "us-west-1"},
            "body": (
                b'<?xml version="1.0" encoding="UTF-8"?>\n'
                b"<Error>"
                b"    <Code>PermanentRedirect</Code>"
                b"    <Message>The bucket you are attempting to access must be"
                b"        addressed using the specified endpoint. Please send "
                b"        all future requests to this endpoint."
                b"    </Message>"
                b"    <Bucket>foo</Bucket>"
                b"    <Endpoint>foo.s3-fips.us-west-1.amazonaws.com</Endpoint>"
                b"</Error>"
            ),
        }
        self.success_response = {
            "status": 200,
            "headers": {},
            "body": (
                b'<?xml version="1.0" encoding="UTF-8"?>\n'
                b"<ListBucketResult>"
                b"    <Name>foo</Name>"
                b"    <Prefix></Prefix>"
                b"    <Marker></Marker>"
                b"    <MaxKeys>1000</MaxKeys>"
                b"    <EncodingType>url</EncodingType>"
                b"    <IsTruncated>false</IsTruncated>"
                b"</ListBucketResult>"
            ),
        }
        self.bad_signing_region_response = {
            "status": 400,
            "headers": {"x-amz-bucket-region": "us-west-1"},
            "body": (
                b'<?xml version="1.0" encoding="UTF-8"?>'
                b"<Error>"
                b"  <Code>AuthorizationHeaderMalformed</Code>"
                b"  <Message>the region us-west-2 is wrong; "
                b"expecting us-west-1</Message>"
                b"  <Region>us-west-1</Region>"
                b"  <RequestId>BD9AA1730D454E39</RequestId>"
                b"  <HostId></HostId>"
                b"</Error>"
            ),
        }

    def test_fips_region_redirect(self):
        self.http_stubber.add_response(**self.redirect_response)
        self.http_stubber.add_response(**self.success_response)
        with self.http_stubber:
            response = self.client.list_objects(Bucket="foo")
        self.assertEqual(response["ResponseMetadata"]["HTTPStatusCode"], 200)
        self.assertEqual(len(self.http_stubber.requests), 2)

        initial_url = (
            "https://foo.s3-fips.us-west-2.amazonaws.com/?encoding-type=url"
        )
        self.assertEqual(self.http_stubber.requests[0].url, initial_url)

        fixed_url = (
            "https://foo.s3-fips.us-west-1.amazonaws.com/?encoding-type=url"
        )
        self.assertEqual(self.http_stubber.requests[1].url, fixed_url)

    def test_fips_region_redirect_cache(self):
        self.http_stubber.add_response(**self.redirect_response)
        self.http_stubber.add_response(**self.success_response)
        self.http_stubber.add_response(**self.success_response)

        with self.http_stubber:
            first_response = self.client.list_objects(Bucket="foo")
            second_response = self.client.list_objects(Bucket="foo")

        self.assertEqual(
            first_response["ResponseMetadata"]["HTTPStatusCode"], 200
        )
        self.assertEqual(
            second_response["ResponseMetadata"]["HTTPStatusCode"], 200
        )

        self.assertEqual(len(self.http_stubber.requests), 3)
        initial_url = (
            "https://foo.s3-fips.us-west-2.amazonaws.com/?encoding-type=url"
        )
        self.assertEqual(self.http_stubber.requests[0].url, initial_url)

        fixed_url = (
            "https://foo.s3-fips.us-west-1.amazonaws.com/?encoding-type=url"
        )
        self.assertEqual(self.http_stubber.requests[1].url, fixed_url)
        self.assertEqual(self.http_stubber.requests[2].url, fixed_url)

    def test_fips_resign_request_with_region_when_needed(self):
        # Create a client with no explicit configuration so we can
        # verify the default behavior.
        client = self.session.create_client("s3", "fips-us-west-2")
        with ClientHTTPStubber(client) as http_stubber:
            http_stubber.add_response(**self.bad_signing_region_response)
            http_stubber.add_response(**self.success_response)
            first_response = client.list_objects(Bucket="foo")
            self.assertEqual(
                first_response["ResponseMetadata"]["HTTPStatusCode"], 200
            )

            self.assertEqual(len(http_stubber.requests), 2)
            initial_url = (
                "https://foo.s3-fips.us-west-2.amazonaws.com/"
                "?encoding-type=url"
            )
            self.assertEqual(http_stubber.requests[0].url, initial_url)

            fixed_url = (
                "https://foo.s3-fips.us-west-1.amazonaws.com/"
                "?encoding-type=url"
            )
            self.assertEqual(http_stubber.requests[1].url, fixed_url)

    def test_fips_resign_request_in_us_east_1(self):
        region_headers = {"x-amz-bucket-region": "us-east-2"}

        # Verify that the default behavior in us-east-1 will redirect
        client = self.session.create_client("s3", "fips-us-east-1")
        with ClientHTTPStubber(client) as http_stubber:
            http_stubber.add_response(status=400)
            http_stubber.add_response(status=400, headers=region_headers)
            http_stubber.add_response(headers=region_headers)
            http_stubber.add_response()
            response = client.head_object(Bucket="foo", Key="bar")
            self.assertEqual(
                response["ResponseMetadata"]["HTTPStatusCode"], 200
            )

            self.assertEqual(len(http_stubber.requests), 4)
            initial_url = "https://foo.s3-fips.us-east-1.amazonaws.com/bar"
            self.assertEqual(http_stubber.requests[0].url, initial_url)

            fixed_url = "https://foo.s3-fips.us-east-2.amazonaws.com/bar"
            self.assertEqual(http_stubber.requests[-1].url, fixed_url)

    def test_fips_resign_request_in_us_east_1_fails(self):
        region_headers = {"x-amz-bucket-region": "us-east-2"}

        # Verify that the final 400 response is propagated
        # back to the user.
        client = self.session.create_client("s3", "fips-us-east-1")
        with ClientHTTPStubber(client) as http_stubber:
            http_stubber.add_response(status=400)
            http_stubber.add_response(status=400, headers=region_headers)
            http_stubber.add_response(headers=region_headers)
            # The final request still fails with a 400.
            http_stubber.add_response(status=400)
            with self.assertRaises(ClientError):
                client.head_object(Bucket="foo", Key="bar")
            self.assertEqual(len(http_stubber.requests), 4)

    def test_fips_no_region_redirect_for_accesspoint(self):
        self.http_stubber.add_response(**self.redirect_response)
        accesspoint_arn = (
            "arn:aws:s3:us-west-2:123456789012:accesspoint:myendpoint"
        )
        with self.http_stubber:
            try:
                self.client.list_objects(Bucket=accesspoint_arn)
            except self.client.exceptions.ClientError as e:
                self.assertEqual(
                    e.response["Error"]["Code"], "PermanentRedirect"
                )
            else:
                self.fail("PermanentRedirect error should have been raised")


class TestGeneratePresigned(BaseS3OperationTest):
    def assert_is_v2_presigned_url(self, url):
        qs_components = parse_qs(urlsplit(url).query)
        # Assert that it looks like a v2 presigned url by asserting it does
        # not have a couple of the v4 qs components and assert that it has the
        # v2 Signature component.
        self.assertNotIn("X-Amz-Credential", qs_components)
        self.assertNotIn("X-Amz-Algorithm", qs_components)
        self.assertIn("Signature", qs_components)

    def test_generate_unauthed_url(self):
        config = Config(signature_version=botocore.UNSIGNED)
        client = self.session.create_client("s3", self.region, config=config)
        url = client.generate_presigned_url(
            ClientMethod="get_object", Params={"Bucket": "foo", "Key": "bar"}
        )
        self.assertEqual(url, "https://foo.s3.amazonaws.com/bar")

    def test_generate_unauthed_post(self):
        config = Config(signature_version=botocore.UNSIGNED)
        client = self.session.create_client("s3", self.region, config=config)
        parts = client.generate_presigned_post(Bucket="foo", Key="bar")
        expected = {
            "fields": {"key": "bar"},
            "url": "https://foo.s3.amazonaws.com/",
        }
        self.assertEqual(parts, expected)

    def test_default_presign_uses_sigv2(self):
        url = self.client.generate_presigned_url(ClientMethod="list_buckets")
        self.assertNotIn("Algorithm=AWS4-HMAC-SHA256", url)

    def test_sigv4_presign(self):
        config = Config(signature_version="s3v4")
        client = self.session.create_client("s3", self.region, config=config)
        url = client.generate_presigned_url(ClientMethod="list_buckets")
        self.assertIn("Algorithm=AWS4-HMAC-SHA256", url)

    def test_sigv2_presign(self):
        config = Config(signature_version="s3")
        client = self.session.create_client("s3", self.region, config=config)
        url = client.generate_presigned_url(ClientMethod="list_buckets")
        self.assertNotIn("Algorithm=AWS4-HMAC-SHA256", url)

    def test_uses_sigv4_for_unknown_region(self):
        client = self.session.create_client("s3", "us-west-88")
        url = client.generate_presigned_url(ClientMethod="list_buckets")
        self.assertIn("Algorithm=AWS4-HMAC-SHA256", url)

    def test_default_presign_sigv4_in_sigv4_only_region(self):
        client = self.session.create_client("s3", "us-east-2")
        url = client.generate_presigned_url(ClientMethod="list_buckets")
        self.assertIn("Algorithm=AWS4-HMAC-SHA256", url)

    def test_presign_unsigned(self):
        config = Config(signature_version=botocore.UNSIGNED)
        client = self.session.create_client("s3", "us-east-2", config=config)
        url = client.generate_presigned_url(ClientMethod="list_buckets")
        self.assertEqual("https://s3.amazonaws.com/", url)

    def test_presign_url_with_ssec(self):
        config = Config(signature_version="s3")
        client = self.session.create_client("s3", "us-east-1", config=config)
        url = client.generate_presigned_url(
            ClientMethod="get_object",
            Params={
                "Bucket": "mybucket",
                "Key": "mykey",
                "SSECustomerKey": "a" * 32,
                "SSECustomerAlgorithm": "AES256",
            },
        )
        # The md5 of the sse-c key will be injected when parameters are
        # built so it should show up in the presigned url as well.
        self.assertIn("x-amz-server-side-encryption-customer-key-md5=", url)

    def test_presign_s3_accelerate(self):
        config = Config(
            signature_version=botocore.UNSIGNED,
            s3={"use_accelerate_endpoint": True},
        )
        client = self.session.create_client("s3", "us-east-1", config=config)
        url = client.generate_presigned_url(
            ClientMethod="get_object",
            Params={"Bucket": "mybucket", "Key": "mykey"},
        )
        # The url should be the accelerate endpoint
        self.assertEqual(
            "https://mybucket.s3-accelerate.amazonaws.com/mykey", url
        )

    def test_presign_s3_accelerate_fails_with_fips(self):
        config = Config(
            signature_version=botocore.UNSIGNED,
            s3={"use_accelerate_endpoint": True},
        )
        client = self.session.create_client(
            "s3", "fips-us-east-1", config=config
        )
        expected_exception = UnsupportedS3ConfigurationError
        with self.assertRaisesRegex(
            expected_exception, "Accelerate cannot be used with FIPS"
        ):
            client.generate_presigned_url(
                ClientMethod="get_object",
                Params={"Bucket": "mybucket", "Key": "mykey"},
            )

    def test_presign_post_s3_accelerate(self):
        config = Config(
            signature_version=botocore.UNSIGNED,
            s3={"use_accelerate_endpoint": True},
        )
        client = self.session.create_client("s3", "us-east-1", config=config)
        parts = client.generate_presigned_post(Bucket="mybucket", Key="mykey")
        # The url should be the accelerate endpoint
        expected = {
            "fields": {"key": "mykey"},
            "url": "https://mybucket.s3-accelerate.amazonaws.com/",
        }
        self.assertEqual(parts, expected)

    def test_presign_uses_v2_for_aws_global(self):
        client = self.session.create_client("s3", "aws-global")
        url = client.generate_presigned_url(
            "get_object", {"Bucket": "mybucket", "Key": "mykey"}
        )
        self.assert_is_v2_presigned_url(url)

    def test_presign_uses_v2_for_default_region_with_us_east_1_regional(self):
        config = Config(s3={"us_east_1_regional_endpoint": "regional"})
        client = self.session.create_client("s3", config=config)
        url = client.generate_presigned_url(
            "get_object", {"Bucket": "mybucket", "Key": "mykey"}
        )
        self.assert_is_v2_presigned_url(url)

    def test_presign_uses_v2_for_aws_global_with_us_east_1_regional(self):
        config = Config(s3={"us_east_1_regional_endpoint": "regional"})
        client = self.session.create_client("s3", "aws-global", config=config)
        url = client.generate_presigned_url(
            "get_object", {"Bucket": "mybucket", "Key": "mykey"}
        )
        self.assert_is_v2_presigned_url(url)

    def test_presign_uses_v2_for_us_east_1(self):
        client = self.session.create_client("s3", "us-east-1")
        url = client.generate_presigned_url(
            "get_object", {"Bucket": "mybucket", "Key": "mykey"}
        )
        self.assert_is_v2_presigned_url(url)

    def test_presign_uses_v2_for_us_east_1_with_us_east_1_regional(self):
        config = Config(s3={"us_east_1_regional_endpoint": "regional"})
        client = self.session.create_client("s3", "us-east-1", config=config)
        url = client.generate_presigned_url(
            "get_object", {"Bucket": "mybucket", "Key": "mykey"}
        )
        self.assert_is_v2_presigned_url(url)


CHECKSUM_TEST_CASES = [
    ("put_bucket_tagging", {"Bucket": "foo", "Tagging": {"TagSet": []}}),
    (
        "put_bucket_lifecycle",
        {"Bucket": "foo", "LifecycleConfiguration": {"Rules": []}},
    ),
    (
        "put_bucket_lifecycle_configuration",
        {"Bucket": "foo", "LifecycleConfiguration": {"Rules": []}},
    ),
    (
        "put_bucket_cors",
        {"Bucket": "foo", "CORSConfiguration": {"CORSRules": []}},
    ),
    (
        "delete_objects",
        {"Bucket": "foo", "Delete": {"Objects": [{"Key": "bar"}]}},
    ),
    (
        "put_bucket_replication",
        {
            "Bucket": "foo",
            "ReplicationConfiguration": {"Role": "", "Rules": []},
        },
    ),
    ("put_bucket_acl", {"Bucket": "foo", "AccessControlPolicy": {}}),
    ("put_bucket_logging", {"Bucket": "foo", "BucketLoggingStatus": {}}),
    (
        "put_bucket_notification",
        {"Bucket": "foo", "NotificationConfiguration": {}},
    ),
    ("put_bucket_policy", {"Bucket": "foo", "Policy": "<bucket-policy>"}),
    (
        "put_bucket_request_payment",
        {"Bucket": "foo", "RequestPaymentConfiguration": {"Payer": ""}},
    ),
    (
        "put_bucket_versioning",
        {"Bucket": "foo", "VersioningConfiguration": {}},
    ),
    ("put_bucket_website", {"Bucket": "foo", "WebsiteConfiguration": {}}),
    (
        "put_object_acl",
        {"Bucket": "foo", "Key": "bar", "AccessControlPolicy": {}},
    ),
    (
        "put_object_legal_hold",
        {"Bucket": "foo", "Key": "bar", "LegalHold": {"Status": "ON"}},
    ),
    (
        "put_object_retention",
        {
            "Bucket": "foo",
            "Key": "bar",
            "Retention": {"RetainUntilDate": "2020-11-05"},
        },
    ),
    (
        "put_object_lock_configuration",
        {"Bucket": "foo", "ObjectLockConfiguration": {}},
    ),
]

accesspoint_arn = "arn:aws:s3:us-west-2:123456789012:accesspoint:myendpoint"
accesspoint_arn_cn = (
    "arn:aws-cn:s3:cn-north-1:123456789012:accesspoint:myendpoint"
)
accesspoint_arn_gov = (
    "arn:aws-us-gov:s3:us-gov-west-1:123456789012:accesspoint:myendpoint"
)
accesspoint_cross_region_arn_gov = (
    "arn:aws-us-gov:s3:us-gov-east-1:123456789012:accesspoint:myendpoint"
)


@pytest.mark.parametrize("operation, operation_kwargs", CHECKSUM_TEST_CASES)
def test_checksums_included_in_expected_operations(
    operation, operation_kwargs
):
    """Validate expected calls include Content-MD5 header"""
    client = _create_s3_client()
    with ClientHTTPStubber(client) as stub:
        stub.add_response()
        call = getattr(client, operation)
        call(**operation_kwargs)
        assert "x-amz-checksum-crc32" in stub.requests[-1].headers


@pytest.mark.parametrize(
    "content_encoding, expected_header",
    [("foo", b"foo,aws-chunked"), (None, b"aws-chunked")],
)
def test_checksum_content_encoding(content_encoding, expected_header):
    op_kwargs = {
        "Bucket": "mybucket",
        "Key": "mykey",
        "Body": b"foo",
        "ChecksumAlgorithm": "sha256",
    }
    if content_encoding is not None:
        op_kwargs["ContentEncoding"] = content_encoding
    s3 = _create_s3_client()
    with ClientHTTPStubber(s3) as http_stubber:
        http_stubber.add_response()
        s3.put_object(**op_kwargs)
        request_headers = http_stubber.requests[-1].headers
        assert request_headers["Content-Encoding"] == expected_header


def _s3_addressing_test_cases():
    # The default behavior for sigv2. DNS compatible buckets
    yield dict(
        region="us-west-2",
        bucket="bucket",
        key="key",
        signature_version="s3",
        expected_url="https://bucket.s3.us-west-2.amazonaws.com/key",
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        signature_version="s3",
        expected_url="https://bucket.s3.amazonaws.com/key",
    )
    yield dict(
        region="us-west-1",
        bucket="bucket",
        key="key",
        signature_version="s3",
        expected_url="https://bucket.s3.us-west-1.amazonaws.com/key",
    )
    yield dict(
        region="us-west-1",
        bucket="bucket",
        key="key",
        signature_version="s3",
        is_secure=False,
        expected_url="http://bucket.s3.us-west-1.amazonaws.com/key",
    )

    # Virtual host addressing is independent of signature version.
    yield dict(
        region="us-west-2",
        bucket="bucket",
        key="key",
        signature_version="s3v4",
        expected_url="https://bucket.s3.us-west-2.amazonaws.com/key",
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        signature_version="s3v4",
        expected_url="https://bucket.s3.amazonaws.com/key",
    )
    yield dict(
        region="us-west-1",
        bucket="bucket",
        key="key",
        signature_version="s3v4",
        expected_url="https://bucket.s3.us-west-1.amazonaws.com/key",
    )
    yield dict(
        region="us-west-1",
        bucket="bucket",
        key="key",
        signature_version="s3v4",
        is_secure=False,
        expected_url="http://bucket.s3.us-west-1.amazonaws.com/key",
    )
    yield dict(
        region="us-west-1",
        bucket="bucket-with-num-1",
        key="key",
        signature_version="s3v4",
        is_secure=False,
        expected_url="http://bucket-with-num-1.s3.us-west-1.amazonaws.com/key",
    )

    # Regions outside of the 'aws' partition.
    # These should still default to virtual hosted addressing
    # unless explicitly configured otherwise.
    yield dict(
        region="cn-north-1",
        bucket="bucket",
        key="key",
        signature_version="s3v4",
        expected_url="https://bucket.s3.cn-north-1.amazonaws.com.cn/key",
    )
    # This isn't actually supported because cn-north-1 is sigv4 only,
    # but we'll still double check that our internal logic is correct
    # when building the expected url.
    yield dict(
        region="cn-north-1",
        bucket="bucket",
        key="key",
        signature_version="s3",
        expected_url="https://bucket.s3.cn-north-1.amazonaws.com.cn/key",
    )
    # If the request is unsigned, we should have the default
    # fix_s3_host behavior which is to use virtual hosting where
    # possible but fall back to path style when needed.
    yield dict(
        region="cn-north-1",
        bucket="bucket",
        key="key",
        signature_version=UNSIGNED,
        expected_url="https://bucket.s3.cn-north-1.amazonaws.com.cn/key",
    )
    yield dict(
        region="cn-north-1",
        bucket="bucket.dot",
        key="key",
        signature_version=UNSIGNED,
        expected_url="https://s3.cn-north-1.amazonaws.com.cn/bucket.dot/key",
    )

    # And of course you can explicitly specify which style to use.
    virtual_hosting = {"addressing_style": "virtual"}
    yield dict(
        region="cn-north-1",
        bucket="bucket",
        key="key",
        signature_version=UNSIGNED,
        s3_config=virtual_hosting,
        expected_url="https://bucket.s3.cn-north-1.amazonaws.com.cn/key",
    )

    path_style = {"addressing_style": "path"}
    yield dict(
        region="cn-north-1",
        bucket="bucket",
        key="key",
        signature_version=UNSIGNED,
        s3_config=path_style,
        expected_url="https://s3.cn-north-1.amazonaws.com.cn/bucket/key",
    )

    # If you don't have a DNS compatible bucket, we use path style.
    yield dict(
        region="us-west-2",
        bucket="bucket.dot",
        key="key",
        expected_url="https://s3.us-west-2.amazonaws.com/bucket.dot/key",
    )
    yield dict(
        region="us-east-1",
        bucket="bucket.dot",
        key="key",
        expected_url="https://s3.amazonaws.com/bucket.dot/key",
    )
    yield dict(
        region="us-east-1",
        bucket="BucketName",
        key="key",
        expected_url="https://s3.amazonaws.com/BucketName/key",
    )
    yield dict(
        region="us-west-1",
        bucket="bucket_name",
        key="key",
        expected_url="https://s3.us-west-1.amazonaws.com/bucket_name/key",
    )
    yield dict(
        region="us-west-1",
        bucket="-bucket-name",
        key="key",
        expected_url="https://s3.us-west-1.amazonaws.com/-bucket-name/key",
    )
    yield dict(
        region="us-west-1",
        bucket="bucket-name-",
        key="key",
        expected_url="https://s3.us-west-1.amazonaws.com/bucket-name-/key",
    )
    yield dict(
        region="us-west-1",
        bucket="aa",
        key="key",
        expected_url="https://s3.us-west-1.amazonaws.com/aa/key",
    )
    yield dict(
        region="us-west-1",
        bucket="a" * 64,
        key="key",
        expected_url=(
            "https://s3.us-west-1.amazonaws.com/%s/key" % ("a" * 64)
        ),
    )

    # Custom endpoint url should always be used.
    yield dict(
        customer_provided_endpoint="https://my-custom-s3/",
        bucket="foo",
        key="bar",
        expected_url="https://my-custom-s3/foo/bar",
    )
    yield dict(
        customer_provided_endpoint="https://my-custom-s3/",
        bucket="bucket.dots",
        key="bar",
        expected_url="https://my-custom-s3/bucket.dots/bar",
    )
    # Doesn't matter what region you specify, a custom endpoint url always
    # wins.
    yield dict(
        customer_provided_endpoint="https://my-custom-s3/",
        region="us-west-2",
        bucket="foo",
        key="bar",
        expected_url="https://my-custom-s3/foo/bar",
    )

    # Explicitly configuring "virtual" addressing_style.
    virtual_hosting = {"addressing_style": "virtual"}
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        s3_config=virtual_hosting,
        expected_url="https://bucket.s3.amazonaws.com/key",
    )
    yield dict(
        region="us-west-2",
        bucket="bucket",
        key="key",
        s3_config=virtual_hosting,
        expected_url="https://bucket.s3.us-west-2.amazonaws.com/key",
    )
    yield dict(
        region="eu-central-1",
        bucket="bucket",
        key="key",
        s3_config=virtual_hosting,
        expected_url="https://bucket.s3.eu-central-1.amazonaws.com/key",
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        s3_config=virtual_hosting,
        customer_provided_endpoint="https://foo.amazonaws.com",
        expected_url="https://bucket.foo.amazonaws.com/key",
    )
    yield dict(
        region="unknown",
        bucket="bucket",
        key="key",
        s3_config=virtual_hosting,
        expected_url="https://bucket.s3.unknown.amazonaws.com/key",
    )

    # Test us-gov with virtual addressing.
    yield dict(
        region="us-gov-west-1",
        bucket="bucket",
        key="key",
        s3_config=virtual_hosting,
        expected_url="https://bucket.s3.us-gov-west-1.amazonaws.com/key",
    )

    yield dict(
        region="us-gov-west-1",
        bucket="bucket",
        key="key",
        signature_version="s3",
        expected_url="https://bucket.s3.us-gov-west-1.amazonaws.com/key",
    )
    yield dict(
        region="fips-us-gov-west-1",
        bucket="bucket",
        key="key",
        signature_version="s3",
        expected_url="https://bucket.s3-fips.us-gov-west-1.amazonaws.com/key",
    )

    # Test path style addressing.
    path_style = {"addressing_style": "path"}
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        s3_config=path_style,
        expected_url="https://s3.amazonaws.com/bucket/key",
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        s3_config=path_style,
        customer_provided_endpoint="https://foo.amazonaws.com/",
        expected_url="https://foo.amazonaws.com/bucket/key",
    )
    yield dict(
        region="unknown",
        bucket="bucket",
        key="key",
        s3_config=path_style,
        expected_url="https://s3.unknown.amazonaws.com/bucket/key",
    )

    # S3 accelerate
    use_accelerate = {"use_accelerate_endpoint": True}
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        s3_config=use_accelerate,
        expected_url="https://bucket.s3-accelerate.amazonaws.com/key",
    )
    yield dict(
        # region is ignored with S3 accelerate.
        region="us-west-2",
        bucket="bucket",
        key="key",
        s3_config=use_accelerate,
        expected_url="https://bucket.s3-accelerate.amazonaws.com/key",
    )
    # Provided endpoints still get recognized as accelerate endpoints.
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        customer_provided_endpoint="https://s3-accelerate.amazonaws.com",
        expected_url="https://bucket.s3-accelerate.amazonaws.com/key",
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        customer_provided_endpoint="http://s3-accelerate.amazonaws.com",
        expected_url="http://bucket.s3-accelerate.amazonaws.com/key",
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        s3_config=use_accelerate,
        is_secure=False,
        # Note we're using http://  because is_secure=False.
        expected_url="http://bucket.s3-accelerate.amazonaws.com/key",
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        # s3-accelerate must be the first part of the url.
        customer_provided_endpoint="https://foo.s3-accelerate.amazonaws.com",
        expected_url="https://foo.s3-accelerate.amazonaws.com/bucket/key",
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        # The endpoint must be an Amazon endpoint.
        customer_provided_endpoint="https://s3-accelerate.notamazon.com",
        expected_url="https://s3-accelerate.notamazon.com/bucket/key",
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        # Extra components must be whitelisted.
        customer_provided_endpoint="https://s3-accelerate.foo.amazonaws.com",
        expected_url="https://s3-accelerate.foo.amazonaws.com/bucket/key",
    )
    yield dict(
        region="unknown",
        bucket="bucket",
        key="key",
        s3_config=use_accelerate,
        expected_url="https://bucket.s3-accelerate.amazonaws.com/key",
    )
    # Use virtual even if path is specified for s3 accelerate because
    # path style will not work with S3 accelerate.
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        s3_config={
            "use_accelerate_endpoint": True,
            "addressing_style": "path",
        },
        expected_url="https://bucket.s3-accelerate.amazonaws.com/key",
    )

    # S3 dual stack endpoints.
    use_dualstack = {"use_dualstack_endpoint": True}
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        s3_config=use_dualstack,
        signature_version="s3",
        # Still default to virtual hosted when possible on sigv2.
        expected_url="https://bucket.s3.dualstack.us-east-1.amazonaws.com/key",
    )
    yield dict(
        region=None,
        bucket="bucket",
        key="key",
        s3_config=use_dualstack,
        # Uses us-east-1 for no region set.
        expected_url="https://bucket.s3.dualstack.us-east-1.amazonaws.com/key",
    )
    yield dict(
        region="aws-global",
        bucket="bucket",
        key="key",
        s3_config=use_dualstack,
        # The aws-global pseudo region does not support dualstack and should
        # be resolved to us-east-1.
        expected_url=(
            "https://bucket.s3.dualstack.us-east-1.amazonaws.com/key"
        ),
    )
    yield dict(
        region="us-west-2",
        bucket="bucket",
        key="key",
        s3_config=use_dualstack,
        signature_version="s3",
        # Still default to virtual hosted when possible on sigv2.
        expected_url="https://bucket.s3.dualstack.us-west-2.amazonaws.com/key",
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        s3_config=use_dualstack,
        signature_version="s3v4",
        expected_url="https://bucket.s3.dualstack.us-east-1.amazonaws.com/key",
    )
    yield dict(
        region="us-west-2",
        bucket="bucket",
        key="key",
        s3_config=use_dualstack,
        signature_version="s3v4",
        expected_url="https://bucket.s3.dualstack.us-west-2.amazonaws.com/key",
    )
    yield dict(
        region="unknown",
        bucket="bucket",
        key="key",
        s3_config=use_dualstack,
        signature_version="s3v4",
        expected_url="https://bucket.s3.dualstack.unknown.amazonaws.com/key",
    )
    # Non DNS compatible buckets use path style for dual stack.
    yield dict(
        region="us-west-2",
        bucket="bucket.dot",
        key="key",
        s3_config=use_dualstack,
        # Still default to virtual hosted when possible.
        expected_url=(
            "https://s3.dualstack.us-west-2.amazonaws.com/bucket.dot/key"
        ),
    )
    # Supports is_secure (use_ssl=False in create_client()).
    yield dict(
        region="us-west-2",
        bucket="bucket.dot",
        key="key",
        is_secure=False,
        s3_config=use_dualstack,
        # Still default to virtual hosted when possible.
        expected_url=(
            "http://s3.dualstack.us-west-2.amazonaws.com/bucket.dot/key"
        ),
    )

    # Is path style is requested, we should use it, even if the bucket is
    # DNS compatible.
    force_path_style = {
        "use_dualstack_endpoint": True,
        "addressing_style": "path",
    }
    yield dict(
        region="us-west-2",
        bucket="bucket",
        key="key",
        s3_config=force_path_style,
        # Still default to virtual hosted when possible.
        expected_url="https://s3.dualstack.us-west-2.amazonaws.com/bucket/key",
    )

    # Accelerate + dual stack
    use_accelerate_dualstack = {
        "use_accelerate_endpoint": True,
        "use_dualstack_endpoint": True,
    }
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        s3_config=use_accelerate_dualstack,
        expected_url=(
            "https://bucket.s3-accelerate.dualstack.amazonaws.com/key"
        ),
    )
    yield dict(
        # Region is ignored with S3 accelerate.
        region="us-west-2",
        bucket="bucket",
        key="key",
        s3_config=use_accelerate_dualstack,
        expected_url=(
            "https://bucket.s3-accelerate.dualstack.amazonaws.com/key"
        ),
    )
    # Only s3-accelerate overrides a customer endpoint.
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        s3_config=use_dualstack,
        customer_provided_endpoint="https://s3-accelerate.amazonaws.com",
        expected_url=("https://bucket.s3-accelerate.amazonaws.com/key"),
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        # Dualstack is whitelisted.
        customer_provided_endpoint=(
            "https://s3-accelerate.dualstack.amazonaws.com"
        ),
        expected_url=(
            "https://bucket.s3-accelerate.dualstack.amazonaws.com/key"
        ),
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        # Even whitelisted parts cannot be duplicated.
        customer_provided_endpoint=(
            "https://s3-accelerate.dualstack.dualstack.amazonaws.com"
        ),
        expected_url=(
            "https://s3-accelerate.dualstack.dualstack"
            ".amazonaws.com/bucket/key"
        ),
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        # More than two extra parts is not allowed.
        customer_provided_endpoint=(
            "https://s3-accelerate.dualstack.dualstack.dualstack"
            ".amazonaws.com"
        ),
        expected_url=(
            "https://s3-accelerate.dualstack.dualstack.dualstack.amazonaws.com"
            "/bucket/key"
        ),
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        # Extra components must be whitelisted.
        customer_provided_endpoint="https://s3-accelerate.foo.amazonaws.com",
        expected_url="https://s3-accelerate.foo.amazonaws.com/bucket/key",
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        s3_config=use_accelerate_dualstack,
        is_secure=False,
        # Note we're using http://  because is_secure=False.
        expected_url=(
            "http://bucket.s3-accelerate.dualstack.amazonaws.com/key"
        ),
    )
    # Use virtual even if path is specified for s3 accelerate because
    # path style will not work with S3 accelerate.
    use_accelerate_dualstack["addressing_style"] = "path"
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        s3_config=use_accelerate_dualstack,
        expected_url=(
            "https://bucket.s3-accelerate.dualstack.amazonaws.com/key"
        ),
    )

    # Access-point arn cases
    yield dict(
        region="us-west-2",
        bucket=accesspoint_arn,
        key="key",
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint."
            "us-west-2.amazonaws.com/key"
        ),
    )
    yield dict(
        region="us-west-2",
        bucket=accesspoint_arn,
        key="key",
        s3_config={"use_arn_region": True},
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint."
            "us-west-2.amazonaws.com/key"
        ),
    )
    yield dict(
        region="us-west-2",
        bucket=accesspoint_arn,
        key="myendpoint/key",
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint."
            "us-west-2.amazonaws.com/myendpoint/key"
        ),
    )
    yield dict(
        region="us-west-2",
        bucket=accesspoint_arn,
        key="foo/myendpoint/key",
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint."
            "us-west-2.amazonaws.com/foo/myendpoint/key"
        ),
    )
    yield dict(
        # Note: The access-point arn has us-west-2 and the client's region is
        # us-east-1, for the defauldict the access-point arn region is used.
        region="us-east-1",
        bucket=accesspoint_arn,
        key="key",
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint."
            "us-west-2.amazonaws.com/key"
        ),
    )
    yield dict(
        region="s3-external-1",
        bucket=accesspoint_arn,
        key="key",
        s3_config={"use_arn_region": True},
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint."
            "us-west-2.amazonaws.com/key"
        ),
    )

    yield dict(
        region="aws-global",
        bucket=accesspoint_arn,
        key="key",
        s3_config={"use_arn_region": True},
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint."
            "us-west-2.amazonaws.com/key"
        ),
    )
    yield dict(
        region="unknown",
        bucket=accesspoint_arn,
        key="key",
        s3_config={"use_arn_region": True},
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint."
            "us-west-2.amazonaws.com/key"
        ),
    )
    yield dict(
        region="cn-north-1",
        bucket=accesspoint_arn_cn,
        key="key",
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint."
            "cn-north-1.amazonaws.com.cn/key"
        ),
    )
    yield dict(
        region="cn-northwest-1",
        bucket=accesspoint_arn_cn,
        key="key",
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint."
            "cn-north-1.amazonaws.com.cn/key"
        ),
    )
    yield dict(
        region="us-gov-west-1",
        bucket=accesspoint_arn_gov,
        key="key",
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint."
            "us-gov-west-1.amazonaws.com/key"
        ),
    )
    yield dict(
        region="fips-us-gov-west-1",
        bucket=accesspoint_arn_gov,
        key="key",
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint-fips."
            "us-gov-west-1.amazonaws.com/key"
        ),
    )
    yield dict(
        region="fips-us-gov-west-1",
        bucket=accesspoint_arn_gov,
        key="key",
        s3_config={"use_arn_region": False},
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint-fips."
            "us-gov-west-1.amazonaws.com/key"
        ),
    )
    yield dict(
        region="fips-us-gov-west-1",
        bucket=accesspoint_cross_region_arn_gov,
        s3_config={"use_arn_region": True},
        key="key",
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint-fips."
            "us-gov-east-1.amazonaws.com/key"
        ),
    )
    yield dict(
        region="us-gov-west-1",
        bucket=accesspoint_arn_gov,
        key="key",
        s3_config={"use_arn_region": False},
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint-fips."
            "us-gov-west-1.amazonaws.com/key"
        ),
        use_fips_endpoint=True,
    )
    yield dict(
        region="us-gov-west-1",
        bucket=accesspoint_cross_region_arn_gov,
        key="key",
        s3_config={"use_arn_region": True},
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint-fips."
            "us-gov-east-1.amazonaws.com/key"
        ),
        use_fips_endpoint=True,
    )

    yield dict(
        region="us-west-2",
        bucket=accesspoint_arn,
        key="key",
        is_secure=False,
        expected_url=(
            "http://myendpoint-123456789012.s3-accesspoint."
            "us-west-2.amazonaws.com/key"
        ),
    )
    # Dual-stack with access-point arn
    yield dict(
        # Note: The access-point arn has us-west-2 and the client's region is
        # us-east-1, for the defauldict the access-point arn region is used.
        region="us-east-1",
        bucket=accesspoint_arn,
        key="key",
        s3_config={
            "use_dualstack_endpoint": True,
        },
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint.dualstack."
            "us-west-2.amazonaws.com/key"
        ),
    )
    yield dict(
        region="us-gov-west-1",
        bucket=accesspoint_arn_gov,
        key="key",
        s3_config={
            "use_dualstack_endpoint": True,
        },
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint.dualstack."
            "us-gov-west-1.amazonaws.com/key"
        ),
    )
    yield dict(
        region="fips-us-gov-west-1",
        bucket=accesspoint_arn_gov,
        key="key",
        s3_config={
            "use_arn_region": True,
            "use_dualstack_endpoint": True,
        },
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint-fips.dualstack."
            "us-gov-west-1.amazonaws.com/key"
        ),
    )
    # None of the various s3 settings related to paths should affect what
    # endpoint to use when an access-point is provided.
    yield dict(
        region="us-west-2",
        bucket=accesspoint_arn,
        key="key",
        s3_config={"addressing_style": "auto"},
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint."
            "us-west-2.amazonaws.com/key"
        ),
    )
    yield dict(
        region="us-west-2",
        bucket=accesspoint_arn,
        key="key",
        s3_config={"addressing_style": "virtual"},
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint."
            "us-west-2.amazonaws.com/key"
        ),
    )
    yield dict(
        region="us-west-2",
        bucket=accesspoint_arn,
        key="key",
        s3_config={"addressing_style": "path"},
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint."
            "us-west-2.amazonaws.com/key"
        ),
    )

    # Use us-east-1 regional endpoindicts: regional
    us_east_1_regional_endpoint = {"us_east_1_regional_endpoint": "regional"}
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        s3_config=us_east_1_regional_endpoint,
        expected_url=("https://bucket.s3.us-east-1.amazonaws.com/key"),
    )
    yield dict(
        region="us-west-2",
        bucket="bucket",
        key="key",
        s3_config=us_east_1_regional_endpoint,
        expected_url=("https://bucket.s3.us-west-2.amazonaws.com/key"),
    )
    yield dict(
        region=None,
        bucket="bucket",
        key="key",
        s3_config=us_east_1_regional_endpoint,
        expected_url=("https://bucket.s3.amazonaws.com/key"),
    )
    yield dict(
        region="unknown",
        bucket="bucket",
        key="key",
        s3_config=us_east_1_regional_endpoint,
        expected_url=("https://bucket.s3.unknown.amazonaws.com/key"),
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        s3_config={
            "us_east_1_regional_endpoint": "regional",
            "use_dualstack_endpoint": True,
        },
        expected_url=(
            "https://bucket.s3.dualstack.us-east-1.amazonaws.com/key"
        ),
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        s3_config={
            "us_east_1_regional_endpoint": "regional",
            "use_accelerate_endpoint": True,
        },
        expected_url=("https://bucket.s3-accelerate.amazonaws.com/key"),
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        s3_config={
            "us_east_1_regional_endpoint": "regional",
            "use_accelerate_endpoint": True,
            "use_dualstack_endpoint": True,
        },
        expected_url=(
            "https://bucket.s3-accelerate.dualstack.amazonaws.com/key"
        ),
    )

    # Use us-east-1 regional endpoindicts: legacy
    us_east_1_regional_endpoint_legacy = {
        "us_east_1_regional_endpoint": "legacy"
    }
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        s3_config=us_east_1_regional_endpoint_legacy,
        expected_url=("https://bucket.s3.amazonaws.com/key"),
    )

    yield dict(
        region=None,
        bucket="bucket",
        key="key",
        s3_config=us_east_1_regional_endpoint_legacy,
        expected_url=("https://bucket.s3.amazonaws.com/key"),
    )

    yield dict(
        region="unknown",
        bucket="bucket",
        key="key",
        s3_config=us_east_1_regional_endpoint_legacy,
        expected_url=("https://bucket.s3.unknown.amazonaws.com/key"),
    )

    s3_object_lambda_arn_gov = (
        "arn:aws-us-gov:s3-object-lambda:us-gov-west-1:"
        "123456789012:accesspoint:mybanner"
    )
    yield dict(
        region="fips-us-gov-west-1",
        bucket=s3_object_lambda_arn_gov,
        key="key",
        expected_url=(
            "https://mybanner-123456789012.s3-object-lambda-fips."
            "us-gov-west-1.amazonaws.com/key"
        ),
    )
    yield dict(
        region="us-gov-west-1",
        bucket=s3_object_lambda_arn_gov,
        key="key",
        expected_url=(
            "https://mybanner-123456789012.s3-object-lambda-fips."
            "us-gov-west-1.amazonaws.com/key"
        ),
        use_fips_endpoint=True,
    )
    s3_object_lambda_cross_region_arn_gov = (
        "arn:aws-us-gov:s3-object-lambda:us-gov-east-1:"
        "123456789012:accesspoint:mybanner"
    )
    yield dict(
        region="fips-us-gov-west-1",
        bucket=s3_object_lambda_cross_region_arn_gov,
        key="key",
        s3_config={"use_arn_region": True},
        expected_url=(
            "https://mybanner-123456789012.s3-object-lambda-fips."
            "us-gov-east-1.amazonaws.com/key"
        ),
    )
    yield dict(
        region="us-gov-west-1",
        bucket=s3_object_lambda_cross_region_arn_gov,
        key="key",
        s3_config={"use_arn_region": True},
        expected_url=(
            "https://mybanner-123456789012.s3-object-lambda-fips."
            "us-gov-east-1.amazonaws.com/key"
        ),
        use_fips_endpoint=True,
    )

    s3_object_lambda_arn = (
        "arn:aws:s3-object-lambda:us-east-1:"
        "123456789012:accesspoint:mybanner"
    )
    yield dict(
        region="aws-global",
        bucket=s3_object_lambda_arn,
        key="key",
        s3_config={"use_arn_region": True},
        expected_url=(
            "https://mybanner-123456789012.s3-object-lambda."
            "us-east-1.amazonaws.com/key"
        ),
    )


def _s3_addressing_invalid_test_cases():
    # client region does not match access point ARN region and use_arn_region
    # is False. If sent to service, this results in an "invalid access point"
    # response. We expect it to be caught by the S3 endpoints ruleset.
    yield dict(
        region="us-east-1",
        bucket=accesspoint_arn,
        key="key",
        s3_config={"use_arn_region": False},
        expected_exception_type=UnsupportedS3AccesspointConfigurationError,
        expected_exception_regex=(
            "region from ARN `us-west-2` does not match client region "
            "`us-east-1`"
        ),
    )
    yield dict(
        region="cn-northwest-1",
        bucket=accesspoint_arn_cn,
        key="key",
        s3_config={"use_arn_region": False},
        expected_exception_type=UnsupportedS3AccesspointConfigurationError,
        expected_exception_regex=(
            "region from ARN `cn-north-1` does not match client region "
            "`cn-northwest-1`"
        ),
    )
    yield dict(
        region="unknown",
        bucket=accesspoint_arn,
        key="key",
        s3_config={"use_arn_region": False},
        expected_exception_type=UnsupportedS3AccesspointConfigurationError,
        expected_exception_regex=None,
    )
    yield dict(
        region="us-east-1",
        bucket=accesspoint_arn,
        key="key",
        s3_config={"use_arn_region": False},
        expected_exception_type=UnsupportedS3AccesspointConfigurationError,
        expected_exception_regex=None,
    )


@pytest.mark.parametrize("test_case", _s3_addressing_test_cases())
def test_correct_url_used_for_s3(test_case):
    # Test that given various sets of config options and bucket names,
    # we construct the expect endpoint url.
    _verify_expected_endpoint_url(**test_case)


@pytest.mark.parametrize("test_case", _s3_addressing_invalid_test_cases())
def test_correct_exception_raise_for_s3(test_case):
    # Test that invalid sets of config options and bucket names, result in
    # appropriate exceptions.
    _verify_expected_exception(**test_case)


def _verify_expected_endpoint_url(
    region=None,
    bucket="bucket",
    key="key",
    s3_config=None,
    is_secure=True,
    customer_provided_endpoint=None,
    expected_url=None,
    signature_version=None,
    use_fips_endpoint=None,
):
    s3 = _create_s3_client(
        region=region,
        is_secure=is_secure,
        endpoint_url=customer_provided_endpoint,
        s3_config=s3_config,
        signature_version=signature_version,
        use_fips_endpoint=use_fips_endpoint,
    )
    with ClientHTTPStubber(s3) as http_stubber:
        http_stubber.add_response()
        s3.put_object(Bucket=bucket, Key=key, Body=b"bar")
        assert http_stubber.requests[0].url == expected_url


def _verify_expected_exception(
    expected_exception_type,
    expected_exception_regex=None,
    region=None,
    bucket="bucket",
    key="key",
    s3_config=None,
    is_secure=True,
    customer_provided_endpoint=None,
    signature_version=None,
    use_fips_endpoint=None,
):
    s3 = _create_s3_client(
        region=region,
        is_secure=is_secure,
        endpoint_url=customer_provided_endpoint,
        s3_config=s3_config,
        signature_version=signature_version,
        use_fips_endpoint=use_fips_endpoint,
    )
    with ClientHTTPStubber(s3) as http_stubber:
        http_stubber.add_response()
        with pytest.raises(
            expected_exception_type, match=expected_exception_regex
        ):
            s3.put_object(Bucket=bucket, Key=key, Body=b"bar")


def _create_s3_client(
    region="us-west-2",
    is_secure=True,
    endpoint_url=None,
    s3_config=None,
    signature_version="s3v4",
    use_fips_endpoint=None,
):
    environ = {}
    with mock.patch("os.environ", environ):
        environ["AWS_ACCESS_KEY_ID"] = "access_key"
        environ["AWS_SECRET_ACCESS_KEY"] = "secret_key"
        environ["AWS_CONFIG_FILE"] = "no-exist-foo"
        environ["AWS_SHARED_CREDENTIALS_FILE"] = "no-exist-foo"
        session = create_session()
        session.config_filename = "no-exist-foo"
        config = Config(
            signature_version=signature_version,
            s3=s3_config,
            use_fips_endpoint=use_fips_endpoint,
        )
        s3 = session.create_client(
            "s3",
            region_name=region,
            use_ssl=is_secure,
            config=config,
            endpoint_url=endpoint_url,
        )
        return s3


def _addressing_for_presigned_url_test_cases():
    # us-east-1, or the "global" endpoint. A signature version of
    # None means the user doesn't have signature version configured.
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        signature_version=None,
        expected_url="https://bucket.s3.amazonaws.com/key",
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        signature_version="s3",
        expected_url="https://bucket.s3.amazonaws.com/key",
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        signature_version="s3v4",
        expected_url="https://bucket.s3.amazonaws.com/key",
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        signature_version="s3v4",
        s3_config={"addressing_style": "path"},
        expected_url="https://s3.amazonaws.com/bucket/key",
    )

    # A region that supports both 's3' and 's3v4'.
    yield dict(
        region="us-west-2",
        bucket="bucket",
        key="key",
        signature_version=None,
        expected_url="https://bucket.s3.amazonaws.com/key",
    )
    yield dict(
        region="us-west-2",
        bucket="bucket",
        key="key",
        signature_version="s3",
        expected_url="https://bucket.s3.amazonaws.com/key",
    )
    yield dict(
        region="us-west-2",
        bucket="bucket",
        key="key",
        signature_version="s3v4",
        expected_url="https://bucket.s3.amazonaws.com/key",
    )
    yield dict(
        region="us-west-2",
        bucket="bucket",
        key="key",
        signature_version="s3v4",
        s3_config={"addressing_style": "path"},
        expected_url="https://s3.us-west-2.amazonaws.com/bucket/key",
    )

    # An 's3v4' only region.
    yield dict(
        region="us-east-2",
        bucket="bucket",
        key="key",
        signature_version=None,
        expected_url="https://bucket.s3.amazonaws.com/key",
    )
    yield dict(
        region="us-east-2",
        bucket="bucket",
        key="key",
        signature_version="s3",
        expected_url="https://bucket.s3.amazonaws.com/key",
    )
    yield dict(
        region="us-east-2",
        bucket="bucket",
        key="key",
        signature_version="s3v4",
        expected_url="https://bucket.s3.amazonaws.com/key",
    )
    yield dict(
        region="us-east-2",
        bucket="bucket",
        key="key",
        signature_version="s3v4",
        s3_config={"addressing_style": "path"},
        expected_url="https://s3.us-east-2.amazonaws.com/bucket/key",
    )

    # Dualstack endpoints
    yield dict(
        region="us-west-2",
        bucket="bucket",
        key="key",
        signature_version=None,
        s3_config={"use_dualstack_endpoint": True},
        expected_url="https://bucket.s3.dualstack.us-west-2.amazonaws.com/key",
    )
    yield dict(
        region="us-west-2",
        bucket="bucket",
        key="key",
        signature_version="s3",
        s3_config={"use_dualstack_endpoint": True},
        expected_url="https://bucket.s3.dualstack.us-west-2.amazonaws.com/key",
    )
    yield dict(
        region="us-west-2",
        bucket="bucket",
        key="key",
        signature_version="s3v4",
        s3_config={"use_dualstack_endpoint": True},
        expected_url="https://bucket.s3.dualstack.us-west-2.amazonaws.com/key",
    )

    # Accelerate
    yield dict(
        region="us-west-2",
        bucket="bucket",
        key="key",
        signature_version=None,
        s3_config={"use_accelerate_endpoint": True},
        expected_url="https://bucket.s3-accelerate.amazonaws.com/key",
    )

    # A region that we don't know about.
    yield dict(
        region="boto-west-1",
        bucket="bucket",
        key="key",
        signature_version=None,
        expected_url="https://bucket.s3.amazonaws.com/key",
    )

    # Customer provided URL results in us leaving the host untouched.
    yield dict(
        region="us-west-2",
        bucket="bucket",
        key="key",
        signature_version=None,
        customer_provided_endpoint="https://foo.com/",
        expected_url="https://foo.com/bucket/key",
    )

    # Access-point
    yield dict(
        region="us-west-2",
        bucket=accesspoint_arn,
        key="key",
        expected_url=(
            "https://myendpoint-123456789012.s3-accesspoint."
            "us-west-2.amazonaws.com/key"
        ),
    )

    # Use us-east-1 regional endpoint configuration cases
    us_east_1_regional_endpoint = {"us_east_1_regional_endpoint": "regional"}
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        s3_config=us_east_1_regional_endpoint,
        signature_version="s3",
        expected_url=("https://bucket.s3.us-east-1.amazonaws.com/key"),
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        s3_config=us_east_1_regional_endpoint,
        signature_version="s3v4",
        expected_url=("https://bucket.s3.us-east-1.amazonaws.com/key"),
    )
    # Bucket names that contain dots or are otherwise not virtual host style
    # compatible should always resolve to a regional endpoint.
    # https://github.com/boto/botocore/issues/2798
    yield dict(
        region="us-west-1",
        bucket="foo.bar.biz",
        key="key",
        signature_version="s3",
        expected_url="https://s3.us-west-1.amazonaws.com/foo.bar.biz/key",
    )
    # Bucket names that contain dots and subcomponents that are less than
    # 3 characters should still use virtual host style addressing if
    # configured by the customer and they provide their own ``endpoint_url``
    # that is insecure. https://github.com/boto/botocore/issues/2938
    yield dict(
        bucket="foo.b.biz",
        key="key",
        s3_config={"addressing_style": "virtual"},
        customer_provided_endpoint="http://s3.us-west-2.amazonaws.com",
        expected_url="http://foo.b.biz.s3.us-west-2.amazonaws.com/key",
    )
    yield dict(
        bucket="foo.b.biz",
        key="key",
        s3_config={"addressing_style": "virtual"},
        customer_provided_endpoint="https://s3.us-west-2.amazonaws.com",
        expected_url="https://s3.us-west-2.amazonaws.com/foo.b.biz/key",
    )

    # virtual style addressing expicitly requested always uses
    # regional endpoints except for us-east-1 and aws-global
    yield dict(
        region="us-west-2",
        bucket="bucket",
        key="key",
        signature_version="s3",
        s3_config={"addressing_style": "virtual"},
        expected_url="https://bucket.s3.us-west-2.amazonaws.com/key",
    )
    yield dict(
        region="us-east-2",
        bucket="bucket",
        key="key",
        signature_version="s3v4",
        s3_config={"addressing_style": "virtual"},
        expected_url="https://bucket.s3.us-east-2.amazonaws.com/key",
    )
    yield dict(
        region="us-west-2",
        bucket="bucket",
        key="key",
        s3_config={"addressing_style": "virtual"},
        expected_url="https://bucket.s3.us-west-2.amazonaws.com/key",
    )
    yield dict(
        region="us-east-1",
        bucket="bucket",
        key="key",
        s3_config={"addressing_style": "virtual"},
        expected_url="https://bucket.s3.amazonaws.com/key",
    )
    yield dict(
        region="aws-global",
        bucket="bucket",
        key="key",
        s3_config={"addressing_style": "virtual"},
        expected_url="https://bucket.s3.amazonaws.com/key",
    )


@pytest.mark.parametrize(
    "test_case", _addressing_for_presigned_url_test_cases()
)
def test_addressing_for_presigned_urls(test_case):
    # Here we're just focusing on the addressing mode used for presigned URLs.
    # We special case presigned URLs due to backward compatibility.
    _verify_presigned_url_addressing(**test_case)


def _verify_presigned_url_addressing(
    region=None,
    bucket="bucket",
    key="key",
    s3_config=None,
    is_secure=True,
    customer_provided_endpoint=None,
    expected_url=None,
    signature_version=None,
):
    s3 = _create_s3_client(
        region=region,
        is_secure=is_secure,
        endpoint_url=customer_provided_endpoint,
        s3_config=s3_config,
        signature_version=signature_version,
    )
    url = s3.generate_presigned_url(
        "get_object", {"Bucket": bucket, "Key": key}
    )
    # We're not trying to verify the params for URL presigning,
    # those are tested elsewhere.  We just care about the hostname/path.
    parts = urlsplit(url)
    actual = "{}://{}{}".format(*parts[:3])
    assert actual == expected_url


class TestS3XMLPayloadEscape(BaseS3OperationTest):
    def assert_correct_crc32_checksum(self, request):
        checksum = get_checksum_cls()()
        crc32_checksum = checksum.handle(request.body).encode()
        self.assertEqual(
            crc32_checksum, request.headers["x-amz-checksum-crc32"]
        )

    def test_escape_keys_in_xml_delete_objects(self):
        self.http_stubber.add_response()
        with self.http_stubber:
            self.client.delete_objects(
                Bucket="mybucket",
                Delete={"Objects": [{"Key": "some\r\n\rkey"}]},
            )
        request = self.http_stubber.requests[0]
        self.assertNotIn(b"\r\n\r", request.body)
        self.assertIn(b"&#xD;&#xA;&#xD;", request.body)
        self.assert_correct_crc32_checksum(request)

    def test_escape_keys_in_xml_put_bucket_lifecycle_configuration(self):
        self.http_stubber.add_response()
        with self.http_stubber:
            self.client.put_bucket_lifecycle_configuration(
                Bucket="mybucket",
                LifecycleConfiguration={
                    "Rules": [
                        {
                            "Prefix": "my\r\n\rprefix",
                            "Status": "ENABLED",
                        }
                    ]
                },
            )
        request = self.http_stubber.requests[0]
        self.assertNotIn(b"my\r\n\rprefix", request.body)
        self.assertIn(b"my&#xD;&#xA;&#xD;prefix", request.body)
        self.assert_correct_crc32_checksum(request)


class TestExpectContinueBehavior(BaseSessionTest):
    def test_sets_100_continute_with_body(self):
        op_kwargs = {
            "Bucket": "mybucket",
            "Key": "mykey",
            "Body": b"foo",
        }
        s3 = _create_s3_client()
        with ClientHTTPStubber(s3) as http_stubber:
            http_stubber.add_response()
            s3.put_object(**op_kwargs)
            expect_header = http_stubber.requests[-1].headers.get("Expect")
            self.assertIsNotNone(expect_header)
            self.assertEqual(expect_header, b"100-continue")

    def test_does_not_set_100_continute_with_empty_body(self):
        environ = {'BOTO_EXPERIMENTAL__NO_EMPTY_CONTINUE': "True"}
        self.environ_patch = mock.patch('os.environ', environ)
        self.environ_patch.start()
        op_kwargs = {"Bucket": "mybucket", "Key": "mykey", "Body": ""}
        s3 = _create_s3_client()
        with ClientHTTPStubber(s3) as http_stubber:
            http_stubber.add_response()
            s3.put_object(**op_kwargs)
            expect_header = http_stubber.requests[-1].headers.get("Expect")
            self.assertIsNone(expect_header)


class TestParameterInjection(BaseS3OperationTest):
    BUCKET = "foo"
    KEY = "bar"

    def test_parameter_injection(self):
        self.http_stubber.add_response()
        self.client.meta.events.register(
            'before-sign.s3', self._verify_bucket_and_key_in_context
        )
        with self.http_stubber:
            self.client.put_object(
                Bucket=self.BUCKET,
                Key=self.KEY,
            )

    def _verify_bucket_and_key_in_context(self, request, **kwargs):
        self.assertEqual(
            request.context['input_params']['Bucket'], self.BUCKET
        )
        self.assertEqual(request.context['input_params']['Key'], self.KEY)


@pytest.mark.parametrize(
    "bucket, key, expected_path, expected_hostname",
    [
        (
            "mybucket",
            "../key.txt",
            "/../key.txt",
            "mybucket.s3.us-west-2.amazonaws.com",
        ),
        (
            "mybucket",
            "foo/../key.txt",
            "/foo/../key.txt",
            "mybucket.s3.us-west-2.amazonaws.com",
        ),
        (
            "mybucket",
            "foo/../../key.txt",
            "/foo/../../key.txt",
            "mybucket.s3.us-west-2.amazonaws.com",
        ),
    ],
)
def test_dot_segments_preserved_in_url_path(
    patched_session, bucket, key, expected_path, expected_hostname
):
    s3 = patched_session.create_client(
        's3',
        'us-west-2',
        config=Config(
            s3={"addressing_style": "virtual"},
        ),
    )
    with ClientHTTPStubber(s3) as http_stubber:
        http_stubber.add_response()
        s3.get_object(Bucket=bucket, Key=key)
        url_parts = urlsplit(http_stubber.requests[0].url)
        assert url_parts.path == expected_path
        assert url_parts.hostname == expected_hostname