diff --git a/oci/remote/install_oci.sh b/oci/remote/install_oci.sh index 7d1d8e85..88d5f0fe 100755 --- a/oci/remote/install_oci.sh +++ b/oci/remote/install_oci.sh @@ -1091,7 +1091,7 @@ ensure_runtime_home() { passwd_file="${rootfs}/etc/passwd" mount_ct_rootfs if ! home=$(python3 "$OCI_RUNTIME_RESOLVER" --home "$passwd_file" "$uid" 2>>"${OCI_LOG:-/dev/stderr}"); then - oci_quiet pct unmount "$VMID" || true + oci_quiet pct unmount "$VMID" || die "$(translate "Could not unmount the container filesystem")" return 0 fi oci_quiet pct unmount "$VMID" || die "$(translate "Could not unmount the container filesystem")" diff --git a/oci/remote/oci_runtime.py b/oci/remote/oci_runtime.py index ee3b9d56..3d842fd3 100644 --- a/oci/remote/oci_runtime.py +++ b/oci/remote/oci_runtime.py @@ -4,7 +4,9 @@ from __future__ import annotations import json +import os import shlex +import stat import sys import tarfile from pathlib import Path @@ -90,14 +92,39 @@ def effective_entrypoint( def home_directory(passwd_path: Path, uid: int) -> str | None: - """Return the safe home directory recorded for a container UID.""" - for line in passwd_path.read_text(encoding="utf-8").splitlines(): + """Read a regular image passwd file; only an absent UID defaults to /.""" + if uid < 0: + return None + # The caller supplies /etc/passwd. Pin etc so an image + # symlink or a rename during lookup cannot redirect the host-side read. + directory = os.open(passwd_path.parent, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + try: + if not stat.S_ISREG(os.stat(passwd_path.name, dir_fd=directory, follow_symlinks=False).st_mode): + return None + descriptor = os.open( + passwd_path.name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory, + ) + try: + # Recheck the opened object, not its name; a FIFO replacement + # must neither block open() nor reach a read operation. + if not stat.S_ISREG(os.fstat(descriptor).st_mode): + return None + with os.fdopen(descriptor, encoding="utf-8", closefd=False) as stream: + content = stream.read() + finally: + os.close(descriptor) + finally: + os.close(directory) + for line in content.splitlines(): fields = line.split(":") - if len(fields) < 7 or fields[2] != str(uid): + if len(fields) < 3 or fields[2] != str(uid): continue + if len(fields) != 7: + return None home = fields[5] - return home if home.startswith("/") else None - return None + safe = home.startswith("/") and not any(ord(char) < 32 or ord(char) == 127 for char in home) + return home if safe else None + return "/" def main() -> int: diff --git a/oci/tests/test_runtime_home_contract.py b/oci/tests/test_runtime_home_contract.py index c14e8e26..25e9ca9e 100644 --- a/oci/tests/test_runtime_home_contract.py +++ b/oci/tests/test_runtime_home_contract.py @@ -31,14 +31,14 @@ class RuntimeHomeResolutionTests(unittest.TestCase): ) self.assertEqual(oci_runtime.home_directory(passwd, 1001), "/srv/app") - def test_unknown_or_unsafe_home_is_not_invented(self): + def test_unsafe_record_is_rejected_but_unknown_uid_defaults_to_root(self): passwd = self.passwd( "root:x:0:0:root:relative-home:/bin/sh\n" "app:x:1001:1001:App user::/sbin/nologin\n" ) self.assertIsNone(oci_runtime.home_directory(passwd, 0)) self.assertIsNone(oci_runtime.home_directory(passwd, 1001)) - self.assertIsNone(oci_runtime.home_directory(passwd, 1234)) + self.assertEqual(oci_runtime.home_directory(passwd, 1234), "/") class RuntimeHomeInstallerContractTests(unittest.TestCase): diff --git a/oci/tests/test_runtime_home_hardening.py b/oci/tests/test_runtime_home_hardening.py new file mode 100644 index 00000000..6418709d --- /dev/null +++ b/oci/tests/test_runtime_home_hardening.py @@ -0,0 +1,260 @@ +"""Exercise HOME resolution without reading image symlinks or changing guests.""" + +from contextlib import redirect_stderr, redirect_stdout +import io +import os +from pathlib import Path +import re +import subprocess +import sys +import tempfile +import unittest +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "remote")) +import oci_runtime + + +class HomeHardeningTests(unittest.TestCase): + def setUp(self): + temporary = tempfile.TemporaryDirectory() + self.addCleanup(temporary.cleanup) + self.root = Path(temporary.name) + (self.root / "etc").mkdir() + self.passwd = self.root / "etc/passwd" + self.passwd.write_text("root:x:0:0:root:/root:/bin/sh\n", encoding="utf-8") + + def cli(self, uid="1234"): + output, error = io.StringIO(), io.StringIO() + with patch.object(sys, "argv", ["oci_runtime.py", "--home", str(self.passwd), uid]), \ + redirect_stdout(output), redirect_stderr(error): + status = oci_runtime.main() + return status, output.getvalue(), error.getvalue() + + def assert_rejected_without_read(self): + with patch.object(os, "fdopen", side_effect=AssertionError("unsafe file read")): + status, output, _ = self.cli() + self.assertEqual(status, 1) + self.assertEqual(output, "") + + def test_unknown_uid_uses_root_only_after_reading_a_regular_database(self): + self.assertEqual(self.cli(), (0, "/\n", "")) + + def test_empty_regular_database_uses_root(self): + self.passwd.write_text("") + self.assertEqual(self.cli(), (0, "/\n", "")) + + def test_final_symlink_is_rejected_without_reading_its_target(self): + target = self.root / "outside" + target.write_text("app:x:1234:1234::/host-secret:/bin/sh\n") + self.passwd.unlink() + self.passwd.symlink_to(target) + self.assert_rejected_without_read() + self.assertEqual(target.read_text(), "app:x:1234:1234::/host-secret:/bin/sh\n") + + def test_symlinked_etc_is_rejected_without_reading_its_target(self): + directory = self.root / "etc" + directory.rename(self.root / "outside-etc") + directory.symlink_to(self.root / "outside-etc", target_is_directory=True) + self.assert_rejected_without_read() + self.assertEqual(self.cli("0")[:2], (1, "")) + + def test_pinned_etc_does_not_follow_a_later_directory_replacement(self): + original_open = os.open + outside = self.root / "outside" + outside.mkdir() + (outside / "passwd").write_text("root:x:0:0::/outside:/bin/sh\n") + replaced = [] + def replace(path, flags, *args, **kwargs): + if str(path) == "passwd": + (self.root / "etc").rename(self.root / "original-etc") + (self.root / "etc").symlink_to(outside, target_is_directory=True) + replaced.append(True) + return original_open(path, flags, *args, **kwargs) + with patch.object(os, "open", side_effect=replace): + self.assertEqual(self.cli("0"), (0, "/root\n", "")) + self.assertEqual(replaced, [True]) + + def test_directory_is_rejected_without_reading(self): + self.passwd.unlink() + self.passwd.mkdir() + self.assert_rejected_without_read() + + def test_fifo_is_rejected_without_reading_or_waiting_for_a_writer(self): + self.passwd.unlink() + os.mkfifo(self.passwd) + done = subprocess.run( + [sys.executable, "-B", str(ROOT / "remote/oci_runtime.py"), "--home", str(self.passwd), "0"], + capture_output=True, text=True, timeout=3, + ) + self.assertEqual(done.returncode, 1) + self.assertEqual(done.stdout, "") + self.assert_rejected_without_read() + + def test_missing_database_does_not_masquerade_as_unknown_uid(self): + self.passwd.unlink() + self.assert_rejected_without_read() + + def test_permission_failure_does_not_masquerade_as_unknown_uid(self): + with patch.object(os, "open", side_effect=PermissionError("denied")): + self.assert_rejected_without_read() + + def test_invalid_encoding_does_not_produce_a_fallback(self): + self.passwd.write_bytes(b"\xff\xfe") + status, output, _ = self.cli() + self.assertEqual((status, output), (1, "")) + + def test_partial_matching_record_does_not_produce_a_fallback(self): + self.passwd.write_text("app:x:1234:1234\n") + self.assertEqual(self.cli()[:2], (1, "")) + + def test_read_error_closes_both_descriptors_without_a_fallback(self): + descriptors = [] + original_open = os.open + def opened(*args, **kwargs): + fd = original_open(*args, **kwargs) + descriptors.append(fd) + return fd + with patch.object(os, "open", side_effect=opened), \ + patch.object(os, "fdopen", side_effect=OSError("read failed")): + self.assertEqual(self.cli()[:2], (1, "")) + self.assertEqual(len(descriptors), 2) + for fd in descriptors: + with self.assertRaises(OSError): + os.fstat(fd) + + def test_matching_unsafe_home_is_not_replaced_with_root(self): + for value in ("", "relative", "/bad\tvalue", "/bad\x00value", "/bad\x7fvalue"): + with self.subTest(value=repr(value)): + self.passwd.write_text(f"app:x:1234:1234::{value}:/bin/sh\n") + self.assertEqual(self.cli()[:2], (1, "")) + + def test_invalid_uid_does_not_open_the_database(self): + for uid in ("-1", "nobody"): + with self.subTest(uid=uid), patch.object(os, "open") as opened: + self.assertEqual(self.cli(uid)[:2], (1, "")) + opened.assert_not_called() + + def test_symlink_replacement_between_check_and_open_is_not_read(self): + original_open = os.open + outside = self.root / "outside" + outside.write_text("app:x:1234:1234::/outside:/bin/sh\n") + def replace(path, flags, *args, **kwargs): + if str(path) == "passwd": + self.passwd.unlink() + self.passwd.symlink_to(outside) + return original_open(path, flags, *args, **kwargs) + with patch.object(os, "open", side_effect=replace): + self.assert_rejected_without_read() + + def test_fifo_replacement_between_check_and_open_is_not_read(self): + original_open = os.open + def replace(path, flags, *args, **kwargs): + if str(path) == "passwd": + self.passwd.unlink() + os.mkfifo(self.passwd) + return original_open(path, flags, *args, **kwargs) + with patch.object(os, "open", side_effect=replace): + self.assert_rejected_without_read() + + +class HomeInstallerExecutionTests(unittest.TestCase): + """Run the real shell HOME function and Python CLI with fake mount boundaries.""" + def setUp(self): + temporary = tempfile.TemporaryDirectory() + self.addCleanup(temporary.cleanup) + self.root = Path(temporary.name) + (self.root / "etc").mkdir() + self.passwd = self.root / "etc/passwd" + self.passwd.write_text("root:x:0:0:root:/root:/bin/sh\napp:x:1001:1001::/srv/app:/bin/sh\n") + self.conf = self.root / "guest.conf" + self.trace = self.root / "trace" + self.source = (ROOT / "remote/install_oci.sh").read_text() + + def run_home(self, config="unrelated: keep\n", unmount_status=0, mount_status=0): + self.conf.write_text(config) + self.trace.write_text("") + functions = [] + for name in ("ensure_runtime_home", "set_runtime_environment"): + found = re.search(rf"^{name}\(\) \{{\n.*?^\}}\n", self.source, re.M | re.S) + self.assertIsNotNone(found, name) + functions.append(found.group(0)) + script = r''' +set -euo pipefail +CONF=$1 +FIXTURE_ROOT=$2 +TRACE=$3 +OCI_RUNTIME_RESOLVER=$4 +PYTHON=$5 +UNMOUNT_STATUS=$6 +MOUNT_STATUS=$7 +OCI_LOG="$FIXTURE_ROOT/resolver.log" +VMID=123456789 +translate() { printf '%s' "$1"; } +die() { printf '%s\n' "$*" >&2; exit 1; } +mount_ct_rootfs() { + printf 'mount\n' >> "$TRACE" + [ "$MOUNT_STATUS" = 0 ] || die 'mount failed' + rootfs=$FIXTURE_ROOT + passwd_file="$rootfs/etc/passwd" +} +oci_quiet() { printf '%s\n' "$*" >> "$TRACE"; return "$UNMOUNT_STATUS"; } +python3() { printf 'resolve\n' >> "$TRACE"; command "$PYTHON" -B "$@"; } +''' + done = subprocess.run( + ["bash", "-c", script + "\n".join(functions) + "\nensure_runtime_home\nprintf 'continued\\n'\n", + "bash", str(self.conf), str(self.root), str(self.trace), str(ROOT / "remote/oci_runtime.py"), + sys.executable, str(unmount_status), str(mount_status)], + capture_output=True, text=True, timeout=5, + ) + return done, self.conf.read_text(), self.trace.read_text().splitlines() + + def test_explicit_home_including_empty_never_mounts_or_resolves(self): + for home in ("/custom", ""): + config = f"unrelated: keep\nlxc.environment.runtime: HOME={home}\n" + done, after, trace = self.run_home(config) + self.assertEqual((done.returncode, after, trace), (0, config, [])) + + def test_default_and_effective_uid_write_the_resolved_home_after_unmount(self): + for uid, expected in ((None, "/root"), (1001, "/srv/app"), (9999, "/")): + config = "unrelated: keep\n" + (f"lxc.init.uid: {uid}\n" if uid is not None else "") + done, after, trace = self.run_home(config) + self.assertEqual(done.returncode, 0, done.stderr) + self.assertEqual(after, config + f"lxc.environment.runtime: HOME={expected}\n") + self.assertEqual(trace, ["mount", "resolve", "pct unmount 123456789"]) + + def test_invalid_uid_leaves_config_unchanged_without_mount(self): + config = "lxc.init.uid: not-a-uid\n" + done, after, trace = self.run_home(config) + self.assertEqual((done.returncode, after, trace), (0, config, [])) + + def test_rejected_input_unmounts_and_preserves_config(self): + self.passwd.unlink() + self.passwd.symlink_to(self.root / "missing") + config = "unrelated: keep\n" + done, after, trace = self.run_home(config) + self.assertEqual((done.returncode, after), (0, config)) + self.assertEqual(trace, ["mount", "resolve", "pct unmount 123456789"]) + + def test_failed_unmount_stops_before_config_write_for_success_and_failure(self): + for valid in (True, False): + if not valid: + self.passwd.unlink() + config = "unrelated: keep\n" + done, after, trace = self.run_home(config, unmount_status=1) + self.assertNotEqual(done.returncode, 0) + self.assertEqual(after, config) + self.assertNotIn("continued", done.stdout) + self.assertEqual(trace, ["mount", "resolve", "pct unmount 123456789"]) + + def test_failed_mount_never_reads_or_writes(self): + config = "unrelated: keep\n" + done, after, trace = self.run_home(config, mount_status=1) + self.assertNotEqual(done.returncode, 0) + self.assertEqual(after, config) + self.assertEqual(trace, ["mount"]) + + +if __name__ == "__main__": + unittest.main()