Add validation for invalid ChromeDriver executable path

py/selenium/webdriver/common/service.py

@@ -35,17 +35,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,
@@ -65,7 +54,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
@@ -74,12 +62,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
@@ -91,12 +77,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)
@@ -106,26 +86,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:
@@ -137,8 +111,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()
@@ -154,13 +126,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,
@@ -185,11 +150,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:
@@ -198,11 +158,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":
@@ -232,8 +199,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/chrome/chrome_service_tests.py

@@ -109,6 +109,9 @@ def test_log_output_null_default(driver, capfd) -> None:
 
 @pytest.mark.no_driver_after_test
 def test_driver_is_stopped_if_browser_cant_start(clean_driver, clean_options, driver_executable) -> None:
+    """
+    Tests that the ChromeDriver process stops cleanly when browser startup fails.
+    """
     clean_options.add_argument("--user-data-dir=/no/such/location")
     service = Service(executable_path=driver_executable)
     with pytest.raises(SessionNotCreatedException):
@@ -117,6 +120,15 @@ def test_driver_is_stopped_if_browser_cant_start(clean_driver, clean_options, dr
     assert service.process.poll() is not None
 
 
+def test_invalid_executable_raises_exception():
+    """
+    Test that Service raises ValueError when given a non-existent or non-executable path.
+    """
+    bad_path = "/invalid/path/to/chromedriver"
+    with pytest.raises(ValueError):
+        Service(executable_path=bad_path)
+
+
 @pytest.fixture
 def service():
     return Service()
@@ -137,5 +149,5 @@ def test_uses_path_from_env_variable(self, service):
 
     def test_updates_path_after_setting_env_variable(self, service):
         service.executable_path = self.service_path  # Simulating the update
-        with patch.dict("os.environ", {"SE_CHROMEDRIVER": "/foo/bar"}):
-            assert "chromedriver" in service.executable_path
+        with patch.dict("os.environ", {"SE_CHROMEDRIVER": "/foo/bar/chromedriver"}):
+            assert "chromedriver" in service.executable_path