Fix REST signing for S3 bulk deletes

pyiceberg/io/fsspec.py

@@ -125,7 +125,7 @@ def __init__(self, properties: Properties) -> None:
         super().__init__(properties)
         self._session = requests.Session()
 
-    def __call__(self, request: "AWSRequest", **_: Any) -> None:
+    def __call__(self, request: "AWSRequest", operation_name: str | None = None, **_: Any) -> None:
         signer_url = self.properties.get(S3_SIGNER_URI, self.properties[URI]).rstrip("/")  # type: ignore
         signer_endpoint = self.properties.get(S3_SIGNER_ENDPOINT, S3_SIGNER_ENDPOINT_DEFAULT)
 
@@ -148,6 +148,9 @@ def __call__(self, request: "AWSRequest", **_: Any) -> None:
             "uri": request.url,
             "headers": {key: [val] for key, val in request.headers.items()},
         }
+        # The signing service needs the DeleteObjects body to authorize the keys being deleted
+        if operation_name == "DeleteObjects":
+            signer_body["body"] = request.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,10 +20,14 @@
 import tempfile
 import threading
 import uuid
+from typing import Any
 from unittest import mock
 
 import pytest
+from botocore.auth import S3SigV4Auth
 from botocore.awsrequest import AWSRequest
+from botocore.config import Config
+from botocore.credentials import Credentials
 from fsspec.implementations.local import LocalFileSystem
 from fsspec.spec import AbstractFileSystem
 from requests_mock import Mocker
@@ -995,6 +999,43 @@ def test_s3v4_rest_signer(requests_mock: Mocker) -> None:
     }
 
 
+@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"""
+    credentials = Credentials(request.config.getoption("--s3.access-key-id"), request.config.getoption("--s3.secret-access-key"))
+
+    def sign(signer_request: Any, _: Any) -> dict[str, Any]:
+        signer_body = signer_request.json()
+        aws_request = AWSRequest(
+            method=signer_body["method"],
+            url=signer_body["uri"],
+            headers={key: values[0] for key, values in signer_body["headers"].items()},
+        )
+        # Upload bodies are not forwarded for signing, so leave the payload unsigned
+        aws_request.context["client_config"] = Config(s3={"payload_signing_enabled": False})
+        S3SigV4Auth(credentials, "s3", signer_body["region"]).add_auth(aws_request)
+        signed_headers = ("Authorization", "X-Amz-Content-SHA256", "X-Amz-Date")
+        return {"uri": signer_body["uri"], "headers": {key: [aws_request.headers[key]] for key in signed_headers}}
+
+    requests_mock.post(f"{TEST_URI}/v1/aws/s3/sign", json=sign)
+    fileio = FsspecFileIO(
+        properties={"s3.endpoint": request.config.getoption("--s3.endpoint"), "s3.signer": "S3V4RestSigner", "uri": TEST_URI}
+    )
+    filename = str(uuid.uuid4())
+    location = f"s3://warehouse/{filename}"
+
+    with fileio.new_output(location).create() as f:
+        f.write(b"foo")
+    assert fileio.new_input(location).open().read() == b"foo"
+    fileio.delete(location)
+    assert not fileio.new_input(location).exists()
+
+    signer_bodies = [signer_request.json() for signer_request in requests_mock.request_history]
+    request_bodies = [signer_body["body"] for signer_body in signer_bodies if "body" in signer_body]
+    assert len(request_bodies) == 1
+    assert f"<Key>{filename}</Key>" in request_bodies[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"