diff --git a/conf/daemon/consolecallback b/conf/daemon/consolecallback index 60585846..ea9eccf5 100755 --- a/conf/daemon/consolecallback +++ b/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))) diff --git a/console/socketiod b/console/socketiod index 707b81ca..63c8683c 100755 --- a/console/socketiod +++ b/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 diff --git a/console/tests/test_socketiod.py b/console/tests/test_socketiod.py index 4957b851..dc997db3 100644 --- a/console/tests/test_socketiod.py +++ b/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']")