Fix connection leak issue on windows

undetected_chromedriver/__init__.py

@@ -16,7 +16,6 @@
 """
 from __future__ import annotations
 
-
 __version__ = "3.5.3"
 
 import json
@@ -33,20 +32,17 @@
 
 import selenium.webdriver.chrome.service
 import selenium.webdriver.chrome.webdriver
-from selenium.webdriver.common.by import By
 import selenium.webdriver.chromium.service
 import selenium.webdriver.remote.command
 import selenium.webdriver.remote.webdriver
+from selenium.webdriver.common.utils import free_port
 
 from .cdp import CDP
 from .dprocess import start_detached
 from .options import ChromeOptions
-from .patcher import IS_POSIX
-from .patcher import Patcher
+from .patcher import IS_POSIX, Patcher
 from .reactor import Reactor
-from .webelement import UCWebElement
-from .webelement import WebElement
-
+from .webelement import UCWebElement, WebElement
 
 __all__ = (
     "Chrome",
@@ -270,14 +266,10 @@ def __init__(
         except AttributeError:
             pass
 
-        options._session = self
+        options._session = self  # type: ignore
 
         if not options.debugger_address:
-            debug_port = (
-                port
-                if port != 0
-                else selenium.webdriver.common.service.utils.free_port()
-            )
+            debug_port = port if port != 0 else free_port()
             debug_host = "127.0.0.1"
             options.debugger_address = "%s:%d" % (debug_host, debug_port)
         else:
@@ -299,13 +291,13 @@ def __init__(
 
         # see if a custom user profile is specified in options
         for arg in options.arguments:
-
             if any([_ in arg for _ in ("--headless", "headless")]):
                 options.arguments.remove(arg)
                 options.headless = True
 
             if "lang" in arg:
                 m = re.search("(?:--)?lang(?:[ =])?(.*)", arg)
+                assert m is not None
                 try:
                     language = m[1]
                 except IndexError:
@@ -314,6 +306,7 @@ def __init__(
 
             if "user-data-dir" in arg:
                 m = re.search("(?:--)?user-data-dir(?:[ =])?(.*)", arg)
+                assert m is not None
                 try:
                     user_data_dir = m[1]
                     logger.debug(
@@ -360,7 +353,10 @@ def __init__(
             try:
                 import locale
 
-                language = locale.getdefaultlocale()[0].replace("_", "-")
+                default_langs = locale.getdefaultlocale()
+                assert len(default_langs) > 1
+                assert default_langs[0]
+                language = default_langs[0].replace("_", "-")
             except Exception:
                 pass
             if not language:
@@ -369,21 +365,26 @@ def __init__(
         options.add_argument("--lang=%s" % language)
 
         if not options.binary_location:
-            options.binary_location = (
+            options.binary_location = (  # type: ignore
                 browser_executable_path or find_chrome_executable()
             )
 
-        if not options.binary_location or not \
-                pathlib.Path(options.binary_location).exists():
-                raise FileNotFoundError(
-                    "\n---------------------\n"
-                    "Could not determine browser executable."
-                    "\n---------------------\n"
-                    "Make sure your browser is installed in the default location (path).\n"
-                    "If you are sure about the browser executable, you can specify it using\n"
-                    "the `browser_executable_path='{}` parameter.\n\n"
-                    .format("/path/to/browser/executable" if IS_POSIX else "c:/path/to/your/browser.exe")
+        if (
+            not options.binary_location
+            or not pathlib.Path(options.binary_location).exists()
+        ):
+            raise FileNotFoundError(
+                "\n---------------------\n"
+                "Could not determine browser executable."
+                "\n---------------------\n"
+                "Make sure your browser is installed in the default location (path).\n"
+                "If you are sure about the browser executable, you can specify it using\n"
+                "the `browser_executable_path='{}` parameter.\n\n".format(
+                    "/path/to/browser/executable"
+                    if IS_POSIX
+                    else "c:/path/to/your/browser.exe"
                 )
+            )
 
         self._delay = 3
 
@@ -396,15 +397,18 @@ def __init__(
             options.arguments.extend(["--no-sandbox", "--test-type"])
 
         if headless or options.headless:
-            #workaround until a better checking is found
+            # workaround until a better checking is found
             try:
-                if self.patcher.version_main < 108:
-                    options.add_argument("--headless=chrome")
-                elif self.patcher.version_main >= 108:
-                    options.add_argument("--headless=new")
-            except:
-                logger.warning("could not detect version_main."
-                               "therefore, we are assuming it is chrome 108 or higher")
+                if isinstance(self.patcher.version_main, int):
+                    if self.patcher.version_main < 108:
+                        options.add_argument("--headless=chrome")
+                    elif self.patcher.version_main >= 108:
+                        options.add_argument("--headless=new")
+            except Exception:
+                logger.warning(
+                    "could not detect version_main."
+                    "therefore, we are assuming it is chrome 108 or higher"
+                )
                 options.add_argument("--headless=new")
 
         options.add_argument("--window-size=1920,1080")
@@ -424,7 +428,7 @@ def __init__(
         # fix exit_type flag to prevent tab-restore nag
         try:
             with open(
-                os.path.join(user_data_dir, "Default/Preferences"),
+                os.path.join(user_data_dir, "Default/Preferences"),  # type: ignore
                 encoding="latin1",
                 mode="r+",
             ) as fs:
@@ -436,7 +440,7 @@ def __init__(
                 json.dump(config, fs)
                 fs.truncate()  # the file might be shorter
                 logger.debug("fixed exit_type flag")
-        except Exception as e:
+        except Exception:
             logger.debug("did not find a bad exit_type flag ")
 
         self.options = options
@@ -458,14 +462,13 @@ def __init__(
             )
             self.browser_pid = browser.pid
 
-
         service = selenium.webdriver.chromium.service.ChromiumService(
             self.patcher.executable_path
         )
 
         super(Chrome, self).__init__(
-            service=service,
-            options=options,
+            service=service,  # type: ignore
+            options=options,  # type: ignore
             keep_alive=keep_alive,
         )
 
@@ -739,6 +742,7 @@ def find_elements_recursive(self, by, value):
             value: str
         Returns: Generator[webelement.WebElement]
         """
+
         def search_frame(f=None):
             if not f:
                 # ensure we are on main content frame
@@ -754,7 +758,7 @@ def search_frame(f=None):
         for elem in search_frame():
             yield elem
         # get iframes
-        frames = self.find_elements('css selector', 'iframe')
+        frames = self.find_elements("css selector", "iframe")
 
         # search per frame
         for f in frames:
@@ -768,10 +772,12 @@ def quit(self):
         except (AttributeError, RuntimeError, OSError):
             pass
         try:
-            self.reactor.event.set()
-            logger.debug("shutting down reactor")
+            if self.reactor:
+                self.reactor.event.set()
+                logger.debug("shutting down reactor")
         except AttributeError:
             pass
+
         try:
             os.kill(self.browser_pid, 15)
             logger.debug("gracefully closed browser")
@@ -781,6 +787,7 @@ def quit(self):
             hasattr(self, "keep_user_data_dir")
             and hasattr(self, "user_data_dir")
             and not self.keep_user_data_dir
+            and self.user_data_dir
         ):
             for _ in range(5):
                 try:
@@ -866,7 +873,9 @@ def find_chrome_executable():
     """
     candidates = set()
     if IS_POSIX:
-        for item in os.environ.get("PATH").split(os.pathsep):
+        path = os.environ.get("PATH")
+        assert path is not None
+        for item in path.split(os.pathsep):
             for subitem in (
                 "google-chrome",
                 "chromium",
@@ -888,12 +897,10 @@ def find_chrome_executable():
             ("PROGRAMFILES", "PROGRAMFILES(X86)", "LOCALAPPDATA", "PROGRAMW6432"),
         ):
             if item is not None:
-                for subitem in (
-                    "Google/Chrome/Application",
-                ):
+                for subitem in ("Google/Chrome/Application",):
                     candidates.add(os.sep.join((item, subitem, "chrome.exe")))
     for candidate in candidates:
-        logger.debug('checking if %s exists and is executable' % candidate)
+        logger.debug("checking if %s exists and is executable" % candidate)
         if os.path.exists(candidate) and os.access(candidate, os.X_OK):
-            logger.debug('found! using %s' % candidate)
+            logger.debug("found! using %s" % candidate)
             return os.path.normpath(candidate)

undetected_chromedriver/cdp.py

@@ -7,7 +7,6 @@
 import requests
 import websockets
 
-
 log = logging.getLogger(__name__)
 
 

undetected_chromedriver/devtool.py

@@ -1,16 +1,11 @@
 import asyncio
-from collections.abc import Mapping
-from collections.abc import Sequence
-from functools import wraps
 import logging
 import threading
 import time
 import traceback
-from typing import Any
-from typing import Awaitable
-from typing import Callable
-from typing import List
-from typing import Optional
+from collections.abc import Mapping, Sequence
+from functools import wraps
+from typing import Any, Awaitable, Callable, List, Optional
 
 
 class Structure(dict):
@@ -101,12 +96,14 @@ def function_reached_timeout():
 
 
 def test():
-    import sys, os
+    import os
+    import sys
 
     sys.path.insert(0, os.path.abspath(os.path.dirname(__file__)))
-    import undetected_chromedriver as uc
     import threading
 
+    import undetected_chromedriver as uc
+
     def collector(
         driver: uc.Chrome,
         stop_event: threading.Event,

undetected_chromedriver/dprocess.py

@@ -4,10 +4,8 @@
 import os
 import platform
 import signal
-from subprocess import PIPE
-from subprocess import Popen
 import sys
-
+from subprocess import PIPE, Popen
 
 CREATE_NEW_PROCESS_GROUP = 0x00000200
 DETACHED_PROCESS = 0x00000008

undetected_chromedriver/patcher.py

@@ -1,7 +1,6 @@
 #!/usr/bin/env python3
 # this module is part of undetected_chromedriver
 
-from distutils.version import LooseVersion
 import io
 import json
 import logging
@@ -14,10 +13,11 @@
 import string
 import sys
 import time
-from urllib.request import urlopen
-from urllib.request import urlretrieve
 import zipfile
+from distutils.version import LooseVersion
 from multiprocessing import Lock
+from typing import Optional
+from urllib.request import urlopen, urlretrieve
 
 logger = logging.getLogger(__name__)
 
@@ -45,7 +45,7 @@ def __init__(
         self,
         executable_path=None,
         force=False,
-        version_main: int = 0,
+        version_main: Optional[int] = 0,
         user_multi_procs=False,
     ):
         """
@@ -197,7 +197,6 @@ def driver_binary_in_use(self, path: str = None) -> bool:
             with open(p, mode="a+b") as fs:
                 exc = []
                 try:
-
                     fs.seek(0, 0)
                 except PermissionError as e:
                     exc.append(e)  # since some systems apprently allow seeking
@@ -208,7 +207,6 @@ def driver_binary_in_use(self, path: str = None) -> bool:
                     exc.append(e)
 
                 if exc:
-
                     return True
                 return False
             # ok safe to assume this is in use
@@ -260,7 +258,9 @@ def fetch_release_number(self):
             response = conn.read().decode()
 
         major_versions = json.loads(response)
-        return LooseVersion(major_versions["milestones"][str(self.version_main)]["version"])
+        return LooseVersion(
+            major_versions["milestones"][str(self.version_main)]["version"]
+        )
 
     def parse_exe_version(self):
         with io.open(self.executable_path, "rb") as f:
@@ -277,10 +277,16 @@ def fetch_package(self):
         """
         zip_name = f"chromedriver_{self.platform_name}.zip"
         if self.is_old_chromedriver:
-            download_url = "%s/%s/%s" % (self.url_repo, self.version_full.vstring, zip_name)
+            download_url = "%s/%s/%s" % (
+                self.url_repo,
+                self.version_full.vstring,
+                zip_name,
+            )
         else:
             zip_name = zip_name.replace("_", "-", 1)
-            download_url = "https://edgedl.me.gvt1.com/edgedl/chrome/chrome-for-testing/%s/%s/%s"
+            download_url = (
+                "https://edgedl.me.gvt1.com/edgedl/chrome/chrome-for-testing/%s/%s/%s"
+            )
             download_url %= (self.version_full.vstring, self.platform_name, zip_name)
 
         logger.debug("downloading from %s" % download_url)