feat(core): resolve run_command executables by name (#12)

Executables were run through hardcoded absolute paths, it only works on some distributions.
`ss` is in `/usr/bin` on the GitHub Ubuntu runners, instead of in `/usr/sbin`
on XCP-ng so the LINSTOR controller tests are marked as failed.

So a `find_executable` helper is used now to search for a program in:
`/usr/sbin`, `/sbin`, `/usr/bin` and `/bin`.

All `_EXEC_PATH_*` are renamed to `_EXECUTABLE_*` and they now hold only a name.

Signed-off-by: Ronan Abhamon <ronan.abhamon@vates.tech>

Author: Wescoeur

semgrep.yml

@@ -31,3 +31,21 @@ rules:
       - pattern-not-inside: |
           class $FLAG(..., IntFlag, ...):
               ...
+
+  - id: python-executable-constant-must-be-a-name
+    languages: [python]
+    severity: WARNING
+    message: >-
+      `$CONST` must contain only an executable name, not a path: `$VAL`.
+    patterns:
+      - pattern-either:
+          - pattern: "$CONST = $VAL"
+          - pattern: "$CONST: $TYPE = $VAL"
+
+      - metavariable-regex:
+          metavariable: $CONST
+          regex: "^_?EXECUTABLE_[A-Z0-9_]+$"
+
+      - metavariable-regex:
+          metavariable: $VAL
+          regex: "^[\"'](?![A-Za-z0-9][A-Za-z0-9._+-]*[\"']$)"

src/xcp_storage/backends/drbd/__init__.py

@@ -44,7 +44,7 @@
 
 # ------------------------------------------------------------------------------
 
-_EXEC_PATH_DRBDSETUP: Final = "/usr/sbin/drbdsetup"
+_EXECUTABLE_DRBDSETUP: Final = "drbdsetup"
 
 _REGEX_DRBD_OPENER_LINE: Final = re.compile(r"(.*)\s+(\d+)\s+(\d+)")
 
@@ -90,7 +90,7 @@ def _handle_drbd_json_error() -> Iterator[None]:
 def _get_drbd_status(resource_name: str) -> Dict[str, Any]:
     try:
         stdout, stderr, ret_code = run_command([
-            _EXEC_PATH_DRBDSETUP, "status", resource_name, "--json"
+            _EXECUTABLE_DRBDSETUP, "status", resource_name, "--json"
         ], simple=False)
         if ret_code != 0:
             logger.warning(
@@ -224,7 +224,7 @@ def demote(resource_name: str) -> bool:
         _check_drbd_resource_name(resource_name)
         error_message = ""
         try:
-            _stdout, stderr, ret_code = run_command([_EXEC_PATH_DRBDSETUP, "secondary", resource_name], simple=False)
+            _stdout, stderr, ret_code = run_command([_EXECUTABLE_DRBDSETUP, "secondary", resource_name], simple=False)
             if not ret_code:
                 return True
             error_message = stderr

src/xcp_storage/backends/linstor/controller.py

@@ -30,7 +30,7 @@
 
 # ------------------------------------------------------------------------------
 
-_EXEC_PATH_SS: Final = "/usr/sbin/ss"
+_EXECUTABLE_SS: Final = "ss"
 
 _SERVICE_LINSTOR_CONTROLLER: Final = "linstor-controller"
 
@@ -40,7 +40,7 @@ class LinstorController:
     @staticmethod
     def get_addresses() -> List[str]:
         stdout = run_command([
-            _EXEC_PATH_SS, "-tnpH", "state", "established",
+            _EXECUTABLE_SS, "-tnpH", "state", "established",
             f"( sport = :{LINSTOR_SATELLITE_PORT_PLAIN} or sport = :{LINSTOR_SATELLITE_PORT_SSL} )"
         ], expected_ret_code=0)
         return [

src/xcp_storage/network/iptables.py

@@ -26,8 +26,8 @@
 
 _MAIN_FIREWALL_INPUT_CHAIN: Final = "INPUT"
 
-_EXEC_PATH_IPTABLES: Final = "/usr/sbin/iptables"
-_EXEC_PATH_SERVICE: Final = "/usr/sbin/service"
+_EXECUTABLE_IPTABLES: Final = "iptables"
+_EXECUTABLE_SERVICE: Final = "service"
 
 _PROTOCOL_TCP: Final = "tcp"
 
@@ -42,7 +42,7 @@ def __init__(self, message: str, code: Optional[int] = None) -> None:
 
 def _save_iptables_changes() -> None:
     try:
-        run_command([_EXEC_PATH_SERVICE, "iptables", "save"], expected_ret_code=0)
+        run_command([_EXECUTABLE_SERVICE, "iptables", "save"], expected_ret_code=0)
     except CommandError as e:
         raise IptablesError(f"Failed to save iptables changes: `{e.reason}`.", e.code) from None
 
@@ -62,7 +62,7 @@ def _is_fatal_iptables_code(code: int) -> bool:
 def has_iptables_rule(rule: List[str]) -> bool:
     code: Optional[int]
     try:
-        _stdout, reason, code = run_command([_EXEC_PATH_IPTABLES, "-C"] + rule, simple=False)
+        _stdout, reason, code = run_command([_EXECUTABLE_IPTABLES, "-C"] + rule, simple=False)
         if not _is_fatal_iptables_code(code):
             return not code
     except CommandError as e:
@@ -73,19 +73,19 @@ def has_iptables_rule(rule: List[str]) -> bool:
 
 def _ensure_iptables_chain_exists(chain: str) -> bool:
     try:
-        _stdout, _stderr, ret_code = run_command([_EXEC_PATH_IPTABLES, "-N", chain], simple=False)
+        _stdout, _stderr, ret_code = run_command([_EXECUTABLE_IPTABLES, "-N", chain], simple=False)
         if _is_fatal_iptables_code(ret_code):
             raise IptablesError(f"Failed to test existence of iptables chain `{chain}`: `{_stderr}`.", ret_code)
         updated = not ret_code
 
         return_rule = [chain, "-j", "RETURN"]
         if not has_iptables_rule(return_rule):
-            run_command([_EXEC_PATH_IPTABLES, "-A"] + return_rule, expected_ret_code=0)
+            run_command([_EXECUTABLE_IPTABLES, "-A"] + return_rule, expected_ret_code=0)
             updated = True
 
         jump_rule = [_MAIN_FIREWALL_INPUT_CHAIN, "-j", chain]
         if not has_iptables_rule(jump_rule):
-            run_command([_EXEC_PATH_IPTABLES, "-I"] + jump_rule, expected_ret_code=0)
+            run_command([_EXECUTABLE_IPTABLES, "-I"] + jump_rule, expected_ret_code=0)
             updated = True
     except CommandError as e:
         raise IptablesError(f"Failed to set up iptables chain `{chain}`: `{e.reason}`.", e.code) from None
@@ -107,14 +107,14 @@ def _update_iptables_ports(protocol: str, str_ports: str, *, open_ports: bool, s
     if open_ports:
         _ensure_iptables_chain_exists(chain)
         try:
-            run_command([_EXEC_PATH_IPTABLES, "-I"] + rule, expected_ret_code=0)
+            run_command([_EXECUTABLE_IPTABLES, "-I"] + rule, expected_ret_code=0)
         except CommandError as e:
             raise IptablesError(
                 f"Failed to open {protocol.upper()} port(s): `{e.reason}`.", e.code
             ) from None
     else:
         try:
-            run_command([_EXEC_PATH_IPTABLES, "-D"] + rule, expected_ret_code=0)
+            run_command([_EXECUTABLE_IPTABLES, "-D"] + rule, expected_ret_code=0)
         except CommandError as e:
             raise IptablesError(
                 f"Failed to close {protocol.upper()} port(s): `{e.reason}`.", e.code

src/xcp_storage/utils/process.py

@@ -12,13 +12,16 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
+import os
 from pathlib import Path
+import shutil
 import subprocess
 
 import xcp_storage.log as log
 
 from xcp_storage.typing import (
     Callable,
+    Final,
     List,
     Literal,
     Optional,
@@ -33,6 +36,14 @@
 
 # ------------------------------------------------------------------------------
 
+# Trusted directories, searched in this order, to find the executables run by `run_command`.
+_EXECUTABLE_DIRS: Final = ("/usr/sbin", "/usr/bin", "/sbin", "/bin")
+
+def find_executable(name: str) -> Optional[str]:
+    return shutil.which(name, path=os.pathsep.join(_EXECUTABLE_DIRS))
+
+# ------------------------------------------------------------------------------
+
 class CommandError(Exception):
     def __init__(self, code: Optional[int], cmd: str, reason: str) -> None:
         super().__init__("Command execution error.")
@@ -90,6 +101,17 @@ def run_internal_command(
     ret_code_callback: Callable[[str, str, int], int] = default_ret_code_callback,
     quiet: bool = False
 ) -> CommandResultType:
+    executable_name = args[0]
+
+    # A bare program name is resolved to an absolute path; an explicit path is kept as is.
+    if os.sep not in executable_name:
+        exec_path = find_executable(executable_name)
+        if exec_path is None:
+            raise CommandError(
+                None, str(args), reason=f"Executable `{executable_name}` not found in `{', '.join(_EXECUTABLE_DIRS)}`."
+            )
+        args = [exec_path, *args[1:]]
+
     try:
         # TODO(XCPNG-3032): Remove noqa warn and use `capture_output` after we discontinue support for Python 3.6.
         result = subprocess.run( # noqa: UP022

src/xcp_storage/utils/service.py

@@ -22,7 +22,7 @@
 
 # ==============================================================================
 
-_EXEC_PATH_SYSTEMCTL: Final = "/usr/bin/systemctl"
+_EXECUTABLE_SYSTEMCTL: Final = "systemctl"
 
 # ------------------------------------------------------------------------------
 
@@ -34,7 +34,7 @@ def __init__(self, message: str, code: Optional[int] = None) -> None:
 # ------------------------------------------------------------------------------
 
 def _run_service_command(service_name: str, action: List[str], *, quiet: bool = False) -> None:
-    args = [_EXEC_PATH_SYSTEMCTL, *action, service_name]
+    args = [_EXECUTABLE_SYSTEMCTL, *action, service_name]
     try:
         run_command(args, expected_ret_code=0, quiet=quiet)
     except CommandError as e:
@@ -77,6 +77,6 @@ def disable_and_stop_service(service_name: str) -> None:
 
 def reload_service_conf() -> None:
     try:
-        run_command([_EXEC_PATH_SYSTEMCTL, "daemon-reload"], expected_ret_code=0)
+        run_command([_EXECUTABLE_SYSTEMCTL, "daemon-reload"], expected_ret_code=0)
     except CommandError as e:
         raise ServiceError(e.reason, e.code) from None

tests/backends/test_drbd.py

@@ -22,6 +22,7 @@
 
 from xcp_storage.backends.drbd import (
     _get_drbd_status,
+    _EXECUTABLE_DRBDSETUP,
     Drbd,
     DrbdOpener,
 )
@@ -87,7 +88,7 @@ def test_success(self, mock_run_command: MagicMock, drbd_json_status_primary: st
         assert status["name"] == "xcp-volume-patate"
         assert status["role"] == "Primary"
         mock_run_command.assert_called_once_with(
-            ["/usr/sbin/drbdsetup", "status", "xcp-volume-patate", "--json"], simple=False
+            [_EXECUTABLE_DRBDSETUP, "status", "xcp-volume-patate", "--json"], simple=False
         )
 
     def test_command_failure(self, mock_run_command: MagicMock, caplog: pytest.LogCaptureFixture) -> None:

tests/network/test_iptables.py

@@ -22,8 +22,8 @@
 import pytest
 
 from xcp_storage.network.iptables import (
-    _EXEC_PATH_IPTABLES,
-    _EXEC_PATH_SERVICE,
+    _EXECUTABLE_IPTABLES,
+    _EXECUTABLE_SERVICE,
     _MAIN_FIREWALL_INPUT_CHAIN,
     _PROTOCOL_TCP,
     DEFAULT_FIREWALL_INPUT_CHAIN,
@@ -74,22 +74,22 @@ def get_tcp_rule(ports_str: str, *, stateful: bool = True, chain: str = DEFAULT_
 # ------------------------------------------------------------------------------
 
 def get_has_chain_cmd(chain: str = DEFAULT_FIREWALL_INPUT_CHAIN) -> List[str]:
-    return [_EXEC_PATH_IPTABLES, "-N", chain]
+    return [_EXECUTABLE_IPTABLES, "-N", chain]
 
 def get_has_rule_cmd(rule: List[str]) -> List[str]:
-    return [_EXEC_PATH_IPTABLES, "-C"] + rule
+    return [_EXECUTABLE_IPTABLES, "-C"] + rule
 
 def get_add_rule_cmd(rule: List[str]) -> List[str]:
-    return [_EXEC_PATH_IPTABLES, "-A"] + rule
+    return [_EXECUTABLE_IPTABLES, "-A"] + rule
 
 def get_insert_rule_cmd(rule: List[str]) -> List[str]:
-    return [_EXEC_PATH_IPTABLES, "-I"] + rule
+    return [_EXECUTABLE_IPTABLES, "-I"] + rule
 
 def get_destroy_rule_cmd(rule: List[str]) -> List[str]:
-    return [_EXEC_PATH_IPTABLES, "-D"] + rule
+    return [_EXECUTABLE_IPTABLES, "-D"] + rule
 
 def get_save_iptables_cmd() -> List[str]:
-    return [_EXEC_PATH_SERVICE, "iptables", "save"]
+    return [_EXECUTABLE_SERVICE, "iptables", "save"]
 
 # ------------------------------------------------------------------------------
 
@@ -159,7 +159,7 @@ def impl(*args: P.args, **kwargs: P.kwargs) -> CommandResultType:
         cmd_args = cast(List[str], args[0])
 
         program_name = cmd_args[0]
-        assert program_name in (_EXEC_PATH_IPTABLES, _EXEC_PATH_SERVICE)
+        assert program_name in (_EXECUTABLE_IPTABLES, _EXECUTABLE_SERVICE)
 
         simple = kwargs.get("simple", True)
 

tests/utils/test_service.py

@@ -18,7 +18,7 @@
 )
 
 from xcp_storage.utils.service import (
-    _EXEC_PATH_SYSTEMCTL,
+    _EXECUTABLE_SYSTEMCTL,
     disable_and_stop_service,
     enable_and_start_service,
     escape_service_instance,
@@ -54,7 +54,7 @@ def _assert_service_command(
         quiet: bool
     ) -> None:
         args.append(self.SERVICE_NAME)
-        args.insert(0, _EXEC_PATH_SYSTEMCTL)
+        args.insert(0, _EXECUTABLE_SYSTEMCTL)
         mock_run_command.assert_called_once_with(args, expected_ret_code=expected_ret_code, quiet=quiet)
 
     def _assert_query_command(self, mock_run_command: MagicMock, args: List[str]) -> None:
@@ -97,4 +97,4 @@ def test_disable_and_stop_service(self, mock_run_command: MagicMock) -> None:
 
     def test_reload_service_conf(self, mock_run_command: MagicMock) -> None:
         reload_service_conf()
-        mock_run_command.assert_called_once_with([_EXEC_PATH_SYSTEMCTL, "daemon-reload"], expected_ret_code=0)
+        mock_run_command.assert_called_once_with([_EXECUTABLE_SYSTEMCTL, "daemon-reload"], expected_ret_code=0)