googleapis
diff --git a/‎packages/google-auth-oauthlib/google_auth_oauthlib/flow.py‎
Lines changed: 26 additions & 3 deletions b/‎packages/google-auth-oauthlib/google_auth_oauthlib/flow.py‎
Lines changed: 26 additions & 3 deletions
diff --git a/‎packages/google-auth-oauthlib/tests/unit/test_flow.py‎
Lines changed: 83 additions & 4 deletions b/‎packages/google-auth-oauthlib/tests/unit/test_flow.py‎
Lines changed: 83 additions & 4 deletions
@@ -58,7 +58,9 @@
 except ImportError:  # pragma: NO COVER
     from random import SystemRandom
 
+import socket
 from string import ascii_letters, digits
+import sys
 import webbrowser
 import wsgiref.simple_server
 import wsgiref.util
@@ -432,10 +434,14 @@ def run_local_server(
                 authorization server.
         """
         wsgi_app = _RedirectWSGIApp(success_message)
-        # Fail fast if the address is occupied
-        wsgiref.simple_server.WSGIServer.allow_reuse_address = False
+        # Use _ExclusiveWSGIServer to fail fast if the address/port is occupied,
+        # and to prevent other apps from binding to it on Windows.
         local_server = wsgiref.simple_server.make_server(
-            bind_addr or host, port, wsgi_app, handler_class=_WSGIRequestHandler
+            bind_addr or host,
+            port,
+            wsgi_app,
+            server_class=_ExclusiveWSGIServer,
+            handler_class=_WSGIRequestHandler,
         )
 
         try:
@@ -478,6 +484,23 @@ def run_local_server(
         return self.credentials
 
 
+class _ExclusiveWSGIServer(wsgiref.simple_server.WSGIServer):
+    """Custom WSGIServer.
+
+    Enforces exclusive address binding on Windows.
+    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.
+    """
+
+    allow_reuse_address = False
+
+    def server_bind(self):
+        if sys.platform == "win32" and hasattr(socket, "SO_EXCLUSIVEADDRUSE"):
+            self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
+        super().server_bind()
+
+
 class _WSGIRequestHandler(wsgiref.simple_server.WSGIRequestHandler):
     """Custom WSGIRequestHandler.
 
 
@@ -20,9 +20,11 @@
 import os
 import re
 import socket
+import time
 from unittest import mock
 import urllib
 import webbrowser
+import wsgiref.simple_server
 
 import pytest
 import requests
@@ -444,10 +446,56 @@ def test_run_local_server_occupied_port(
         self, webbrowser_mock, instance, mock_fetch_token, port, socket
     ):
         # socket fixture is already bound to http://localhost:port
-        instance.run_local_server
-        with pytest.raises(OSError) as exc:
-            instance.run_local_server(port=port)
-            assert "address already in use" in exc.strerror.lower()
+        with pytest.raises(OSError):
+            instance.run_local_server(port=port, timeout_seconds=1)
+
+    @pytest.mark.webtest
+    @mock.patch("google_auth_oauthlib.flow.webbrowser", autospec=True)
+    def test_run_local_server_exclusive_port(
+        self, webbrowser_mock, instance, mock_fetch_token, port
+    ):
+        """Verify that while run_local_server is running, another socket cannot bind to its port."""
+        auth_redirect_url = urllib.parse.urljoin(
+            f"http://localhost:{port}", self.REDIRECT_REQUEST_PATH
+        )
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
+            future = pool.submit(partial(instance.run_local_server, port=port))
+            time.sleep(0.2)
+
+            hijack_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            hijack_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            try:
+                with pytest.raises(OSError):
+                    hijack_socket.bind(("localhost", port))
+            finally:
+                hijack_socket.close()
+                while not future.done():
+                    try:
+                        requests.get(auth_redirect_url)
+                    except requests.ConnectionError:  # pragma: NO COVER
+                        pass
+
+            credentials = future.result()
+
+        assert credentials.token == mock.sentinel.access_token
+
+    @mock.patch("google_auth_oauthlib.flow.webbrowser", autospec=True)
+    @mock.patch("wsgiref.simple_server.make_server", autospec=True)
+    def test_run_local_server_uses_exclusive_server_class(
+        self, make_server_mock, webbrowser_mock, instance
+    ):
+        server_mock = mock.MagicMock()
+        make_server_mock.return_value = server_mock
+
+        with pytest.raises(Exception):
+            instance.run_local_server(port=0)
+
+        make_server_mock.assert_called_once()
+        assert (
+            make_server_mock.call_args.kwargs.get("server_class")
+            is flow._ExclusiveWSGIServer
+        )
 
     @mock.patch("google_auth_oauthlib.flow.webbrowser.get", autospec=True)
     @mock.patch("wsgiref.simple_server.make_server", autospec=True)
@@ -514,3 +562,34 @@ def test_run_local_server_timeout(
 
         webbrowser_mock.get.assert_called_with(None)
         webbrowser_mock.get.return_value.open.assert_called_once()
+
+
+class TestExclusiveWSGIServer(object):
+    def test_exclusive_wsgi_server_bind_windows(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 = getattr(socket, "SO_EXCLUSIVEADDRUSE", 1)
+
+            server = flow._ExclusiveWSGIServer(
+                ("localhost", 0), flow._WSGIRequestHandler, bind_and_activate=False
+            )
+            server.socket = mock.Mock()
+
+            with mock.patch.object(wsgiref.simple_server.WSGIServer, "server_bind"):
+                server.server_bind()
+                server.socket.setsockopt.assert_called_once_with(
+                    mock_socket.SOL_SOCKET, mock_socket.SO_EXCLUSIVEADDRUSE, 1
+                )
+
+    def test_exclusive_wsgi_server_bind_non_windows(self):
+        with mock.patch("sys.platform", "linux"):
+            server = flow._ExclusiveWSGIServer(
+                ("localhost", 0), flow._WSGIRequestHandler, bind_and_activate=False
+            )
+            server.socket = mock.Mock()
+
+            with mock.patch.object(wsgiref.simple_server.WSGIServer, "server_bind"):
+                server.server_bind()
+                server.socket.setsockopt.assert_not_called()