Fix code review issues in DIGEST-MD5 delegation token auth

Address all findings from code review:

Critical:
- Rewrite VInt decoder to match Java WritableUtils.readVLong exactly,
  using signed-byte interpretation and correct prefix/length semantics

High:
- Catch OSError (not just FileNotFoundError) when reading token file
- Reject unknown auth mechanisms with HiveAuthError instead of silently
  falling back to unauthenticated TBufferedTransport
- Replace monkey-patching sasl.process in _DigestMD5SaslTransport with
  a clean send_sasl_msg override (thread-safe, no shared state mutation)

Medium:
- Fix kerberos_service_name default from config key to actual value
- Wrap UnicodeDecodeError in HiveAuthError for invalid UTF-8 in tokens
- Rewrite VInt test encoder to match real Hadoop encoding format
- Fix dead kerberos backward-compat tests to actually exercise __init__

Low:
- Add upper bound to pure-sasl dependency (<1.0.0)
- Fix tmp_path typing from object to pathlib.Path
- Fix docs to say pure-sasl (pip package name) not puresasl

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: ShreyeshArangath

mkdocs/docs/configuration.md

@@ -687,7 +687,7 @@ catalog:
 | ugi                 | t-1234:secret                    | Hadoop UGI for Hive client.                                                                        |
 | hive.metastore.authentication | DIGEST-MD5 | Auth mechanism: `NONE` (default), `KERBEROS`, or `DIGEST-MD5` |
 
-When using DIGEST-MD5 authentication, PyIceberg reads a Hive delegation token from the file pointed to by the `$HADOOP_TOKEN_FILE_LOCATION` environment variable. This is the standard mechanism used in secure Hadoop environments where delegation tokens are distributed to jobs. Install PyIceberg with `pip install "pyiceberg[hive]"` to get the required `puresasl` dependency.
+When using DIGEST-MD5 authentication, PyIceberg reads a Hive delegation token from the file pointed to by the `$HADOOP_TOKEN_FILE_LOCATION` environment variable. This is the standard mechanism used in secure Hadoop environments where delegation tokens are distributed to jobs. Install PyIceberg with `pip install "pyiceberg[hive]"` to get the required `pure-sasl` dependency.
 
 When using Hive 2.x, make sure to set the compatibility flag:
 

pyiceberg/catalog/hive.py

@@ -154,21 +154,8 @@ class _DigestMD5SaslTransport(TTransport.TSaslClientTransport):
     coerces ``None`` to ``b""`` so the SASL handshake proceeds normally.
     """
 
-    def open(self) -> None:
-        # Intercept sasl.process to coerce the initial None response
-        original_process = self.sasl.process
-
-        def _patched_process(challenge: bytes | None = None) -> bytes | None:
-            result = original_process(challenge)
-            if result is None:
-                return b""
-            return result
-
-        self.sasl.process = _patched_process
-        try:
-            super().open()
-        finally:
-            self.sasl.process = original_process
+    def send_sasl_msg(self, status: int, body: bytes | None) -> None:  # type: ignore[override]
+        super().send_sasl_msg(status, body if body is not None else b"")
 
 
 class _HiveClient:
@@ -182,7 +169,7 @@ def __init__(
         uri: str,
         ugi: str | None = None,
         kerberos_auth: bool | None = HIVE_KERBEROS_AUTH_DEFAULT,
-        kerberos_service_name: str | None = HIVE_KERBEROS_SERVICE_NAME,
+        kerberos_service_name: str | None = HIVE_KERBEROS_SERVICE_NAME_DEFAULT,
         auth_mechanism: str | None = None,
     ):
         self._uri = uri
@@ -204,7 +191,9 @@ def _init_thrift_transport(self) -> TTransport:
         url_parts = urlparse(self._uri)
         socket = TSocket.TSocket(url_parts.hostname, url_parts.port)
 
-        if self._auth_mechanism == "KERBEROS":
+        if self._auth_mechanism == "NONE":
+            return TTransport.TBufferedTransport(socket)
+        elif self._auth_mechanism == "KERBEROS":
             return TTransport.TSaslClientTransport(socket, host=url_parts.hostname, service=self._kerberos_service_name)
         elif self._auth_mechanism == "DIGEST-MD5":
             identifier, password = read_hive_delegation_token()
@@ -217,7 +206,10 @@ def _init_thrift_transport(self) -> TTransport:
                 password=password,
             )
         else:
-            return TTransport.TBufferedTransport(socket)
+            raise HiveAuthError(
+                f"Unknown auth mechanism: {self._auth_mechanism!r}. "
+                f"Valid values: NONE, KERBEROS, DIGEST-MD5"
+            )
 
     def _client(self) -> Client:
         protocol = TBinaryProtocol.TBinaryProtocol(self._transport)

pyiceberg/utils/hadoop_credentials.py

@@ -36,29 +36,32 @@
 
 
 def _read_hadoop_vint(stream: BytesIO) -> int:
-    """Decode a Hadoop WritableUtils VInt/VLong from a byte stream."""
+    """Decode a Hadoop WritableUtils VInt/VLong from a byte stream.
+
+    Matches the encoding in Java's ``WritableUtils.readVInt``/``readVLong``:
+    - If the first byte (interpreted as signed) is >= -112, it *is* the value.
+    - Otherwise the first byte encodes both a negativity flag and the number
+      of additional big-endian payload bytes that carry the actual value.
+    """
     first = stream.read(1)
     if not first:
         raise HiveAuthError("Unexpected end of token file while reading VInt")
+    # Reinterpret as signed byte to match Java's signed-byte semantics
     b = first[0]
-    if b <= 0x7F:
+    if b > 127:
+        b -= 256
+    if b >= -112:
         return b
-    # Number of additional bytes is encoded in leading 1-bits
-    num_extra = 0
-    mask = 0x80
-    while b & mask:
-        num_extra += 1
-        mask >>= 1
-    # First byte contributes the remaining bits
-    result = b & (mask - 1)
-    extra = stream.read(num_extra)
-    if len(extra) != num_extra:
+    negative = b < -120
+    length = (-119 - b) if negative else (-111 - b)
+    extra = stream.read(length)
+    if len(extra) != length:
         raise HiveAuthError("Unexpected end of token file while reading VInt")
-    for byte in extra:
-        result = (result << 8) | byte
-    # Sign-extend if negative (high bit of decoded value is set)
-    if result >= (1 << (8 * num_extra + (8 - num_extra - 1) - 1)):
-        result -= 1 << (8 * num_extra + (8 - num_extra - 1))
+    result = 0
+    for byte_val in extra:
+        result = (result << 8) | byte_val
+    if negative:
+        result = ~result
     return result
 
 
@@ -75,7 +78,11 @@ def _read_hadoop_bytes(stream: BytesIO) -> bytes:
 
 def _read_hadoop_text(stream: BytesIO) -> str:
     """Read a VInt-prefixed UTF-8 string from a Hadoop token stream."""
-    return _read_hadoop_bytes(stream).decode("utf-8")
+    raw = _read_hadoop_bytes(stream)
+    try:
+        return raw.decode("utf-8")
+    except UnicodeDecodeError as e:
+        raise HiveAuthError(f"Token file contains invalid UTF-8 in text field: {e}") from e
 
 
 def read_hive_delegation_token() -> tuple[str, str]:
@@ -99,8 +106,8 @@ def read_hive_delegation_token() -> tuple[str, str]:
     try:
         with open(token_file, "rb") as f:
             data = f.read()
-    except FileNotFoundError:
-        raise HiveAuthError(f"Hadoop token file not found: {token_file}")
+    except OSError as e:
+        raise HiveAuthError(f"Cannot read Hadoop token file {token_file}: {e}") from e
 
     stream = BytesIO(data)
 

pyproject.toml

@@ -76,7 +76,7 @@ polars = ["polars>=1.21.0,<2"]
 snappy = ["python-snappy>=0.6.0,<1.0.0"]
 hive = [
     "thrift>=0.13.0,<1.0.0",
-    "pure-sasl>=0.6.0",
+    "pure-sasl>=0.6.0,<1.0.0",
 ]
 hive-kerberos = [
     "thrift>=0.13.0,<1.0.0",

tests/catalog/test_hive.py

@@ -1429,15 +1429,13 @@ def test_auth_mechanism_none_creates_buffered_transport_explicit() -> None:
     assert client._auth_mechanism == "NONE"
 
 
-def test_auth_mechanism_kerberos_resolved() -> None:
-    """When auth_mechanism is KERBEROS, _auth_mechanism is set correctly.
-
-    We don't fully instantiate because TSaslClientTransport with GSSAPI
-    requires the kerberos C module which may not be installed.
-    """
-    client = _HiveClient.__new__(_HiveClient)
-    client._auth_mechanism = "KERBEROS"
+def test_auth_mechanism_kerberos_resolved(monkeypatch: pytest.MonkeyPatch) -> None:
+    """When auth_mechanism is KERBEROS, _auth_mechanism is resolved correctly."""
+    # Stub TSaslClientTransport.__init__ to avoid requiring the kerberos C module
+    monkeypatch.setattr(TTransport.TSaslClientTransport, "__init__", lambda *a, **kw: None)
+    client = _HiveClient(uri="thrift://localhost:9083", auth_mechanism="KERBEROS")
     assert client._auth_mechanism == "KERBEROS"
+    assert isinstance(client._transport, TTransport.TSaslClientTransport)
 
 
 def test_auth_mechanism_digest_md5_creates_digest_transport(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -1447,12 +1445,12 @@ def test_auth_mechanism_digest_md5_creates_digest_transport(monkeypatch: pytest.
     assert isinstance(client._transport, _DigestMD5SaslTransport)
 
 
-def test_legacy_kerberos_auth_backward_compat() -> None:
+def test_legacy_kerberos_auth_backward_compat(monkeypatch: pytest.MonkeyPatch) -> None:
     """Legacy kerberos_auth=True resolves to KERBEROS auth_mechanism."""
-    client = _HiveClient.__new__(_HiveClient)
-    # Replicate the constructor's mechanism resolution logic
-    client._auth_mechanism = "KERBEROS"  # what kerberos_auth=True produces
+    monkeypatch.setattr(TTransport.TSaslClientTransport, "__init__", lambda *a, **kw: None)
+    client = _HiveClient(uri="thrift://localhost:9083", kerberos_auth=True)
     assert client._auth_mechanism == "KERBEROS"
+    assert isinstance(client._transport, TTransport.TSaslClientTransport)
 
 
 def test_auth_mechanism_overrides_kerberos_auth() -> None:
@@ -1462,6 +1460,14 @@ def test_auth_mechanism_overrides_kerberos_auth() -> None:
     assert client._auth_mechanism == "NONE"
 
 
+def test_auth_mechanism_unknown_raises() -> None:
+    """Unknown auth mechanism should raise HiveAuthError, not silently fall back."""
+    from pyiceberg.exceptions import HiveAuthError
+
+    with pytest.raises(HiveAuthError, match="Unknown auth mechanism.*PLAIN"):
+        _HiveClient(uri="thrift://localhost:9083", auth_mechanism="PLAIN")
+
+
 def test_auth_mechanism_case_insensitive(monkeypatch: pytest.MonkeyPatch) -> None:
     """Auth mechanism should be case-insensitive."""
     monkeypatch.setattr("pyiceberg.catalog.hive.read_hive_delegation_token", _fake_read_token)
@@ -1480,8 +1486,8 @@ def test_create_hive_client_passes_auth_mechanism(monkeypatch: pytest.MonkeyPatc
     assert client._auth_mechanism == "DIGEST-MD5"
 
 
-def test_digest_md5_transport_coerces_none_to_empty_bytes(monkeypatch: pytest.MonkeyPatch) -> None:
-    """_DigestMD5SaslTransport.open() coerces None initial sasl.process() to b''."""
+def test_digest_md5_transport_send_sasl_msg_coerces_none(monkeypatch: pytest.MonkeyPatch) -> None:
+    """_DigestMD5SaslTransport.send_sasl_msg coerces None body to b''."""
     monkeypatch.setattr("pyiceberg.catalog.hive.read_hive_delegation_token", _fake_read_token)
 
     transport = _DigestMD5SaslTransport(
@@ -1493,36 +1499,18 @@ def test_digest_md5_transport_coerces_none_to_empty_bytes(monkeypatch: pytest.Mo
         password="dGVzdC1wdw==",
     )
 
-    # Build a simple sasl stand-in that returns None on first call
-    class FakeSasl:
-        def __init__(self) -> None:
-            self._call_count = 0
-            self.complete = True
-
-        def process(self, challenge: bytes | None = None) -> bytes | None:
-            self._call_count += 1
-            if self._call_count == 1:
-                return None  # DIGEST-MD5 initial response is None
-            return b"response-data"
-
-    original_sasl = transport.sasl
-    transport.sasl = FakeSasl()  # type: ignore[assignment]
-
-    # Capture what the patched process returns during open()
-    captured_results: list[bytes | None] = []
+    # Capture what the parent send_sasl_msg receives
+    captured_calls: list[tuple[int, bytes | None]] = []
 
-    def fake_super_open(self: TTransport.TSaslClientTransport) -> None:
-        captured_results.append(self.sasl.process(None))
-        captured_results.append(self.sasl.process(b"challenge"))
+    def capture_send(self: TTransport.TSaslClientTransport, status: int, body: bytes | None) -> None:
+        captured_calls.append((status, body))
 
-    monkeypatch.setattr(TTransport.TSaslClientTransport, "open", fake_super_open)
+    monkeypatch.setattr(TTransport.TSaslClientTransport, "send_sasl_msg", capture_send)
 
-    try:
-        transport.open()
-    finally:
-        transport.sasl = original_sasl
+    # Send with None body — should be coerced to b""
+    transport.send_sasl_msg(1, None)
+    # Send with real body — should pass through unchanged
+    transport.send_sasl_msg(2, b"real-data")
 
-    # None from first process() should have been coerced to b""
-    assert captured_results[0] == b""
-    # Non-None result passes through unchanged
-    assert captured_results[1] == b"response-data"
+    assert captured_calls[0] == (1, b""), "None body should be coerced to b''"
+    assert captured_calls[1] == (2, b"real-data"), "Non-None body should pass through"