fix(gstfsd): find the guest OS by inspection, answer once, and stop a change at its deadline

gstfsd mounted every partition it listed: a root on LVM was never found,
a disk with two OS roots got the change twice and two concatenated
replies, and no root, an error or a request longer than 1024 bytes ended
the connection without a reply. It now lets libguestfs inspect the guest,
mounts its one OS the way the guest does (a broken filesystem off the
changed path is left out), and sends exactly one reply, keeping the first
error. The root entry is found on any line of /etc/shadow and must be
unique; a key needs no shadow file. A request is at most 64 KiB. Each
change runs in a worker process with a directly started appliance; both
are killed when the request is not done 100 s after gstfsd accepted it
(GSTFSD_DEADLINE may only lower that), before the panel's 120 s timeout
releases the VM, or when the panel's connection ends, and the reply waits
until they are gone, so nothing touches the disk once the panel has given
up. A worker that dies says why. The panel
connects over IPv4 or IPv6, sends the whole request and reads the reply
until gstfsd closes.

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

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())

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")

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])
 

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: