Allow custom endpoints when addressing ARN resources

botocore/utils.py

@@ -1543,13 +1543,6 @@ def _use_accesspoint_endpoint(self, request):
         return 's3_accesspoint' in request.context
 
     def _validate_accesspoint_supported(self, request):
-        if self._endpoint_url:
-            raise UnsupportedS3AccesspointConfigurationError(
-                msg=(
-                    'Client cannot use a custom "endpoint_url" when '
-                    'specifying an access-point ARN.'
-                )
-            )
         if self._use_accelerate_endpoint:
             raise UnsupportedS3AccesspointConfigurationError(
                 msg=(
@@ -1609,19 +1602,26 @@ def _get_accesspoint_netloc(self, request_context, region_name):
         accesspoint_netloc_components = [
             '%s-%s' % (s3_accesspoint['name'], s3_accesspoint['account']),
         ]
-        if 'outpost_name' in s3_accesspoint:
-            outpost_host = [s3_accesspoint['outpost_name'], 's3-outposts']
-            accesspoint_netloc_components.extend(outpost_host)
+        outpost_name = s3_accesspoint.get('outpost_name')
+        if self._endpoint_url:
+            if outpost_name:
+                accesspoint_netloc_components.append(outpost_name)
+            endpoint_url_netloc = urlsplit(self._endpoint_url).netloc
+            accesspoint_netloc_components.append(endpoint_url_netloc)
         else:
-            accesspoint_netloc_components.append('s3-accesspoint')
-        if self._s3_config.get('use_dualstack_endpoint'):
-            accesspoint_netloc_components.append('dualstack')
-        accesspoint_netloc_components.extend(
-            [
-                region_name,
-                self._get_dns_suffix(region_name)
-            ]
-        )
+            if outpost_name:
+                outpost_host = [outpost_name, 's3-outposts']
+                accesspoint_netloc_components.extend(outpost_host)
+            else:
+                accesspoint_netloc_components.append('s3-accesspoint')
+            if self._s3_config.get('use_dualstack_endpoint'):
+                accesspoint_netloc_components.append('dualstack')
+            accesspoint_netloc_components.extend(
+                [
+                    region_name,
+                    self._get_dns_suffix(region_name)
+                ]
+            )
         return '.'.join(accesspoint_netloc_components)
 
     def _get_accesspoint_path(self, original_path, request_context):
@@ -1770,7 +1770,6 @@ def _use_endpoint_from_outpost_id(self, request):
         return 'outpost_id' in request.context
 
     def _validate_endpoint_from_arn_details_supported(self, request):
-        self._validate_no_custom_endpoint()
         if not self._s3_config.get('use_arn_region', False):
             arn_region = request.context['arn_details']['region']
             if arn_region != self._region:
@@ -1797,17 +1796,7 @@ def _validate_endpoint_from_arn_details_supported(self, request):
         if 'outpost_name' in request.context['arn_details']:
             self._validate_outpost_redirection_valid(request)
 
-    def _validate_no_custom_endpoint(self):
-        if self._endpoint_url:
-            raise UnsupportedS3ControlConfigurationError(
-                msg=(
-                    'Client cannot use a custom "endpoint_url" when '
-                    'specifying a resource ARN.'
-                )
-            )
-
     def _validate_outpost_redirection_valid(self, request):
-        self._validate_no_custom_endpoint()
         if self._s3_config.get('use_dualstack_endpoint'):
             raise UnsupportedS3ControlConfigurationError(
                 msg=(
@@ -1865,22 +1854,29 @@ def _validate_host_labels(self, *labels):
 
     def _construct_s3_control_endpoint(self, region_name, account):
         self._validate_host_labels(region_name, account)
-        netloc = [
-            account,
-            's3-control',
-        ]
-        self._add_dualstack(netloc)
-        dns_suffix = self._get_dns_suffix(region_name)
-        netloc.extend([region_name, dns_suffix])
+        if self._endpoint_url:
+            endpoint_url_netloc = urlsplit(self._endpoint_url).netloc
+            netloc = [account, endpoint_url_netloc]
+        else:
+            netloc = [
+                account,
+                's3-control',
+            ]
+            self._add_dualstack(netloc)
+            dns_suffix = self._get_dns_suffix(region_name)
+            netloc.extend([region_name, dns_suffix])
         return self._construct_netloc(netloc)
 
     def _construct_outpost_endpoint(self, region_name):
         self._validate_host_labels(region_name)
-        netloc = [
-            's3-outposts',
-            region_name,
-            self._get_dns_suffix(region_name),
-        ]
+        if self._endpoint_url:
+            return urlsplit(self._endpoint_url).netloc
+        else:
+            netloc = [
+                's3-outposts',
+                region_name,
+                self._get_dns_suffix(region_name),
+            ]
         return self._construct_netloc(netloc)
 
     def _construct_netloc(self, netloc):

tests/functional/test_s3.py

@@ -562,12 +562,24 @@ def test_accesspoint_arn_with_custom_endpoint(self):
         accesspoint_arn = (
             'arn:aws:s3:us-west-2:123456789012:accesspoint:myendpoint'
         )
-        self.client, _ = self.create_stubbed_s3_client(
+        self.client, http_stubber = self.create_stubbed_s3_client(
             endpoint_url='https://custom.com')
-        with self.assertRaises(
-                botocore.exceptions.
-                UnsupportedS3AccesspointConfigurationError):
-            self.client.list_objects(Bucket=accesspoint_arn)
+        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 = (
@@ -745,6 +757,24 @@ def test_basic_outpost_arn(self):
         )
         self.assert_endpoint(request, expected_endpoint)
 
+    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(
+            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_with_s3_accelerate(self):
         outpost_arn = (
             'arn:aws:s3-outposts:us-west-2:123456789012:outpost:'

tests/functional/test_s3_control_redirects.py

@@ -420,16 +420,24 @@ def test_outpost_id_redirection_list_regional_buckets(self):
     def test_outpost_redirection_custom_endpoint(self):
         self._bootstrap_client(endpoint_url='https://outpost.foo.com/')
         self.stubber.add_response()
-        with self.assertRaises(UnsupportedS3ControlConfigurationError):
-            with self.stubber:
-                self.client.create_bucket(Bucket='foo', OutpostId='op-123')
+        with self.stubber:
+            self.client.create_bucket(Bucket='foo', OutpostId='op-123')
+        _assert_netloc(self.stubber, 'outpost.foo.com')
+        _assert_header(self.stubber, 'x-amz-outpost-id', 'op-123')
 
     def test_normal_ap_request_has_correct_endpoint(self):
         self.stubber.add_response()
         with self.stubber:
             self.client.get_access_point_policy(Name='MyAp', AccountId='1234')
         _assert_netloc(self.stubber, '1234.s3-control.us-west-2.amazonaws.com')
 
+    def test_normal_ap_request_custom_endpoint(self):
+        self._bootstrap_client(endpoint_url='https://example.com/')
+        self.stubber.add_response()
+        with self.stubber:
+            self.client.get_access_point_policy(Name='MyAp', AccountId='1234')
+        _assert_netloc(self.stubber, '1234.example.com')
+
     def test_normal_bucket_request_has_correct_endpoint(self):
         self.stubber.add_response()
         with self.stubber:

tests/unit/test_utils.py

@@ -2002,12 +2002,15 @@ def test_uses_region_of_client_if_use_arn_disabled(self):
         )
         self.assertEqual(request.url, expected_url)
 
-    def test_accesspoint_errors_for_custom_endpoint(self):
+    def test_accesspoint_supports_custom_endpoint(self):
         endpoint_setter = self.get_endpoint_setter(
             endpoint_url='https://custom.com')
         request = self.get_s3_accesspoint_request()
-        with self.assertRaises(UnsupportedS3AccesspointConfigurationError):
-            self.call_set_endpoint(endpoint_setter, request=request)
+        self.call_set_endpoint(endpoint_setter, request=request)
+        expected_url = 'https://%s-%s.custom.com/' % (
+            self.accesspoint_name, self.account,
+        )
+        self.assertEqual(request.url, expected_url)
 
     def test_errors_for_mismatching_partition(self):
         endpoint_setter = self.get_endpoint_setter(partition='aws-cn')