From 2a008e399518ca96de0ace01609ed526c6af731b Mon Sep 17 00:00:00 2001
From: Sreesh Maheshwar <maheshwarsreesh@gmail.com>
Date: Thu, 15 Oct 2026 20:35:09 +0100
Subject: [PATCH] Send the DeleteObjects body to the S3 REST signer

s3fs deletes files with DeleteObjects, which lists the object keys in the
request body. S3V4RestSigner never forwarded the body, so a signing
service could not authorize the keys before signing.

Also add an S3 test that writes, reads, and deletes through the REST
signer, which previously had only mocked unit tests.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 pyiceberg/io/fsspec.py  |  5 ++++-
 tests/io/test_fsspec.py | 41 +++++++++++++++++++++++++++++++++++++++++
 2 files changed, 45 insertions(+), 1 deletion(-)

diff --git a/pyiceberg/io/fsspec.py b/pyiceberg/io/fsspec.py
index 09bbe6f1d6..83d8e64701 100644
--- a/pyiceberg/io/fsspec.py
+++ b/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:
diff --git a/tests/io/test_fsspec.py b/tests/io/test_fsspec.py
index 45835a08eb..efd3ff1ddc 100644
--- a/tests/io/test_fsspec.py
+++ b/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"