diff options
Diffstat (limited to 'tests/test_canonical_subids.py')
| -rw-r--r-- | tests/test_canonical_subids.py | 174 |
1 files changed, 174 insertions, 0 deletions
diff --git a/tests/test_canonical_subids.py b/tests/test_canonical_subids.py new file mode 100644 index 0000000..3cf16f6 --- /dev/null +++ b/tests/test_canonical_subids.py @@ -0,0 +1,174 @@ +import tempfile +import unittest +from contextlib import nullcontext +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from scripts import canonical_subids as subids + + +class SubidTests(unittest.TestCase): + def setUp(self): + self.directory = tempfile.TemporaryDirectory() + self.addCleanup(self.directory.cleanup) + self.etc = Path(self.directory.name) + for name, text in { + "nsswitch.conf": "passwd: files authd\ngroup: files authd\n", + "login.defs": "SUB_UID_MIN 1000000\nSUB_GID_MIN 1000000\n", + "subuid": "ruifm:1000000:65536\n", + "subgid": "ruifm:1000000:65536\n", + }.items(): + (self.etc / name).write_text(text) + self.uid = 10000 + self.name = "rui.marques@canonical.com" + + def plan(self): + return subids.plan_allocations(self.etc, self.name, self.uid, {10000}, {10000}) + + def test_allocates_after_existing_ranges(self): + changes = self.plan() + for name in ("subuid", "subgid"): + self.assertEqual( + changes[self.etc / name], + f"ruifm:1000000:65536\n{self.name}:1065536:65536\n", + ) + + def test_preserves_existing_name_and_numeric_allocations(self): + (self.etc / "subuid").write_text(f"{self.name}:2000000:1000\n") + (self.etc / "subgid").write_text(f"{self.uid}:3000000:65536\n") + self.assertEqual(self.plan(), {}) + + def test_fills_only_missing_file_and_is_idempotent(self): + (self.etc / "subgid").write_text(f"{self.name}:2000000:65536\n") + changes = self.plan() + self.assertEqual(list(changes), [self.etc / "subuid"]) + for path, text in changes.items(): + subids.write_atomic(path, text) + self.assertEqual(self.plan(), {}) + + def test_skips_real_user_and_group_ids(self): + changes = subids.plan_allocations( + self.etc, self.name, self.uid, {1065536}, {1065537} + ) + self.assertIn(f"{self.name}:1065537:65536", changes[self.etc / "subuid"]) + self.assertIn(f"{self.name}:1065538:65536", changes[self.etc / "subgid"]) + + def test_rejects_external_provider(self): + (self.etc / "nsswitch.conf").write_text("subid: sss\n") + with self.assertRaisesRegex(ValueError, "provider"): + self.plan() + + def test_rejects_exhausted_range(self): + with (self.etc / "login.defs").open("a") as stream: + stream.write("SUB_UID_MAX 1065535\n") + with self.assertRaisesRegex(ValueError, "free"): + self.plan() + + def test_rejects_disabled_allocation(self): + with (self.etc / "login.defs").open("a") as stream: + stream.write("SUB_UID_COUNT 0\n") + with self.assertRaisesRegex(ValueError, "disabled"): + self.plan() + + def test_rejects_malformed_ranges_before_writing(self): + (self.etc / "subgid").write_text("broken entry\n") + with self.assertRaises(ValueError): + self.plan() + self.assertEqual((self.etc / "subuid").read_text(), "ruifm:1000000:65536\n") + + def test_handles_missing_file_and_missing_final_newline(self): + (self.etc / "subuid").unlink() + (self.etc / "subgid").write_text("ruifm:1000000:65536") + changes = self.plan() + self.assertEqual(changes[self.etc / "subuid"], f"{self.name}:1000000:65536\n") + self.assertIn("65536\n" + self.name, changes[self.etc / "subgid"]) + + def test_atomic_write_preserves_mode_and_leaves_no_temporary_file(self): + target = self.etc / "subuid" + target.chmod(0o640) + subids.write_atomic(target, "replacement\n") + self.assertEqual(target.read_text(), "replacement\n") + self.assertEqual(target.stat().st_mode & 0o777, 0o640) + self.assertEqual(len(list(self.etc.iterdir())), 4) + + def test_failed_replace_preserves_file(self): + target = self.etc / "subuid" + with ( + patch.object(subids.os, "replace", side_effect=OSError("failure")), + self.assertRaises(OSError), + ): + subids.write_atomic(target, "replacement\n") + self.assertEqual(target.read_text(), "ruifm:1000000:65536\n") + self.assertEqual(len(list(self.etc.iterdir())), 4) + + def test_rejects_symlink_without_touching_its_target(self): + path = self.etc / "subuid" + path.unlink() + path.symlink_to(self.etc / "subgid") + with self.assertRaisesRegex(ValueError, "symlink"): + self.plan() + self.assertEqual((self.etc / "subgid").read_text(), "ruifm:1000000:65536\n") + + def test_respects_larger_configured_range_count(self): + with (self.etc / "login.defs").open("a") as stream: + stream.write("SUB_UID_COUNT 131072\n") + self.assertIn(f"{self.name}:1065536:131072", self.plan()[self.etc / "subuid"]) + + def test_lock_is_released_on_error(self): + libc = MagicMock() + libc.lckpwdf.return_value = 0 + with ( + patch.object(subids.ctypes, "CDLL", return_value=libc), + self.assertRaises(ValueError), + subids.account_lock(), + ): + raise ValueError("failure") + libc.ulckpwdf.assert_called_once_with() + + def test_failed_lock_does_not_enter_or_unlock(self): + libc = MagicMock() + libc.lckpwdf.return_value = -1 + with ( + patch.object(subids.ctypes, "CDLL", return_value=libc), + self.assertRaises(OSError), + subids.account_lock(), + ): + self.fail("Entered without the account lock") + libc.ulckpwdf.assert_not_called() + + def test_configure_verifies_both_ranges_with_system_commands(self): + with ( + patch.object( + subids.pwd, + "getpwnam", + return_value=SimpleNamespace(pw_uid=10000, pw_gid=10000), + ), + patch.object(subids.pwd, "getpwall", return_value=[]), + patch.object(subids.grp, "getgrall", return_value=[]), + patch.object(subids, "account_lock", return_value=nullcontext()), + patch.object(subids, "plan_allocations", return_value={}) as plan, + patch.object(subids, "write_atomic") as write, + patch.object(subids.subprocess, "run") as run, + ): + subids.configure(self.name) + plan.assert_called_once_with(Path("/etc"), self.name, 10000, set(), {10000}) + write.assert_not_called() + self.assertEqual(run.call_count, 2) + run.assert_any_call(["/usr/bin/getsubids", self.name], check=True) + run.assert_any_call(["/usr/bin/getsubids", "-g", self.name], check=True) + + def test_root_account_is_rejected_before_file_access(self): + with ( + patch.object( + subids.pwd, "getpwnam", return_value=SimpleNamespace(pw_uid=0) + ), + patch.object(subids, "account_lock") as lock, + self.assertRaises(ValueError), + ): + subids.configure("root") + lock.assert_not_called() + + +if __name__ == "__main__": + unittest.main() |