diff --git a/backend/deploy/VM_CONTACT.md b/backend/deploy/VM_CONTACT.md index ffac303c..b67d1bf4 100644 --- a/backend/deploy/VM_CONTACT.md +++ b/backend/deploy/VM_CONTACT.md @@ -405,7 +405,10 @@ require(all((target[k].startswith('/') for k in ('root', 'admin_socket', 'token_ root = pathlib.Path(target["root"]).resolve() install = pathlib.Path("/opt/vm-mcp").resolve() require(root != pathlib.Path('/') and root != install and (root not in install.parents) and (install not in root.parents), 'workspace overlaps install') -require(root.is_dir() and any((p.name != '.vm-mcp-home' for p in root.iterdir())), 'workspace must exist and be nonempty') +# A preserved root (the inventoried one) may be empty; a newly chosen root must hold data. +preserved = root == pathlib.Path(state['home']).parent.resolve() +require(root.is_dir(), 'workspace must exist and be a directory') +require(preserved or any((p.name != '.vm-mcp-home' for p in root.iterdir())), 'a changed workspace root must exist and be nonempty') require(not re.search('[\\s%\\\\"\']', target['root']), 'hmmm: pinned installer cannot represent this workspace drop-in safely') # Verify the completed archive without touching the live deployment. h = hashlib.sha256() @@ -463,6 +466,11 @@ PY ) ``` +The workspace guard accepts the preserved (inventoried) root whether or not it holds +data; it must still exist and be a directory. Only a root changed in `target.json` +must also be nonempty, so an edit cannot silently point the service at an empty one. +A root counts as preserved when it resolves to the parent of the U1 state's service-user home. + Expect `rc=0`. The Python result is initialized by the actual invocation, so an interactive shell's old `rc` cannot turn success into failure. On any failure after stopping, keep using the human recovery path and use U6 or fix forward. Do not infer @@ -521,7 +529,10 @@ def check(label, passed): for key, field in (("VM_MCP_ROOT", "root"), ("VM_MCP_PROFILE", "profile"), ("VM_MCP_PORT", "port"), ("VM_MCP_ADMIN_SOCKET", "admin_socket")): check(key + " preserved", env.get(key) == target[field]) check("target service home", pwd.getpwnam("vmmcp").pw_dir == target["root"].rstrip("/") + "/.vm-mcp-home") -check("workspace not empty", any(p.name != ".vm-mcp-home" for p in pathlib.Path(target["root"]).iterdir())) +state = json.loads((B / "state.json").read_text()) +preserved = pathlib.Path(target["root"]).resolve() == pathlib.Path(state["home"]).parent.resolve() +check("workspace exists" + (" (preserved root, may be empty)" if preserved else " and is not empty"), + pathlib.Path(target["root"]).is_dir() and (preserved or any(p.name != ".vm-mcp-home" for p in pathlib.Path(target["root"]).iterdir()))) unit_result = subprocess.run(["systemctl", "show", "vm-mcp.service", "--property=ReadWritePaths,ProtectSystem"], check=True, text=True, capture_output=True) unit = dict(line.split("=", 1) for line in unit_result.stdout.splitlines() if "=" in line) diff --git a/backend/tests/test_vm_contact_upgrade.py b/backend/tests/test_vm_contact_upgrade.py index b1dbe1a0..3672d783 100644 --- a/backend/tests/test_vm_contact_upgrade.py +++ b/backend/tests/test_vm_contact_upgrade.py @@ -198,6 +198,11 @@ def __init__(self, *parts): if type(Path()).__init__ is not object.__init__: super().__init__(vm.path(Path(*parts))) + def with_segments(self, *parts): + # 3.12+ derives child/glob paths through this hook; they are already + # sandbox-resolved, so return plain paths (matching 3.11 behaviour). + return type(Path())(*parts) + def open(self, *args, **kwargs): return vm.path(self).open(*args, **kwargs) @@ -312,6 +317,74 @@ def test_failed_new_snapshot_keeps_existing_latest(self): self.assertEqual(self.vm.backup(), old) self.assertEqual((old / "COMPLETE").read_text(), "complete\n") + def set_target_root(self, root): + target = self.vm.target() + target["root"] = root + (self.vm.backup() / "target.json").write_text(json.dumps(target)) + return target + + def installed(self): + return any(args[0] == "bash" for args, _ in self.vm.commands) + + def test_u4_accepts_preserved_empty_root(self): + self.snapshot_target() + self.vm.path("/srv/existing/data.txt").unlink() + self.vm.commands.clear() + self.vm.execute("U4") + self.assertTrue(self.installed()) + self.assertEqual(self.vm.installer_env["VM_MCP_ROOT"], "/srv/existing") + + def test_u4_refuses_preserved_missing_root(self): + self.snapshot_target() + shutil.rmtree(self.vm.path("/srv/existing")) + self.vm.commands.clear() + with self.assertRaisesRegex(AssertionError, "must exist and be a directory"): + self.vm.execute("U4") + self.assertEqual(self.destructive(), []) + + def test_u4_refuses_preserved_root_that_is_a_file(self): + self.snapshot_target() + shutil.rmtree(self.vm.path("/srv/existing")) + self.vm.write("/srv/existing", "not a directory") + self.vm.commands.clear() + with self.assertRaisesRegex(AssertionError, "must exist and be a directory"): + self.vm.execute("U4") + self.assertEqual(self.destructive(), []) + + def test_u4_refuses_changed_empty_root(self): + self.snapshot_target() + self.vm.path("/srv/other").mkdir(parents=True) + self.set_target_root("/srv/other") + self.vm.commands.clear() + with self.assertRaisesRegex(AssertionError, "changed workspace root must exist and be nonempty"): + self.vm.execute("U4") + self.assertEqual(self.destructive(), []) + + def test_u4_changed_nonempty_root_passes_the_workspace_guard(self): + self.snapshot_target() + self.vm.write("/srv/other/data.txt", "fixture") + self.set_target_root("/srv/other") + self.vm.commands.clear() + self.vm.execute("U4") + self.assertEqual(self.vm.installer_env["VM_MCP_ROOT"], "/srv/other") + + def test_u5_fails_on_changed_empty_root(self): + self.snapshot_target() + self.vm.path("/srv/other").mkdir(parents=True) + target = self.set_target_root("/srv/other") + self.vm.write("/etc/vm-mcp.env", "".join(key + "=" + shlex.quote(target[field]) + "\n" for key, field in + (("VM_MCP_ROOT", "root"), ("VM_MCP_PROFILE", "profile"), + ("VM_MCP_PORT", "port"), ("VM_MCP_ADMIN_SOCKET", "admin_socket")))) + self.vm.home = "/srv/other/.vm-mcp-home" + self.vm.write_paths = "/srv/other" + with self.assertRaises(SystemExit) as failure: + self.vm.execute("U5") + self.assertEqual(failure.exception.code, 1) + self.vm.write("/srv/other/data.txt", "fixture") + with self.assertRaises(SystemExit) as ok: + self.vm.execute("U5") + self.assertEqual(ok.exception.code, 0) + def test_shell_metacharacters_and_spaces_roundtrip_without_evaluation(self): root = '/srv/a workspace;$(touch /tmp/never-created)`id`' self.vm.home = root + "/.vm-mcp-home" @@ -416,7 +489,12 @@ def environment(): self.vm.execute("U5") self.assertEqual(failure.exception.code, 1) self.vm.home = target["root"] + "/.vm-mcp-home" + # The preserved root may be empty, but it must still exist. self.vm.path("/srv/existing/data.txt").unlink() + with self.assertRaises(SystemExit) as ok: + self.vm.execute("U5") + self.assertEqual(ok.exception.code, 0) + shutil.rmtree(self.vm.path("/srv/existing")) with self.assertRaises(SystemExit) as failure: self.vm.execute("U5") self.assertEqual(failure.exception.code, 1)