Replace S3 remote signer headers instead of appending them

pyiceberg/io/fsspec.py

@@ -160,6 +160,8 @@ def __call__(self, request: "AWSRequest", operation_name: str | None = None, **_
             raise SignError(f"Failed to sign request {response.status_code}: {signer_body}") from e
 
         for key, value in response_json["headers"].items():
+            # Replace instead of append, so a header the request already has is not sent twice
+            del request.headers[key]
             request.headers.add_header(key, ", ".join(value))
 
         request.url = response_json["uri"]

tests/io/test_fsspec.py

@@ -1030,6 +1030,42 @@ def test_s3v4_rest_signer(requests_mock: Mocker) -> None:
     }
 
 
+def test_s3v4_rest_signer_replaces_existing_headers(requests_mock: Mocker) -> None:
+    requests_mock.post(
+        f"{TEST_URI}/v1/aws/s3/sign",
+        json={
+            "uri": "https://bucket/metadata/file.avro",
+            "headers": {
+                "Authorization": ["AWS4-HMAC-SHA256 Credential=..., SignedHeaders=..., Signature=..."],
+                "x-amz-content-sha256": ["UNSIGNED-PAYLOAD"],
+                "User-Agent": ["Botocore/1.27.59"],
+            },
+            "extensions": {},
+        },
+        status_code=200,
+    )
+
+    request = AWSRequest(
+        method="PUT",
+        url="https://bucket/metadata/file.avro",
+        headers={
+            "User-Agent": "Botocore/1.27.59",
+            "X-Amz-Content-SHA256": "STREAMING-UNSIGNED-PAYLOAD-TRAILER",
+        },
+        data=b"",
+        params={},
+        auth_path="/metadata/file.avro",
+    )
+    request.context = {"client_region": "us-west-2"}
+
+    signer = S3V4RestSigner(properties={"token": "abc", "uri": TEST_URI})
+    signer(request)
+
+    assert request.headers.get_all("x-amz-content-sha256") == ["UNSIGNED-PAYLOAD"]
+    assert request.headers.get_all("User-Agent") == ["Botocore/1.27.59"]
+    assert len(request.headers.get_all("Authorization")) == 1
+
+
 @pytest.mark.s3
 def test_s3v4_rest_signer_write_read_delete(request: pytest.FixtureRequest, requests_mock: Mocker) -> None:
     """Test writing, reading, and deleting an S3 file through a REST signing service"""