[collect] update for strict confinement for juju

sos/collector/transports/juju.py

@@ -10,10 +10,12 @@
 
 
 import subprocess
+import os
+import shutil
 
 from sos.collector.exceptions import JujuNotInstalledException
 from sos.collector.transports import RemoteTransport
-from sos.utilities import sos_get_command_output
+from sos.utilities import sos_get_command_output, parse_version
 
 
 class JujuSSH(RemoteTransport):
@@ -79,12 +81,44 @@ def _copy_file_to_remote(self, fname, dest):
         res = sos_get_command_output(cmd, timeout=15)
         return res["status"] == 0
 
+    def _get_juju_version(self):
+        """Grab the version of juju"""
+        res = sos_get_command_output("juju version")
+        return res['output'].split("-", maxsplit=1)[0]
+
     def _retrieve_file(self, fname, dest):
         self._chmod(fname)  # juju scp needs the archive to be world-readable
         model, unit = self.address.split(":")
         model_option = f"-m {model}" if model else ""
-        cmd = f"juju scp {model_option} -- -r {unit}:{fname} {dest}"
-        res = sos_get_command_output(cmd)
+
+        if parse_version(self._get_juju_version()) >= parse_version("3"):
+            # From juju 3.0 onwards the juju client is a strictly confined
+            # snap. Strict confinement prevents the snap from writing to
+            # arbitrary paths on the host (such as sos' tmpdir under /tmp or
+            # the snap's own private tmp namespace). It can, however, write
+            # into the invoking user's $HOME thanks to the 'home' snap
+            # interface. So we scp the file into a staging directory under
+            # $HOME and then move it to the requested destination ourselves.
+            #
+            # This avoids reaching into the snap's private confinement dir
+            # (/tmp/snap-private-tmp/...) and removes the previous requirement
+            # of running sos collect as root/with sudo for juju.
+            staging_dir = os.path.join(
+                os.path.expanduser("~"), ".cache", "sos-collect-juju"
+            )
+            os.makedirs(staging_dir, exist_ok=True)
+            staged_file = os.path.join(staging_dir, os.path.basename(fname))
+
+            cmd = (
+                f"juju scp {model_option} -- -r "
+                f"{unit}:{fname} {staging_dir}"
+            )
+            res = sos_get_command_output(cmd)
+            if res["status"] == 0:
+                shutil.move(staged_file, dest)
+        else:
+            cmd = f"juju scp {model_option} -- -r {unit}:{fname} {dest}"
+            res = sos_get_command_output(cmd)
         return res["status"] == 0
 
 

tests/unittests/juju/juju_transports_test.py

@@ -33,6 +33,14 @@ def setUp(self):
             address="model_abc:unit_abc",
         )
 
+    # pylint: disable=no-method-argument
+    def get_juju_version():
+        return "2.9.45"
+
+    # pylint: disable=no-method-argument
+    def get_juju_version_3():
+        return "3.1.0"
+
     @patch("sos.collector.transports.juju.subprocess.check_output")
     def test_check_juju_installed_err(self, mock_subprocess_check_output):
         """Raise error if juju is not installed."""
@@ -72,17 +80,63 @@ def test_remote_exec(self):
             self.juju_ssh.remote_exec == "juju ssh -m model_abc unit_abc"
         )
 
+    @patch(
+        "sos.collector.transports.juju.JujuSSH._get_juju_version",
+        side_effect=get_juju_version,
+    )
     @patch(
         "sos.collector.transports.juju.sos_get_command_output",
         return_value={"status": 0},
     )
     @patch("sos.collector.transports.juju.JujuSSH._chmod", return_value=True)
     # pylint: disable=unused-argument
-    def test_retrieve_file(self, mock_chmod, mock_sos_get_cmd_output):
+    def test_retrieve_file(
+        self,
+        mock_chmod,
+        mock_sos_get_cmd_output,
+        mock_get_juju_version
+    ):
         self.juju_ssh._retrieve_file(fname="file_abc", dest="/tmp/sos-juju/")
         mock_sos_get_cmd_output.assert_called_with(
             "juju scp -m model_abc -- -r unit_abc:file_abc /tmp/sos-juju/"
         )
 
+    @patch("sos.collector.transports.juju.shutil.move")
+    @patch("sos.collector.transports.juju.os.makedirs")
+    @patch(
+        "sos.collector.transports.juju.os.path.expanduser",
+        return_value="/home/user_abc",
+    )
+    @patch(
+        "sos.collector.transports.juju.JujuSSH._get_juju_version",
+        side_effect=get_juju_version_3,
+    )
+    @patch(
+        "sos.collector.transports.juju.sos_get_command_output",
+        return_value={"status": 0},
+    )
+    @patch("sos.collector.transports.juju.JujuSSH._chmod", return_value=True)
+    # pylint: disable=unused-argument
+    def test_retrieve_file_juju_3(
+        self,
+        mock_chmod,
+        mock_sos_get_cmd_output,
+        mock_get_juju_version,
+        mock_expanduser,
+        mock_makedirs,
+        mock_move,
+    ):
+        """For juju 3+ the file is staged under $HOME (confinement-safe) and
+        then moved to the destination, without sudo or private-tmp hacks."""
+        staging_dir = "/home/user_abc/.cache/sos-collect-juju"
+        self.juju_ssh._retrieve_file(fname="file_abc", dest="/tmp/sos-juju/")
+        mock_makedirs.assert_called_with(staging_dir, exist_ok=True)
+        mock_sos_get_cmd_output.assert_called_with(
+            f"juju scp -m model_abc -- -r unit_abc:file_abc {staging_dir}"
+        )
+        mock_move.assert_called_with(
+            f"{staging_dir}/file_abc", "/tmp/sos-juju/"
+        )
+
 
 # vim: set et ts=4 sw=4 :