fix(rest): keep a table-scoped token off the shared session

commit_table bound self._session.headers and wrote the table's token into
it, so the token outlived the commit and was sent with later requests for
other tables. Build the request headers from a copy instead; the request
itself is unchanged.

Author: ghoshp83

pyiceberg/catalog/rest/__init__.py

@@ -1603,7 +1603,10 @@ def commit_table(
         table_identifier = TableIdentifier(namespace=identifier[:-1], name=identifier[-1])
         table_request = CommitTableRequest(identifier=table_identifier, requirements=requirements, updates=updates)
 
-        headers = self._session.headers
+        # A table-scoped token applies to this request only, so it is layered onto a
+        # copy: assigning into self._session.headers would leave it on the session and
+        # send it with every later request, including ones for other tables.
+        headers = dict(self._session.headers)
         if table_token := table.config.get(TOKEN):
             headers[AUTHORIZATION_HEADER] = f"{BEARER_PREFIX} {table_token}"
 

tests/catalog/test_rest.py

@@ -3467,3 +3467,39 @@ def test_load_table_without_storage_credentials(
     )
     assert actual.metadata.model_dump() == expected.metadata.model_dump()
     assert actual == expected
+
+
+def test_commit_table_does_not_leak_table_token_onto_session(
+    rest_mock: Mocker, example_table_metadata_v2: dict[str, Any]
+) -> None:
+    """A table-scoped token must not outlive the commit that used it.
+
+    The catalog session is shared by every table, so a token left on it would be
+    sent with later requests for other tables.
+    """
+    table_token = "table_scoped_token"
+    metadata_location = "s3://warehouse/database/table/metadata.json"
+
+    rest_mock.get(
+        f"{TEST_URI}v1/namespaces/namespace/tables/table_name",
+        json={
+            "metadata-location": metadata_location,
+            "metadata": example_table_metadata_v2,
+            "config": {"token": table_token},
+        },
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+    table = catalog.load_table(("namespace", "table_name"))
+
+    rest_mock.post(
+        f"{TEST_URI}v1/namespaces/namespace/tables/table_name",
+        json={"metadata-location": metadata_location, "metadata": example_table_metadata_v2},
+        status_code=200,
+    )
+
+    table.update_schema().add_column("new_col", StringType()).commit()
+
+    assert "Authorization" not in catalog._session.headers