fix(mounts): escape special characters in fstab mount paths

cloudinit/config/cc_mounts.py

@@ -369,7 +369,7 @@ def parse_fstab() -> Tuple[List[str], Dict[str, str], List[str]]:
                 continue
             toks = line.split()
             if toks:
-                fstab_devs[toks[0]] = line
+                fstab_devs[util.unescape_fstab_field(toks[0])] = line
                 fstab_lines.append(line)
     return fstab_lines, fstab_devs, fstab_removed
 
@@ -584,7 +584,18 @@ def handle(name: str, cfg: Config, cloud: Cloud, args: list) -> None:
         LOG.debug("No modifications to fstab needed")
         return
 
-    cfg_lines = ["\t".join(entry) for entry in updated_cfg]
+    # Only fs_spec (device) and fs_file (mount point) can contain values that
+    # need escaping; the remaining fstab fields never do, so we skip them.
+    cfg_lines = [
+        "\t".join(
+            [
+                util.escape_fstab_field(entry[0]),
+                util.escape_fstab_field(entry[1]),
+                *entry[2:],
+            ]
+        )
+        for entry in updated_cfg
+    ]
 
     dirs = [d[1] for d in updated_cfg if d[1].startswith("/")]
 

cloudinit/util.py

@@ -1910,6 +1910,35 @@ def unmounter(umount):
             subp.subp(umount_cmd)
 
 
+# Per fstab(5), fstab fields are whitespace-separated, so these characters
+# must be octal-escaped when they appear inside a field. Backslash must come
+# first so we don't double-escape the escapes we introduce.
+FSTAB_ESCAPES = (
+    ("\\", "\\134"),
+    (" ", "\\040"),
+    ("\t", "\\011"),
+    ("\n", "\\012"),
+)
+
+
+def escape_fstab_field(value: str) -> str:
+    """Octal-escape special characters for safe writing to fstab."""
+    for char, escaped in FSTAB_ESCAPES:
+        value = value.replace(char, escaped)
+    return value
+
+
+def unescape_fstab_field(value: str) -> str:
+    """Reverse escape_fstab_field.
+
+    Iterate FSTAB_ESCAPES in reverse so backslash is decoded last; this
+    avoids mis-decoding a field whose original value contained a backslash.
+    """
+    for char, escaped in reversed(FSTAB_ESCAPES):
+        value = value.replace(escaped, char)
+    return value
+
+
 def mounts():
     mounted = {}
     try:
@@ -1938,9 +1967,9 @@ def mounts():
                 mp = m.group(2)
                 fstype = m.group(3)
                 opts = m.group(4)
-            # If the name of the mount point contains spaces these
-            # can be escaped as '\040', so undo that..
-            mp = mp.replace("\\040", " ")
+            # Mount points may contain octal-escaped characters (e.g. a
+            # space as '\040'); undo that so callers see the real path.
+            mp = unescape_fstab_field(mp)
             mounted[dev] = {
                 "fstype": fstype,
                 "mountpoint": mp,

doc/module-docs/cc_mounts/data.yaml

@@ -11,7 +11,12 @@ cc_mounts:
     omitted. Any mounts that do not appear to either an attached block device
     or network resource will be skipped with a log like "Ignoring nonexistent
     mount ...".
-
+
+    Whitespace and other special characters in the ``fs_spec`` and ``fs_file``
+    fields are automatically octal-escaped when written to ``/etc/fstab`` (for
+    example, a space becomes ``\040``), so mount points containing spaces work
+    without any manual escaping in the config.
+
     Cloud-init will attempt to add the following mount directives if available
     and unconfigured in ``/etc/fstab``:
 
@@ -65,5 +70,11 @@ cc_mounts:
       Example 2: Create a 2 GB swap file at ``/swapfile`` using human-readable
       values.
     file: cc_mounts/example2.yaml
+  - comment: >
+      Example 3: Mount a device at a mount point whose path contains a space.
+      Whitespace and other special characters in the ``fs_spec`` and
+      ``fs_file`` fields are automatically escaped (e.g. a space is written as
+      ``\040``) so that the resulting ``/etc/fstab`` entry can be parsed.
+    file: cc_mounts/example3.yaml
   name: Mounts
   title: Configure mount points and swap files

doc/module-docs/cc_mounts/example3.yaml

@@ -0,0 +1,3 @@
+#cloud-config
+mounts:
+- [ /dev/sr0, "/mnt/Cdrom Drive", auto, "uid=1000,gid=1000", "0", "0" ]

tests/integration_tests/bugs/test_gh6911.py

@@ -0,0 +1,45 @@
+"""Integration test for gh-6911.
+
+cc_mounts historically wrote /etc/fstab entries without escaping, so a
+space in the device (fs_spec) or mount point (fs_file) produced an
+unparseable line and broke ``mount -a``. Verify that a mount point
+containing a space is octal-escaped in /etc/fstab, that cloud-init logs
+stay clean, and that the real (unescaped) directory is created.
+
+A network fs_spec (contains ``:``) with ``noauto`` keeps the entry in
+fstab without cloud-init attempting the actual mount, so the test does
+not depend on any real device or NFS server.
+
+https://github.com/canonical/cloud-init/issues/3603
+"""
+
+import pytest
+
+from tests.integration_tests.instances import IntegrationInstance
+from tests.integration_tests.util import verify_clean_log
+
+USER_DATA = """\
+#cloud-config
+mounts:
+- - server.example:/export
+  - /mnt/Cdrom Drive
+  - nfs
+  - defaults,noauto
+  - "0"
+  - "0"
+"""
+
+
+@pytest.mark.user_data(USER_DATA)
+def test_mount_point_with_space_is_escaped(client: IntegrationInstance):
+    # cloud-init should have configured mounts without any errors.
+    log = client.read_from_file("/var/log/cloud-init.log")
+    verify_clean_log(log)
+    assert "SUCCESS: config-mounts ran successfully" in log
+
+    # The space must be octal-escaped (\040) in the /etc/fstab entry.
+    fstab = client.read_from_file("/etc/fstab")
+    assert "/mnt/Cdrom\\040Drive" in fstab
+
+    # The real directory (with a literal space) must have been created.
+    assert client.execute("test -d '/mnt/Cdrom Drive'").ok

tests/unittests/config/test_cc_mounts.py

@@ -535,6 +535,52 @@ def test_fstab_mounts_combinations(self):
             ).strip()
         )
 
+    def test_fstab_mountpoint_with_spaces(self):
+        """Spaces in the fs_spec and fs_file fields are octal-escaped.
+
+        Reproduces GH-3603: an unescaped space in the device or mount point
+        breaks `mount -a` parsing of /etc/fstab.
+        """
+        cfg = {
+            "mounts": [
+                [
+                    "/dev/sr0 spaced",
+                    "/mnt/Cdrom Drive",
+                    "auto",
+                    "uid=1000,gid=1000",
+                    "0",
+                    "0",
+                ],
+            ]
+        }
+        cc_mounts.handle("", cfg, self.mock_cloud, [])
+        with open(cc_mounts.FSTAB_PATH, "r") as fd:
+            fstab_new_content = fd.read()
+        assert (
+            "/dev/sr0\\040spaced\t/mnt/Cdrom\\040Drive\tauto\t"
+            "uid=1000,gid=1000,comment=cloudconfig\t0\t0\n"
+        ) in fstab_new_content
+
+    def test_fstab_mountpoint_with_spaces_idempotent(self):
+        """A second run must not duplicate an escaped mount entry."""
+        cfg = {
+            "mounts": [
+                [
+                    "/dev/sr0",
+                    "/mnt/Cdrom Drive",
+                    "auto",
+                    "uid=1000,gid=1000",
+                    "0",
+                    "0",
+                ],
+            ]
+        }
+        cc_mounts.handle("", cfg, self.mock_cloud, [])
+        cc_mounts.handle("", cfg, self.mock_cloud, [])
+        with open(cc_mounts.FSTAB_PATH, "r") as fd:
+            fstab_new_content = fd.read()
+        assert fstab_new_content.count("/mnt/Cdrom\\040Drive") == 1
+
 
 class TestCreateSwapfile:
     @pytest.mark.parametrize("fstype", ("xfs", "btrfs", "ext4", "other"))

tests/unittests/test_util.py

@@ -2145,6 +2145,24 @@ def test_none_returns_default(self):
         ]
 
 
+class TestFstabEscaping:
+    @pytest.mark.parametrize(
+        "raw, escaped",
+        [
+            ("/mnt/Cdrom Drive", "/mnt/Cdrom\\040Drive"),
+            ("/mnt/a\tb", "/mnt/a\\011b"),
+            ("/mnt/a\nb", "/mnt/a\\012b"),
+            ("/mnt/a\\b", "/mnt/a\\134b"),
+            # backslash escaped first so existing escapes aren't doubled
+            ("/mnt/a \\b", "/mnt/a\\040\\134b"),
+            ("/mnt/plain", "/mnt/plain"),
+        ],
+    )
+    def test_escape_unescape_roundtrip(self, raw, escaped):
+        assert util.escape_fstab_field(raw) == escaped
+        assert util.unescape_fstab_field(escaped) == raw
+
+
 class TestMountinfoParsing:
     def test_invalid_mountinfo(self):
         line = (