REST: Allow disabling remote signing with s3.remote-signing-enabled

mkdocs/docs/configuration.md

@@ -407,6 +407,7 @@ catalog:
 | warehouse           | myWarehouse                      | Warehouse location or identifier to request from the catalog service. May be used to determine server-side overrides, such as the warehouse location. |
 | snapshot-loading-mode | refs                           | The snapshots to return in the body of the metadata. Setting the value to `all` would return the full set of snapshots currently valid for the table. Setting the value to `refs` would load all snapshots referenced by branches or tags. |
 | `header.X-Iceberg-Access-Delegation` | `vended-credentials` | Signal to the server that the client supports delegated access via a comma-separated list of access mechanisms. The server may choose to supply access via any or none of the requested mechanisms. When using `vended-credentials`, the server provides temporary credentials to the client. When using `remote-signing`, the server signs requests on behalf of the client. (default: `vended-credentials`) |
+| s3.remote-signing-enabled | false | Set to `false` to turn off S3 remote signing for this catalog, even when the server returns `s3.signer` in its config. Useful when the client already has its own S3 credentials. (default: `true`) |
 | view-endpoints-supported | false                           | For backwards compatibility with older REST servers. Set to `true` if the server supports view endpoints but doesn't send the `endpoints` field in the ConfigResponse. |
 | scan-planning-mode | client | When set to `server`, and the catalog advertises the plan-table-scan endpoint, `table.scan()` uses REST server-side scan planning. May be set by the client, returned by the catalog via `GET /v1/config` defaults/overrides, or returned per table in the `config` of the `loadTable` response. The `loadTable` value takes precedence over the catalog-level value, which lets a server request server-side planning only for specific tables. Async plans (`status=submitted`) are polled via `GET .../plan/{plan-id}` until terminal state (completed / failed / cancelled). |
 | rest-scan-planning.poll-timeout-ms | 300000 | Maximum time to wait for an async scan plan to complete before failing (default: 5 minutes). |

pyiceberg/catalog/rest/__init__.py

@@ -73,6 +73,8 @@
     AWS_REGION,
     AWS_SECRET_ACCESS_KEY,
     AWS_SESSION_TOKEN,
+    S3_REMOTE_SIGNING_ENABLED,
+    S3_SIGNER,
     FileIO,
     load_file_io,
 )
@@ -645,6 +647,10 @@ def _load_file_io(
         table_properties: Properties = EMPTY_DICT,
     ) -> FileIO:
         merged_properties = {**table_properties, **self.properties, **properties}
+        # s3.remote-signing-enabled=false in the catalog config opts out of remote signing,
+        # even when the server config or vended credentials set a signer.
+        if not property_as_bool(self.properties, S3_REMOTE_SIGNING_ENABLED, True):
+            merged_properties.pop(S3_SIGNER, None)
         if self._auth_manager:
             merged_properties[AUTH_MANAGER] = self._auth_manager
         return load_file_io(merged_properties, location)

pyiceberg/io/__init__.py

@@ -74,6 +74,7 @@ def _is_local_path(path: str) -> bool:
 S3_CONNECT_TIMEOUT = "s3.connect-timeout"
 S3_REQUEST_TIMEOUT = "s3.request-timeout"
 S3_SIGNER = "s3.signer"
+S3_REMOTE_SIGNING_ENABLED = "s3.remote-signing-enabled"
 S3_SIGNER_URI = "s3.signer.uri"
 S3_SIGNER_ENDPOINT = "s3.signer.endpoint"
 S3_SIGNER_ENDPOINT_DEFAULT = "v1/aws/s3/sign"

tests/catalog/test_rest.py

@@ -3481,6 +3481,47 @@ def test_load_table_catalog_config_outranks_table_properties(
     assert table.io.properties["s3.proxy-uri"] == "http://table-only-proxy"
 
 
+def test_load_table_remote_signing_disabled_in_catalog_config(
+    rest_mock: Mocker, example_table_metadata_with_snapshot_v1: dict[str, Any]
+) -> None:
+    metadata_location = "s3://warehouse/database/table/metadata/00001.metadata.json"
+    rest_mock.get(
+        f"{TEST_URI}v1/namespaces/fokko/tables/table",
+        json={
+            "metadata-location": metadata_location,
+            "metadata": example_table_metadata_with_snapshot_v1,
+            "config": {"s3.signer": "S3V4RestSigner", "s3.signer.uri": "http://signer"},
+        },
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN, **{"s3.remote-signing-enabled": "false"})
+    table = catalog.load_table(("fokko", "table"))
+
+    assert "s3.signer" not in table.io.properties
+
+
+@pytest.mark.parametrize("catalog_properties", [{}, {"s3.remote-signing-enabled": "true"}])
+def test_load_table_server_signer_kept_unless_disabled(
+    rest_mock: Mocker, example_table_metadata_with_snapshot_v1: dict[str, Any], catalog_properties: dict[str, str]
+) -> None:
+    metadata_location = "s3://warehouse/database/table/metadata/00001.metadata.json"
+    rest_mock.get(
+        f"{TEST_URI}v1/namespaces/fokko/tables/table",
+        json={
+            "metadata-location": metadata_location,
+            "metadata": example_table_metadata_with_snapshot_v1,
+            "config": {"s3.signer": "S3V4RestSigner", "s3.signer.uri": "http://signer"},
+        },
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN, **catalog_properties)
+    table = catalog.load_table(("fokko", "table"))
+
+    assert table.io.properties["s3.signer"] == "S3V4RestSigner"
+
+
 def test_load_credentials_with_longest_prefix(rest_mock: Mocker) -> None:
     rest_mock.get(
         f"{TEST_URI}v1/namespaces/fokko/tables/table/credentials",