Add validation for invalid ChromeDriver executable path

py/selenium/webdriver/common/service.py

@@ -24,13 +24,7 @@
 from platform import system
 from subprocess import PIPE
 from time import sleep
-from typing import IO
-from typing import Any
-from typing import List
-from typing import Mapping
-from typing import Optional
-from typing import Union
-from typing import cast
+from typing import IO, Any, List, Mapping, Optional, Union, cast
 from urllib import request
 from urllib.error import URLError
 
@@ -42,17 +36,6 @@
 
 
 class Service(ABC):
-    """The abstract base class for all service objects.  Services typically
-    launch a child program in a new process as an interim process to
-    communicate with a browser.
-
-    :param executable: install path of the executable.
-    :param port: Port for the service to run on, defaults to 0 where the operating system will decide.
-    :param log_output: (Optional) int representation of STDOUT/DEVNULL, any IO instance or String path to file.
-    :param env: (Optional) Mapping of environment variables for the new process, defaults to `os.environ`.
-    :param driver_path_env_key: (Optional) Environment variable to use to get the path to the driver executable.
-    """
-
     def __init__(
         self,
         executable_path: Optional[str] = None,
@@ -72,7 +55,6 @@ def __init__(
             self.log_output = log_output
 
         self.port = port or utils.free_port()
-        # Default value for every python subprocess: subprocess.Popen(..., creationflags=0)
         self.popen_kw = kwargs.pop("popen_kw", {})
         self.creation_flags = self.popen_kw.pop("creation_flags", 0)
         self.env = env or os.environ
@@ -81,12 +63,10 @@ def __init__(
 
     @property
     def service_url(self) -> str:
-        """Gets the url of the Service."""
         return f"http://{utils.join_host_port('localhost', self.port)}"
 
     @abstractmethod
     def command_line_args(self) -> List[str]:
-        """A List of program arguments (excluding the executable)."""
         raise NotImplementedError("This method needs to be implemented in a sub class")
 
     @property
@@ -98,12 +78,6 @@ def path(self, value: str) -> None:
         self._path = str(value)
 
     def start(self) -> None:
-        """Starts the Service.
-
-        :Exceptions:
-         - WebDriverException : Raised either when it can't start the service
-           or when it can't connect to the service
-        """
         if self._path is None:
             raise WebDriverException("Service path cannot be None.")
         self._start_process(self._path)
@@ -113,26 +87,20 @@ def start(self) -> None:
             self.assert_process_still_running()
             if self.is_connectable():
                 break
-            # sleep increasing: 0.01, 0.06, 0.11, 0.16, 0.21, 0.26, 0.31, 0.36, 0.41, 0.46, 0.5
             sleep(min(0.01 + 0.05 * count, 0.5))
             count += 1
             if count == 70:
                 raise WebDriverException(f"Can not connect to the Service {self._path}")
 
     def assert_process_still_running(self) -> None:
-        """Check if the underlying process is still running."""
         return_code = self.process.poll()
         if return_code:
             raise WebDriverException(f"Service {self._path} unexpectedly exited. Status code was: {return_code}")
 
     def is_connectable(self) -> bool:
-        """Establishes a socket connection to determine if the service running
-        on the port is accessible."""
         return utils.is_connectable(self.port)
 
     def send_remote_shutdown_command(self) -> None:
-        """Dispatch an HTTP request to the shutdown endpoint for the service in
-        an attempt to stop it."""
         try:
             request.urlopen(f"{self.service_url}/shutdown")
         except URLError:
@@ -144,8 +112,6 @@ def send_remote_shutdown_command(self) -> None:
             sleep(1)
 
     def stop(self) -> None:
-        """Stops the service."""
-
         if self.log_output not in {PIPE, subprocess.DEVNULL}:
             if isinstance(self.log_output, IOBase):
                 self.log_output.close()
@@ -161,13 +127,6 @@ def stop(self) -> None:
                 self._terminate_process()
 
     def _terminate_process(self) -> None:
-        """Terminate the child process.
-
-        On POSIX this attempts a graceful SIGTERM followed by a SIGKILL,
-        on a Windows OS kill is an alias to terminate.  Terminating does
-        not raise itself if something has gone wrong but (currently)
-        silently ignores errors here.
-        """
         try:
             stdin, stdout, stderr = (
                 self.process.stdin,
@@ -192,11 +151,6 @@ def _terminate_process(self) -> None:
             logger.error("Error terminating service process.", exc_info=True)
 
     def __del__(self) -> None:
-        # `subprocess.Popen` doesn't send signal on `__del__`;
-        # so we attempt to close the launched process when `__del__`
-        # is triggered.
-        # do not use globals here; interpreter shutdown may have already cleaned them up
-        # and they would be `None`. This goes for anything this method is referencing internally.
         try:
             self.stop()
         except Exception:
@@ -205,11 +159,18 @@ def __del__(self) -> None:
     def _start_process(self, path: str) -> None:
         """Creates a subprocess by executing the command provided.
 
-        :param cmd: full command to execute
+        :param path: full command to execute
         """
+        # Pre-check: validate path exists and is executable
+        if not os.path.isfile(path):
+            raise WebDriverException(f"The executable at path '{path}' does not exist.")
+        if not os.access(path, os.X_OK):
+            raise WebDriverException(f"The executable at path '{path}' is not executable.")
+
         cmd = [path]
         cmd.extend(self.command_line_args())
         close_file_descriptors = self.popen_kw.pop("close_fds", system() != "Windows")
+
         try:
             start_info = None
             if system() == "Windows":
@@ -239,8 +200,6 @@ def _start_process(self, path: str) -> None:
             raise
         except OSError as err:
             if err.errno == errno.EACCES:
-                if self._path is None:
-                    raise WebDriverException("Service path cannot be None.")
                 raise WebDriverException(
                     f"'{os.path.basename(self._path)}' executable may have wrong permissions."
                 ) from err

py/test/selenium/webdriver/common/test_service.py

@@ -0,0 +1,9 @@
+def test_reusing_closed_stdout_fails():
+    import sys
+    from selenium.webdriver.chrome.service import Service
+    from selenium.common.exceptions import WebDriverException
+
+    service = Service(log_output=sys.stdout)
+    service._log_output.close()
+    with pytest.raises(ValueError):
+        Service(log_output=sys.stdout)