ssh: only allow string command

List of strings was previously accepted, but was giving the false feeling
that we could avoid the command being parsed by the remote shell.
ssh doesn't support passing separated parameters. It reconstructs a string
with the arguments it gets from the command line separated by a single space.
This string is then passed to the ssh servers that passes it to the shell.

With this change, the string passed to the ssh() function is the exact
string received by the shell on the remote host.

Signed-off-by: Gaëtan Lehmann <gaetan.lehmann@vates.tech>

Author: glehmann

conftest.py

@@ -681,7 +681,7 @@ def uefi_vm(imported_vm):
     yield vm
 
 @pytest.fixture(scope='session')
-def additional_repos(request, hosts):
+def additional_repos(request, hosts: list[Host]):
     if request.param is None:
         yield []
         return
@@ -707,7 +707,7 @@ def additional_repos(request, hosts):
 
     for host in hosts:
         for host_ in host.pool.hosts:
-            host_.ssh(['rm', '-f', repo_file])
+            host_.ssh('rm -f {repo_file}')
 
 @pytest.fixture(scope='session')
 def second_network(pytestconfig, host):

jobs.py

@@ -544,7 +544,7 @@ def build_pytest_cmd(job_data, hosts=None, host_version=None, pytest_args=[]):
     if hosts is not None:
         try:
             host = hosts.split(',')[0]
-            cmd = ["lsb_release", "-sr"]
+            cmd = "lsb_release -sr"
             host_version = ssh(host, cmd)
         except Exception as e:
             print(e, file=sys.stderr)

lib/commands.py

@@ -1,14 +1,18 @@
+from __future__ import annotations
+
 import base64
 import logging
 import os
 import platform
 import subprocess
 
 import lib.config as config
-from lib.common import HostAddress
 from lib.netutil import wrap_ip
 
-from typing import List, Literal, Union, overload
+from typing import TYPE_CHECKING, List, Literal, Union, overload
+
+if TYPE_CHECKING:
+    from lib.common import HostAddress
 
 class BaseCommandFailed(Exception):
     __slots__ = 'returncode', 'stdout', 'cmd'
@@ -63,8 +67,17 @@ def _ellide_log_lines(log):
         reduced_message.append("(...)")
     return "\n{}".format("\n".join(reduced_message))
 
-def _ssh(hostname_or_ip, cmd, check, simple_output, suppress_fingerprint_warnings,
-         background, decode, options, multiplexing) -> Union[SSHResult, SSHCommandFailed, str, bytes, None]:
+def _ssh(
+    hostname_or_ip: str,
+    cmd: str,
+    check: bool,
+    simple_output: bool,
+    suppress_fingerprint_warnings: bool,
+    background: bool,
+    decode: bool,
+    options: list[str],
+    multiplexing: bool,
+) -> SSHResult | SSHCommandFailed | str | bytes | None:
     opts = list(options)
     opts += ['-o', 'BatchMode yes']
     opts += ['-o', 'PubkeyAcceptedAlgorithms +ssh-rsa']
@@ -87,12 +100,7 @@ def _ssh(hostname_or_ip, cmd, check, simple_output, suppress_fingerprint_warning
     else:
         opts += ['-o', 'ControlMaster no']
 
-    if isinstance(cmd, str):
-        command = [cmd]
-    else:
-        command = cmd
-
-    ssh_cmd = ['ssh', f'root@{hostname_or_ip}'] + opts + command
+    ssh_cmd = ['ssh', f'root@{hostname_or_ip}'] + opts + [cmd]
 
     # Fetch banner and remove it to avoid stdout/stderr pollution.
     banner_res = None
@@ -104,7 +112,7 @@ def _ssh(hostname_or_ip, cmd, check, simple_output, suppress_fingerprint_warning
             check=False
         )
 
-    logging.debug(f"[{hostname_or_ip}] {' '.join(command)}")
+    logging.debug(f"[{hostname_or_ip}] {cmd}")
     process = subprocess.Popen(
         ssh_cmd,
         stdout=subprocess.PIPE,
@@ -127,20 +135,20 @@ def _ssh(hostname_or_ip, cmd, check, simple_output, suppress_fingerprint_warning
 
     # Even if check is False, we still raise in case of return code 255, which means a SSH error.
     if res.returncode == 255:
-        return SSHCommandFailed(255, "SSH Error: %s" % output_for_errors, command)
+        return SSHCommandFailed(255, "SSH Error: %s" % output_for_errors, cmd)
 
     output: Union[bytes, str] = res.stdout
     if banner_res:
         if banner_res.returncode == 255:
-            return SSHCommandFailed(255, "SSH Error: %s" % banner_res.stdout.decode(errors='replace'), command)
+            return SSHCommandFailed(255, "SSH Error: %s" % banner_res.stdout.decode(errors='replace'), cmd)
         output = output[len(banner_res.stdout):]
 
     if decode:
         assert isinstance(output, bytes)
         output = output.decode()
 
     if res.returncode and check:
-        return SSHCommandFailed(res.returncode, output_for_errors, command)
+        return SSHCommandFailed(res.returncode, output_for_errors, cmd)
 
     if simple_output:
         return output.strip()
@@ -150,37 +158,37 @@ def _ssh(hostname_or_ip, cmd, check, simple_output, suppress_fingerprint_warning
 # This function is kept short for shorter pytest traces upon SSH failures, which are common,
 # as pytest prints the whole function definition that raised the SSHCommandFailed exception
 @overload
-def ssh(hostname_or_ip: HostAddress, cmd: Union[str, List[str]], *, check: bool = True,
+def ssh(hostname_or_ip: HostAddress, cmd: str, *, check: bool = True,
         simple_output: Literal[True] = True,
         suppress_fingerprint_warnings: bool = True, background: Literal[False] = False,
         decode: Literal[True] = True, options: List[str] = [], multiplexing=True) -> str:
     ...
 @overload
-def ssh(hostname_or_ip: HostAddress, cmd: Union[str, List[str]], *, check: bool = True,
+def ssh(hostname_or_ip: HostAddress, cmd: str, *, check: bool = True,
         simple_output: Literal[True] = True,
         suppress_fingerprint_warnings: bool = True, background: Literal[False] = False,
         decode: Literal[False], options: List[str] = [], multiplexing=True) -> bytes:
     ...
 @overload
-def ssh(hostname_or_ip: HostAddress, cmd: Union[str, List[str]], *, check: bool = True,
+def ssh(hostname_or_ip: HostAddress, cmd: str, *, check: bool = True,
         simple_output: Literal[False],
         suppress_fingerprint_warnings: bool = True, background: Literal[False] = False,
         decode: bool = True, options: List[str] = [], multiplexing=True) -> SSHResult:
     ...
 @overload
-def ssh(hostname_or_ip: HostAddress, cmd: Union[str, List[str]], *, check: bool = True,
+def ssh(hostname_or_ip: HostAddress, cmd: str, *, check: bool = True,
         simple_output: Literal[False],
         suppress_fingerprint_warnings: bool = True, background: Literal[True],
         decode: bool = True, options: List[str] = [], multiplexing=True) -> None:
     ...
 @overload
-def ssh(hostname_or_ip: HostAddress, cmd: Union[str, List[str]], *, check=True,
+def ssh(hostname_or_ip: HostAddress, cmd: str, *, check=True,
         simple_output: bool = True,
         suppress_fingerprint_warnings=True, background: bool = False,
         decode: bool = True, options: List[str] = [], multiplexing=True) \
         -> Union[str, bytes, SSHResult, None]:
     ...
-def ssh(hostname_or_ip, cmd, *, check=True, simple_output=True,
+def ssh(hostname_or_ip: HostAddress, cmd: str, *, check=True, simple_output=True,
         suppress_fingerprint_warnings=True,
         background=False, decode=True, options=[], multiplexing=True):
     result_or_exc = _ssh(hostname_or_ip, cmd, check, simple_output, suppress_fingerprint_warnings,
@@ -190,7 +198,7 @@ def ssh(hostname_or_ip, cmd, *, check=True, simple_output=True,
     else:
         return result_or_exc
 
-def ssh_with_result(hostname_or_ip, cmd, suppress_fingerprint_warnings=True,
+def ssh_with_result(hostname_or_ip: HostAddress, cmd: str, suppress_fingerprint_warnings=True,
                     background=False, decode=True, options=[], multiplexing=True) -> SSHResult:
     result_or_exc = _ssh(hostname_or_ip, cmd, False, False, suppress_fingerprint_warnings,
                          background, decode, options, multiplexing)

lib/common.py

@@ -35,7 +35,8 @@
 )
 
 if TYPE_CHECKING:
-    import lib.host
+    from lib.host import Host
+
 
 KiB = 2**10
 MiB = KiB**2
@@ -217,7 +218,7 @@ def strip_suffix(string, suffix):
         return string[:-len(suffix)]
     return string
 
-def setup_formatted_and_mounted_disk(host, sr_disk, fs_type, mountpoint):
+def setup_formatted_and_mounted_disk(host: Host, sr_disk, fs_type, mountpoint):
     if fs_type == 'ext4':
         option_force = '-F'
     elif fs_type == 'xfs':
@@ -226,29 +227,25 @@ def setup_formatted_and_mounted_disk(host, sr_disk, fs_type, mountpoint):
         raise Exception(f"Unsupported fs_type '{fs_type}' in this function")
     device = '/dev/' + sr_disk
     logging.info(f">> Format sr_disk {sr_disk} and mount it on host {host}")
-    host.ssh(['mkfs.' + fs_type, option_force, device])
-    host.ssh(['rm', '-rf', mountpoint]) # Remove any existing leftover to ensure rmdir will not fail in teardown
-    host.ssh(['mkdir', '-p', mountpoint])
-    host.ssh(['cp', '-f', '/etc/fstab', '/etc/fstab.orig'])
-    ssh_client = host.ssh(['echo', '$SSH_CLIENT']).split()[0]
+    host.ssh(f'mkfs.{fs_type} {option_force} {device}')
+    host.ssh(f'rm -rf {mountpoint}') # Remove any existing leftover to ensure rmdir will not fail in teardown
+    host.ssh(f'mkdir -p {mountpoint}')
+    host.ssh('cp -f /etc/fstab /etc/fstab.orig')
+    ssh_client = host.ssh('echo $SSH_CLIENT').split()[0]
     now = datetime.now().isoformat()
-    host.ssh([
-        'echo',
-        f'"# added by {ssh_client} on {now}\n{device} {mountpoint} {fs_type} defaults 0 0"',
-        '>>/etc/fstab',
-    ])
+    host.ssh(f'echo "# added by {ssh_client} on {now}\n{device} {mountpoint} {fs_type} defaults 0 0" >>/etc/fstab')
     try:
-        host.ssh(['mount', mountpoint])
+        host.ssh(f'mount {mountpoint}')
     except Exception:
         # restore fstab then re-raise
-        host.ssh(['cp', '-f', '/etc/fstab.orig', '/etc/fstab'])
+        host.ssh('cp -f /etc/fstab.orig /etc/fstab')
         raise
 
-def teardown_formatted_and_mounted_disk(host, mountpoint):
+def teardown_formatted_and_mounted_disk(host: Host, mountpoint):
     logging.info(f"<< Restore fstab and unmount {mountpoint} on host {host}")
-    host.ssh(['cp', '-f', '/etc/fstab.orig', '/etc/fstab'])
-    host.ssh(['umount', mountpoint])
-    host.ssh(['rmdir', mountpoint])
+    host.ssh('cp -f /etc/fstab.orig /etc/fstab')
+    host.ssh(f'umount {mountpoint}')
+    host.ssh(f'rmdir {mountpoint}')
 
 def exec_nofail(func):
     """ Execute a function, log a warning if it fails, and return eiter [] or [e] where e is the exception. """
@@ -305,21 +302,21 @@ def randid(length=6):
     return ''.join(random.choices(characters, k=length))
 
 @overload
-def _param_get(host: 'lib.host.Host', xe_prefix: str, uuid: str, param_name: str, key: Optional[str] = ...,
+def _param_get(host: Host, xe_prefix: str, uuid: str, param_name: str, key: Optional[str] = ...,
                accept_unknown_key: Literal[False] = ...) -> str:
     ...
 
 @overload
-def _param_get(host: 'lib.host.Host', xe_prefix: str, uuid: str, param_name: str, key: Optional[str] = ...,
+def _param_get(host: Host, xe_prefix: str, uuid: str, param_name: str, key: Optional[str] = ...,
                accept_unknown_key: Literal[True] = ...) -> Optional[str]:
     ...
 
 @overload
-def _param_get(host: 'lib.host.Host', xe_prefix: str, uuid: str, param_name: str, key: Optional[str] = ...,
+def _param_get(host: Host, xe_prefix: str, uuid: str, param_name: str, key: Optional[str] = ...,
                accept_unknown_key: bool = ...) -> Optional[str]:
     ...
 
-def _param_get(host: 'lib.host.Host', xe_prefix: str, uuid: str, param_name: str, key: Optional[str] = None,
+def _param_get(host: Host, xe_prefix: str, uuid: str, param_name: str, key: Optional[str] = None,
                accept_unknown_key: bool = False) -> Optional[str]:
     """ Common implementation for param_get. """
     import lib.commands as commands

lib/fistpoint.py

@@ -40,7 +40,7 @@ def enable_exit_on_fistpoint(host: Host):
 
     @staticmethod
     def disable_exit_on_fistpoint(host: Host):
-        host.ssh(["rm", FistPoint._get_path(LVHDRT_EXIT_FIST)])
+        host.ssh(f'rm {FistPoint._get_path(LVHDRT_EXIT_FIST)}')
 
     @staticmethod
     def _get_name(name: str) -> str:
@@ -60,7 +60,7 @@ def enable(self):
     def disable(self):
         logging.info(f"Disabling fistpoint {self.fistpointName}")
         try:
-            self.host.ssh(["rm", self._get_path(self.fistpointName)])
+            self.host.ssh(f'rm {self._get_path(self.fistpointName)}')
         except SSHCommandFailed as e:
             logging.info(f"Failed trying to disable fistpoint {self._get_path(self.fistpointName)} with error {e}")
             raise