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"{filename}" 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"