fix(rest): handle empty body in AWS SigV4 signing (#2827)

# Rationale for this change
The x-amz-content-sha256 header is required for AWS requests. It
provides a hash of the request payload. If there is no payload, you must
provide the hash of an empty string.

## Are these changes tested?

## Are there any user-facing changes?

---------

Co-authored-by: Fokko Driesprong <fokko@apache.org>

Author: plusplusjiajia

pyiceberg/catalog/rest/__init__.py

@@ -216,6 +216,7 @@ class IdentifierKind(Enum):
 SIGV4 = "rest.sigv4-enabled"
 SIGV4_REGION = "rest.signing-region"
 SIGV4_SERVICE = "rest.signing-name"
+EMPTY_BODY_SHA256: str = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
 OAUTH2_SERVER_URI = "oauth2-server-uri"
 SNAPSHOT_LOADING_MODE = "snapshot-loading-mode"
 AUTH = "auth"
@@ -560,7 +561,11 @@ def add_headers(self, request: PreparedRequest, **kwargs: Any) -> None:  # pylin
                 params = dict(parse.parse_qsl(query))
 
                 # remove the connection header as it will be updated after signing
-                del request.headers["connection"]
+                if "connection" in request.headers:
+                    del request.headers["connection"]
+                # For empty bodies, explicitly set the content hash header to the SHA256 of an empty string
+                if not request.body:
+                    request.headers["x-amz-content-sha256"] = EMPTY_BODY_SHA256
 
                 aws_request = AWSRequest(
                     method=request.method, url=url, params=params, data=request.body, headers=dict(request.headers)

tests/catalog/test_rest.py

@@ -22,12 +22,21 @@
 from unittest import mock
 
 import pytest
+from requests import Request
+from requests.adapters import HTTPAdapter
 from requests.exceptions import HTTPError
 from requests_mock import Mocker
 
 import pyiceberg
 from pyiceberg.catalog import PropertiesUpdateSummary, load_catalog
-from pyiceberg.catalog.rest import DEFAULT_ENDPOINTS, OAUTH2_SERVER_URI, SNAPSHOT_LOADING_MODE, Capability, RestCatalog
+from pyiceberg.catalog.rest import (
+    DEFAULT_ENDPOINTS,
+    EMPTY_BODY_SHA256,
+    OAUTH2_SERVER_URI,
+    SNAPSHOT_LOADING_MODE,
+    Capability,
+    RestCatalog,
+)
 from pyiceberg.exceptions import (
     AuthorizationExpiredError,
     NamespaceAlreadyExistsError,
@@ -451,6 +460,62 @@ def test_list_tables_200_sigv4(rest_mock: Mocker) -> None:
     assert rest_mock.called
 
 
+def test_sigv4_sign_request_without_body(rest_mock: Mocker) -> None:
+    existing_token = "existing_token"
+
+    catalog = RestCatalog(
+        "rest",
+        **{
+            "uri": TEST_URI,
+            "token": existing_token,
+            "rest.sigv4-enabled": "true",
+            "rest.signing-region": "us-west-2",
+            "client.access-key-id": "id",
+            "client.secret-access-key": "secret",
+        },
+    )
+
+    prepared = catalog._session.prepare_request(Request("GET", f"{TEST_URI}v1/config"))
+    adapter = catalog._session.adapters[catalog.uri]
+    assert isinstance(adapter, HTTPAdapter)
+    adapter.add_headers(prepared)
+
+    assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256")
+    assert prepared.headers["Original-Authorization"] == f"Bearer {existing_token}"
+    assert prepared.headers["x-amz-content-sha256"] == EMPTY_BODY_SHA256
+
+
+def test_sigv4_sign_request_with_body(rest_mock: Mocker) -> None:
+    existing_token = "existing_token"
+
+    catalog = RestCatalog(
+        "rest",
+        **{
+            "uri": TEST_URI,
+            "token": existing_token,
+            "rest.sigv4-enabled": "true",
+            "rest.signing-region": "us-west-2",
+            "client.access-key-id": "id",
+            "client.secret-access-key": "secret",
+        },
+    )
+
+    prepared = catalog._session.prepare_request(
+        Request(
+            "POST",
+            f"{TEST_URI}v1/namespaces",
+            data={"namespace": "asdfasd"},
+        )
+    )
+    adapter = catalog._session.adapters[catalog.uri]
+    assert isinstance(adapter, HTTPAdapter)
+    adapter.add_headers(prepared)
+
+    assert prepared.headers["Authorization"].startswith("AWS4-HMAC-SHA256")
+    assert prepared.headers["Original-Authorization"] == f"Bearer {existing_token}"
+    assert prepared.headers.get("x-amz-content-sha256") != EMPTY_BODY_SHA256
+
+
 def test_list_tables_404(rest_mock: Mocker) -> None:
     namespace = "examples"
     rest_mock.get(