1#!/usr/bin/env python3
2import importlib
3import json
4import os
5from pathlib import Path
6import subprocess
7import shutil
8import tempfile
9import unittest
10import uuid
11from unittest.mock import patch
12
13import release
14
15runs = importlib.import_module("dashboard-run")
16host = importlib.import_module("dashboard-host")
17
18
19class DeploymentBoundary(unittest.TestCase):
20 def setUp(self):
21 self.temporary = tempfile.TemporaryDirectory(prefix="studio-deploy-boundary-")
22 self.addCleanup(self.temporary.cleanup)
23 self.root = Path(self.temporary.name).resolve()
24 self.state = self.root / "state"
25 self.state.mkdir()
26 self.releases = self.root / "releases"
27 self.releases.mkdir()
28 self.history = self.state / "deployments.json"
29 self.history.write_text("[]")
30 for name, value in [("ROOT", self.root), ("STATE", self.state), ("RELEASES", self.releases), ("HISTORY", self.history)]:
31 context = patch.object(release, name, value)
32 context.start()
33 self.addCleanup(context.stop)
34 self.host = host.Host("studio-demo")
35 state_context = patch.object(runs, "HOST_STATE", self.state / "host")
36 state_context.start()
37 self.addCleanup(state_context.stop)
38 self.version = "a" * 16
39 self.snapshot = self.releases / self.version
40 self.snapshot.mkdir()
41 for name in release.SOURCES:
42 path = self.snapshot / name
43 if "." in name:
44 path.write_text("fixture")
45 else:
46 path.mkdir()
47 (self.snapshot / "tools/studio.py").write_text("print('fixture')")
48 (self.snapshot / ".studio-release.json").write_text(json.dumps({"id": self.version, "digest": release.tree_digest(self.snapshot), "version": 2, "main": {"commit": "b" * 40, "description": "Fixture main"}}))
49 (self.state / "stages").mkdir()
50 self.stage = "fixture-preview"
51 self.stage_file = self.state / "stages" / (self.stage + ".json")
52 self.stage_file.write_text(json.dumps({"sourceId": "fixture", "release": self.version, "ready": True,
53 "clone": None, "mount": "/srv/staging/fixture", "overrides": []}))
54 self.history.write_text(json.dumps([{"release": self.version, "source": "fixture", "time": 123, "legacy": False}]))
55 (self.state / "managed-jobs.json").write_text('["fixture"]')
56 (self.root / "current").symlink_to(self.snapshot)
57 (self.root / "main").symlink_to(self.snapshot)
58
59 def call(self, operation, **kwargs):
60 return self.host.handle({"operation": operation, **kwargs})
61
62 def test_requests_cannot_select_commands_paths_or_environment(self):
63 requests = [{"operation": "iam.request", "path": "/users", "method": "GET", "body": None},
64 {"operation": "deploy.start", "action": "destroy", "target": self.stage, "argv": ["sh"]},
65 {"operation": "deploy.start", "action": "deploy", "target": self.version, "env": {"PATH": "/tmp"}},
66 {"operation": "deploy.release", "release": "../private"},
67 {"operation": "deploy.run", "id": "../../nomad.token"},
68 {"operation": "deploy.history", "path": "/etc/shadow"}]
69 for action, target in [("sh", self.stage), ([], self.stage), ("destroy", "../escape"),
70 ("destroy", "--root"), ("rollback", self.version), ("rollback", "0"), ("rollback", [])]:
71 requests.append({"operation": "deploy.start", "action": action, "target": target})
72 with patch.object(runs.subprocess, "run") as executed:
73 for request in requests:
74 with self.subTest(request=request), self.assertRaises((host.Rejected, runs.Error)):
75 self.host.handle(request)
76 executed.assert_not_called()
77
78 def test_commands_come_from_verified_releases_and_history(self):
79 self.assertEqual(runs.command("destroy", self.stage)[1:], [str(self.snapshot / "tools/studio.py"), "destroy", self.stage])
80 self.assertEqual(runs.command("deploy", self.version)[1:], [str(Path(runs.__file__).with_name("release.py")), "deploy", self.version])
81 self.assertEqual(runs.command("rollback", "1")[1:], [str(Path(runs.__file__).with_name("release.py")), "rollback", self.version])
82 self.assertEqual(runs.command("start", "fixture")[1:], [str(self.snapshot / "tools/studio.py"), "deploy", "fixture"])
83 self.assertEqual(runs.command("stop", "fixture"), ["nomad", "job", "stop", "-yes", "fixture"])
84 self.assertEqual(runs.command("restart", "fixture"), ["nomad", "job", "restart", "-yes", "fixture"])
85 (self.snapshot / "tools/studio.py").write_text("changed")
86 for action, target in [("destroy", self.stage), ("deploy", self.version), ("rollback", "1"), ("start", "fixture")]:
87 with self.subTest(action=action), self.assertRaisesRegex(ValueError, "contents changed"):
88 runs.command(action, target)
89
90 def test_destroy_verifies_releases_from_before_guest_sources_were_added(self):
91 with patch.object(release, "SOURCES", tuple(name for name in release.SOURCES if name != "guest")):
92 digest = release.tree_digest(self.snapshot)
93 shutil.rmtree(self.snapshot / "guest")
94 manifest = self.snapshot / ".studio-release.json"
95 details = json.loads(manifest.read_text())
96 details["digest"] = digest
97 manifest.write_text(json.dumps(details))
98 self.assertEqual(runs.command("destroy", self.stage)[1:], [str(self.snapshot / "tools/studio.py"), "destroy", self.stage])
99 shutil.rmtree(self.snapshot / "tools")
100 with self.assertRaisesRegex(ValueError, "contents changed"):
101 runs.command("destroy", self.stage)
102
103 def test_service_control_is_limited_to_managed_names(self):
104 self.assertEqual(self.call("deploy.managed"), ["fixture"])
105 for action in ["start", "stop", "restart"]:
106 for target, status in [("unmanaged", 404), ("../escape", 400), ("--purge", 400), ([], 400)]:
107 with self.subTest(action=action, target=target), self.assertRaises(runs.Error) as error:
108 runs.command(action, target)
109 self.assertEqual(error.exception.status, status)
110 (self.state / "managed-jobs.json").write_text('["--purge"]')
111 with self.assertRaisesRegex(ValueError, "invalid managed"):
112 self.call("deploy.managed")
113 for data in [b'{"argv":["sh"]}', b'[1]', b'x' * (runs.MAX_METADATA + 1)]:
114 (self.state / "managed-jobs.json").write_bytes(data)
115 with self.subTest(data=data[:30]), self.assertRaises(ValueError):
116 self.call("deploy.managed")
117 (self.state / "managed-jobs.json").unlink()
118 self.assertEqual(self.call("deploy.managed"), [])
119
120 def test_main_deployment_ignores_stage_state_and_rejects_stale_candidates(self):
121 self.stage_file.write_text(json.dumps({"sourceId": "fixture", "ready": False, "overrides": ["PASSWORD=private"]}))
122 self.assertEqual(runs.command("deploy", self.version)[-2:], ["deploy", self.version])
123 for target in [self.stage, "c" * 16, "../escape", []]:
124 with self.subTest(target=target), self.assertRaises(runs.Error) as error:
125 runs.command("deploy", target)
126 self.assertEqual(error.exception.status, 409)
127 with self.assertRaises(runs.Error):
128 runs.command("promote", self.stage)
129 (self.root / "main").unlink()
130 self.assertIsNone(self.call("deploy.main"))
131 with self.assertRaises(runs.Error):
132 runs.command("deploy", self.version)
133
134 def test_main_metadata_and_deployment_are_checked_again_under_lock(self):
135 import sys
136 with patch.object(sys, "argv", ["release.py", "deploy", "c" * 16]), patch.object(release, "activate") as activate:
137 with self.assertRaisesRegex(ValueError, "main changed"):
138 release.main()
139 activate.assert_not_called()
140 manifest = self.snapshot / ".studio-release.json"
141 original = json.loads(manifest.read_text())
142 for revision in [None, {"commit": "b" * 40, "description": " "}, {"commit": "escape", "description": "main"}]:
143 manifest.write_text(json.dumps({**original, "main": revision}))
144 with self.subTest(revision=revision), self.assertRaises(ValueError):
145 self.call("deploy.main")
146
147 def test_metadata_and_legacy_logs_are_read_without_running_commands(self):
148 directory = self.snapshot / "service/fixture"
149 directory.mkdir()
150 (directory / "service.pkl").write_text("fixture definition")
151 identity = str(uuid.uuid4())
152 directory = self.state / "runs"
153 directory.mkdir()
154 (directory / (identity + ".log")).write_bytes(b"legacy\n\xff\n")
155 (directory / (identity + ".exit")).write_text("0")
156 with patch.object(runs.subprocess, "run") as executed:
157 self.assertEqual(self.call("deploy.current"), self.version)
158 self.assertEqual(len(self.call("deploy.history")), 1)
159 self.assertEqual(self.call("deploy.release", release=self.version), {"fixture": "fixture definition"})
160 self.assertEqual(self.call("deploy.run", id=identity), {"lines": ["legacy", "\ufffd"], "code": 0})
161 executed.assert_not_called()
162
163 def test_first_activation_exposes_current_before_starting_host_units(self):
164 (self.root / "current").unlink()
165
166 def execute(argv, **kwargs):
167 if argv[:2] == ["nixos-rebuild", "switch"]:
168 self.assertEqual((self.root / "current").resolve(), self.snapshot)
169 output = 'job "fixture" {\n' if argv[-1] == "render" else '"leader"'
170 return subprocess.CompletedProcess(argv, 0, stdout=output, stderr="")
171
172 with patch.object(release.subprocess, "run", side_effect=execute):
173 self.assertIsNone(release.activate(self.version, initial=True))
174 self.assertEqual(release.current_release(), self.version)
175
176 def test_state_reads_reject_symlinks_and_large_files(self):
177 source = self.root / "fixture"
178 source.write_bytes(b"x" * 200)
179 link = self.root / "link"
180 link.symlink_to(source)
181 with self.assertRaises(OSError):
182 runs.read(link)
183 with self.assertRaisesRegex(ValueError, "too large"):
184 runs.read(source, 100)
185
186 def test_output_selection_ignores_unrelated_and_old_large_logs(self):
187 directory = self.state / "runs"
188 directory.mkdir()
189 (directory / "unrelated.log").write_bytes(b"x" * (runs.MAX_LOG + 1))
190 old = directory / (str(uuid.uuid4()) + ".log")
191 old.write_bytes(b"x" * (runs.MAX_LOG + 1))
192 selected = directory / "new-prod-fixture.log"
193 selected.write_text("approved deployment\n")
194 os.utime(selected, (123, 123))
195 self.assertEqual(self.call("deploy.output", kind="history", target="1"), ["approved deployment"])
196 self.assertIsNone(self.call("deploy.output", kind="stages", target=self.stage))
197
198 def test_run_state_survives_restart_and_never_forwards_request_environment(self):
199 with patch.object(runs.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, stdout="", stderr="")) as executed:
200 started = self.call("deploy.start", action="destroy", target=self.stage)
201 argv = executed.call_args.args[0]
202 self.assertIn("--property=RuntimeMaxSec=3600", argv)
203 self.assertEqual(argv[-3:], [started["id"], "destroy", self.stage])
204 self.assertNotIn("--setenv=NOMAD_TOKEN", " ".join(argv))
205 root = runs.HOST_STATE
206 (root / "runs").mkdir(exist_ok=True)
207 (root / "runs" / (started["id"] + ".log")).write_text("fixture\n")
208 restored = host.Host("studio-demo")
209 active = subprocess.CompletedProcess([], 0, stdout="ActiveState=active\nExecMainStatus=0\nLoadState=loaded\n")
210 with patch.object(runs.subprocess, "run", return_value=active):
211 self.assertEqual(restored.handle({"operation": "deploy.last"}), {**started, "code": None})
212 with self.assertRaises(runs.Error) as error:
213 restored.handle({"operation": "deploy.start", "action": "deploy", "target": self.version})
214 self.assertEqual(error.exception.status, 409)
215 (root / "runs" / (started["id"] + ".exit")).write_text("0")
216 self.assertEqual(restored.handle({"operation": "deploy.last"}), {**started, "code": 0})
217 (root / "runs" / (started["id"] + ".exit")).unlink()
218 missing = subprocess.CompletedProcess([], 1, stdout="LoadState=not-found\nActiveState=inactive\nExecMainStatus=0\n")
219 with patch.object(runs.subprocess, "run", return_value=missing):
220 self.assertEqual(restored.handle({"operation": "deploy.run", "id": started["id"]})["code"], 1)
221
222 def test_worker_output_and_child_cleanup_are_bounded(self):
223 identity = str(uuid.uuid4())
224 root = runs.HOST_STATE
225 root.mkdir()
226 (root / "last-run.json").write_text(json.dumps({"id": identity, "action": "destroy", "target": self.stage}))
227 import sys
228 argv = [sys.executable, "-c", "import os,time; os.write(1,b'x'*200000); time.sleep(10)"]
229 with patch.object(runs, "command", return_value=argv), patch.object(runs, "MAX_LOG", 4096), patch.object(runs.signal, "signal"):
230 self.assertEqual(runs.worker(identity, "destroy", self.stage), 1)
231 self.assertEqual((root / "runs" / (identity + ".exit")).read_text(), "1")
232 log = (root / "runs" / (identity + ".log")).read_text()
233 self.assertLess(len(log), 8192)
234 self.assertIn("output exceeded its size limit", log)
235
236 def test_management_token_is_loaded_only_in_the_host_worker(self):
237 identity = str(uuid.uuid4())
238 root = runs.HOST_STATE
239 root.mkdir()
240 (root / "last-run.json").write_text(json.dumps({"id": identity, "action": "stop", "target": "fixture"}))
241 (self.state / "nomad.token").write_text("private-fixture-token\n")
242 import sys
243 argv = [sys.executable, "-c", "import os; assert os.environ['NOMAD_TOKEN']==open(" + repr(str(self.state / "nomad.token")) + ").read().strip(); print('stopped')"]
244 with patch.object(runs, "command", return_value=argv), patch.object(runs.signal, "signal"):
245 self.assertEqual(runs.worker(identity, "stop", "fixture"), 0)
246 log = (root / "runs" / (identity + ".log")).read_text()
247 self.assertIn("stopped", log)
248 self.assertNotIn("private-fixture-token", log)
249
250 def test_secret_scope_and_compare_and_set_preserve_siblings(self):
251 (self.state / "nomad.token").write_text("private-fixture-token")
252 job = {"Meta": {"studio_secrets": json.dumps([{"name": "token", "generated": True, "bytes": 16}])}}
253 existing = {"ModifyIndex": 7, "Items": {"token": "old", "sibling": "keep", "undeclared": "private"}}
254 with patch.object(runs, "nomad", side_effect=[job, existing]) as requested:
255 self.assertEqual(runs.secret("get", "fixture", "token"), "old")
256 with patch.object(runs, "nomad", return_value=job) as requested, self.assertRaises(runs.Error):
257 runs.secret("get", "fixture", "undeclared")
258 requested.assert_called_once_with("job/fixture")
259 with patch.object(runs, "nomad", side_effect=[job, existing, {}]) as requested, patch.object(runs.subprocess, "run", return_value=subprocess.CompletedProcess([], 0)):
260 runs.secret("set", "fixture", "token", "new-secret")
261 self.assertEqual(requested.call_args.args, ("var/nomad/jobs/fixture?cas=7", "PUT", {"Namespace": "default", "Path": "nomad/jobs/fixture", "Items": {**existing["Items"], "token": "new-secret"}}))
262 with patch.object(runs, "nomad", side_effect=[job, existing, runs.Error(409, "conflict")]), patch.object(runs.subprocess, "run") as restarted, self.assertRaises(runs.Error):
263 runs.secret("set", "fixture", "token", "new-secret")
264 restarted.assert_not_called()
265
266 def test_rotations_use_the_recorded_size_and_refuse_external_secrets(self):
267 (self.state / "nomad.token").write_text("private-fixture-token")
268 for generated, count in [(False, 16), (True, True), (True, 4097), (True, 0)]:
269 job = {"Meta": {"studio_secrets": json.dumps([{"name": "token", "generated": generated, "bytes": count}])}}
270 with patch.object(runs, "nomad", side_effect=[job, None]) as requested, self.assertRaises(runs.Error):
271 runs.secret("rotate", "fixture", "token")
272 self.assertEqual(requested.call_count, 2)
273 job = {"Meta": {"studio_secrets": '[{"name":"token","generated":true,"bytes":16}]'}}
274 with patch.object(runs, "nomad", side_effect=[job, None, {}]) as requested, patch.object(runs.subprocess, "run", return_value=subprocess.CompletedProcess([], 0)):
275 runs.secret("rotate", "fixture", "token")
276 value = requested.call_args.args[2]["Items"]["token"]
277 self.assertEqual(len(value), 32)
278 self.assertRegex(value, "^[a-f0-9]+$")
279
280 def test_secret_input_is_private_never_forwarded_and_removed_after_consumption(self):
281 import stat
282 import sys
283 with patch.object(runs.subprocess, "run", return_value=subprocess.CompletedProcess([], 0)) as dispatched:
284 saved = self.call("deploy.secret.set", service="fixture", key="token", value="private-fixture-value")
285 self.assertNotIn("private-fixture-value", " ".join(dispatched.call_args.args[0]))
286 self.assertNotIn("value", saved)
287 source = runs.HOST_STATE / "runs" / (saved["id"] + ".input")
288 self.assertEqual(stat.S_IMODE(source.stat().st_mode), 0o600)
289 argv = [sys.executable, "-c", "import sys; assert len(sys.stdin.read())==21; print('consumed')"]
290 with patch.object(runs, "command", return_value=argv), patch.object(runs.signal, "signal"):
291 self.assertEqual(runs.worker(saved["id"], "secret-set", "fixture", "token"), 0)
292 self.assertFalse(source.exists())
293 self.assertNotIn("private-fixture-value", (source.with_suffix(".log")).read_text())
294 with patch.object(runs.subprocess, "run", side_effect=OSError("fixture dispatch failed")), self.assertRaises(OSError):
295 self.call("deploy.secret.set", service="fixture", key="token", value="private-fixture-value")
296 self.assertFalse(list(source.parent.glob("*.input")))
297
298
299if __name__ == "__main__":
300 unittest.main()