ProxMenux/oci/tests/test_readonly_volume_preparation.py

"""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 keep ordinary directories and run real 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
SCRIPT_DIR="$REMOTE"
oci_quiet() { printf "WRITE cleanup %s\n" "${!#}"; "$@"; }
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), REMOTE=str(ROOT / "remote"),
                         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.assertTrue((protected / "lost+found").is_dir())
            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 ("cleanup", "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()