From 659137b1aec1aa40df7cc4807a3de8963494fc87 Mon Sep 17 00:00:00 2001
From: kgala2 <galakp@google.com>
Date: Wed, 14 Oct 2026 00:21:41 +0000
Subject: [PATCH] fix(psycopg): prevent proxy deadlock on TLS 1.3
 post-handshake messages

In TLS 1.3 (RFC 8446 Section 4.6.1), the server sends a post-handshake NewSessionTicket record asynchronously after the handshake completes. When this record arrives on the remote SSLSocket before psycopg writes its PostgreSQL StartupMessage to the local socket (or during an idle window between queries), selectors.DefaultSelector() wakes up with EVENT_READ on remote. Calling recv() on a blocking SSLSocket consumes the NewSessionTicket internally and then blocks indefinitely inside SSL_read waiting for application data from the server, deadlocking the single-threaded proxy until psycopg's 30s connect_timeout expires.

Temporarily put the remote socket in non-blocking mode during selector-driven recv() calls and return True on SSLWantReadError/BlockingIOError so the selector loop continues without blocking.
---
 google/cloud/sql/connector/psycopg.py | 16 ++++++++--
 tests/unit/test_psycopg.py            | 46 +++++++++++++++++++++++++++
 2 files changed, 60 insertions(+), 2 deletions(-)

diff --git a/google/cloud/sql/connector/psycopg.py b/google/cloud/sql/connector/psycopg.py
index a1d9f3da..c9a12c97 100644
--- a/google/cloud/sql/connector/psycopg.py
+++ b/google/cloud/sql/connector/psycopg.py
@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import contextlib
 import logging
 import os
 import platform
@@ -52,6 +53,8 @@ def _forward_bytes(
     dst: socket.socket,
     src_name: str,
     dst_name: str,
+    *,
+    nonblocking: bool = False,
 ) -> bool:
     """Read a chunk of bytes from src and send it to dst.
 
@@ -59,8 +62,17 @@ def _forward_bytes(
         bool: True if bytes were forwarded, False on EOF or OSError.
     """
     try:
-        data = src.recv(_BUFFER_SIZE)
+        if nonblocking:
+            src.setblocking(False)
+        try:
+            data = src.recv(_BUFFER_SIZE)
+        finally:
+            if nonblocking:
+                with contextlib.suppress(OSError):
+                    src.setblocking(True)
     except OSError as e:
+        if nonblocking and isinstance(e, (ssl.SSLWantReadError, BlockingIOError)):
+            return True
         logger.debug(f"Failed to receive from {src_name} socket: {e}")
         return False
     if not data:
@@ -118,7 +130,7 @@ def _proxy(local: socket.socket, remote: "ssl.SSLSocket") -> None:
                 ):
                     return
                 if key.data == "remote" and not _forward_bytes(
-                    remote, local, "remote", "local"
+                    remote, local, "remote", "local", nonblocking=True
                 ):
                     return
     except OSError as e:
diff --git a/tests/unit/test_psycopg.py b/tests/unit/test_psycopg.py
index e419d3cf..c745ccb0 100644
--- a/tests/unit/test_psycopg.py
+++ b/tests/unit/test_psycopg.py
@@ -813,3 +813,49 @@ def socket_side_effect(family, type, proto=0, fileno=None):
 
     mock_local_sock.close.assert_called_once()
     assert mock_remote_sock.close.call_count >= 1
+
+
+def test_proxy_tls13_post_handshake_want_read() -> None:
+    """Test that _proxy does not deadlock or exit on SSLWantReadError (e.g. TLS 1.3 NewSessionTicket)."""
+    local_client, local_server = socket.socketpair()
+    remote_client, remote_server = socket.socketpair()
+
+    class MockTLS13SSLSocket:
+        def __init__(self, sock: socket.socket) -> None:
+            self._sock = sock
+            self._want_read_once = True
+
+        def pending(self) -> int:
+            return 0
+
+        def recv(self, bufsize: int, flags: int = 0) -> bytes:
+            if self._want_read_once:
+                self._want_read_once = False
+                # Drain the 1-byte dummy ticket wakeup from the underlying socket
+                self._sock.recv(1)
+                raise ssl.SSLWantReadError()
+            return self._sock.recv(bufsize, flags)
+
+        def __getattr__(self, name: str) -> Any:
+            return getattr(self._sock, name)
+
+    wrapped_remote = MockTLS13SSLSocket(remote_client)
+
+    # Simulate TLS 1.3 NewSessionTicket arriving on remote BEFORE client sends StartupMessage
+    remote_server.sendall(b"T")
+
+    proxy_thread = threading.Thread(
+        target=_proxy, args=(local_server, wrapped_remote), daemon=True
+    )
+    proxy_thread.start()
+
+    # Client sends StartupMessage and receives response without deadlocking
+    local_client.sendall(b"StartupMessage")
+    assert remote_server.recv(1024) == b"StartupMessage"
+    remote_server.sendall(b"AuthenticationOk")
+    assert local_client.recv(1024) == b"AuthenticationOk"
+
+    local_client.close()
+    proxy_thread.join(timeout=2.0)
+    assert not proxy_thread.is_alive()
+    remote_server.close()