fix(google-auth-oauthlib): prevent ipv6 address reuse (#18463)

Extending the solution in
#18166

We previously added a guard to prevent external processes sitting on the
same port, and hijacking oauth calls. This PR extends that logic to also
prevent processes using the ipv6 address

Fixes: #18296

Author: daniel-sanche

packages/google-auth-oauthlib/google_auth_oauthlib/flow.py

@@ -49,6 +49,7 @@
 
 """
 
+import errno
 import hashlib
 import json
 import logging
@@ -492,14 +493,66 @@ class _ExclusiveWSGIServer(wsgiref.simple_server.WSGIServer):
     Setting `WSGIServer.allow_reuse_address` is not enough, since it sets `SO_REUSEADDR`
     and not `SO_EXCLUSIVEADDRUSE`. `SO_REUSEADDR` alone allows other processes to bind
     to the same address and port on Windows.
+
+    When bound to `localhost`, also reserves the IPv6 loopback (`::1`) so
+    another process listening on `::1` cannot intercept the OAuth callback.
     """
 
     allow_reuse_address = False
+    _PROBE_TIMEOUT_SECONDS = 0.1
 
-    def server_bind(self):
+    def __init__(self, *args, **kwargs):
+        self._ipv6_socket = None
+        super().__init__(*args, **kwargs)
+
+    @staticmethod
+    def is_listener_present(family: int, addr: str, port: int) -> bool:
+        """Check if another process is already listening on (addr, port) by
+        attempting a test connection.
+
+        This is needed because on Windows, `bind()` on `::1` succeeds even when
+        another process from the same user is already listening on all
+        interfaces (`[::]`).
+        """
+        try:
+            with socket.socket(family, socket.SOCK_STREAM) as probe:
+                probe.settimeout(_ExclusiveWSGIServer._PROBE_TIMEOUT_SECONDS)
+                return probe.connect_ex((addr, port)) == 0
+        except OSError:
+            return False
+
+    @staticmethod
+    def _set_exclusive_addr_use(sock: socket.socket) -> None:
         if sys.platform == "win32" and hasattr(socket, "SO_EXCLUSIVEADDRUSE"):
-            self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
+            sock.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
+
+    def _close_ipv6_socket(self):
+        if self._ipv6_socket is not None:
+            self._ipv6_socket.close()
+            self._ipv6_socket = None
+
+    def server_bind(self):
+        host = self.server_address[0]
+        self._set_exclusive_addr_use(self.socket)
         super().server_bind()
+        port = self.server_address[1]
+        # Reserve IPv6 loopback (::1) so another process cannot intercept localhost callbacks.
+        if host == "localhost" and port and hasattr(socket, "AF_INET6"):
+            # base class (TCPServer) calls server_close on error
+            if self.is_listener_present(socket.AF_INET6, "::1", port):
+                raise OSError(errno.EADDRINUSE, "Address already in use")
+            # Hold `::1` without calling `listen()` so no other process can claim
+            # the port while the browser falls back from `::1` to `127.0.0.1`.
+            try:
+                self._ipv6_socket = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
+                self._set_exclusive_addr_use(self._ipv6_socket)
+                self._ipv6_socket.bind(("::1", port))
+            except OSError:
+                self._close_ipv6_socket()
+
+    def server_close(self):
+        self._close_ipv6_socket()
+        super().server_close()
 
 
 class _WSGIRequestHandler(wsgiref.simple_server.WSGIRequestHandler):

packages/google-auth-oauthlib/google_auth_oauthlib/interactive.py

@@ -50,6 +50,13 @@ def is_port_open(port):
             is_open = False
         else:
             is_open = True
+    # Also verify that no process is already listening on IPv6 loopback (::1).
+    if is_open and hasattr(socket, "AF_INET6"):
+        is_open = (
+            not google_auth_oauthlib.flow._ExclusiveWSGIServer.is_listener_present(
+                socket.AF_INET6, "::1", port
+            )
+        )
     return is_open
 
 

packages/google-auth-oauthlib/tests/unit/test_flow.py

@@ -572,9 +572,13 @@ def test_exclusive_wsgi_server_bind_windows(self):
         ):
             mock_socket.SOL_SOCKET = socket.SOL_SOCKET
             mock_socket.SO_EXCLUSIVEADDRUSE = getattr(socket, "SO_EXCLUSIVEADDRUSE", 1)
+            mock_socket.AF_INET = socket.AF_INET
+            mock_socket.AF_INET6 = socket.AF_INET6
+            mock_socket.SOCK_STREAM = socket.SOCK_STREAM
+            mock_socket.socket.return_value.__enter__.return_value.connect_ex.return_value = 1
 
             server = flow._ExclusiveWSGIServer(
-                ("localhost", 0), flow._WSGIRequestHandler, bind_and_activate=False
+                ("localhost", 8085), flow._WSGIRequestHandler, bind_and_activate=False
             )
             server.socket = mock.Mock()
 
@@ -583,6 +587,12 @@ def test_exclusive_wsgi_server_bind_windows(self):
                 server.socket.setsockopt.assert_called_once_with(
                     mock_socket.SOL_SOCKET, mock_socket.SO_EXCLUSIVEADDRUSE, 1
                 )
+                server._ipv6_socket.setsockopt.assert_called_once_with(
+                    mock_socket.SOL_SOCKET, mock_socket.SO_EXCLUSIVEADDRUSE, 1
+                )
+                server._ipv6_socket.bind.assert_called_once_with(("::1", 8085))
+                server.server_close()
+                assert server._ipv6_socket is None
 
     def test_exclusive_wsgi_server_bind_non_windows(self):
         with mock.patch("sys.platform", "linux"):
@@ -594,3 +604,84 @@ def test_exclusive_wsgi_server_bind_non_windows(self):
             with mock.patch.object(wsgiref.simple_server.WSGIServer, "server_bind"):
                 server.server_bind()
                 server.socket.setsockopt.assert_not_called()
+                server.server_close()
+
+    def test_exclusive_wsgi_server_detects_existing_listeners(self):
+        with (
+            mock.patch("sys.platform", "win32"),
+            mock.patch("google_auth_oauthlib.flow.socket") as mock_socket,
+        ):
+            mock_socket.SOL_SOCKET = socket.SOL_SOCKET
+            mock_socket.SO_EXCLUSIVEADDRUSE = 1
+            mock_socket.AF_INET = socket.AF_INET
+            mock_socket.AF_INET6 = socket.AF_INET6
+            mock_socket.SOCK_STREAM = socket.SOCK_STREAM
+
+            server = flow._ExclusiveWSGIServer(
+                ("localhost", 8085), flow._WSGIRequestHandler, bind_and_activate=False
+            )
+            server.socket = mock.Mock()
+
+            # 1. Pre-existing IPv6 listener on ::1 / [::]
+            mock_socket.socket.return_value.__enter__.return_value.connect_ex.return_value = 0
+            with mock.patch.object(wsgiref.simple_server.WSGIServer, "server_bind"):
+                with pytest.raises(OSError):
+                    server.server_bind()
+
+            # 2. is_listener_present returns False when socket() raises OSError
+            mock_socket.socket.side_effect = OSError("socket error")
+            assert not flow._ExclusiveWSGIServer.is_listener_present(
+                socket.AF_INET6, "::1", 8085
+            )
+
+    def test_exclusive_wsgi_server_ipv4_literal_ignores_ipv6(self):
+        # An address literal is served verbatim in redirect_uri, so the browser
+        # never resolves `localhost` and `::1` is irrelevant.
+        server = flow._ExclusiveWSGIServer(
+            ("127.0.0.1", 8085), flow._WSGIRequestHandler, bind_and_activate=False
+        )
+        server.socket = mock.Mock()
+
+        with (
+            mock.patch.object(wsgiref.simple_server.WSGIServer, "server_bind"),
+            mock.patch.object(
+                flow._ExclusiveWSGIServer, "is_listener_present", return_value=True
+            ) as is_listener_present,
+        ):
+            server.server_bind()
+
+        is_listener_present.assert_not_called()
+        assert server._ipv6_socket is None
+
+    def test_exclusive_wsgi_server_ipv6_bind_errors(self):
+        import errno
+
+        server = flow._ExclusiveWSGIServer(
+            ("localhost", 8085), flow._WSGIRequestHandler, bind_and_activate=False
+        )
+        server.socket = mock.Mock()
+
+        with (
+            mock.patch.object(wsgiref.simple_server.WSGIServer, "server_bind"),
+            mock.patch.object(
+                flow._ExclusiveWSGIServer, "is_listener_present", return_value=False
+            ),
+            mock.patch("google_auth_oauthlib.flow.socket") as mock_socket,
+        ):
+            mock_socket.AF_INET6 = socket.AF_INET6
+            mock_socket.SOCK_STREAM = socket.SOCK_STREAM
+
+            # 1. Ignored when socket(AF_INET6) itself raises EAFNOSUPPORT
+            mock_socket.socket.side_effect = OSError(
+                errno.EAFNOSUPPORT, "IPv6 disabled in kernel"
+            )
+            server.server_bind()
+            assert server._ipv6_socket is None
+
+            # 2. Non-fatal when bind(("::1", ...)) fails but no listener is present
+            mock_socket.socket.side_effect = None
+            mock_socket.socket.return_value.bind.side_effect = OSError(
+                errno.EADDRINUSE, "Held by non-listening socket"
+            )
+            server.server_bind()
+            assert server._ipv6_socket is None

packages/google-auth-oauthlib/tests/unit/test_interactive.py

@@ -55,6 +55,33 @@ def mock_socket(family, type_):
     socket_instance.listen.assert_has_calls(mock.call(1) for _ in range(100))
 
 
+def test_is_port_open_ipv6_checks(monkeypatch):
+    import errno
+
+    from google_auth_oauthlib import interactive as module_under_test
+
+    sock4 = mock.create_autospec(socket.socket, instance=True)
+    sock6 = mock.MagicMock()
+
+    def make_sock(family, type_):
+        if family == socket.AF_INET6:
+            return sock6
+        return sock4
+
+    monkeypatch.setattr(socket, "socket", make_sock)
+
+    sock6.__enter__.return_value.connect_ex.return_value = 0
+    assert not module_under_test.is_port_open(8085)
+
+    def mock_socket_fn(family, type_):
+        if family == socket.AF_INET6:
+            raise OSError(errno.EAFNOSUPPORT, "IPv6 disabled")
+        return sock4
+
+    monkeypatch.setattr(socket, "socket", mock_socket_fn)
+    assert module_under_test.is_port_open(8085)
+
+
 def test_get_user_credentials():
     from google_auth_oauthlib import flow
     from google_auth_oauthlib import interactive as module_under_test