use basic auth instead of post

Author: sungwy

pyiceberg/catalog/rest/auth.py

@@ -17,10 +17,11 @@
 
 import base64
 import importlib
+import logging
 import threading
 import time
-import logging
 from abc import ABC, abstractmethod
+from functools import cached_property
 from typing import Any, Dict, List, Optional, Type
 
 import requests
@@ -158,16 +159,19 @@ def __init__(
         self._expires_at = 0
         self._lock = threading.Lock()
 
+    @cached_property
+    def _client_secret_header(self) -> str:
+        creds = f"{self.client_id}:{self.client_secret}"
+        creds_bytes = creds.encode("utf-8")
+        b64_creds = base64.b64encode(creds_bytes).decode("utf-8")
+        return f"Basic {b64_creds}"
+
     def _refresh_token(self) -> None:
-        data = {
-            "grant_type": "client_credentials",
-            "client_id": self.client_id,
-            "client_secret": self.client_secret,
-        }
+        data = {}
         if self.scope:
             data["scope"] = self.scope
 
-        response = requests.post(self.token_url, data=data)
+        response = requests.post(self.token_url, data=data, headers={"Authorization": self._client_secret_header})
         response.raise_for_status()
         result = response.json()
 
@@ -177,11 +181,11 @@ def _refresh_token(self) -> None:
             raise ValueError(
                 "The expiration time of the Token must be provided by the Server in the Access Token Response in `expires_in` field, or by the PyIceberg Client."
             )
-        self._expires_at = time.time() + expires_in - self.refresh_margin
+        self._expires_at = time.monotonic() + expires_in - self.refresh_margin
 
     def get_token(self) -> str:
         with self._lock:
-            if not self._token or time.time() >= self._expires_at:
+            if not self._token or time.monotonic() >= self._expires_at:
                 self._refresh_token()
             if self._token is None:
                 raise ValueError("Authorization token is None after refresh")
@@ -211,6 +215,8 @@ def __init__(
 
     def auth_header(self) -> str:
         return f"Bearer {self.token_provider.get_token()}"
+
+
 class GoogleAuthManager(AuthManager):
     """An auth manager that is responsible for handling Google credentials."""
 

tests/catalog/test_rest.py

@@ -21,6 +21,7 @@
 from unittest import mock
 
 import pytest
+from requests.exceptions import HTTPError
 from requests_mock import Mocker
 
 import pyiceberg
@@ -1680,6 +1681,29 @@ def test_rest_catalog_with_oauth2_auth_type(requests_mock: Mocker) -> None:
     assert catalog.uri == TEST_URI
 
 
+def test_rest_catalog_oauth2_non_200_token_response(requests_mock: Mocker) -> None:
+    requests_mock.post(
+        f"{TEST_URI}oauth2/token",
+        json={"error": "invalid_client"},
+        status_code=401,
+    )
+    catalog_properties = {
+        "uri": TEST_URI,
+        "auth": {
+            "type": "oauth2",
+            "oauth2": {
+                "client_id": "bad_client_id",
+                "client_secret": "bad_client_secret",
+                "token_url": f"{TEST_URI}oauth2/token",
+                "scope": "read",
+            },
+        },
+    }
+
+    with pytest.raises(HTTPError):
+        RestCatalog("rest", **catalog_properties)  # type: ignore
+
+
 EXAMPLE_ENV = {"PYICEBERG_CATALOG__PRODUCTION__URI": TEST_URI}