CVE-2023-1786: Insertion of Sensitive Information into Log File (CWE-532)
value: Sensitive data could be exposed in logs of cloud-init before version 23.1.2. An attacker could use this information to find hashed passwords and possibly escalate their privilege.
CWE description: Information written to log files can be of a sensitive nature and give valuable guidance to an attacker or expose sensitive user information.
CVSS v3 base score: 5.5
Published: 2023-04-26T23:15Z
Repository: https://github.com/canonical/cloud-init
Language: Python
Files changed: cloudinit/sources/DataSourceLXD.py, cloudinit/sources/DataSourceVultr.py, cloudinit/sources/__init__.py, cloudinit/stages.py, tests/unittests/sources/test_init.py, tests/unittests/test_stages.py
Commit date: 2023-04-26 15:11:55 -0500
Commit message: Make user/vendor data sensitive and remove log permissions (#2144)

Because user data and vendor data may contain sensitive information,
this commit ensures that any user data or vendor data written to
instance-data.json gets redacted and is only available to root user.

Also, modify the permissions of cloud-init.log to be 640, so that
sensitive data leaked to the log isn't world readable.
Additionally, remove the logging of user data and vendor data to
cloud-init.log from the Vultr datasource.

LP: #2013967
CVE: CVE-2023-1786
Diff: diff --git a/cloudinit/sources/DataSourceLXD.py b/cloudinit/sources/DataSourceLXD.py
index 2643149b3..6a41fbcc4 100644
--- a/cloudinit/sources/DataSourceLXD.py
+++ b/cloudinit/sources/DataSourceLXD.py
@@ -12,11 +12,11 @@ import os
 import socket
 import stat
 import time
 from enum import Flag, auto
 from json.decoder import JSONDecodeError
-from typing import Any, Dict, List, Optional, Union, cast
+from typing import Any, Dict, List, Optional, Tuple, Union, cast
 
 import requests
 from requests.adapters import HTTPAdapter
 
 # Note: `urllib3` is transitively installed by `requests`.
@@ -166,15 +166,18 @@ class DataSourceLXD(sources.DataSource):
     dsname = "LXD"
 
     _network_config: Union[Dict, str] = sources.UNSET
     _crawled_metadata: Union[Dict, str] = sources.UNSET
 
-    sensitive_metadata_keys = (
-        "merged_cfg",
+    sensitive_metadata_keys: Tuple[
+        str, ...
+    ] = sources.DataSource.sensitive_metadata_keys + (
         "user.meta-data",
         "user.vendor-data",
         "user.user-data",
+        "cloud-init.user-data",
+        "cloud-init.vendor-data",
     )
 
     skip_hotplug_detect = True
 
     def _unpickle(self, ci_pkl_version: int) -> None:
diff --git a/cloudinit/sources/DataSourceVultr.py b/cloudinit/sources/DataSourceVultr.py
index 7b7cc696a..4141be90d 100644
--- a/cloudinit/sources/DataSourceVultr.py
+++ b/cloudinit/sources/DataSourceVultr.py
@@ -3,10 +3,12 @@
 # This file is part of cloud-init. See LICENSE file for license information.
 
 # Vultr Metadata API:
 # https://www.vultr.com/metadata/
 
+from typing import Tuple
+
 import cloudinit.sources.helpers.vultr as vultr
 from cloudinit import log as log
 from cloudinit import sources, stages, util, version
 
 LOG = log.getLogger(__name__)
@@ -26,10 +28,14 @@ BUILTIN_DS_CONFIG = {
 
 class DataSourceVultr(sources.DataSource):
 
     dsname = "Vultr"
 
+    sensitive_metadata_keys: Tuple[
+        str, ...
+    ] = sources.DataSource.sensitive_metadata_keys + ("startup-script",)
+
     def __init__(self, sys_cfg, distro, paths):
         super(DataSourceVultr, self).__init__(sys_cfg, distro, paths)
         self.ds_cfg = util.mergemanydict(
             [
                 util.get_cfg_by_path(sys_cfg, ["datasource", "Vultr"], {}),
@@ -52,17 +58,12 @@ class DataSourceVultr(sources.DataSource):
 
         # Generate config and process data
         self.get_datasource_data(self.metadata)
 
         # Dump some data so diagnosing failures is manageable
-        LOG.debug("Vultr Vendor Config:")
-        LOG.debug(util.json_dumps(self.metadata["vendor-data"]))
         LOG.debug("SUBID: %s", self.metadata["instance-id"])
         LOG.debug("Hostname: %s", self.metadata["local-hostname"])
-        if self.userdata_raw is not None:
-            LOG.debug("User-Data:")
-            LOG.debug(self.userdata_raw)
 
         return True
 
     # Process metadata
     def get_datasource_data(self, md):
@@ -153,8 +154,5 @@ if __name__ == "__main__":
         BUILTIN_DS_CONFIG["wait"],
         BUILTIN_DS_CONFIG["user-agent"],
     )
     config = md["vendor-data"]
     sysinfo = vultr.get_sysinfo()
-
-    print(util.json_dumps(sysinfo))
-    print(util.json_dumps(config))
diff --git a/cloudinit/sources/__init__.py b/cloudinit/sources/__init__.py
index 90521ba2f..9d91512f2 100644
--- a/cloudinit/sources/__init__.py
+++ b/cloudinit/sources/__init__.py
@@ -109,11 +109,14 @@ def process_instance_metadata(metadata, key_path="", sensitive_keys=()):
     for key, val in metadata.items():
         if key_path:
             sub_key_path = key_path + "/" + key
         else:
             sub_key_path = key
-        if key in sensitive_keys or sub_key_path in sensitive_keys:
+        if (
+            key.lower() in sensitive_keys
+            or sub_key_path.lower() in sensitive_keys
+        ):
             sens_keys.append(sub_key_path)
         if isinstance(val, str) and val.startswith("ci-b64:"):
             base64_encoded_keys.append(sub_key_path)
             md_copy[key] = val.replace("ci-b64:", "")
         if isinstance(val, dict):
@@ -131,20 +134,31 @@ def process_instance_metadata(metadata, key_path="", sensitive_keys=()):
 def redact_sensitive_keys(metadata, redact_value=REDACT_SENSITIVE_VALUE):
     """Redact any sensitive keys from to provided metadata dictionary.
 
     Replace any keys values listed in 'sensitive_keys' with redact_value.
     """
+    # While 'sensitive_keys' should already sanitized to only include what
+    # is in metadata, it is possible keys will overlap. For example, if
+    # "merged_cfg" and "merged_cfg/ds/userdata" both match, it's possible that
+    # "merged_cfg" will get replaced first, meaning "merged_cfg/ds/userdata"
+    # no longer represents a valid key.
+    # Thus, we still need to do membership checks in this function.
     if not metadata.get("sensitive_keys", []):
         return metadata
     md_copy = copy.deepcopy(metadata)
     for key_path in metadata.get("sensitive_keys"):
         path_parts = key_path.split("/")
         obj = md_copy
         for path in path_parts:
-            if isinstance(obj[path], dict) and path != path_parts[-1]:
+            if (
+                path in obj
+                and isinstance(obj[path], dict)
+                and path != path_parts[-1]
+            ):
                 obj = obj[path]
-        obj[path] = redact_value
+        if path in obj:
+            obj[path] = redact_value
     return md_copy
 
 
 URLParams = namedtuple(
     "URLParams",
@@ -248,10 +262,18 @@ class DataSource(CloudInitPickleMixin, metaclass=abc.ABCMeta):
     # N-tuple of keypaths or keynames redact from instance-data.json for
     # non-root users
     sensitive_metadata_keys: Tuple[str, ...] = (
         "merged_cfg",
         "security-credentials",
+        "userdata",
+        "user-data",
+        "user_data",
+        "vendordata",
+        "vendor-data",
+        # Provide ds/vendor_data to avoid redacting top-level
+        #  "vendor_data": {enabled: True}
+        "ds/vendor_data",
     )
 
     # True on datasources that may not see hotplugged devices reflected
     # in the updated metadata
     skip_hotplug_detect = False
diff --git a/cloudinit/stages.py b/cloudinit/stages.py
index 65f952e7a..509d8f7f3 100644
--- a/cloudinit/stages.py
+++ b/cloudinit/stages.py
@@ -201,11 +201,13 @@ class Init:
 
     def _initialize_filesystem(self):
         util.ensure_dirs(self._initial_subdirs())
         log_file = util.get_cfg_option_str(self.cfg, "def_log_file")
         if log_file:
-            util.ensure_file(log_file, mode=0o640, preserve_mode=True)
+            # At this point the log file should have already been created
+            # in the setupLogging function of log.py
+            util.ensure_file(log_file, mode=0o640, preserve_mode=False)
             perms = self.cfg.get("syslog_fix_perms")
             if not perms:
                 perms = {}
             if not isinstance(perms, list):
                 perms = [perms]
diff --git a/tests/unittests/sources/test_init.py b/tests/unittests/sources/test_init.py
index 96e4dd90c..005a571b8 100644
--- a/tests/unittests/sources/test_init.py
+++ b/tests/unittests/sources/test_init.py
@@ -456,16 +456,28 @@ class TestDataSource(CiTestCase):
                     "security-credentials": {
                         "cred1": "sekret",
                         "cred2": "othersekret",
                     }
                 },
+                "someother": {
+                    "nested": {
+                        "userData": "HIDE ME",
+                    }
+                },
+                "VENDOR-DAta": "HIDE ME TOO",
             },
         )
         self.assertCountEqual(
             (
                 "merged_cfg",
                 "security-credentials",
+                "userdata",
+                "user-data",
+                "user_data",
+                "vendordata",
+                "vendor-data",
+                "ds/vendor_data",
             ),
             datasource.sensitive_metadata_keys,
         )
         sys_info = {
             "python": "3.7",
@@ -488,20 +500,23 @@ class TestDataSource(CiTestCase):
         redacted = util.load_json(util.load_file(json_file))
         expected = {
             "base64_encoded_keys": [],
             "merged_cfg": REDACT_SENSITIVE_VALUE,
             "sensitive_keys": [
+                "ds/meta_data/VENDOR-DAta",
                 "ds/meta_data/some/security-credentials",
+                "ds/meta_data/someother/nested/userData",
                 "merged_cfg",
             ],
             "sys_info": sys_info,
             "v1": {
                 "_beta_keys": ["subplatform"],
                 "availability-zone": "myaz",
                 "availability_zone": "myaz",
                 "cloud-name": "subclasscloudname",
                 "cloud_name": "subclasscloudname",
+                "cloud_id": "subclasscloudname",
                 "distro": "ubuntu",
                 "distro_release": "focal",
                 "distro_version": "20.04",
                 "instance-id": "iid-datasource",
                 "instance_id": "iid-datasource",
@@ -520,18 +535,22 @@ class TestDataSource(CiTestCase):
                 "variant": "ubuntu",
             },
             "ds": {
                 "_doc": EXPERIMENTAL_TEXT,
                 "meta_data": {
+                    "VENDOR-DAta": REDACT_SENSITIVE_VALUE,
                     "availability_zone": "myaz",
                     "local-hostname": "test-subclass-hostname",
                     "region": "myregion",
                     "some": {"security-credentials": REDACT_SENSITIVE_VALUE},
+                    "someother": {
+                        "nested": {"userData": REDACT_SENSITIVE_VALUE}
+                    },
                 },
             },
         }
-        self.assertCountEqual(expected, redacted)
+        self.assertEqual(expected, redacted)
         file_stat = os.stat(json_file)
         self.assertEqual(0o644, stat.S_IMODE(file_stat.st_mode))
 
     def test_get_data_writes_json_instance_data_sensitive(self):
         """
@@ -572,10 +591,16 @@ class TestDataSource(CiTestCase):
 
         self.assertCountEqual(
             (
                 "merged_cfg",
                 "security-credentials",
+                "userdata",
+                "user-data",
+                "user_data",
+                "vendordata",
+                "vendor-data",
+                "ds/vendor_data",
             ),
             datasource.sensitive_metadata_keys,
         )
         with mock.patch("cloudinit.util.system_info", return_value=sys_info):
             with mock.patch(
diff --git a/tests/unittests/test_stages.py b/tests/unittests/test_stages.py
index 15a7e973d..a61f9df9b 100644
--- a/tests/unittests/test_stages.py
+++ b/tests/unittests/test_stages.py
@@ -604,21 +604,25 @@ class TestInit_InitializeFilesystem:
 
         assert log_file.exists()
         # Assert we create it 0o640  by default if it doesn't already exist
         assert 0o640 == stat.S_IMODE(log_file.stat().mode)
 
-    def test_existing_file_permissions_are_not_modified(self, init, tmpdir):
-        """If the log file already exists, we should not modify its permissions
+    def test_existing_file_permissions(self, init, tmpdir):
+        """Test file permissions are set as expected.
+
+        CIS Hardening requires 640 permissions. These permissions are
+        currently hardcoded on every boot, but if there's ever a reason
+        to change this, we need to then ensure that they
+        are *not* set every boot.
 
         See https://bugs.launchpad.net/cloud-init/+bug/1900837.
         """
-        # Use a mode that will never be made the default so this test will
-        # always be valid
-        mode = 0o606
         log_file = tmpdir.join("cloud-init.log")
         log_file.ensure()
-        log_file.chmod(mode)
+        # Use a mode that will never be made the default so this test will
+        # always be valid
+        log_file.chmod(0o606)
         init._cfg = {"def_log_file": str(log_file)}
 
         init._initialize_filesystem()
 
-        assert mode == stat.S_IMODE(log_file.stat().mode)
+        assert 0o640 == stat.S_IMODE(log_file.stat().mode)

Vulnerable code: from typing import Any, Dict, List, Optional, Union, cast
    sensitive_metadata_keys = (
        "merged_cfg",
        LOG.debug("Vultr Vendor Config:")
        LOG.debug(util.json_dumps(self.metadata["vendor-data"]))
        if self.userdata_raw is not None:
            LOG.debug("User-Data:")
            LOG.debug(self.userdata_raw)

    print(util.json_dumps(sysinfo))
    print(util.json_dumps(config))
        if key in sensitive_keys or sub_key_path in sensitive_keys:
            if isinstance(obj[path], dict) and path != path_parts[-1]:
        obj[path] = redact_value
            util.ensure_file(log_file, mode=0o640, preserve_mode=True)
        self.assertCountEqual(expected, redacted)
    def test_existing_file_permissions_are_not_modified(self, init, tmpdir):
        """If the log file already exists, we should not modify its permissions
        # Use a mode that will never be made the default so this test will
        # always be valid
        mode = 0o606
        log_file.chmod(mode)
        assert mode == stat.S_IMODE(log_file.stat().mode)
Fixed code: from typing import Any, Dict, List, Optional, Tuple, Union, cast
    sensitive_metadata_keys: Tuple[
        str, ...
    ] = sources.DataSource.sensitive_metadata_keys + (
        "cloud-init.user-data",
        "cloud-init.vendor-data",
from typing import Tuple

    sensitive_metadata_keys: Tuple[
        str, ...
    ] = sources.DataSource.sensitive_metadata_keys + ("startup-script",)

        if (
            key.lower() in sensitive_keys
            or sub_key_path.lower() in sensitive_keys
        ):
    # While 'sensitive_keys' should already sanitized to only include what
    # is in metadata, it is possible keys will overlap. For example, if
    # "merged_cfg" and "merged_cfg/ds/userdata" both match, it's possible that
    # "merged_cfg" will get replaced first, meaning "merged_cfg/ds/userdata"
    # no longer represents a valid key.
    # Thus, we still need to do membership checks in this function.
            if (
                path in obj
                and isinstance(obj[path], dict)
                and path != path_parts[-1]
            ):
        if path in obj:
            obj[path] = redact_value
        "userdata",
        "user-data",
        "user_data",
        "vendordata",
        "vendor-data",
        # Provide ds/vendor_data to avoid redacting top-level
        #  "vendor_data": {enabled: True}
        "ds/vendor_data",
            # At this point the log file should have already been created
            # in the setupLogging function of log.py
            util.ensure_file(log_file, mode=0o640, preserve_mode=False)
                "someother": {
                    "nested": {
                        "userData": "HIDE ME",
                    }
                },
                "VENDOR-DAta": "HIDE ME TOO",
                "userdata",
                "user-data",
                "user_data",
                "vendordata",
                "vendor-data",
                "ds/vendor_data",
                "ds/meta_data/VENDOR-DAta",
                "ds/meta_data/someother/nested/userData",
                "cloud_id": "subclasscloudname",
                    "VENDOR-DAta": REDACT_SENSITIVE_VALUE,
                    "someother": {
                        "nested": {"userData": REDACT_SENSITIVE_VALUE}
                    },
        self.assertEqual(expected, redacted)
                "userdata",
                "user-data",
                "user_data",
                "vendordata",
                "vendor-data",
                "ds/vendor_data",
    def test_existing_file_permissions(self, init, tmpdir):
        """Test file permissions are set as expected.

        CIS Hardening requires 640 permissions. These permissions are
        currently hardcoded on every boot, but if there's ever a reason
        to change this, we need to then ensure that they
        are *not* set every boot.
        # Use a mode that will never be made the default so this test will
        # always be valid
        log_file.chmod(0o606)
        assert 0o640 == stat.S_IMODE(log_file.stat().mode)