fix(console): a serial paste and its echo no longer wait on each other

- socketiod wrote input to the worker's terminal while holding the
  session's lock, and the write could wait for the worker; the output
  forwarder needs that lock. A large paste into a guest that echoes it
  could stop both for good. The terminal is now non-blocking: input is
  written as far as the worker takes it, and the wait for room is outside
  the lock.
- The terminal is raw from the start. Until the worker made it raw, it
  kept at most one 4 KiB line of a paste, and the worker's own setraw
  dropped what was typed meanwhile; the worker no longer sets it.
- A superuser's takeover ends the old worker, and waits for it, before
  starting the new one: libvirt gives a VM's console to one at a time.
- A resize while the worker ends no longer raises.

conf/daemon/consolecallback

@@ -3,11 +3,12 @@
 # (socketiod runs it in a pty, one per client). It opens the console through
 # the compute's connection (its stored credentials); the connection's event
 # loop thread runs the stream callbacks. The console survives guest reboots.
+# socketiod made the terminal raw before starting it (setting it here would
+# drop what the user typed meanwhile).
 
 import os
 import sys
 import time
-import tty
 
 import django
 
@@ -117,7 +118,6 @@ def main():
     c = instance.compute
     try:
         console = Console(connection_manager.get_connection(c.hostname, c.login, c.password, c.type), instance.uuid)
-        tty.setraw(0)  # before the console opens: nothing typed is echoed by the terminal
         console.check()
     except libvirt.libvirtError as err:
         say("Cannot open the console of %s: %s" % (instance.name, public_error_text(err)))

console/socketiod

@@ -16,6 +16,7 @@ import struct
 import sys
 import termios
 import threading
+import tty
 
 import django
 
@@ -40,7 +41,9 @@ sio = socketio.Server(async_mode="threading")  # the same origin only
 # Handlers run on their own threads. sessions (sid -> {"pid", "fd", "vm",
 # "lock", "open"}) changes only under sessions_lock; a session's fd is used
 # and closed only under its own lock, while it is open, so a closed fd's
-# number, reused by a new session, never reaches another console.
+# number, reused by a new session, never reaches another console. The fd
+# is non-blocking: no one waits on it while holding the lock (a paste and
+# the guest's echo of it would wait on each other).
 sessions = {}
 sessions_lock = threading.Lock()
 
@@ -71,6 +74,8 @@ def connect(sid, environ, auth):
             raise refused("Too many consoles are open on this server: try again later")
         if open_sid is not None:  # a superuser takes over a console left open
             taken = open_sid, sessions.pop(open_sid)
+            # gone before the new worker starts: libvirt gives a console to one at a time
+            end(taken[1], wait=True)
         pid, fd = pty.fork()
         if pid == 0:
             try:
@@ -80,13 +85,16 @@ def connect(sid, environ, auth):
                 os.execv(sys.executable, [sys.executable, CONSOLECALLBACK, str(instance.pk), str(user.pk)])
             finally:
                 os._exit(1)
+        os.set_blocking(fd, False)
+        # raw at once, not when the worker gets to it: until then the terminal
+        # keeps at most one 4 KiB line of a paste
+        tty.setraw(fd)
         sessions[sid] = session = {"pid": pid, "fd": fd, "vm": instance.pk, "lock": threading.Lock(), "open": True}
     logger.info("serial console of VM %s opened by %s%s", instance.name, user.username,
                 " (taken over)" if taken else "")
     if taken:
-        old_sid, old_session = taken
+        old_sid, _ = taken
         sio.emit("pty_output", {"output": "\r\n[An administrator opened this console]\r\n"}, to=old_sid)
-        end(old_session)
         sio.disconnect(old_sid)
     sio.start_background_task(forward, sid, session)
 
@@ -107,6 +115,8 @@ def forward(sid, session):
                 return
             try:
                 output = os.read(session["fd"], 20480)
+            except BlockingIOError:  # nothing after all
+                continue
             except OSError:  # EIO: the worker ended
                 output = b""
         if not output:
@@ -125,8 +135,8 @@ def stop(sid):
         end(session)
 
 
-def end(session):
-    """End a session taken out of sessions."""
+def end(session, wait=False):
+    """End a session taken out of sessions; with wait, once its worker is gone."""
     try:
         os.kill(session["pid"], signal.SIGKILL)  # first: a write blocked on its terminal ends
     except ProcessLookupError:
@@ -137,20 +147,35 @@ def end(session):
             os.close(session["fd"])
         except OSError:
             pass
-    sio.start_background_task(os.waitpid, session["pid"], 0)
+    if wait:
+        try:
+            os.waitpid(session["pid"], 0)
+        except ChildProcessError:  # reaped already
+            pass
+    else:
+        sio.start_background_task(os.waitpid, session["pid"], 0)
 
 
 @sio.event
 def pty_input(sid, message):
     session = sessions.get(sid)
     if session and isinstance(message, dict) and isinstance(message.get("input"), str):
-        data = message["input"].encode(errors="replace")  # a lone surrogate from the browser
-        with session["lock"]:
-            if session["open"]:
+        data = memoryview(message["input"].encode(errors="replace"))  # a lone surrogate from the browser
+        while data:
+            with session["lock"]:
+                if not session["open"]:
+                    return
                 try:
-                    os.write(session["fd"], data)
-                except OSError:
+                    data = data[os.write(session["fd"], data):]
+                    continue
+                except BlockingIOError:  # the worker has not read the rest yet
                     pass
+                except OSError:
+                    return
+            try:
+                select.select([], [session["fd"]], [], 0.1)  # outside the lock: output goes on meanwhile
+            except (OSError, ValueError):  # closed meanwhile: the loop sees it
+                pass
 
 
 @sio.event
@@ -162,7 +187,10 @@ def resize(sid, message):
     if all(isinstance(v, int) and 1 <= v <= 500 for v in (rows, cols)):
         with session["lock"]:
             if session["open"]:
-                fcntl.ioctl(session["fd"], termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols, 0, 0))
+                try:
+                    fcntl.ioctl(session["fd"], termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols, 0, 0))
+                except OSError:  # the worker has just ended
+                    pass
 
 
 @sio.event

console/tests/test_socketiod.py

@@ -1,6 +1,7 @@
 """socketiod: who may connect, and each session's console kept to itself."""
 
 import os
+import select
 import tempfile
 import time
 from importlib.machinery import SourceFileLoader
@@ -67,6 +68,7 @@ class SocketiodTestCase(Base):
         super().setUp()
         self.d = load_socketiod()
         self.started = []
+        self.killed_before_fork = []
         self.forks = iter(range(1000, 2000))
         patches = [
             patch.object(self.d.pty, "fork", side_effect=self.fork),
@@ -80,9 +82,10 @@ class SocketiodTestCase(Base):
             self.addCleanup(p.stop)
 
     def fork(self):
-        read, write = os.pipe()
-        self.addCleanup(lambda: [os.close(fd) for fd in (read, write) if _open(fd)])
-        return next(self.forks), read
+        self.killed_before_fork.append(self.d.os.kill.call_count)
+        master, worker_side = os.openpty()
+        self.addCleanup(lambda: [os.close(fd) for fd in (master, worker_side) if _open(fd)])
+        return next(self.forks), master
 
     @override_settings(SERIAL_CONSOLE_ENABLED=True)
     def test_a_valid_token_starts_a_console_for_its_session(self):
@@ -116,7 +119,8 @@ class SocketiodTestCase(Base):
     def test_input_and_resize_go_to_the_session_and_are_checked(self):
         self.d.connect("sid-1", {}, {"token": self.token()})
         fd = self.d.sessions["sid-1"]["fd"]
-        with patch.object(self.d.os, "write") as write, patch.object(self.d.fcntl, "ioctl") as ioctl:
+        with patch.object(self.d.os, "write", side_effect=lambda fd, data: len(data)) as write, \
+                patch.object(self.d.fcntl, "ioctl") as ioctl:
             self.d.pty_input("sid-1", {"input": "ls\r"})
             for bad in (None, "ls", {"input": 7}):
                 self.d.pty_input("sid-1", bad)
@@ -124,7 +128,8 @@ class SocketiodTestCase(Base):
             self.d.resize("sid-1", {"rows": 24, "cols": 80})
             for bad in ({"rows": 0, "cols": 80}, {"rows": 24, "cols": 501}, {"rows": "24", "cols": 80}, "x"):
                 self.d.resize("sid-1", bad)
-        write.assert_called_once_with(fd, b"ls\r")
+        self.assertEqual(write.call_count, 1)
+        self.assertEqual((write.call_args.args[0], bytes(write.call_args.args[1])), (fd, b"ls\r"))
         self.assertEqual(ioctl.call_count, 1)
 
     @override_settings(SERIAL_CONSOLE_ENABLED=True)
@@ -153,6 +158,8 @@ class SocketiodTestCase(Base):
             self.assertIsNone(self.d.connect("admin-tab", {}, {"token": self.token(admin)}))
         self.assertEqual(list(self.d.sessions), ["admin-tab"])
         self.assertFalse(old["open"])
+        # the old worker was killed before the new one started: libvirt gives the console to one
+        self.assertEqual(self.killed_before_fork, [0, 1])
         self.assertIn("An administrator opened this console", emit.call_args.args[1]["output"])
         disconnect.assert_called_once_with("owner-tab")
 
@@ -180,6 +187,40 @@ class SocketiodTestCase(Base):
         self.assertEqual(len(self.d.sessions), 1)
 
 
+class PasteTestCase(Base):
+    """A real worker that echoes every byte twice: a paste far larger than the
+    terminal's buffers goes through while its echo comes back. With the
+    session lock held during a blocking write, the paste and its echo
+    waited on each other for ever."""
+
+    @override_settings(SERIAL_CONSOLE_ENABLED=True)
+    def test_a_large_paste_and_its_echo_do_not_wait_on_each_other(self):
+        import threading
+
+        d = load_socketiod()
+        with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as script:
+            # raw already: socketiod made the terminal raw before starting it
+            script.write("import os\nwhile (data := os.read(0, 4096)):\n    os.write(1, data * 2)\n")
+        self.addCleanup(os.unlink, script.name)
+        echoed = []
+        patches = [patch.object(d, "CONSOLECALLBACK", script.name), patch.object(d.db, "close_old_connections"),
+                   patch.object(d.sio, "emit", side_effect=lambda event, data, to: echoed.append(len(data["output"]))),
+                   patch.object(d.sio, "disconnect")]
+        for p in patches:
+            p.start()
+            self.addCleanup(p.stop)
+        d.connect("sid-1", {}, {"token": self.token()})
+        self.addCleanup(d.stop, "sid-1")
+        paste = threading.Thread(target=d.pty_input, args=("sid-1", {"input": "x" * 1_000_000}), daemon=True)
+        paste.start()
+        paste.join(30)
+        self.assertFalse(paste.is_alive(), "the paste never finished")
+        deadline = time.monotonic() + 30
+        while sum(echoed) < 2_000_000 and time.monotonic() < deadline:
+            time.sleep(0.05)
+        self.assertEqual(sum(echoed), 2_000_000)
+
+
 class WorkerIsolationTestCase(Base):
     """A real fork: a session's worker holds no other session's console."""
 
@@ -197,9 +238,10 @@ class WorkerIsolationTestCase(Base):
             d.connect("first", {}, {"token": self.token()})
             d.connect("second", {}, {"token": self.token(vm=other)})
         self.addCleanup(lambda: [d.stop(sid) for sid in list(d.sessions)])
-        output = b""
+        output, fd = b"", d.sessions["second"]["fd"]
         while b"]" not in output:
-            output += os.read(d.sessions["second"]["fd"], 1024)
+            select.select([fd], [], [], 5)
+            output += os.read(fd, 1024)
         # 0-2: its terminal, 3: the listing itself
         self.assertEqual(output.decode().strip(), "['0', '1', '2', '3']")