Allow custom s3 endpoint_url for use with boto3[crt]

boto3/crt.py

@@ -53,10 +53,17 @@ def _create_crt_client(session, config, region_name, cred_provider):
     return create_s3_crt_client(**create_crt_client_kwargs)
 
 
-def _create_crt_request_serializer(session, region_name):
-    return BotocoreCRTRequestSerializer(
-        session, {'region_name': region_name, 'endpoint_url': None}
-    )
+def _create_crt_request_serializer(session, region_name, endpoint_url=None):
+    from botocore.config import Config
+
+    client_kwargs = {'region_name': region_name, 'endpoint_url': endpoint_url}
+
+    # When using custom endpoints, force path-style addressing to avoid
+    # invalid DNS names (e.g., bucket.custom-endpoint.com)
+    if endpoint_url is not None:
+        client_kwargs['config'] = Config(s3={'addressing_style': 'path'})
+
+    return BotocoreCRTRequestSerializer(session, client_kwargs)
 
 
 def _create_crt_s3_client(
@@ -84,8 +91,11 @@ def _initialize_crt_transfer_primatives(client, config):
     session = Session()
     region_name = client.meta.region_name
     credentials = client._get_credentials()
+    endpoint_url = client.meta.endpoint_url
 
-    serializer = _create_crt_request_serializer(session, region_name)
+    serializer = _create_crt_request_serializer(
+        session, region_name, endpoint_url
+    )
     s3_client = _create_crt_s3_client(
         session, config, region_name, credentials, lock
     )
@@ -149,10 +159,14 @@ def compare_identity(boto3_creds, crt_s3_creds):
     except botocore.exceptions.NoCredentialsError:
         return False
 
+    # Normalize tokens: treat empty string and None as equivalent
+    boto3_token = boto3_creds.token if boto3_creds.token else None
+    crt_token = crt_creds.session_token if crt_creds.session_token else None
+
     is_matching_identity = (
         boto3_creds.access_key == crt_creds.access_key_id
         and boto3_creds.secret_key == crt_creds.secret_access_key
-        and boto3_creds.token == crt_creds.session_token
+        and boto3_token == crt_token
     )
     return is_matching_identity
 

tests/unit/test_crt.py

@@ -59,13 +59,16 @@ def mock_serializer_singleton(monkeypatch):
     yield None
 
 
-def create_test_client(service_name='s3', region_name="us-east-1"):
+def create_test_client(
+    service_name='s3', region_name="us-east-1", endpoint_url=None
+):
     return boto3.client(
         service_name,
         region_name=region_name,
         aws_access_key_id="access",
         aws_secret_access_key="secret",
         aws_session_token="token",
+        endpoint_url=endpoint_url,
     )
 
 
@@ -157,6 +160,27 @@ def test_create_crt_transfer_manager_w_client_in_wrong_region(
                 ("access", "secret", "notoken"),
                 False,
             ),
+            # Token normalization: None and empty string should be treated as equivalent
+            (
+                ("access", "secret", None),
+                ("access", "secret", ""),
+                True,
+            ),
+            (
+                ("access", "secret", ""),
+                ("access", "secret", None),
+                True,
+            ),
+            (
+                ("access", "secret", None),
+                ("access", "secret", None),
+                True,
+            ),
+            (
+                ("access", "secret", ""),
+                ("access", "secret", ""),
+                True,
+            ),
         ),
     )
     @requires_crt()
@@ -213,3 +237,71 @@ def test_get_crt_s3_client_w_wrong_region(
         )
         assert use1_crt_s3_client is crt_s3_client
         assert use1_crt_s3_client.region == "us-west-2"
+
+    @requires_crt()
+    def test_create_crt_transfer_manager_with_custom_endpoint(
+        self,
+        mock_crt_process_lock,
+        mock_crt_client_singleton,
+        mock_serializer_singleton,
+    ):
+        """Test that custom endpoints are properly passed through to the serializer"""
+        custom_endpoint = "https://custom-s3.example.com"
+        custom_client = create_test_client(
+            region_name="us-west-2", endpoint_url=custom_endpoint
+        )
+
+        tm = boto3.crt.create_crt_transfer_manager(custom_client, None)
+        assert isinstance(tm, s3transfer.crt.CRTTransferManager)
+
+        # Verify the serializer was created with the custom endpoint
+        serializer = boto3.crt.BOTOCORE_CRT_SERIALIZER
+        assert serializer is not None
+        # Check that the client_kwargs passed to the serializer includes the endpoint_url
+        assert hasattr(serializer, '_client')
+        # The serializer's internal client should have the custom endpoint
+        assert serializer._client.meta.endpoint_url == custom_endpoint
+
+    @requires_crt()
+    def test_create_crt_request_serializer_with_custom_endpoint(
+        self,
+        mock_crt_process_lock,
+        mock_crt_client_singleton,
+        mock_serializer_singleton,
+    ):
+        """Test that _create_crt_request_serializer properly handles custom endpoints"""
+        from botocore.session import Session
+
+        custom_endpoint = "https://custom-s3.example.com"
+        session = Session()
+
+        serializer = boto3.crt._create_crt_request_serializer(
+            session, "us-west-2", endpoint_url=custom_endpoint
+        )
+
+        # Verify serializer was created and has the custom endpoint in its client
+        assert serializer is not None
+        assert hasattr(serializer, '_client')
+        assert serializer._client.meta.endpoint_url == custom_endpoint
+        # Verify path-style addressing is configured for custom endpoints
+        assert serializer._client.meta.config.s3['addressing_style'] == 'path'
+
+    @requires_crt()
+    def test_create_crt_request_serializer_without_custom_endpoint(
+        self,
+        mock_crt_process_lock,
+        mock_crt_client_singleton,
+        mock_serializer_singleton,
+    ):
+        """Test that _create_crt_request_serializer works without custom endpoint (default behavior)"""
+        from botocore.session import Session
+
+        session = Session()
+
+        serializer = boto3.crt._create_crt_request_serializer(
+            session, "us-west-2", endpoint_url=None
+        )
+
+        # Verify serializer was created
+        assert serializer is not None
+        assert hasattr(serializer, '_client')