1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
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()
|
