diff --git a/.github/workflows/test-oci-registries.yml b/.github/workflows/test-oci-registries.yml index 9612c0d3..dd39bded 100644 --- a/.github/workflows/test-oci-registries.yml +++ b/.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' diff --git a/oci/remote/install_oci.sh b/oci/remote/install_oci.sh index 7d1d8e85..5d39523a 100755 --- a/oci/remote/install_oci.sh +++ b/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 diff --git a/oci/tests/test_readonly_volume_preparation.py b/oci/tests/test_readonly_volume_preparation.py new file mode 100644 index 00000000..37f67011 --- /dev/null +++ b/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()