fix(aws): omit static key/secret/token for RefreshableCredentials in …

dlt/common/configuration/specs/aws_credentials.py

@@ -131,6 +131,21 @@ def to_session_credentials(self) -> Dict[str, str]:
             )
         return super().to_session_credentials()
 
+    def to_s3fs_credentials(self) -> Dict[str, Optional[str]]:
+        """Omit static key/secret/token when underlying provider is refreshable
+        so s3fs uses its own aiobotocore default chain and rotates the token."""
+        creds = super().to_s3fs_credentials()
+        if self.has_default_credentials():
+            try:
+                from botocore.credentials import RefreshableCredentials
+
+                if isinstance(self.default_credentials().get_credentials(), RefreshableCredentials):
+                    for k in ("key", "secret", "token"):
+                        creds.pop(k, None)
+            except ImportError:
+                pass
+        return creds
+
     def to_sts_credentials(self) -> Dict[str, str]:
         """Return session credentials with a session token, generating one via STS if needed."""
         sess_creds = self.to_session_credentials()

tests/load/filesystem/test_aws_credentials.py

@@ -161,6 +161,42 @@ def test_aws_credentials_with_endpoint_url(environment: Dict[str, str]) -> None:
     assert "config_kwargs" in s3fs_creds
 
 
+def test_aws_credentials_to_s3fs_omits_refreshable_token() -> None:
+    """RefreshableCredentials must not freeze into s3fs static kwargs (#4003)."""
+    import botocore.session
+    from botocore.credentials import RefreshableCredentials
+
+    def _refresh() -> Dict[str, str]:
+        return {
+            "access_key": "refreshable_access_key",
+            "secret_key": "refreshable_secret_key",
+            "token": "refreshable_session_token",
+            "expiry_time": "2099-01-01T00:00:00Z",
+        }
+
+    refreshable = RefreshableCredentials.create_from_metadata(
+        metadata=_refresh(),
+        refresh_using=_refresh,
+        method="custom",
+    )
+
+    session = botocore.session.get_session()
+    session._credentials = refreshable
+    session.set_config_variable("region", "eu-central-1")
+
+    c = AwsCredentials.from_session(session)
+    assert c.has_default_credentials()
+
+    s3fs_creds = c.to_s3fs_credentials()
+    # static key/secret/token must NOT be present so s3fs uses its own
+    # refreshable default chain instead of frozen strings.
+    assert "key" not in s3fs_creds
+    assert "secret" not in s3fs_creds
+    assert "token" not in s3fs_creds
+    # non-credential kwargs are preserved.
+    assert s3fs_creds["client_kwargs"] == {"region_name": "eu-central-1"}
+
+
 def test_explicit_filesystem_credentials() -> None:
     import dlt
     from dlt.destinations import filesystem