fix(oci): skip preparation of read-only volumes

.github/workflows/test-oci-registries.yml

@@ -28,6 +28,13 @@ jobs:
           python-version: '3.13'
       - name: Install OCI test dependencies
         run: python3 -m pip install -r oci/requirements.txt
+      - name: Run read-only volume preparation regressions
+        env:
+          PYTHONDONTWRITEBYTECODE: '1'
+        run: |
+          test -x /bin/bash
+          test -x /usr/bin/jq
+          python3 -m unittest discover -s oci/tests -p test_readonly_volume_preparation.py -v
       - name: Run offline registry account tests
         env:
           PYTHONDONTWRITEBYTECODE: '1'

oci/remote/install_oci.sh

@@ -1230,6 +1230,17 @@ apply_installer_profile() {
       fi
       [[ -d ${rootfs}${target} ]] \
         || { oci_log "The volume to prepare does not exist: $target"; failed=1; break; }
+      # A reused volume may intentionally be mounted read-only (for example,
+      # /run/secrets). Do not remove files or change ownership/mode on it.
+      # Use the most specific containing mount: an RW child can override RO.
+      if [[ $(jq -r --arg target "$target" '
+        [.mounts[]? | select(.container_path as $path |
+          $target == $path or ($target | startswith($path + "/")))]
+        | sort_by(.container_path | length) | last | .read_only // false' \
+        "$DEPLOYMENT_FILE") == true ]]; then
+        oci_log "Skipping preparation of read-only volume: $target"
+        continue
+      fi
       if [[ $remove_lost_found == true ]]; then
         rm -rf "${rootfs}${target}/lost+found" || { failed=1; break; }
       fi

oci/tests/test_readonly_volume_preparation.py

@@ -0,0 +1,187 @@
+"""Run the installer's volume preparation loop against filesystem fixtures.
+
+Write doubles reject every mutation of the protected directory, including
+lost+found removal. Writable paths still run real rm/chmod; chown is recorded
+without requiring root. These tests never invoke Proxmox or touch a container.
+"""
+
+import json
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+
+
+ROOT = Path(__file__).resolve().parents[1]
+INSTALLER = ROOT / "remote/install_oci.sh"
+
+
+def preparation_loop():
+    source = INSTALLER.read_text()
+    start = source.index("  if (( failed == 0 )) && [[ $(jq -r '.stack_managed")
+    end = source.index("\n  if (( failed == 0 && tls_count == 1 )); then", start)
+    return source[start:end]
+
+
+def preparation(target):
+    return {
+        "container_path": target,
+        "remove_lost_found": True,
+        "owner_strategy": "mapped-application-user",
+        "mode": "0750",
+    }
+
+
+class ReadOnlyVolumePreparationTests(unittest.TestCase):
+    def run_case(self, mounts, *, target="/run/secrets", read_only=True,
+                 preparations=None):
+        preparations = preparations if preparations is not None else [preparation(target)]
+        with tempfile.TemporaryDirectory() as directory:
+            fixture = Path(directory)
+            rootfs = fixture / "rootfs"
+            for item in preparations:
+                path = rootfs / item["container_path"].lstrip("/")
+                path.mkdir(parents=True, exist_ok=True)
+                (path / "lost+found").mkdir(exist_ok=True)
+            protected = rootfs / target.lstrip("/")
+            protected.mkdir(parents=True, exist_ok=True)
+            sentinel = protected / "sentinel"
+            sentinel.write_bytes(b"existing secret\n")
+            sentinel.chmod(0o600)
+            protected.chmod(0o700)
+            before = sentinel.stat()
+            template = {"proxmox": {"installer_profile": {
+                "volume_preparations": preparations,
+            }}}
+            (fixture / "template.json").write_text(json.dumps(template))
+            (fixture / "deployment.json").write_text(json.dumps({"mounts": mounts}))
+            script = r'''set -Eeuo pipefail
+rootfs="$FIXTURE/rootfs"
+TEMPLATE_FILE="$FIXTURE/template.json"
+DEPLOYMENT_FILE="$FIXTURE/deployment.json"
+HOST_BIND_UID=110001; HOST_BIND_GID=110001
+HOST_ROOT_UID=100000; HOST_ROOT_GID=100000
+failed=0; volumes_prepared=0
+oci_log() { printf 'log %s\n' "$*"; }
+write_op() {
+  local operation=$1
+  shift
+  local path=${!#}
+  printf 'WRITE %s %s\n' "$operation" "$path"
+  if [[ -n $PROTECTED && ( $path == "$PROTECTED" || $path == "$PROTECTED/"* ) ]]; then
+    printf 'Read-only file system\n' >&2
+    return 1
+  fi
+  case "$operation" in
+    rm|chmod) command "$operation" "$@" ;;
+    chown) return 0 ;;
+  esac
+}
+chown() { write_op chown "$@"; }
+chmod() { write_op chmod "$@"; }
+rm() { write_op rm "$@"; }
+''' + preparation_loop() + r'''
+printf 'failed=%s prepared=%s\n' "$failed" "$volumes_prepared"
+test "$failed" = 0
+'''
+            result = subprocess.run(
+                ["bash", "-c", script], capture_output=True, text=True,
+                env=dict(os.environ, FIXTURE=str(fixture),
+                         PROTECTED=str(protected) if read_only else ""),
+                timeout=15,
+            )
+            self.assertEqual(sentinel.read_bytes(), b"existing secret\n")
+            after = sentinel.stat()
+            self.assertEqual(
+                (before.st_ino, before.st_uid, before.st_gid, before.st_mode, before.st_mtime_ns),
+                (after.st_ino, after.st_uid, after.st_gid, after.st_mode, after.st_mtime_ns),
+            )
+            if read_only:
+                self.assertEqual(protected.stat().st_mode & 0o777, 0o700)
+                self.assertTrue((protected / "lost+found").is_dir())
+            elif "prepared=1" in result.stdout:
+                self.assertEqual(protected.stat().st_mode & 0o777, 0o750)
+                self.assertFalse((protected / "lost+found").exists())
+            return result
+
+    def assert_skipped(self, result):
+        self.assertEqual(result.returncode, 0, result.stderr + result.stdout)
+        self.assertNotIn("WRITE ", result.stdout)
+        self.assertIn("Skipping preparation of read-only volume", result.stdout)
+        self.assertIn("prepared=0", result.stdout)
+
+    def assert_prepared(self, result):
+        self.assertEqual(result.returncode, 0, result.stderr + result.stdout)
+        for operation in ("rm", "chown", "chmod"):
+            self.assertIn(f"WRITE {operation} ", result.stdout)
+        self.assertIn("prepared=1", result.stdout)
+
+    def test_read_only_managed_and_bind_volumes_receive_no_writes(self):
+        for mount_type in ("managed-volume", "host-bind"):
+            with self.subTest(mount_type=mount_type):
+                self.assert_skipped(self.run_case([
+                    {"container_path": "/run/secrets", "type": mount_type, "read_only": True},
+                ]))
+
+    def test_read_only_parent_protects_nested_preparation(self):
+        self.assert_skipped(self.run_case([
+            {"container_path": "/run", "read_only": True},
+        ]))
+
+    def test_writable_child_overrides_read_only_parent_regardless_of_order(self):
+        mounts = [
+            {"container_path": "/run", "read_only": True},
+            {"container_path": "/run/secrets", "read_only": False},
+        ]
+        for ordered in (mounts, mounts[::-1]):
+            with self.subTest(mounts=ordered):
+                self.assert_prepared(self.run_case(ordered, read_only=False))
+
+    def test_child_without_read_only_flag_overrides_read_only_parent(self):
+        self.assert_prepared(self.run_case([
+            {"container_path": "/run", "read_only": True},
+            {"container_path": "/run/secrets"},
+        ], read_only=False))
+
+    def test_read_only_child_overrides_writable_parent_regardless_of_order(self):
+        mounts = [
+            {"container_path": "/run/secrets", "read_only": True},
+            {"container_path": "/run", "read_only": False},
+        ]
+        for ordered in (mounts, mounts[::-1]):
+            with self.subTest(mounts=ordered):
+                self.assert_skipped(self.run_case(ordered))
+
+    def test_writable_volume_is_still_prepared(self):
+        self.assert_prepared(self.run_case([
+            {"container_path": "/data", "read_only": False},
+        ], target="/data", read_only=False))
+
+    def test_missing_read_only_flag_defaults_to_writable(self):
+        self.assert_prepared(self.run_case([
+            {"container_path": "/data"},
+        ], target="/data", read_only=False))
+
+    def test_unrelated_mount_with_similar_prefix_does_not_skip_preparation(self):
+        self.assert_prepared(self.run_case([
+            {"container_path": "/run/secret", "read_only": True},
+        ], read_only=False))
+
+    def test_directory_outside_mounts_is_still_prepared(self):
+        self.assert_prepared(self.run_case([], read_only=False))
+
+    def test_skipping_read_only_volume_does_not_skip_following_writable_volume(self):
+        result = self.run_case([
+            {"container_path": "/run/secrets", "read_only": True},
+            {"container_path": "/data", "read_only": False},
+        ], preparations=[preparation("/run/secrets"), preparation("/data")])
+        self.assert_prepared(result)
+        self.assertIn("Skipping preparation of read-only volume: /run/secrets", result.stdout)
+        for line in result.stdout.splitlines():
+            if line.startswith("WRITE "):
+                self.assertTrue(line.endswith("/data") or line.endswith("/data/lost+found"), line)
+
+
+if __name__ == "__main__":
+    unittest.main()