diff --git a/README.md b/README.md index 103fdd99..1cb896a0 100644 --- a/README.md +++ b/README.md @@ -523,6 +523,7 @@ A rollback leaves the clone on a detached HEAD at the restored commit; the updat - "VM Clone Auto Migrate" is removed; a clone stays on its source host. - A template VM is changed only by superusers and staff owners with change permission; other owners can still view, open the console and clone it. - The log rotates (10 MB × 5) in `data/webvirtcloud.log`. +- gstfsd finds the guest's operating system the way libguestfs inspects it, so a root on LVM works, and it always answers (an error when it finds no or several operating systems). Update `/usr/local/bin/gstfsd` on every compute (see [Adding a Compute Node](#adding-a-compute-node), step 3). - Docker no longer publishes port 6080; the console goes through nginx on the panel's port (`WS_PUBLIC_PORT`, default 80). ## Usage diff --git a/conf/daemon/gstfsd b/conf/daemon/gstfsd index bf1a3754..cb763da9 100644 --- a/conf/daemon/gstfsd +++ b/conf/daemon/gstfsd @@ -1,18 +1,38 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 # # gstfsd - WebVirtCloud daemon for managing VM's filesystem # +# The server reads one JSON request per connection and sends exactly one JSON +# reply. Each change runs in a worker process ("gstfsd --worker") that is +# killed, with its guestfs appliance, when the request is not done DEADLINE +# seconds after the connection was accepted. +# import ipaddress -import os -import socketserver import json +import os import re +import select +import signal import socket +import socketserver +import subprocess import sys +import threading +import time PORT = int(os.environ.get("GSTFSD_PORT", 16510)) ADDRESS = os.environ.get("GSTFSD_BIND_HOST", "127.0.0.1") +MAX_REQUEST = 64 * 1024 +# A request, from accepting the connection to closing the appliance, must +# finish within this many seconds, or it is stopped: the panel gives up 120 s +# after connecting and releases the VM, which may then be started. +DEADLINE = float(os.environ.get("GSTFSD_DEADLINE", 100)) +MAX_DEADLINE = 100 +# Time to send the reply after the deadline. +REPLY_TIMEOUT = 5 +# Time for a killed worker and its appliance to be gone. +STOP_TIMEOUT = 5 def is_loopback(host): @@ -22,71 +42,246 @@ def is_loopback(host): return host == "localhost" -# gstfsd is an unauthenticated root service that rewrites any VM's -# /etc/shadow and authorized_keys, so it only listens on loopback unless -# explicitly allowed. -if not is_loopback(ADDRESS) and os.environ.get("GSTFSD_ALLOW_REMOTE") != "1": - sys.stderr.write( - "gstfsd: refusing to listen on %s (unauthenticated root service). " - "Set GSTFSD_ALLOW_REMOTE=1 to allow it on a trusted network.\n" % ADDRESS +# --- worker: one change to one guest ---------------------------------------- + + +def set_root_password(gfs, passwd): + shadow = gfs.cat('/etc/shadow') + new_shadow, found = re.subn('^root:[^:]*:', lambda m: "root:" + passwd + ":", shadow, flags=re.MULTILINE) + if found != 1: + raise RuntimeError("/etc/shadow has %d root entries, need exactly one" % found) + gfs.write('/etc/shadow', new_shadow) + gfs.chmod(0o640, '/etc/shadow') + + +def add_root_key(gfs, key): + if not gfs.is_dir('/root/.ssh'): + gfs.mkdir('/root/.ssh') + gfs.chmod(0o700, "/root/.ssh") + # Add the key; keep the ones already there. + keys_path = '/root/.ssh/authorized_keys' + keys = gfs.cat(keys_path) if gfs.is_file(keys_path) else '' + new_key = key.strip() + if new_key not in keys.splitlines(): + if keys and not keys.endswith('\n'): + keys += '\n' + gfs.write(keys_path, keys + new_key + '\n') + gfs.chmod(0o600, keys_path) + + +# action: (function, the files it changes) +ACTIONS = { + 'password': (lambda gfs, data: set_root_password(gfs, data['passwd']), ['/etc/shadow']), + 'publickey': (lambda gfs, data: add_root_key(gfs, data['key']), ['/root/.ssh/authorized_keys']), +} + + +def holds(mountpoint, path): + return mountpoint == '/' or path == mountpoint or path.startswith(mountpoint.rstrip('/') + '/') + + +def mount_guest(gfs, paths): + """Mount the guest's one operating system the way it mounts itself (LVM + included), parents first. A filesystem on the way to one of paths must + mount; another one that fails (a broken /data) is left out.""" + roots = gfs.inspect_os() + if len(roots) != 1: + raise RuntimeError("found %d operating systems on the disks, need exactly one" % len(roots)) + mountpoints = gfs.inspect_get_mountpoints(roots[0]) + for mountpoint in sorted(mountpoints, key=len): + try: + gfs.mount(mountpoints[mountpoint], mountpoint) + except RuntimeError: + if any(holds(mountpoint, path) for path in paths): + raise + + +def change_guest(data): + import guestfs + + action, paths = ACTIONS[data['action']] + gfs = guestfs.GuestFS(python_return_dict=True) + launched = False + try: + gfs.add_domain(data['vname']) + gfs.launch() + launched = True + mount_guest(gfs, paths) + if '/etc/shadow' in paths and not gfs.is_file('/etc/shadow'): + raise RuntimeError("the guest has no /etc/shadow") + action(gfs, data) + except Exception: + # the first error is the one to report; cleanup still runs + if launched: + try: + gfs.umount_all() + gfs.shutdown() + except Exception: + pass + gfs.close() + raise + try: + # writes reach the disk at umount and shutdown: their errors count + gfs.umount_all() + gfs.shutdown() + finally: + gfs.close() + + +def worker(): + """Read one request on stdin, change the guest, write the reply on stdout.""" + try: + change_guest(json.loads(sys.stdin.buffer.read())) + reply = {'return': 'success'} + except Exception as err: + reply = {'return': 'error', 'message': str(err)} + sys.stdout.write(json.dumps(reply)) + return 0 + + +# --- server ----------------------------------------------------------------- + + +def read_request(sock, end): + """The client sends one JSON object and waits for the reply: read until + it parses (a long SSH key spans several reads), before end (a + time.monotonic() value) and within MAX_REQUEST bytes.""" + data = b"" + while True: + remaining = end - time.monotonic() + if remaining <= 0: + raise ValueError("the request was not complete within %d seconds" % DEADLINE) + sock.settimeout(remaining) + try: + chunk = sock.recv(min(4096, MAX_REQUEST + 1 - len(data))) + except socket.timeout: + raise ValueError("the request was not complete within %d seconds" % DEADLINE) + if not chunk: + return json.loads(data) + data += chunk + if len(data) > MAX_REQUEST: + raise ValueError("the request is longer than %d bytes" % MAX_REQUEST) + try: + return json.loads(data) + except ValueError: + continue + + +def group_running(pgid): + """Whether a process of the group still runs. A killed appliance whose + parent is gone may stay a zombie until init reaps it; it holds no disk.""" + for pid in os.listdir('/proc'): + try: + with open('/proc/%s/stat' % pid) as f: + fields = f.read().rsplit(')', 1)[1].split() + except (OSError, IndexError): + continue # not a process, or gone meanwhile + if fields[0] != 'Z' and int(fields[2]) == pgid: + return True + return False + + +def stop_worker(proc): + """Kill the worker with its appliance and wait until they are gone.""" + try: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass + # a process in uninterruptible disk I/O dies only when the I/O ends + end = time.monotonic() + STOP_TIMEOUT + try: + proc.wait(timeout=STOP_TIMEOUT) + except subprocess.TimeoutExpired: + pass + while time.monotonic() < end: + if proc.returncode is not None and not group_running(proc.pid): + return + time.sleep(0.05) + raise RuntimeError("the change was stopped, but its appliance is still running") + + +def watch_client(sock, proc, done): + """Kill the worker when the client leaves: the panel then no longer keeps + the VM from being started. The client sends nothing more until the reply.""" + while not done.is_set(): + try: + if select.select([sock], [], [], 0.2)[0] and not sock.recv(4096): + break + except OSError: + break + else: + return + try: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass + + +def run_worker(data, end, sock): + if data.get('action') not in ACTIONS: + raise ValueError("unknown action %r" % data.get('action')) + # Its own session, and the appliance started directly by it (not by + # libvirtd): killing the worker's process group also kills the appliance. + proc = subprocess.Popen( + [sys.executable, os.path.abspath(__file__), '--worker'], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + start_new_session=True, + env=dict(os.environ, LIBGUESTFS_BACKEND='direct'), ) - sys.exit(1) - -import guestfs # noqa: E402 - - -class MyTCPServer(socketserver.ThreadingTCPServer): - allow_reuse_address = True - address_family = socket.AF_INET6 if ":" in ADDRESS else socket.AF_INET + done = threading.Event() + watcher = threading.Thread(target=watch_client, args=(sock, proc, done), daemon=True) + watcher.start() + try: + out, err = proc.communicate(json.dumps(data).encode(), timeout=max(0, end - time.monotonic())) + except subprocess.TimeoutExpired: + stop_worker(proc) + raise RuntimeError("the change took longer than %d seconds and was stopped" % DEADLINE) + finally: + done.set() + watcher.join() + if proc.returncode != 0: + stop_worker(proc) # a crashed or killed worker may leave its appliance + lines = err.decode(errors='replace').strip().splitlines() + raise RuntimeError("the change failed (exit %d): %s" % (proc.returncode, lines[-1] if lines else "no output")) + return json.loads(out) class MyTCPServerHandler(socketserver.StreamRequestHandler): - timeout = 30 # seconds; an idle client must not hold a thread forever - def handle(self): - # recive data - d = self.request.recv(1024).strip() - data = json.loads(d) - - # GuestFS - gfs = guestfs.GuestFS(python_return_dict=True) + # Exactly one reply per request, success or error. + end = time.monotonic() + DEADLINE try: - gfs.add_domain(data['vname']) - gfs.launch() - parts = gfs.list_partitions() - for part in parts: - try: - gfs.mount(part, '/') - if gfs.is_file('/etc/shadow'): - if data['action'] == 'password': - file_shadow = gfs.cat('/etc/shadow') - new_root_hash = "root:" + data['passwd'] + ":" - file_shadow_new = re.sub('^root:.*?:', new_root_hash, file_shadow) - gfs.write('/etc/shadow', file_shadow_new) - gfs.chmod(0o640, '/etc/shadow') - self.request.sendall(json.dumps({'return': 'success'}).encode()) - if data['action'] == 'publickey': - if not gfs.is_dir('/root/.ssh'): - gfs.mkdir('/root/.ssh') - gfs.chmod(0o700, "/root/.ssh") - # Add the key; keep the ones already there. - keys_path = '/root/.ssh/authorized_keys' - keys = gfs.cat(keys_path) if gfs.is_file(keys_path) else '' - new_key = data['key'].strip() - if new_key not in keys.splitlines(): - if keys and not keys.endswith('\n'): - keys += '\n' - gfs.write(keys_path, keys + new_key + '\n') - gfs.chmod(0o600, '/root/.ssh/authorized_keys') - self.request.sendall(json.dumps({'return': 'success'}).encode()) - gfs.umount(part) - except RuntimeError: - pass - gfs.shutdown() - gfs.close() + reply = run_worker(read_request(self.request, end), end, self.request) except Exception as err: - self.request.sendall(bytes(json.dumps({'return': 'error', 'message': str(err)}).encode())) + reply = {'return': 'error', 'message': str(err)} + try: + self.request.settimeout(REPLY_TIMEOUT) + self.request.sendall(json.dumps(reply).encode()) + except OSError: + pass # the client is gone -server = MyTCPServer((ADDRESS, PORT), MyTCPServerHandler) -server.serve_forever() +def serve(): + # gstfsd is an unauthenticated root service that rewrites any VM's + # /etc/shadow and authorized_keys, so it only listens on loopback unless + # explicitly allowed. + if not 0 < DEADLINE <= MAX_DEADLINE: + sys.stderr.write("gstfsd: GSTFSD_DEADLINE must be above 0 and at most %d seconds\n" % MAX_DEADLINE) + return 1 + if not is_loopback(ADDRESS) and os.environ.get("GSTFSD_ALLOW_REMOTE") != "1": + sys.stderr.write( + "gstfsd: refusing to listen on %s (unauthenticated root service). " + "Set GSTFSD_ALLOW_REMOTE=1 to allow it on a trusted network.\n" % ADDRESS + ) + return 1 + + class MyTCPServer(socketserver.ThreadingTCPServer): + allow_reuse_address = True + address_family = socket.AF_INET6 if ":" in ADDRESS else socket.AF_INET + + MyTCPServer((ADDRESS, PORT), MyTCPServerHandler).serve_forever() + return 0 + + +if __name__ == '__main__': + sys.exit(worker() if sys.argv[1:] == ['--worker'] else serve()) diff --git a/instances/test_gstfsd.py b/instances/test_gstfsd.py index 36bc34cc..6030d1e2 100644 --- a/instances/test_gstfsd.py +++ b/instances/test_gstfsd.py @@ -10,7 +10,8 @@ from pathlib import Path GSTFSD = Path(__file__).resolve().parent.parent / "conf" / "daemon" / "gstfsd" -# A stand-in for libguestfs: one partition whose files live in a JSON file. +# A stand-in for libguestfs: the guest's operating systems ("roots", each with +# its mountpoints), its files, and what is mounted live in a JSON file. FAKE_GUESTFS = ''' import json, os STATE = os.environ["FAKE_GUESTFS_STATE"] @@ -21,22 +22,61 @@ def _save(state): with open(STATE, "w") as f: json.dump(state, f) class GuestFS: - def __init__(self, **kwargs): pass - def add_domain(self, name): pass - def launch(self): pass - def list_partitions(self): return ["/dev/sda1"] - def mount(self, part, mountpoint): pass - def umount(self, part): pass - def shutdown(self): pass - def close(self): pass - def is_file(self, path): return path in _load()["files"] - def is_dir(self, path): return path in _load()["dirs"] - def cat(self, path): return _load()["files"][path] + def __init__(self, **kwargs): + if _load().get("crash"): + import sys + sys.stderr.write("guestfs: the appliance crashed\\n") + os._exit(3) + self.launched = False + def add_domain(self, name, **kwargs): + if _load().get("fail_add"): + raise RuntimeError("no domain named " + name) + def launch(self): + import time + if _load().get("appliance"): + # a stand-in for the appliance: a child process, like qemu + import subprocess + child = subprocess.Popen(["sleep", "60"]) + state = _load(); state["appliance_pid"] = child.pid; _save(state) + time.sleep(_load().get("launch_seconds", 0)) + state = _load(); state["backend"] = os.environ.get("LIBGUESTFS_BACKEND"); _save(state) + self.launched = True + def inspect_os(self): return list(_load()["roots"]) + def inspect_get_mountpoints(self, root): return _load()["roots"][root] + def mount(self, device, mountpoint): + state = _load() + if device in state.get("broken", []): + raise RuntimeError("mount: wrong fs type on " + device) + # a mountpoint needs its parent filesystem mounted first + if mountpoint != "/" and "/" not in state["mounted"]: + raise RuntimeError("mount: " + mountpoint + " does not exist") + state["mounted"].append(mountpoint); state["history"].append([device, mountpoint]); _save(state) + def umount_all(self): + if not self.launched: + raise RuntimeError("umount_all: call launch before using this function") + state = _load(); state["mounted"] = []; _save(state) + def shutdown(self): + if not self.launched: + raise RuntimeError("shutdown: call launch before using this function") + def close(self): + state = _load(); state["closed"] = True; _save(state) + def _guest(self): + if "/" not in _load()["mounted"]: + raise RuntimeError("the guest root is not mounted") + def is_file(self, path): self._guest(); return path in _load()["files"] + def is_dir(self, path): self._guest(); return path in _load()["dirs"] + def cat(self, path): self._guest(); return _load()["files"][path] def write(self, path, content): - state = _load(); state["files"][path] = content; _save(state) + self._guest() + import time + time.sleep(_load().get("write_seconds", 0)) + state = _load() + if state.get("fail_write"): + raise RuntimeError("write: read-only file system") + state["files"][path] = content; state["writes"] += 1; _save(state) def mkdir(self, path): - state = _load(); state["dirs"].append(path); _save(state) - def chmod(self, mode, path): pass + self._guest(); state = _load(); state["dirs"].append(path); _save(state) + def chmod(self, mode, path): self._guest() ''' @@ -55,8 +95,13 @@ class GstfsdTestCase(unittest.TestCase): Path(self.tmp.name, "guestfs.py").write_text(FAKE_GUESTFS) self.state = Path(self.tmp.name, "state.json") self.state.write_text(json.dumps({ + # one Linux guest with its root on LVM and a separate /boot + "roots": {"/dev/vg/root": {"/": "/dev/vg/root", "/boot": "/dev/sda1"}}, "files": {"/etc/shadow": "root:x:1::::::\n", "/root/.ssh/authorized_keys": "ssh-rsa OLD old@host\n"}, "dirs": ["/root/.ssh"], + "mounted": [], + "history": [], + "writes": 0, })) def _env(self, **extra): @@ -67,10 +112,10 @@ class GstfsdTestCase(unittest.TestCase): **extra, ) - def _start(self, host, port, family=socket.AF_INET): + def _start(self, host, port, family=socket.AF_INET, **env): proc = subprocess.Popen( [sys.executable, str(GSTFSD)], - env=self._env(GSTFSD_BIND_HOST=host, GSTFSD_PORT=str(port)), + env=self._env(GSTFSD_BIND_HOST=host, GSTFSD_PORT=str(port), **env), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, @@ -93,7 +138,8 @@ class GstfsdTestCase(unittest.TestCase): s.settimeout(10) s.connect((host, port)) s.sendall(json.dumps(data).encode()) - return json.loads(s.recv(1024)) + # everything until gstfsd closes: more than one reply fails to parse + return json.loads(b"".join(iter(lambda: s.recv(4096), b""))) def test_refuses_non_loopback_bind_without_opt_in(self): result = subprocess.run( @@ -117,6 +163,16 @@ class GstfsdTestCase(unittest.TestCase): self.assertEqual(keys.splitlines(), ["ssh-rsa OLD old@host", "ssh-ed25519 NEW new@host"]) @unittest.skipUnless(socket.has_ipv6, "no IPv6") + def test_refuses_a_deadline_the_panel_does_not_wait_for(self): + # the panel gives up after 120 s and releases the VM + proc = subprocess.run( + [sys.executable, str(GSTFSD)], + env=self._env(GSTFSD_PORT=str(free_port()), GSTFSD_DEADLINE="180"), + capture_output=True, text=True, timeout=10, + ) + self.assertNotEqual(proc.returncode, 0) + self.assertIn("GSTFSD_DEADLINE", proc.stderr) + def test_listens_on_ipv6_loopback(self): try: port = free_port(socket.AF_INET6, "::1") @@ -128,3 +184,262 @@ class GstfsdTestCase(unittest.TestCase): socket.AF_INET6, ) self.assertEqual(reply["return"], "success") + + def _update_state(self, **values): + state = json.loads(self.state.read_text()) + state.update(values) + self.state.write_text(json.dumps(state)) + + def _serve(self): + port = free_port() + self._start("127.0.0.1", port) + return port + + def test_root_password_is_set_on_the_guest_root(self): + reply = self._request("127.0.0.1", self._serve(), {"action": "password", "vname": "vm", "passwd": "$6$new"}) + self.assertEqual(reply, {"return": "success"}) + state = json.loads(self.state.read_text()) + self.assertEqual(state["files"]["/etc/shadow"], "root:$6$new:1::::::\n") + self.assertEqual(state["mounted"], []) + self.assertTrue(state["closed"]) + + def test_no_or_several_operating_systems_give_one_error(self): + for roots in ({}, {"/dev/sda1": {"/": "/dev/sda1"}, "/dev/sdb1": {"/": "/dev/sdb1"}}): + with self.subTest(roots=list(roots)): + self._update_state(roots=roots) + reply = self._request("127.0.0.1", self._serve(), {"action": "password", "vname": "vm", "passwd": "$6$x"}) + self.assertEqual(reply["return"], "error") + self.assertEqual(json.loads(self.state.read_text())["files"]["/etc/shadow"], "root:x:1::::::\n") + + def test_a_failed_change_gives_one_error_and_cleans_up(self): + self._update_state(fail_write=True) + reply = self._request("127.0.0.1", self._serve(), {"action": "password", "vname": "vm", "passwd": "$6$x"}) + self.assertEqual(reply["return"], "error") + self.assertIn("read-only", reply["message"]) + state = json.loads(self.state.read_text()) + self.assertEqual(state["mounted"], []) + self.assertTrue(state["closed"]) + + def test_a_long_key_arrives_whole(self): + # longer than one read of the daemon (4096 bytes) + key = "ssh-rsa " + "A" * 6000 + " long@host" + reply = self._request("127.0.0.1", self._serve(), {"action": "publickey", "vname": "vm", "key": key}) + self.assertEqual(reply, {"return": "success"}) + keys = json.loads(self.state.read_text())["files"]["/root/.ssh/authorized_keys"] + self.assertEqual(keys.splitlines()[-1], key) + + def test_the_root_is_mounted_before_the_filesystems_under_it(self): + self._request("127.0.0.1", self._serve(), {"action": "password", "vname": "vm", "passwd": "$6$x"}) + self.assertEqual( + json.loads(self.state.read_text())["history"], [["/dev/vg/root", "/"], ["/dev/sda1", "/boot"]] + ) + + def test_a_broken_unrelated_filesystem_does_not_block_the_change(self): + self._update_state(broken=["/dev/sda1"]) # /boot + reply = self._request("127.0.0.1", self._serve(), {"action": "password", "vname": "vm", "passwd": "$6$y"}) + self.assertEqual(reply, {"return": "success"}) + self.assertEqual(json.loads(self.state.read_text())["files"]["/etc/shadow"], "root:$6$y:1::::::\n") + + def test_a_broken_filesystem_on_the_path_blocks_the_change(self): + state = json.loads(self.state.read_text()) + state["roots"]["/dev/vg/root"]["/root"] = "/dev/vg/home" + state["broken"] = ["/dev/vg/home"] + self.state.write_text(json.dumps(state)) + reply = self._request("127.0.0.1", self._serve(), {"action": "publickey", "vname": "vm", "key": "ssh-ed25519 K k@host"}) + self.assertEqual(reply["return"], "error") + self.assertEqual(json.loads(self.state.read_text())["writes"], 0) + + def test_root_is_found_anywhere_in_shadow(self): + self._update_state(files={"/etc/shadow": "daemon:*:1::::::\nroot:x:1::::::\n"}) + reply = self._request("127.0.0.1", self._serve(), {"action": "password", "vname": "vm", "passwd": "$6$z"}) + self.assertEqual(reply, {"return": "success"}) + self.assertEqual( + json.loads(self.state.read_text())["files"]["/etc/shadow"], "daemon:*:1::::::\nroot:$6$z:1::::::\n" + ) + + def test_a_shadow_without_root_is_an_error(self): + self._update_state(files={"/etc/shadow": "daemon:*:1::::::\n"}) + reply = self._request("127.0.0.1", self._serve(), {"action": "password", "vname": "vm", "passwd": "$6$z"}) + self.assertEqual(reply["return"], "error") + self.assertEqual(json.loads(self.state.read_text())["writes"], 0) + + def test_an_early_failure_keeps_its_own_message(self): + self._update_state(fail_add=True) + reply = self._request("127.0.0.1", self._serve(), {"action": "password", "vname": "vm", "passwd": "$6$x"}) + self.assertEqual(reply["return"], "error") + self.assertIn("no domain named vm", reply["message"]) + + def test_a_request_in_two_pieces(self): + port = self._serve() + body = json.dumps({"action": "password", "vname": "vm", "passwd": "$6$parts"}).encode() + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.settimeout(10) + s.connect(("127.0.0.1", port)) + s.sendall(body[:10]) + time.sleep(0.3) + s.sendall(body[10:]) + reply = json.loads(b"".join(iter(lambda: s.recv(4096), b""))) + self.assertEqual(reply, {"return": "success"}) + + def test_a_request_over_the_limit_is_refused(self): + port = self._serve() + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.settimeout(10) + s.connect(("127.0.0.1", port)) + s.sendall(b'{"action": "publickey", "vname": "vm", "key": "' + b"A" * 70000 + b'"}') + reply = json.loads(b"".join(iter(lambda: s.recv(4096), b""))) + self.assertEqual(reply["return"], "error") + self.assertEqual(json.loads(self.state.read_text())["writes"], 0) + + def test_a_trickling_client_is_cut_off(self): + port = free_port() + self._start("127.0.0.1", port, GSTFSD_DEADLINE="2") + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.connect(("127.0.0.1", port)) + s.settimeout(0.25) + start = time.monotonic() + reply = b"" + # a space every quarter second: each read is quick, the request never completes + while not reply and time.monotonic() - start < 8: + try: + s.sendall(b" ") + reply = s.recv(4096) + except socket.timeout: + pass + except OSError: + break + self.assertTrue(reply, "gstfsd kept reading a request that never completes") + self.assertEqual(json.loads(reply)["return"], "error") + + def test_a_slow_write_is_stopped_at_the_deadline(self): + # the whole change is bounded, not only its start: a write still + # running at the deadline is killed with the appliance + port = free_port() + self._start("127.0.0.1", port, GSTFSD_DEADLINE="1") + self._update_state(write_seconds=3) + start = time.monotonic() + reply = self._request("127.0.0.1", port, {"action": "password", "vname": "vm", "passwd": "$6$late"}) + self.assertLess(time.monotonic() - start, 2.5) + self.assertEqual(reply["return"], "error") + time.sleep(3) # longer than the write would have taken + self.assertEqual(json.loads(self.state.read_text())["writes"], 0) + + def test_a_key_needs_no_shadow_file(self): + self._update_state(files={"/root/.ssh/authorized_keys": ""}) + reply = self._request("127.0.0.1", self._serve(), {"action": "publickey", "vname": "vm", "key": "ssh-ed25519 K k@host"}) + self.assertEqual(reply, {"return": "success"}) + + def test_a_client_that_leaves_does_not_break_the_service(self): + port = free_port() + proc = self._start("127.0.0.1", port, GSTFSD_DEADLINE="1") + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.connect(("127.0.0.1", port)) + s.sendall(b'{"action": ') + time.sleep(1.5) # gstfsd times the request out and answers a closed socket + reply = self._request("127.0.0.1", port, {"action": "password", "vname": "vm", "passwd": "$6$after"}) + self.assertEqual(reply, {"return": "success"}) + proc.kill() + self.assertNotIn("Traceback", proc.stderr.read()) + + def test_nothing_is_written_after_the_deadline(self): + # the panel gives up after its timeout; gstfsd must not write later + port = free_port() + self._start("127.0.0.1", port, GSTFSD_DEADLINE="1") + self._update_state(launch_seconds=2) + reply = self._request("127.0.0.1", port, {"action": "password", "vname": "vm", "passwd": "$6$late"}) + self.assertEqual(reply["return"], "error") + state = json.loads(self.state.read_text()) + self.assertEqual(state["writes"], 0) + self.assertEqual(state["mounted"], []) + + def test_the_appliance_runs_where_the_deadline_can_kill_it(self): + # a libvirt-started appliance is no child of the worker: killing the + # worker's process group would leave it writing to the disks + port = free_port() + self._start("127.0.0.1", port, LIBGUESTFS_BACKEND="libvirt") + reply = self._request("127.0.0.1", port, {"action": "password", "vname": "vm", "passwd": "$6$d"}) + self.assertEqual(reply, {"return": "success"}) + self.assertEqual(json.loads(self.state.read_text())["backend"], "direct") + + def test_reading_the_request_counts_toward_the_deadline(self): + # the panel waits a fixed time from connecting; a slow request leaves + # less time for the change, not more + port = free_port() + self._start("127.0.0.1", port, GSTFSD_DEADLINE="2") + self._update_state(launch_seconds=3) + request = json.dumps({"action": "password", "vname": "vm", "passwd": "$6$slow"}).encode() + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.settimeout(10) + s.connect(("127.0.0.1", port)) + start = time.monotonic() + s.sendall(request[:10]) + time.sleep(1.5) + s.sendall(request[10:]) + reply = json.loads(b"".join(iter(lambda: s.recv(4096), b""))) + self.assertEqual(reply["return"], "error") + self.assertLess(time.monotonic() - start, 2.8) + + def test_a_crashed_worker_tells_why(self): + self._update_state(crash=True) + reply = self._request("127.0.0.1", self._serve(), {"action": "password", "vname": "vm", "passwd": "$6$c"}) + self.assertEqual(reply["return"], "error") + self.assertIn("the appliance crashed", reply["message"]) + + def test_a_client_that_leaves_stops_the_change(self): + # the panel releases the VM when its connection breaks: nothing may + # change the disks after that + port = free_port() + self._start("127.0.0.1", port) + self._update_state(launch_seconds=2) + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.connect(("127.0.0.1", port)) + s.sendall(json.dumps({"action": "password", "vname": "vm", "passwd": "$6$gone"}).encode()) + time.sleep(0.5) + time.sleep(3) # longer than the change would have taken + self.assertEqual(json.loads(self.state.read_text())["writes"], 0) + + def test_the_reply_waits_until_the_appliance_is_gone(self): + port = free_port() + self._start("127.0.0.1", port, GSTFSD_DEADLINE="1") + self._update_state(appliance=True, launch_seconds=3) + reply = self._request("127.0.0.1", port, {"action": "password", "vname": "vm", "passwd": "$6$late"}) + self.assertEqual(reply["return"], "error") + pid = json.loads(self.state.read_text())["appliance_pid"] + try: + with open("/proc/%d/stat" % pid) as f: + state = f.read().rsplit(")", 1)[1].split()[0] + except FileNotFoundError: + state = "gone" + self.assertIn(state, ("gone", "Z")) + + +class GstfsdClientTestCase(unittest.TestCase): + """The panel side sends the whole request and reads the whole reply.""" + + def test_a_reply_in_pieces_is_read_whole(self): + import threading + from unittest.mock import patch + + from instances import views + + server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server.bind(("127.0.0.1", 0)) + server.listen(1) + self.addCleanup(server.close) + received = [] + + def serve(): + conn, _ = server.accept() + with conn: + received.append(conn.recv(65536)) + conn.sendall(b'{"return": ') + time.sleep(0.2) + conn.sendall(b'"success"}') + + thread = threading.Thread(target=serve) + thread.start() + with patch.object(views, "GSTFSD_PORT", server.getsockname()[1]): + reply = views.gstfsd_request("127.0.0.1", {"action": "publickey", "vname": "vm", "key": "k"}) + thread.join(5) + self.assertEqual(reply, {"return": "success"}) + self.assertEqual(json.loads(received[0])["vname"], "vm") diff --git a/instances/test_root_pass.py b/instances/test_root_pass.py index e706c907..d8f75818 100644 --- a/instances/test_root_pass.py +++ b/instances/test_root_pass.py @@ -28,10 +28,12 @@ class SetRootPassTestCase(TestCase): def test_set_root_pass_post(self, mock_proxy, mock_socket_cls): # Renamed on the host: gstfsd must get the name of this UUID's domain mock_proxy.instance.name.return_value = "renamed-on-host" - mock_proxy.get_status.return_value = 5 # status: running + mock_proxy.get_status.return_value = 5 # shut off mock_sock = MagicMock() mock_socket_cls.return_value = mock_sock - mock_sock.recv.return_value = json.dumps({"return": "success"}).encode() + replies = iter([json.dumps({"return": "success"}).encode(), b""] * 2) + # one reply, then gstfsd closes the connection + mock_sock.recv.side_effect = lambda size: next(replies) response = self.client.post( reverse("instances:rootpasswd", args=[self.instance.id]), @@ -41,8 +43,8 @@ class SetRootPassTestCase(TestCase): self.assertEqual(response.status_code, 302) # Check socket sent the command with hashed password - self.assertTrue(mock_sock.send.called) - sent_bytes = mock_sock.send.call_args[0][0] + self.assertTrue(mock_sock.sendall.called) + sent_bytes = mock_sock.sendall.call_args[0][0] data = json.loads(sent_bytes.decode()) self.assertEqual(data["action"], "password") self.assertEqual(data["vname"], "renamed-on-host") @@ -65,13 +67,15 @@ class SetRootPassTestCase(TestCase): mock_proxy.get_status.return_value = 5 mock_sock = MagicMock() mock_socket_cls.return_value = mock_sock - mock_sock.recv.return_value = json.dumps({"return": "success"}).encode() + replies = iter([json.dumps({"return": "success"}).encode(), b""] * 2) + # one reply, then gstfsd closes the connection + mock_sock.recv.side_effect = lambda size: next(replies) url = reverse("instances:rootpasswd", args=[self.instance.id]) self.client.post(url, {"passwd": "samePassword1"}) self.client.post(url, {"passwd": "samePassword1"}) - hashes = [json.loads(c[0][0].decode())["passwd"] for c in mock_sock.send.call_args_list] + hashes = [json.loads(c[0][0].decode())["passwd"] for c in mock_sock.sendall.call_args_list] self.assertEqual(len(hashes), 2) self.assertNotEqual(hashes[0], hashes[1]) diff --git a/instances/views.py b/instances/views.py index 92281284..1706cd78 100755 --- a/instances/views.py +++ b/instances/views.py @@ -707,17 +707,20 @@ def migrate(request, pk): GSTFSD_PORT = 16510 -GSTFSD_TIMEOUT = 120 # gstfsd starts a guestfs appliance, which takes a while +GSTFSD_TIMEOUT = 120 # gstfsd answers within its 100 s limit for a request (GSTFSD_DEADLINE) def gstfsd_request(hostname, data): """Send one request to gstfsd and return its JSON reply, or an error reply.""" - s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - s.settimeout(GSTFSD_TIMEOUT) try: - s.connect((hostname, GSTFSD_PORT)) - s.send(json.dumps(data).encode()) - return json.loads(s.recv(1024).strip()) + # the host name may resolve to IPv4 or IPv6 + s = socket.create_connection((hostname, GSTFSD_PORT), timeout=GSTFSD_TIMEOUT) + except OSError as err: + return {"return": "error", "message": _("gstfsd error: %(err)s") % {"err": err}} + try: + s.sendall(json.dumps(data).encode()) + # gstfsd sends one reply and closes the connection + return json.loads(b"".join(iter(lambda: s.recv(4096), b""))) except (OSError, ValueError) as err: return {"return": "error", "message": _("gstfsd error: %(err)s") % {"err": err}} finally: