WebREPL: an abandoned login no longer leaves the board's session slot held

cli/src/mpftp/webrepl.py

@@ -51,6 +51,68 @@
 #: blocking reads give up after this long instead of hanging the sidecar.
 STALL_TIMEOUT = 20.0
 
+#: Each login step (upgrade reply, password prompt, login reply, banner) waits
+#: at least this long, and 30 times the upgrade's measured round trip when
+#: that is longer, so a lossy link isn't abandoned halfway through the login.
+LOGIN_TIMEOUT_FLOOR = 5.0
+LOGIN_TIMEOUT_RTT_FACTOR = 30.0
+LOGIN_TIMEOUT_CEILING = 60.0
+#: Set this to a number of seconds to fix the login wait instead.
+LOGIN_TIMEOUT_ENV = "MPFTP_WEBREPL_LOGIN_TIMEOUT"
+#: How long to wait for the board to finish closing before dropping the socket.
+CLOSE_DRAIN = 0.5
+
+
+def configured_login_timeout() -> Optional[float]:
+    """The login wait the user fixed, or None to scale it with the link."""
+    raw = os.environ.get(LOGIN_TIMEOUT_ENV, "").strip()
+    try:
+        value = float(raw) if raw else 0.0
+    except ValueError:
+        return None
+    return value if value > 0 else None
+
+
+def release_socket(sock: Any, drain: float = CLOSE_DRAIN) -> None:
+    """Close ``sock`` so the board sees the connection end.
+
+    A bare ``close()`` with bytes the board sent still unread makes the host
+    answer with a RST, which is one packet. On a lossy link that packet can be
+    lost, and then the board never learns the session ended: WebREPL has no
+    timeout and no keepalive, so it holds its one session slot until the board
+    resets. Shutting down the write side sends a FIN, which the host's TCP
+    retransmits until it is acknowledged, and reading until the board's own
+    close arrives (or ``drain`` runs out) leaves nothing unread to turn the
+    final ``close()`` into a RST.
+    """
+    try:
+        sock.shutdown(socket.SHUT_WR)
+    except Exception:
+        pass
+    deadline = time.monotonic() + drain
+    try:
+        sock.setblocking(False)
+        while True:
+            left = deadline - time.monotonic()
+            if left <= 0:
+                break
+            ready, _, _ = select.select([sock], [], [], min(left, 0.2))
+            if not ready:
+                continue
+            try:
+                if not sock.recv(65536):
+                    break
+            except (BlockingIOError, InterruptedError):
+                continue
+            except OSError:
+                break
+    except Exception:
+        pass
+    try:
+        sock.close()
+    except Exception:
+        pass
+
 
 class WebReplError(OSError):
     """The WebREPL connection failed or dropped.
@@ -61,6 +123,15 @@ class WebReplError(OSError):
     """
 
 
+class WebReplBusyError(WebReplError):
+    """The board accepted the upgrade and hung up before the password prompt.
+
+    ``webrepl.py`` does exactly that when ``os.dupterm()`` is already taken:
+    the board serves one session, and it frees the slot only when a read on
+    that session's socket ends. A session lost on a poor link may never end.
+    """
+
+
 class WebReplAuthError(RuntimeError):
     """The board rejected the password. Never retried."""
 
@@ -214,7 +285,7 @@ def __init__(
         *,
         timeout: Optional[float] = None,
         connect_timeout: float = 5.0,
-        login_timeout: float = 5.0,
+        login_timeout: Optional[float] = None,
         create_connection: Callable[..., Any] = socket.create_connection,
     ) -> None:
         self.port = device
@@ -223,9 +294,12 @@ def __init__(
         self.timeout = timeout
         self.write_timeout: Optional[float] = None
         self.connect_timeout = connect_timeout
-        self.login_timeout = login_timeout
+        #: None scales the wait with the link (``LOGIN_TIMEOUT_ENV`` fixes it).
+        self._login_fixed = login_timeout if login_timeout else configured_login_timeout()
+        self.login_timeout = self._login_fixed or LOGIN_TIMEOUT_FLOOR
         self._create_connection = create_connection
         self._sock: Any = None
+        self._upgraded = False
         self._parser = FrameParser()
         self._rx = bytearray()  # text frames: the REPL stream
         self._brx = bytearray()  # binary frames: file-transfer replies
@@ -345,11 +419,8 @@ def close(self) -> None:
             except Exception:
                 pass
         self._closed_reason = self._closed_reason or "closed by mpftp"
-        try:
-            sock.close()
-        except Exception:
-            pass
         self._sock = None
+        release_socket(sock)
 
     # --- connection -------------------------------------------------------
 
@@ -376,16 +447,32 @@ def open(self) -> None:
             self._sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
         except Exception:
             pass
+        self._upgraded = False
         try:
+            t0 = time.monotonic()
             self._upgrade()
+            self._upgraded = True
+            if not self._login_fixed:
+                rtt = time.monotonic() - t0
+                self.login_timeout = min(
+                    LOGIN_TIMEOUT_CEILING,
+                    max(LOGIN_TIMEOUT_FLOOR, LOGIN_TIMEOUT_RTT_FACTOR * rtt),
+                )
             self._login()
         except BaseException:
+            sock, self._sock = self._sock, None
+            # Tell the board, politely, that this login is over: its one
+            # session slot stays taken until it sees the connection end.
+            if sock is not None and not self._closed_reason and self._upgraded:
+                try:
+                    sock.setblocking(True)
+                    sock.settimeout(1.0)
+                    sock.sendall(encode_frame(OP_CLOSE, struct.pack("!H", 1000)))
+                except Exception:
+                    pass
             self._closed_reason = self._closed_reason or "open failed"
-            try:
-                self._sock.close()
-            except Exception:
-                pass
-            self._sock = None
+            if sock is not None:
+                release_socket(sock)
             raise
 
     def _resolve_host(self) -> str:
@@ -475,9 +562,14 @@ def _read_until_any(self, needles: tuple[bytes, ...], limit: float, what: str) -
                         return data[: at + len(n)]
                 if self._closed_reason:
                     if not data:
-                        raise WebReplError(
+                        raise WebReplBusyError(
                             f"{self.port}: the board closed the connection before {what}. "
-                            "Another WebREPL client may be connected (the board allows one)."
+                            "WebREPL serves one session and this one was turned away, so "
+                            "another client holds it, or an earlier session was lost on a "
+                            "weak link and the board still counts it (it has no timeout). "
+                            "Close the other client. If none is open, reset the board, or "
+                            "run `import webrepl; webrepl.stop(); webrepl.start()` on it "
+                            "over USB serial."
                         )
                     raise WebReplError(
                         f"{self.port}: the board closed the connection before {what} "

cli/tests/test_webrepl.py

@@ -8,11 +8,13 @@
 
 import base64
 import hashlib
+import os
 import socket
 import struct
 import threading
 import time
 import unittest
+from unittest import mock
 
 from mpftp import webrepl
 from mpftp.webrepl import (
@@ -236,7 +238,7 @@ def script(board):  # webrepl.py: "Concurrent WebREPL connection ... rejected"
             board.sock.close()
 
         ws, _ = self.connect(script)
-        with self.assertRaisesRegex(WebReplError, "Another WebREPL client"):
+        with self.assertRaisesRegex(WebReplError, "one session and this one was turned away"):
             ws.open()
 
     def test_a_plain_web_server_is_not_mistaken_for_webrepl(self):
@@ -266,7 +268,7 @@ def script(board):
         t0 = time.monotonic()
         with self.assertRaisesRegex(WebReplError, "no the password prompt|password prompt"):
             ws.open()
-        self.assertLess(time.monotonic() - t0, 3.0)
+        self.assertLess(time.monotonic() - t0, 3.5)
 
 
 class SessionTests(SessionCase):
@@ -453,3 +455,112 @@ def script(board):
 
 if __name__ == "__main__":
     unittest.main()
+
+class AbandonedLoginTests(unittest.TestCase):
+    """A login that gives up must end the session in a way the board sees.
+
+    The board keeps one WebREPL slot and frees it only when a read on that
+    session ends. A RST is a single packet; on a lossy link its loss leaves
+    the slot taken for good, so an abandoned login ends with a close frame and
+    a FIN, with nothing unread to turn the final close into a RST.
+    """
+
+    def _listener(self):
+        lsock = socket.socket()
+        lsock.bind(("127.0.0.1", 0))
+        lsock.listen(1)
+        self.addCleanup(lsock.close)
+        return lsock
+
+    def test_a_timed_out_login_ends_with_a_close_frame_and_a_fin_not_a_reset(self):
+        lsock = self._listener()
+        seen: dict = {}
+
+        def board():
+            conn, _ = lsock.accept()
+            self.addCleanup(conn.close)
+            fake = FakeBoard(conn)
+            fake.accept(fake.read_request())
+            time.sleep(0.5)  # a slow link: the prompt comes after the client gave up
+            try:
+                conn.sendall(board_frame(OP_TEXT, b"Password: "))
+                conn.settimeout(3)
+                data = b""
+                while True:
+                    chunk = conn.recv(4096)
+                    if not chunk:
+                        seen["end"] = "eof"
+                        break
+                    data += chunk
+                seen["frames"] = FrameParser().feed(data)
+            except ConnectionResetError:
+                seen["end"] = "reset"
+            except OSError as e:
+                seen["end"] = repr(e)
+
+        t = threading.Thread(target=board, daemon=True)
+        t.start()
+        ws = WebSocketSerial(
+            f"ws://127.0.0.1:{lsock.getsockname()[1]}", PASSWORD, login_timeout=0.2
+        )
+        with self.assertRaisesRegex(WebReplError, "password prompt"):
+            ws.open()
+        t.join(5)
+        self.assertEqual(seen.get("end"), "eof", seen)
+        self.assertIn(OP_CLOSE, [op for op, _ in seen["frames"]])
+
+    def test_release_socket_reads_what_the_board_sent_so_close_is_not_a_reset(self):
+        lsock = self._listener()
+        client = socket.create_connection(lsock.getsockname())
+        conn, _ = lsock.accept()
+        self.addCleanup(conn.close)
+        conn.sendall(b"unread by the client")
+        time.sleep(0.1)
+        webrepl.release_socket(client, drain=0.3)
+        conn.settimeout(3)
+        self.assertEqual(conn.recv(10), b"")  # a FIN; a RST would raise
+
+    def test_a_held_slot_gets_its_own_error_that_says_what_to_do(self):
+        def script(board):
+            board.accept(board.read_request())
+            board.sock.close()
+
+        ws, _ = connect(script)
+        for sock in ws._test_socks:
+            self.addCleanup(sock.close)
+        with self.assertRaises(webrepl.WebReplBusyError) as cm:
+            ws.open()
+        self.assertIn("before the password prompt", str(cm.exception))
+        self.assertIn("reset the board", str(cm.exception))
+
+    def test_login_wait_scales_with_the_link_unless_fixed(self):
+        ws = WebSocketSerial("ws://192.0.2.1", PASSWORD)
+        self.assertEqual(ws.login_timeout, webrepl.LOGIN_TIMEOUT_FLOOR)
+        self.assertEqual(WebSocketSerial("ws://192.0.2.1", PASSWORD, login_timeout=9).login_timeout, 9)
+        with mock.patch.dict(os.environ, {webrepl.LOGIN_TIMEOUT_ENV: "12"}):
+            self.assertEqual(WebSocketSerial("ws://192.0.2.1", PASSWORD).login_timeout, 12)
+
+    def test_a_slow_upgrade_stretches_the_rest_of_the_login(self):
+        def script(board):
+            req = board.read_request()
+            time.sleep(0.4)
+            board.accept(req)
+            board.send(OP_TEXT, b"Password: ")
+            board.login_tail = board.next_data_frame()
+            board.send(OP_TEXT, b"\r\nWebREPL connected\r\n>>> ")
+
+        client_end, board_end = socket.socketpair()
+        self.addCleanup(client_end.close)
+        self.addCleanup(board_end.close)
+        FakeBoard(board_end, script=script).start()
+        ws = WebSocketSerial(
+            "ws://192.0.2.1:8266", PASSWORD, create_connection=lambda a, timeout=None: client_end
+        )
+        with mock.patch.object(webrepl, "LOGIN_TIMEOUT_FLOOR", 0.1), mock.patch.object(
+            webrepl, "LOGIN_TIMEOUT_RTT_FACTOR", 10.0
+        ):
+            ws.login_timeout = 1.0  # the upgrade's wait
+            ws._login_fixed = None
+            # the upgrade takes ~0.4 s; 10x that is the budget for the steps after it
+            ws.open()
+        self.assertGreaterEqual(ws.login_timeout, 3.5)

docs/guides/connect-over-wifi.md

@@ -58,6 +58,20 @@ its logs and refuses them in the address (`ws://:pw@host`).
 The first connect to a board asks for its password and remembers it for that
 board.
 
+## On a weak link
+
+WebREPL serves one session at a time and has no timeout, so a session that was
+lost on a bad link can keep the board's only slot long after you stopped. The
+next login then connects, gets a `101` and is hung up on before the password
+prompt, and mpftp says the board is holding a session.
+
+mpftp ends an abandoned login with a close frame and a clean TCP close, so the
+board sees it end, and it waits longer for each login step when the link is
+slow (at least 5 s, up to 60 s). Set `MPFTP_WEBREPL_LOGIN_TIMEOUT` to a number
+of seconds to fix that wait yourself. If the slot is held anyway, reset the
+board, or run `import webrepl; webrepl.stop(); webrepl.start()` on it over USB
+serial.
+
 ## Limits
 
 - **A loop that never yields.** On stock esp32 firmware, a program that never