fix: defensive access for WS ack message to avoid KeyError on bad payloads (OPS-3613)

CHANGELOG.md

@@ -1,5 +1,10 @@
 # CHANGELOG
 
+## [3.8.2] - 2026-06-19
+
+- [Fixed] `_waiting_connection_ack` (initial WS connection_ack after connection_init) now performs defensive access (`isinstance` + `"type" not in`) and raises a controlled `GQLResponseException` (instead of `KeyError`/`TypeError` on `message["type"]`) when the server sends a non-dict or dict-without-type (e.g. `null`, `{"payload":...}`). This matches the "no defensive access" code-bug pattern (see cq['name'] in subscription callbacks) and is defense-in-depth for subscription paths, similar to the non-dict guards added in OPS-3485. (OPS-3613)
+- [Changed] Added regression test exercising the bad-ack paths (TDD: red then green).
+
 ## [3.8.1] - 2026-06-11
 
 - [Fixed] Stale `httpx.AsyncClient` instances are now best-effort `aclose()`d instead of being dropped to the garbage collector. Previously three paths leaked the client's transports: `_get_async_client` when it detected a closed event loop, `async_execute`'s "Event loop is closed" retry, and `_close()`/`__del__`. Orphaned `_SelectorTransport.__del__` socket finalizers then ran during cyclic-GC sweeps on arbitrary threads, contributing to false TMPRL1101 deadlock detections in Temporal workers (see valiot/python-tooling#151). A new private `_drop_async_client()` helper awaits `aclose()` (swallowing failures when the original loop is gone) and is used by `_get_async_client`, the `async_execute` retry, and `async_cleanup`; `_close()` now schedules `aclose()` on the running loop via `call_soon_threadsafe` when one exists, falling back to GC only when no loop is available.

pygqlc/GraphQLClient.py

@@ -759,9 +759,26 @@ def _conn_init(self):
     def _waiting_connection_ack(self):
         self._conn.settimeout(self.ack_timeout)
         # set timeout to raise Exception websocket.WebSocketTimeoutException
-        message = orjson.loads(self._conn.recv())
+        raw = self._conn.recv()
+        message = orjson.loads(raw)
+        if not isinstance(message, dict) or "type" not in message:
+            raise GQLResponseException(
+                message=f"Invalid connection ack message, expected dict with 'type': {message}",
+                status_code=0,
+                query="connection_init",
+                variables=None,
+                response_body=str(raw),
+            )
         if message["type"] == CONNECTION_ACK_TYPE:
             pass  # Connection Ack with the server
+        else:
+            raise GQLResponseException(
+                message=f"Connection init did not receive ack, got: {message}",
+                status_code=0,
+                query="connection_init",
+                variables=None,
+                response_body=str(raw),
+            )
 
     def _ping_pong(self):
         self.pingTimer = time.time()

pygqlc/__version__.py

@@ -1 +1 @@
-__version__ = "3.8.1"
+__version__ = "3.8.2"

tests/pygqlc/gql_client/test_subscribe.py

@@ -9,7 +9,7 @@
 
 import pytest
 
-from pygqlc import GraphQLClient
+from pygqlc import GraphQLClient, GQLResponseException
 from pygqlc.helper_modules.Singleton import Singleton
 from pygqlc.logging import LogLevel, get_logger, set_logger
 
@@ -299,3 +299,31 @@ def test_new_conn_returns_false_when_no_environment():
     assert result is False
     assert any(level == LogLevel.ERROR for level, _ in records)
     Singleton._instances.pop(GraphQLClient, None)
+
+
+def test_waiting_connection_ack_defensive_no_keyerror_on_bad_ack(routing_client):
+    """OPS-3613: _waiting_connection_ack must not raise KeyError on message["type"]
+    when the server sends non-dict payload (e.g. null) or dict without "type" key
+    for the initial connection_ack after connection_init. This is the direct
+    subscript on untrusted WS data in the subscription path (analogous to
+    cq['name'] non-defensive access in callbacks). We raise controlled
+    GQLResponseException instead so callers (e.g. _new_conn) can handle cleanly.
+    """
+    gql = routing_client
+    conn = MagicMock()
+    gql._conn = conn
+
+    # Case that triggers KeyError today: dict missing "type"
+    conn.recv.return_value = b'{"payload": {"errors": ["bad init"]}}'
+    with pytest.raises(GQLResponseException) as exc:
+        gql._waiting_connection_ack()
+    assert "connection" in str(exc.value).lower()
+
+    # Case non-dict (orjson.loads -> None or list etc)
+    conn.recv.return_value = b"null"
+    with pytest.raises(GQLResponseException):
+        gql._waiting_connection_ack()
+
+    # Good path still succeeds
+    conn.recv.return_value = b'{"type": "connection_ack"}'
+    gql._waiting_connection_ack()  # no raise