Improve environment variable handling for chroot commands

`sudo-rs`, which is the default sudo on `Ubuntu 26.04`, does not implement
the `-E` flag, so proxy variables no longer reached commands run inside
the chroot.

Signed-off-by: Mihaela Balutoiu <mbalutoiu@cloudbasesolutions.com>

coriolis/osmorphing/base.py

@@ -250,7 +250,18 @@ class BaseOSMorphingTools(object, with_metaclass(abc.ABCMeta)):
         pass
 
     def set_environment(self, environment):
-        self._environment = environment
+        """Merges the given environment into the tools' own environment.
+
+        The variables declared by the tools' constructor (such as
+        DEBIAN_FRONTEND for the Debian-based tools) are preserved, unless
+        the given environment explicitly overrides them.
+
+        The variables are copied over instead of the mapping being stored
+        as-is, as the environment provided by the OSMount tools is shared
+        between the OSDetect and the OSMorphing tools of both the export
+        and the import providers.
+        """
+        self._environment.update(environment or {})
 
 
 class BaseLinuxOSMorphingTools(BaseOSMorphingTools):

coriolis/tests/osmorphing/test_base.py

@@ -99,9 +99,26 @@ class BaseOSMorphingToolsTestBase(test_base.CoriolisBaseTestCase):
         )
 
     def test_set_environment(self):
-        self.os_morphing_tools.set_environment(mock.sentinel.environment)
+        environment = {'http_proxy': 'http://10.0.0.1:3128'}
+
+        self.os_morphing_tools.set_environment(environment)
+
+        self.assertEqual(
+            {'http_proxy': 'http://10.0.0.1:3128'},
+            self.os_morphing_tools._environment)
+
+    def test_set_environment_preserves_constructor_variables(self):
+        """Variables declared by the constructor survive set_environment."""
+        self.os_morphing_tools._environment['DEBIAN_FRONTEND'] = (
+            'noninteractive')
+
+        self.os_morphing_tools.set_environment(
+            {'http_proxy': 'http://10.0.0.1:3128'})
+
         self.assertEqual(
-            self.os_morphing_tools._environment, mock.sentinel.environment)
+            {'DEBIAN_FRONTEND': 'noninteractive',
+             'http_proxy': 'http://10.0.0.1:3128'},
+            self.os_morphing_tools._environment)
 
 
 # This class is used to test the BaseLinuxOSMorphingTools class since it is

coriolis/tests/test_utils.py

@@ -6,6 +6,7 @@ import hashlib
 import json
 import logging
 import os
+import shlex
 import socket
 from unittest import mock
 import uuid
@@ -439,23 +440,156 @@ class UtilsTestCase(test_base.CoriolisBaseTestCase):
         self.assertRaises(exception.SSHCommandNotFoundException,
                           utils.exec_ssh_cmd, self.mock_ssh, "command")
 
-    def test_exec_ssh_cmd_chroot(self):
+    def _setup_successful_ssh_cmd(self):
         self.mock_stdout.read.return_value = b'output\n'
         self.mock_stdout.channel.recv_exit_status.return_value = 0
         self.mock_ssh.exec_command.return_value = (None, self.mock_stdout,
                                                    self.mock_stdout)
 
+    def _get_executed_ssh_cmd(self):
+        return self.mock_ssh.exec_command.call_args[0][0]
+
+    def test_get_env_command_prefix(self):
+        self.assertEqual("", utils.get_env_command_prefix(None))
+        self.assertEqual("", utils.get_env_command_prefix({}))
+        self.assertEqual(
+            "env DEBIAN_FRONTEND=noninteractive ",
+            utils.get_env_command_prefix(
+                {"DEBIAN_FRONTEND": "noninteractive"}))
+        # Values holding shell-sensitive characters must be quoted whole.
+        self.assertEqual(
+            "env 'http_proxy=http://a b' ",
+            utils.get_env_command_prefix({"http_proxy": "http://a b"}))
+        self.assertEqual(
+            "env 'http_proxy=http://u:p'\"'\"'w@h:3128' ",
+            utils.get_env_command_prefix(
+                {"http_proxy": "http://u:p'w@h:3128"}))
+
+    def test_exec_ssh_cmd_chroot(self):
+        self._setup_successful_ssh_cmd()
+
         result = utils.exec_ssh_cmd_chroot(
-            self.mock_ssh, "/chroot /bin/bash -c", "command")
+            self.mock_ssh, "/chroot", "command")
 
         self.mock_ssh.exec_command.assert_called_once_with(
-            "sudo -E chroot /chroot /bin/bash -c command",
+            "sudo chroot /chroot command",
             environment=None, get_pty=False, timeout=None)
 
         expected = self.mock_stdout.read.return_value.decode(
             'utf-8', errors='replace')
         self.assertEqual(result, expected)
 
+    def test_exec_ssh_cmd_chroot_no_environment(self):
+        """With no env vars set, no 'env' prefix may be added at all."""
+        self._setup_successful_ssh_cmd()
+
+        for environment in (None, {}):
+            self.mock_ssh.exec_command.reset_mock()
+            utils.exec_ssh_cmd_chroot(
+                self.mock_ssh, "/chroot", "apt-get update -y",
+                environment=environment)
+
+            self.mock_ssh.exec_command.assert_called_once_with(
+                "sudo chroot /chroot apt-get update -y",
+                environment=environment, get_pty=False, timeout=None)
+
+    def test_exec_ssh_cmd_chroot_with_proxy_environment(self):
+        """Proxy vars are passed through env(1), placed before the chroot."""
+        self._setup_successful_ssh_cmd()
+        environment = {
+            "http_proxy": "http://10.0.0.1:3128",
+            "HTTP_PROXY": "http://10.0.0.1:3128",
+            "https_proxy": "http://10.0.0.1:3128",
+            "HTTPS_PROXY": "http://10.0.0.1:3128",
+            "ftp_proxy": "http://10.0.0.1:3128",
+            "FTP_PROXY": "http://10.0.0.1:3128",
+            "no_proxy": "localhost.127.0.0.1",
+        }
+
+        utils.exec_ssh_cmd_chroot(
+            self.mock_ssh, "/chroot", "apt-get update -y",
+            environment=environment)
+
+        cmd = self._get_executed_ssh_cmd()
+        self.assertEqual(
+            "sudo env "
+            "http_proxy=http://10.0.0.1:3128 "
+            "HTTP_PROXY=http://10.0.0.1:3128 "
+            "https_proxy=http://10.0.0.1:3128 "
+            "HTTPS_PROXY=http://10.0.0.1:3128 "
+            "ftp_proxy=http://10.0.0.1:3128 "
+            "FTP_PROXY=http://10.0.0.1:3128 "
+            "no_proxy=localhost.127.0.0.1 "
+            "chroot /chroot apt-get update -y", cmd)
+        # env(1) must run before chroot(1) so that the variables are
+        # inherited by the command running *inside* the chroot.
+        self.assertLess(cmd.index("env "), cmd.index("chroot"))
+        self.assertNotIn("sudo -E", cmd)
+
+    def test_exec_ssh_cmd_chroot_proxy_with_credentials_is_quoted(self):
+        """Shell-sensitive proxy values survive as a single argument."""
+        self._setup_successful_ssh_cmd()
+        proxy = "http://user:p@ss w0rd&$(reboot)@10.0.0.1:3128?a=1"
+
+        utils.exec_ssh_cmd_chroot(
+            self.mock_ssh, "/chroot", "apt-get update -y",
+            environment={"https_proxy": proxy})
+
+        self.assertEqual(
+            ["sudo", "env", "https_proxy=%s" % proxy, "chroot", "/chroot",
+             "apt-get", "update", "-y"],
+            shlex.split(self._get_executed_ssh_cmd()))
+
+    def test_exec_ssh_cmd_chroot_does_not_re_quote_the_command(self):
+        """Commands holding quotes must not get an extra quoting layer.
+
+        Regression test for the GRUB serial command: wrapping 'cmd' in
+        "/bin/bash -c '...'" terminated the outer quoting early, which turned
+        the appended GRUB value into separate 'sed' options and failed with
+        "sed: unrecognized option '--word=8'".
+        """
+        self._setup_successful_ssh_cmd()
+        grub_value = (
+            'serial --word=8 --stop=1 --speed=115200 --parity=no --unit=0')
+        sed_script = '$aGRUB_SERIAL_COMMAND="%s"' % grub_value
+        cmd = "sed -ie %s /tmp/tmp.OIK95wgYUb" % shlex.quote(sed_script)
+
+        utils.exec_ssh_cmd_chroot(
+            self.mock_ssh, "/tmp/tmp.q15QdW45qE", cmd,
+            environment={"http_proxy": "http://10.0.0.1:3128"})
+
+        argv = shlex.split(self._get_executed_ssh_cmd())
+        self.assertEqual(
+            ["sudo", "env", "http_proxy=http://10.0.0.1:3128", "chroot",
+             "/tmp/tmp.q15QdW45qE", "sed", "-ie", sed_script,
+             "/tmp/tmp.OIK95wgYUb"],
+            argv)
+        # No part of the GRUB value may become a standalone sed option.
+        self.assertNotIn("--word=8", argv)
+        self.assertEqual([], [a for a in argv if a.startswith("--")])
+
+    def test_exec_ssh_cmd_chroot_preserves_shell_operators(self):
+        """Host-side shell operators used by callers stay intact."""
+        self._setup_successful_ssh_cmd()
+
+        utils.exec_ssh_cmd_chroot(
+            self.mock_ssh, "/chroot",
+            '[ -f "/etc/default/grub" ] && echo 1 || echo 0')
+
+        self.assertEqual(
+            'sudo chroot /chroot [ -f "/etc/default/grub" ] '
+            '&& echo 1 || echo 0',
+            self._get_executed_ssh_cmd())
+
+    def test_exec_ssh_cmd_chroot_quotes_the_chroot_dir(self):
+        self._setup_successful_ssh_cmd()
+
+        utils.exec_ssh_cmd_chroot(self.mock_ssh, "/tmp/os root dir", "true")
+
+        self.assertEqual(
+            "sudo chroot '/tmp/os root dir' true",
+            self._get_executed_ssh_cmd())
+
     def test_check_fs(self):
         self.mock_stdout.read.return_value.replace.return_value = \
             self.mock_stdout.read.return_value

coriolis/utils.py

@@ -13,6 +13,7 @@ import OpenSSL
 import os
 import pickle
 import re
+import shlex
 import socket
 import string
 import subprocess
@@ -327,6 +328,20 @@ def list_ssh_dir(ssh, remote_path):
         return [f for f in output.splitlines() if f.strip()]
 
 
+def get_env_command_prefix(environment):
+    """
+    Each variable is shell-quoted individually, so values containing
+    spaces or other shell-sensitive characters are passed through
+    verbatim.
+    """
+
+    if not environment:
+        return ""
+    return "env %s " % " ".join(
+        shlex.quote("%s=%s" % (key, value))
+        for key, value in environment.items())
+
+
 def _exec_ssh_cmd(ssh, cmd, environment=None, get_pty=False, timeout=None):
     sanitized_cmd = strutils.mask_password(cmd)
     remote_str = "<undeterminable>"
@@ -401,9 +416,16 @@ def exec_ssh_cmd(
 
 def exec_ssh_cmd_chroot(ssh, chroot_dir, cmd, environment=None, get_pty=False,
                         timeout=None):
-    return exec_ssh_cmd(ssh, "sudo -E chroot %s %s" % (chroot_dir, cmd),
-                        environment=environment, get_pty=get_pty,
-                        timeout=timeout)
+    return exec_ssh_cmd(
+        ssh,
+        "sudo %schroot %s %s" % (
+            get_env_command_prefix(environment),
+            shlex.quote(chroot_dir),
+            cmd),
+        environment=environment,
+        get_pty=get_pty,
+        timeout=timeout,
+    )
 
 
 def check_fs(ssh, fs_type, dev_path):