CVE-2022-23948: Insufficient Information (NVD-CWE-noinfo)
value: A flaw was found in Keylime before 6.3.0. The logic in the Keylime agent for checking for a secure mount can be fooled by previously created unprivileged mounts allowing secrets to be leaked to other processes on the host.
CWE description: There is insufficient information about the issue to classify it; details are unkown or unspecified.
CVSS v3 base score: 7.5
Published: 2022-09-21T19:15Z
Repository: https://github.com/keylime/keylime
Language: Python
Files changed: keylime/keylime_agent.py, keylime/secure_mount.py, test/test_secure_mount.py
Commit date: 2022-01-27 11:09:26 -0500
Commit message: secure_mount: add umount function

The `mount` function is called several times in different places in the
code, but it takes care of not mounting the secure directory if is
already present.  The problem is that this is never umounted.

This patch add the `umount` function, and update the `mount` one to
track when a mount point needs to be umounted.

Also adjust the keylime_agent exit point to make sure that the `umount`
function is called at the end.

Signed-off-by: Alberto Planas <aplanas@suse.com>

Diff: diff --git a/keylime/keylime_agent.py b/keylime/keylime_agent.py
index 47fcb24..2f7de35 100644
--- a/keylime/keylime_agent.py
+++ b/keylime/keylime_agent.py
@@ -744,13 +744,15 @@ def main():
         revocation_process.terminate()
         logger.debug("Shutting down HTTP server...")
         server.shutdown()
         server.server_close()
         serverthread.join()
-        logger.debug("...HTTP server stopped")
+        logger.debug("HTTP server stopped...")
         revocation_process.join()
-        logger.debug("... revocation notifier stopped")
+        logger.debug("Revocation notifier stopped...")
+        secure_mount.umount()
+        logger.debug("Umounting directories...")
         instance_tpm.flush_keys()
         logger.debug("Flushed keys successfully")
         sys.exit(0)
 
     signal.signal(signal.SIGTERM, shutdown_handler)
diff --git a/keylime/secure_mount.py b/keylime/secure_mount.py
index b9ebf5a..4d879f8 100644
--- a/keylime/secure_mount.py
+++ b/keylime/secure_mount.py
@@ -9,15 +9,19 @@ from keylime import keylime_logging
 from keylime import cmd_exec
 from keylime import config
 
 logger = keylime_logging.init_logging('secure_mount')
 
+# Store the mounted directories done by Keylime, so we can unmount
+# them in reverse order
+_MOUNTED = []
+
 
 def check_mounted(secdir):
     """Inspect mountinfo to detect if a directory is mounted."""
     secdir_escaped = secdir.replace(" ", r"\040")
-    for line in open("/proc/self/mountinfo", "r"):
+    for line in open("/proc/self/mountinfo", "r", encoding="utf-8"):
         # /proc/[pid]/mountinfo have 10+ elements separated with
         # spaces (check proc (5) for a complete description)
         #
         # At position 7 there are some optional fields, so we need
         # first to determine the separator mark, and validate the
@@ -27,10 +31,11 @@ def check_mounted(secdir):
             separator = elements.index("-")
         except ValueError:
             msg = "Separator filed not found. " \
                 "Information line cannot be parsed"
             logger.error(msg)
+            # pylint: disable=raise-missing-from
             raise Exception(msg)
 
         if len(elements) < 10 or len(elements) - separator < 4:
             msg = "Mount information line cannot be parsed"
             logger.error(msg)
@@ -54,14 +59,14 @@ def check_mounted(secdir):
     logger.debug("Secure storage location %s not mounted", secdir)
     return False
 
 
 def mount():
-    secdir = config.WORK_DIR + "/secure"
+    secdir = os.path.join(config.WORK_DIR, "secure")
 
     if not config.MOUNT_SECURE:
-        secdir = config.WORK_DIR + "/tmpfs-dev"
+        secdir = os.path.join(config.WORK_DIR, "tmpfs-dev")
         if not os.path.isdir(secdir):
             os.makedirs(secdir)
         return secdir
 
     if not check_mounted(secdir):
@@ -72,7 +77,24 @@ def mount():
         size = config.get('cloud_agent', 'secure_size')
         logger.info("mounting secure storage location %s on tmpfs" % secdir)
         cmd = ('mount', '-t', 'tmpfs', '-o', 'size=%s,mode=0700' % size,
                'tmpfs', secdir)
         cmd_exec.run(cmd)
+        _MOUNTED.append(secdir)
 
     return secdir
+
+
+def umount():
+    """Umount all the devices mounted by Keylime."""
+    while _MOUNTED:
+        directory = _MOUNTED.pop()
+        logger.info("Unmounting %s", directory)
+        if check_mounted(directory):
+            cmd = ("umount", directory)
+            ret = cmd_exec.run(cmd, raiseOnError=False)
+            if ret["code"] != 0:
+                logger.error("%s cannot be umounted. "
+                             "A running process can be keeping it bussy: %s",
+                             directory, str(ret["reterr"]))
+        else:
+            logger.warning("%s already unmounted by another process", directory)
diff --git a/test/test_secure_mount.py b/test/test_secure_mount.py
index f3b9e70..0a91302 100644
--- a/test/test_secure_mount.py
+++ b/test/test_secure_mount.py
@@ -1,81 +1,208 @@
+import tempfile
 import unittest
 from unittest.mock import patch
 
 from keylime import secure_mount
 
 
 class TestSecureMount(unittest.TestCase):
+    def setUp(self):
+        """Remove global state from secure_mount module."""
+        # pylint: disable=protected-access
+        secure_mount._MOUNTED = []
 
     @patch("keylime.secure_mount.logger")
     @patch("keylime.secure_mount.open")
-    def test_check_mounted_not_found(self, open_mock, logger_mock):
+    def test_check_mounted_not_found(self, open_mock, _logger_mock):
         """Test when secdir is not mounted."""
         open_mock.return_value = (
             "23 106 0:21 / /proc rw,nosuid,nodev,noexec,relatime shared:26 - proc proc rw",
-            )
+        )
         self.assertFalse(secure_mount.check_mounted("/secdir"))
 
     @patch("keylime.secure_mount.logger")
     @patch("keylime.secure_mount.open")
-    def test_check_mounted_found(self, open_mock, logger_mock):
+    def test_check_mounted_found(self, open_mock, _logger_mock):
         """Test when secdir is mounted."""
         open_mock.return_value = (
             "23 106 0:21 / /proc rw,nosuid,nodev,noexec,relatime shared:26 - proc proc rw",
             "303 154 0:69 / /secdir rw,relatime shared:130 - tmpfs tmpfs rw,size=1024k,mode=700,inode64",
-            )
+        )
         self.assertTrue(secure_mount.check_mounted("/secdir"))
 
     @patch("keylime.secure_mount.logger")
     @patch("keylime.secure_mount.open")
-    def test_check_mounted_found_zero_optional_fields(self, open_mock, logger_mock):
+    def test_check_mounted_found_zero_optional_fields(self, open_mock, _logger_mock):
         """Test when secdir is mounted when there are no optional fields."""
         open_mock.return_value = (
             "23 106 0:21 / /proc rw,nosuid,nodev,noexec,relatime shared:26 - proc proc rw",
             "303 154 0:69 / /secdir rw,relatime - tmpfs tmpfs rw,size=1024k,mode=700,inode64",
         )
         self.assertTrue(secure_mount.check_mounted("/secdir"))
 
     @patch("keylime.secure_mount.logger")
     @patch("keylime.secure_mount.open")
-    def test_check_mounted_found_extra_optional_fields(self, open_mock, logger_mock):
+    def test_check_mounted_found_extra_optional_fields(self, open_mock, _logger_mock):
         """Test when secdir is mounted when there are extra optional fields."""
         open_mock.return_value = (
             "23 106 0:21 / /proc rw,nosuid,nodev,noexec,relatime shared:26 - proc proc rw",
             "303 154 0:69 / /secdir rw,relatime shared:130 extra:1 - tmpfs tmpfs rw,size=1024k,mode=700,inode64",
         )
         self.assertTrue(secure_mount.check_mounted("/secdir"))
 
     @patch("keylime.secure_mount.logger")
     @patch("keylime.secure_mount.open")
-    def test_check_mounted_found_wrong_fs(self, open_mock, logger_mock):
+    def test_check_mounted_found_wrong_fs(self, open_mock, _logger_mock):
         """Test when secdir is mounted but under a wrong fs."""
         open_mock.return_value = (
             "23 106 0:21 / /proc rw,nosuid,nodev,noexec,relatime shared:26 - proc proc rw",
             "303 154 0:69 / /secdir rw,relatime shared:130 - btrfs /dev/sda2 rw",
-            )
+        )
         with self.assertRaises(Exception) as e:
             secure_mount.check_mounted("/secdir")
         self.assertTrue("wrong file system" in str(e.exception))
 
     @patch("keylime.secure_mount.logger")
     @patch("keylime.secure_mount.open")
-    def test_check_mounted_found_spaces(self, open_mock, logger_mock):
+    def test_check_mounted_found_spaces(self, open_mock, _logger_mock):
         """Test when secdir is mounted and contains spaces."""
         open_mock.return_value = (
             "23 106 0:21 / /proc rw,nosuid,nodev,noexec,relatime shared:26 - proc proc rw",
             r"303 154 0:69 / /sec\040dir rw,relatime shared:130 - tmpfs tmpfs rw,size=1024k,mode=700,inode64",
-            )
+        )
         self.assertTrue(secure_mount.check_mounted("/sec dir"))
 
     @patch("keylime.secure_mount.logger")
     @patch("keylime.secure_mount.open")
-    def test_check_mounted_wrong_format(self, open_mock, logger_mock):
+    def test_check_mounted_wrong_format(self, open_mock, _logger_mock):
         """Test when the mount info lines are wrong."""
         open_mock.return_value = ("invalid line",)
         with self.assertRaises(Exception) as e:
             secure_mount.check_mounted("/secdir")
         self.assertTrue("cannot be parsed" in str(e.exception))
 
+    @patch("keylime.secure_mount.logger")
+    @patch("keylime.secure_mount.config")
+    def test_check_mount_no_secure(self, config_mock, _logger_mock):
+        """Test when mounted outside tmpfs."""
+        config_mock.MOUNT_SECURE = False
+
+        with tempfile.TemporaryDirectory() as tmpdirname:
+            config_mock.WORK_DIR = tmpdirname
+            self.assertEqual(secure_mount.mount(), f"{tmpdirname}/tmpfs-dev")
+            # pylint: disable=protected-access
+            self.assertEqual(secure_mount._MOUNTED, [])
+
+    @patch("keylime.secure_mount.check_mounted")
+    @patch("keylime.secure_mount.config")
+    def test_check_mount_secure_already_mounted(self, config_mock, check_mounted_mock):
+        """Test when mounting in tmpfs but is already present."""
+        config_mock.MOUNT_SECURE = True
+        check_mounted_mock.return_value = True
+
+        with tempfile.TemporaryDirectory() as tmpdirname:
+            config_mock.WORK_DIR = tmpdirname
+            self.assertEqual(secure_mount.mount(), f"{tmpdirname}/secure")
+            # pylint: disable=protected-access
+            self.assertEqual(secure_mount._MOUNTED, [])
+
+    @patch("keylime.secure_mount.logger")
+    @patch("keylime.secure_mount.check_mounted")
+    @patch("keylime.secure_mount.config")
+    @patch("keylime.secure_mount.os.path.exists")
+    @patch("keylime.secure_mount.cmd_exec")
+    def test_check_mount_secure_already_created(
+        self, _cmd_exec_mock, exists_mock, config_mock, check_mounted_mock, _logger_mock
+    ):
+        """Test when mounting in tmpfs but the mount point is present."""
+        exists_mock.return_value = True
+        config_mock.MOUNT_SECURE = True
+        check_mounted_mock.return_value = False
+
+        with tempfile.TemporaryDirectory() as tmpdirname:
+            config_mock.WORK_DIR = tmpdirname
+            self.assertEqual(secure_mount.mount(), f"{tmpdirname}/secure")
+            # pylint: disable=protected-access
+            self.assertEqual(secure_mount._MOUNTED, [f"{tmpdirname}/secure"])
+
+    @patch("keylime.secure_mount.logger")
+    @patch("keylime.secure_mount.check_mounted")
+    @patch("keylime.secure_mount.config")
+    @patch("keylime.secure_mount.os.path.exists")
+    @patch("keylime.secure_mount.os.makedirs")
+    @patch("keylime.secure_mount.cmd_exec")
+    def test_check_mount(
+        self,
+        _cmd_exec_mock,
+        _makedirs_mock,
+        exists_mock,
+        config_mock,
+        check_mounted_mock,
+        _logger_mock,
+    ):
+        """Test when mounting in tmpfs but the mount point is not present."""
+        exists_mock.return_value = False
+        config_mock.MOUNT_SECURE = True
+        check_mounted_mock.return_value = False
+
+        with tempfile.TemporaryDirectory() as tmpdirname:
+            config_mock.WORK_DIR = tmpdirname
+            self.assertEqual(secure_mount.mount(), f"{tmpdirname}/secure")
+            # pylint: disable=protected-access
+            self.assertEqual(secure_mount._MOUNTED, [f"{tmpdirname}/secure"])
+
+    def test_check_umount_empty(self):
+        """Test umount when there are nothing to clean."""
+        secure_mount.umount()
+        # pylint: disable=protected-access
+        self.assertEqual(secure_mount._MOUNTED, [])
+
+    @patch("keylime.secure_mount.logger")
+    @patch("keylime.secure_mount.check_mounted")
+    @patch("keylime.secure_mount.cmd_exec")
+    def test_check_umount_mount(self, cmd_exec_mock, check_mounted_mock, logger_mock):
+        """Test umount for a single mount."""
+        # pylint: disable=protected-access
+        secure_mount._MOUNTED = ["/secdir"]
+        check_mounted_mock.return_value = True
+        cmd_exec_mock.run.return_value = {"code": 0}
+        secure_mount.umount()
+        self.assertEqual(secure_mount._MOUNTED, [])
+        cmd_exec_mock.run.assert_called_once()
+        logger_mock.error.assert_not_called()
+        logger_mock.warning.assert_not_called()
+
+    @patch("keylime.secure_mount.logger")
+    @patch("keylime.secure_mount.check_mounted")
+    @patch("keylime.secure_mount.cmd_exec")
+    def test_check_umount_mount_busy(
+        self, cmd_exec_mock, check_mounted_mock, logger_mock
+    ):
+        """Test umount for a single mount that fail."""
+        # pylint: disable=protected-access
+        secure_mount._MOUNTED = ["/secdir"]
+        check_mounted_mock.return_value = True
+        cmd_exec_mock.run.return_value = {"code": 1, "reterr": "Device busy"}
+        secure_mount.umount()
+        self.assertEqual(secure_mount._MOUNTED, [])
+        cmd_exec_mock.run.assert_called_once()
+        logger_mock.error.assert_called_once()
+        logger_mock.warning.assert_not_called()
+
+    @patch("keylime.secure_mount.logger")
+    @patch("keylime.secure_mount.check_mounted")
+    @patch("keylime.secure_mount.cmd_exec")
+    def test_check_umount_umount(self, cmd_exec_mock, check_mounted_mock, logger_mock):
+        """Test umount for a single umount and no creation."""
+        # pylint: disable=protected-access
+        secure_mount._MOUNTED = ["/secdir"]
+        check_mounted_mock.return_value = False
+        secure_mount.umount()
+        self.assertEqual(secure_mount._MOUNTED, [])
+        cmd_exec_mock.run.assert_not_called()
+        logger_mock.error.assert_not_called()
+        logger_mock.warning.assert_called_once()
+
 
 if __name__ == "__main__":
     unittest.main()

Vulnerable code:         logger.debug("...HTTP server stopped")
        logger.debug("... revocation notifier stopped")
    for line in open("/proc/self/mountinfo", "r"):
    secdir = config.WORK_DIR + "/secure"
        secdir = config.WORK_DIR + "/tmpfs-dev"
    def test_check_mounted_not_found(self, open_mock, logger_mock):
            )
    def test_check_mounted_found(self, open_mock, logger_mock):
            )
    def test_check_mounted_found_zero_optional_fields(self, open_mock, logger_mock):
    def test_check_mounted_found_extra_optional_fields(self, open_mock, logger_mock):
    def test_check_mounted_found_wrong_fs(self, open_mock, logger_mock):
            )
    def test_check_mounted_found_spaces(self, open_mock, logger_mock):
            )
    def test_check_mounted_wrong_format(self, open_mock, logger_mock):
Fixed code:         logger.debug("HTTP server stopped...")
        logger.debug("Revocation notifier stopped...")
        secure_mount.umount()
        logger.debug("Umounting directories...")
# Store the mounted directories done by Keylime, so we can unmount
# them in reverse order
_MOUNTED = []

    for line in open("/proc/self/mountinfo", "r", encoding="utf-8"):
            # pylint: disable=raise-missing-from
    secdir = os.path.join(config.WORK_DIR, "secure")
        secdir = os.path.join(config.WORK_DIR, "tmpfs-dev")
        _MOUNTED.append(secdir)


def umount():
    """Umount all the devices mounted by Keylime."""
    while _MOUNTED:
        directory = _MOUNTED.pop()
        logger.info("Unmounting %s", directory)
        if check_mounted(directory):
            cmd = ("umount", directory)
            ret = cmd_exec.run(cmd, raiseOnError=False)
            if ret["code"] != 0:
                logger.error("%s cannot be umounted. "
                             "A running process can be keeping it bussy: %s",
                             directory, str(ret["reterr"]))
        else:
            logger.warning("%s already unmounted by another process", directory)
import tempfile
    def setUp(self):
        """Remove global state from secure_mount module."""
        # pylint: disable=protected-access
        secure_mount._MOUNTED = []
    def test_check_mounted_not_found(self, open_mock, _logger_mock):
        )
    def test_check_mounted_found(self, open_mock, _logger_mock):
        )
    def test_check_mounted_found_zero_optional_fields(self, open_mock, _logger_mock):
    def test_check_mounted_found_extra_optional_fields(self, open_mock, _logger_mock):
    def test_check_mounted_found_wrong_fs(self, open_mock, _logger_mock):
        )
    def test_check_mounted_found_spaces(self, open_mock, _logger_mock):
        )
    def test_check_mounted_wrong_format(self, open_mock, _logger_mock):
    @patch("keylime.secure_mount.logger")
    @patch("keylime.secure_mount.config")
    def test_check_mount_no_secure(self, config_mock, _logger_mock):
        """Test when mounted outside tmpfs."""
        config_mock.MOUNT_SECURE = False

        with tempfile.TemporaryDirectory() as tmpdirname:
            config_mock.WORK_DIR = tmpdirname
            self.assertEqual(secure_mount.mount(), f"{tmpdirname}/tmpfs-dev")
            # pylint: disable=protected-access
            self.assertEqual(secure_mount._MOUNTED, [])

    @patch("keylime.secure_mount.check_mounted")
    @patch("keylime.secure_mount.config")
    def test_check_mount_secure_already_mounted(self, config_mock, check_mounted_mock):
        """Test when mounting in tmpfs but is already present."""
        config_mock.MOUNT_SECURE = True
        check_mounted_mock.return_value = True

        with tempfile.TemporaryDirectory() as tmpdirname:
            config_mock.WORK_DIR = tmpdirname
            self.assertEqual(secure_mount.mount(), f"{tmpdirname}/secure")
            # pylint: disable=protected-access
            self.assertEqual(secure_mount._MOUNTED, [])

    @patch("keylime.secure_mount.logger")
    @patch("keylime.secure_mount.check_mounted")
    @patch("keylime.secure_mount.config")
    @patch("keylime.secure_mount.os.path.exists")
    @patch("keylime.secure_mount.cmd_exec")
    def test_check_mount_secure_already_created(
        self, _cmd_exec_mock, exists_mock, config_mock, check_mounted_mock, _logger_mock
    ):
        """Test when mounting in tmpfs but the mount point is present."""
        exists_mock.return_value = True
        config_mock.MOUNT_SECURE = True
        check_mounted_mock.return_value = False

        with tempfile.TemporaryDirectory() as tmpdirname:
            config_mock.WORK_DIR = tmpdirname
            self.assertEqual(secure_mount.mount(), f"{tmpdirname}/secure")
            # pylint: disable=protected-access
            self.assertEqual(secure_mount._MOUNTED, [f"{tmpdirname}/secure"])

    @patch("keylime.secure_mount.logger")
    @patch("keylime.secure_mount.check_mounted")
    @patch("keylime.secure_mount.config")
    @patch("keylime.secure_mount.os.path.exists")
    @patch("keylime.secure_mount.os.makedirs")
    @patch("keylime.secure_mount.cmd_exec")
    def test_check_mount(
        self,
        _cmd_exec_mock,
        _makedirs_mock,
        exists_mock,
        config_mock,
        check_mounted_mock,
        _logger_mock,
    ):
        """Test when mounting in tmpfs but the mount point is not present."""
        exists_mock.return_value = False
        config_mock.MOUNT_SECURE = True
        check_mounted_mock.return_value = False

        with tempfile.TemporaryDirectory() as tmpdirname:
            config_mock.WORK_DIR = tmpdirname
            self.assertEqual(secure_mount.mount(), f"{tmpdirname}/secure")
            # pylint: disable=protected-access
            self.assertEqual(secure_mount._MOUNTED, [f"{tmpdirname}/secure"])

    def test_check_umount_empty(self):
        """Test umount when there are nothing to clean."""
        secure_mount.umount()
        # pylint: disable=protected-access
        self.assertEqual(secure_mount._MOUNTED, [])

    @patch("keylime.secure_mount.logger")
    @patch("keylime.secure_mount.check_mounted")
    @patch("keylime.secure_mount.cmd_exec")
    def test_check_umount_mount(self, cmd_exec_mock, check_mounted_mock, logger_mock):
        """Test umount for a single mount."""
        # pylint: disable=protected-access
        secure_mount._MOUNTED = ["/secdir"]
        check_mounted_mock.return_value = True
        cmd_exec_mock.run.return_value = {"code": 0}
        secure_mount.umount()
        self.assertEqual(secure_mount._MOUNTED, [])
        cmd_exec_mock.run.assert_called_once()
        logger_mock.error.assert_not_called()
        logger_mock.warning.assert_not_called()

    @patch("keylime.secure_mount.logger")
    @patch("keylime.secure_mount.check_mounted")
    @patch("keylime.secure_mount.cmd_exec")
    def test_check_umount_mount_busy(
        self, cmd_exec_mock, check_mounted_mock, logger_mock
    ):
        """Test umount for a single mount that fail."""
        # pylint: disable=protected-access
        secure_mount._MOUNTED = ["/secdir"]
        check_mounted_mock.return_value = True
        cmd_exec_mock.run.return_value = {"code": 1, "reterr": "Device busy"}
        secure_mount.umount()
        self.assertEqual(secure_mount._MOUNTED, [])
        cmd_exec_mock.run.assert_called_once()
        logger_mock.error.assert_called_once()
        logger_mock.warning.assert_not_called()

    @patch("keylime.secure_mount.logger")
    @patch("keylime.secure_mount.check_mounted")
    @patch("keylime.secure_mount.cmd_exec")
    def test_check_umount_umount(self, cmd_exec_mock, check_mounted_mock, logger_mock):
        """Test umount for a single umount and no creation."""
        # pylint: disable=protected-access
        secure_mount._MOUNTED = ["/secdir"]
        check_mounted_mock.return_value = False
        secure_mount.umount()
        self.assertEqual(secure_mount._MOUNTED, [])
        cmd_exec_mock.run.assert_not_called()
        logger_mock.error.assert_not_called()
        logger_mock.warning.assert_called_once()