Fix REST signing for S3 bulk deletes

pyiceberg/io/fsspec.py

@@ -29,7 +29,7 @@
     TYPE_CHECKING,
     Any,
 )
-from urllib.parse import ParseResult, urlparse
+from urllib.parse import ParseResult, parse_qs, urlparse
 
 import requests
 from fsspec import AbstractFileSystem
@@ -148,6 +148,9 @@ def __call__(self, request: "AWSRequest", **_: Any) -> None:
             "uri": request.url,
             "headers": {key: [val] for key, val in request.headers.items()},
         }
+        if request.method == "POST" and "delete" in parse_qs(urlparse(request.url).query, keep_blank_values=True):
+            if body := request.body:
+                signer_body["body"] = body.decode("utf-8")
 
         response = self._session.post(f"{signer_url}/{signer_endpoint.strip()}", headers=signer_headers, json=signer_body)
         try:

tests/io/test_fsspec.py

@@ -20,6 +20,7 @@
 import tempfile
 import threading
 import uuid
+from io import BytesIO
 from unittest import mock
 
 import pytest
@@ -995,6 +996,49 @@ def test_s3v4_rest_signer(requests_mock: Mocker) -> None:
     }
 
 
+@pytest.mark.parametrize("query", ["delete", "delete=", "delete=&x-id=DeleteObjects"])
+def test_s3v4_rest_signer_sends_delete_objects_body(requests_mock: Mocker, query: str) -> None:
+    body = "<Delete><Object><Key>data/space + percent%2F-snowman-☃</Key></Object></Delete>"
+    uri = f"https://bucket.s3.us-west-2.amazonaws.com/?{query}"
+    request = AWSRequest(method="POST", url=uri, data=body.encode("utf-8"), headers={"x-amz-checksum-crc32": "AAAAAA=="})
+    request.context["client_region"] = "us-west-2"
+    requests_mock.post(f"{TEST_URI}/v1/aws/s3/sign", json={"uri": uri, "headers": {}})
+
+    S3V4RestSigner(properties={"uri": TEST_URI})(request)
+
+    assert requests_mock.last_request is not None
+    signed_request = requests_mock.last_request.json()
+    assert signed_request["body"] == body
+    assert signed_request["headers"]["x-amz-checksum-crc32"] == ["AAAAAA=="]
+    assert request.body == body.encode("utf-8")
+
+
+@pytest.mark.parametrize(
+    "method,query",
+    [
+        ("PUT", "delete"),
+        ("POST", "uploads"),
+        ("POST", "uploadId=upload"),
+        ("GET", "delete"),
+        ("POST", "undelete"),
+        ("POST", "key=delete"),
+    ],
+)
+def test_s3v4_rest_signer_does_not_read_other_bodies(requests_mock: Mocker, method: str, query: str) -> None:
+    body = BytesIO(b"\x00\xffobject data")
+    uri = f"https://bucket.s3.us-west-2.amazonaws.com/key?{query}"
+    request = AWSRequest(method=method, url=uri, data=body)
+    request.context["client_region"] = "us-west-2"
+    requests_mock.post(f"{TEST_URI}/v1/aws/s3/sign", json={"uri": uri, "headers": {}})
+
+    S3V4RestSigner(properties={"uri": TEST_URI})(request)
+
+    assert requests_mock.last_request is not None
+    assert "body" not in requests_mock.last_request.json()
+    assert request.body is body
+    assert body.tell() == 0
+
+
 def test_s3v4_rest_signer_endpoint(requests_mock: Mocker) -> None:
     new_uri = "https://other-bucket/metadata/snap-8048355899640248710-1-a5c8ea2d-aa1f-48e8-89f4-1fa69db8c742.avro"
     endpoint = "v1/main/s3-sign/foo.bar?e=e&b=b&k=k=k&s=s&w=w"