LINEMODE negotiation fixes (jquast#131)

Finish our "LINEMODE" telnet option negotiation by enabling full feature support in our "telsh"
server shell and design a compliant "LinemodeBuffer" for compliant client shell so that it may be
fully exercised.

Although LINEMODE was mostly implemented, there were a number of bugs that prevented it from
negotiating correctly. However, both our server and client take a passive approach to LINEMODE, and
only negotiate for it when asked.

Author: jquast

docs/history.rst

@@ -1,20 +1,14 @@
 History
 =======
-3.0.3 (unreleased)
-  * new: :func:`~telnetlib3.accessories.make_logger` accepts a ``filemode``
-    parameter (``"a"`` append or ``"w"`` overwrite) so callers can control
-    whether log files are truncated on each run.  The default remains
-    ``"a"`` (append), matching historical behaviour.
-  * new: ``--logfile-mode {append,rewrite}`` CLI flag controls whether the
-    log file is appended to (default) or overwritten on each connection.
-  * new: ``--typescript-mode {append,rewrite}`` CLI flag controls whether the
-    typescript file is appended to (default) or overwritten on each connection.
-  * bugfix: kludge mode not detected when server negotiates WILL ECHO + DO SGA
-    (requesting client WILL SGA) instead of WILL ECHO + WILL SGA.
-    :attr:`~telnetlib3.stream_writer.TelnetWriter.mode` and
-    ``_server_will_sga()`` now check SGA in either direction
-    (``remote_option`` or ``local_option``), so clients correctly switch from
-    line mode to character-at-a-time mode with these servers.
+3.0.3
+  * bugfix: server and client now correctly complete LINEMODE negotiation when prompted to.
+  * new: ``--logfile-mode {append,rewrite}`` and ``--typescript-mode`` CLI flags
+    and :func:`~telnetlib3.accessories.make_logger` ``filemode`` argument control whether the log
+    file is appended to (default) or overwritten on each connection.
+  * new: :class:`~telnetlib3.client_shell.LinemodeBuffer` used by ``telnetlib3-client``, a
+    client-side line buffer for LINEMODE EDIT mode with local erase-char, erase-line, erase-word
+    editing, forwardmask flushing, and TRAPSIG IAC command generation. The default 'telsh' server
+    was also updated to support linemode.
 
 3.0.2
   * bugfix: :meth:`~telnetlib3.stream_writer.TelnetWriter.request_charset` raised :exc:`TypeError`,

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "telnetlib3"
-version = "3.0.2"  # Keep in sync with telnetlib3/accessories.py::get_version !
+version = "3.0.3"  # Keep in sync with telnetlib3/accessories.py::get_version !
 description = " Python Telnet server and client CLI and Protocol library"
 readme = "README.rst"
 license = "ISC"

telnetlib3/accessories.py

@@ -42,7 +42,7 @@
 
 def get_version() -> str:
     """Return the current version of telnetlib3."""
-    return "3.0.2"  # keep in sync with pyproject.toml !
+    return "3.0.3"  # keep in sync with pyproject.toml !
 
 
 def encoding_from_lang(lang: str) -> Optional[str]:

telnetlib3/client_shell.py

@@ -10,12 +10,14 @@
 from dataclasses import dataclass
 
 # local
+from . import slc as slc_module
 from . import accessories
 from ._session_context import TelnetSessionContext
 
 log = logging.getLogger(__name__)
 
 # local
+from .telopt import LINEMODE  # noqa: E402
 from .accessories import TRACE  # noqa: E402
 from .stream_reader import TelnetReader, TelnetReaderUnicode  # noqa: E402
 from .stream_writer import TelnetWriter, TelnetWriterUnicode  # noqa: E402
@@ -180,6 +182,93 @@ class _RawLoopState:
     reactivate_repl: bool = False
 
 
+class LinemodeBuffer:
+    """
+    Client-side line buffer for LINEMODE EDIT mode (RFC 1184 §3.1).
+
+    Accumulates characters typed by the user, applying local SLC editing functions (erase-char,
+    erase-line, erase-word) and transmitting complete lines to the server.  When TRAPSIG is enabled,
+    signal characters (^C etc.) are sent as IAC commands instead of buffered.
+
+    :param slctab: The writer's current SLC character table.
+    :param forwardmask: FORWARDMASK received from server, or None.
+    :param trapsig: When True, signal characters are sent as IAC commands.
+    """
+
+    def __init__(
+        self,
+        slctab: Dict[bytes, slc_module.SLC],
+        forwardmask: Optional[slc_module.Forwardmask] = None,
+        trapsig: bool = False,
+    ) -> None:
+        """Initialize LinemodeBuffer."""
+        from .telopt import IP, AYT, BRK, EOF, IAC, SUSP, ABORT
+
+        self._buf: list[str] = []
+        self.slctab = slctab
+        self.forwardmask = forwardmask
+        self.trapsig = trapsig
+        self._trapsig_map: Dict[bytes, bytes] = {
+            slc_module.SLC_IP: IAC + IP,
+            slc_module.SLC_ABORT: IAC + ABORT,
+            slc_module.SLC_SUSP: IAC + SUSP,
+            slc_module.SLC_EOF: IAC + EOF,
+            slc_module.SLC_BRK: IAC + BRK,
+            slc_module.SLC_AYT: IAC + AYT,
+        }
+
+    def _slc_val(self, func: bytes) -> Optional[int]:
+        """Return the active byte value for SLC function, or None if unsupported."""
+        defn = self.slctab.get(func)
+        if defn is None or defn.nosupport:
+            return None
+        v = defn.val
+        return ord(v) if v and v != slc_module.theNULL else None
+
+    def feed(self, char: str) -> Tuple[str, Optional[bytes]]:
+        """
+        Feed one character into the buffer.
+
+        :returns: ``(echo, data)`` where ``echo`` is text to display locally
+            (may be empty) and ``data`` is bytes to send to server, or None
+            if buffering.
+        """
+        b = ord(char)
+        if self.trapsig:
+            for func, cmd in self._trapsig_map.items():
+                if b == self._slc_val(func):
+                    return ("", cmd)
+        if b == self._slc_val(slc_module.SLC_EC):
+            if self._buf:
+                self._buf.pop()
+                return ("\b \b", None)
+            return ("", None)
+        if b == self._slc_val(slc_module.SLC_EL):
+            n = len(self._buf)
+            self._buf.clear()
+            return ("\b \b" * n, None)
+        if b == self._slc_val(slc_module.SLC_EW):
+            popped = 0
+            # skip trailing spaces (POSIX VWERASE behaviour)
+            while self._buf and self._buf[-1] == " ":
+                self._buf.pop()
+                popped += 1
+            while self._buf and self._buf[-1] != " ":
+                self._buf.pop()
+                popped += 1
+            return ("\b \b" * popped, None)
+        if char in ("\r", "\n"):
+            line = "".join(self._buf) + char
+            self._buf.clear()
+            return (char, line.encode())
+        if self.forwardmask is not None and b in self.forwardmask:
+            data = ("".join(self._buf) + char).encode()
+            self._buf.clear()
+            return (char, data)
+        self._buf.append(char)
+        return (char, None)
+
+
 if sys.platform == "win32":
 
     async def telnet_client_shell(
@@ -314,10 +403,7 @@ def _server_will_sga(self) -> bool:
             from .telopt import SGA
 
             w = self.telnet_writer
-            return bool(
-                w.client
-                and (w.remote_option.enabled(SGA) or w.local_option.enabled(SGA))
-            )
+            return bool(w.client and (w.remote_option.enabled(SGA) or w.local_option.enabled(SGA)))
 
         def check_auto_mode(
             self, switched_to_raw: bool, last_will_echo: bool
@@ -334,6 +420,20 @@ def check_auto_mode(
                 return None
             wecho = self.telnet_writer.will_echo
             wsga = self._server_will_sga()
+            # LINEMODE EDIT: kernel must handle line editing; keep/restore cooked mode.
+            # This takes priority over the SGA/ECHO raw-mode heuristics below.
+            if (
+                self.telnet_writer.local_option.enabled(LINEMODE)
+                and self.telnet_writer.linemode.edit
+            ):
+                if switched_to_raw:
+                    assert self._save_mode is not None
+                    self.set_mode(self._save_mode)
+                    self.telnet_writer.log.debug(
+                        "auto: LINEMODE EDIT confirmed, restoring cooked mode"
+                    )
+                    return (False, wecho, False)
+                return None
             # WILL ECHO alone = line mode with server echo (suppress local echo)
             # WILL SGA (with or without ECHO) = raw/character-at-a-time
             should_go_raw = not switched_to_raw and wsga
@@ -366,20 +466,35 @@ def determine_mode(self, mode: "Terminal.ModeDef") -> "Terminal.ModeDef":
 
             Auto mode (``_raw_mode is None``): follows the server's negotiation.
 
-            =================  ========  ==========  ================================
+            =================  ========  ==========  ========================================
             Server negotiates  ICANON    ECHO        Behavior
-            =================  ========  ==========  ================================
+            =================  ========  ==========  ========================================
             Nothing            on        on          Line mode, local echo
+            LINEMODE EDIT      **on**    on          Cooked mode, kernel handles EC/EL/echo
+            LINEMODE remote    **off**   **off**     Raw, server echoes
             WILL SGA only      **off**   on          Character-at-a-time, local echo
             WILL ECHO only     on        **off**     Line mode, server echoes
             WILL SGA + ECHO    **off**   **off**     Full kludge mode (most common)
-            =================  ========  ==========  ================================
+            =================  ========  ==========  ========================================
             """
             raw_mode = _get_raw_mode(self.telnet_writer)
             will_echo = self.telnet_writer.will_echo
             will_sga = self._server_will_sga()
             # Auto mode (None): follow server negotiation
             if raw_mode is None:
+                if self.telnet_writer.local_option.enabled(LINEMODE):
+                    linemode_mode = self.telnet_writer.linemode
+                    if linemode_mode.edit:
+                        # RFC 1184 / NetBSD reference: LINEMODE EDIT means ICANON on.
+                        # The kernel line discipline handles EC (VERASE), EL (VKILL),
+                        # EW (VWERASE), and echo.  No software line editing needed.
+                        self.telnet_writer.log.debug(
+                            "auto: LINEMODE EDIT, cooked mode (kernel line editing)"
+                        )
+                        self.software_echo = False
+                        return mode  # keep ICANON on; kernel handles EC/EL/EW and echo
+                    self.telnet_writer.log.debug("auto: LINEMODE remote, raw input server echo")
+                    return self._make_raw(mode, suppress_echo=True)
                 if will_echo and will_sga:
                     self.telnet_writer.log.debug("auto: server echo + SGA, kludge mode")
                     return self._make_raw(mode)
@@ -535,6 +650,18 @@ def _ensure_autoreply_engine(
         """Return the autoreply engine from the writer's context, if set."""
         return telnet_writer.ctx.autoreply_engine
 
+    def _get_linemode_buffer(writer: Union[TelnetWriter, TelnetWriterUnicode]) -> "LinemodeBuffer":
+        """Return (or lazily create) the LinemodeBuffer attached to *writer*."""
+        buf: Optional[LinemodeBuffer] = getattr(writer, "_linemode_buf", None)
+        if buf is None:
+            buf = LinemodeBuffer(
+                slctab=writer.slctab,
+                forwardmask=writer.forwardmask,
+                trapsig=writer.linemode.trapsig,
+            )
+            writer._linemode_buf = buf
+        return buf
+
     async def _raw_event_loop(
         telnet_reader: Union[TelnetReader, TelnetReaderUnicode],
         telnet_writer: Union[TelnetWriter, TelnetWriterUnicode],
@@ -593,7 +720,26 @@ async def _raw_event_loop(
                         wait_for.remove(telnet_task)
                     handle_close("Connection closed.")
                     break
-                new_timer, has_pending = _send_stdin(inp, telnet_writer, stdout, state.local_echo)
+                linemode_edit = (
+                    telnet_writer.local_option.enabled(LINEMODE) and telnet_writer.linemode.edit
+                )
+                if linemode_edit and state.switched_to_raw:
+                    # Raw PTY or non-TTY: kernel not doing line editing, use LinemodeBuffer
+                    lmbuf = _get_linemode_buffer(telnet_writer)
+                    for ch in inp.decode(errors="replace"):
+                        echo, data = lmbuf.feed(ch)
+                        if echo:
+                            stdout.write(echo.encode())
+                        if data:
+                            telnet_writer._write(data)
+                    new_timer, has_pending = None, False
+                elif linemode_edit:
+                    # Cooked PTY: kernel already handled EC/EL/echo; forward line directly
+                    new_timer, has_pending = _send_stdin(inp, telnet_writer, stdout, False)
+                else:
+                    new_timer, has_pending = _send_stdin(
+                        inp, telnet_writer, stdout, state.local_echo
+                    )
                 if has_pending and esc_timer_task not in wait_for:
                     esc_timer_task = new_timer
                     if esc_timer_task is not None:

telnetlib3/server.py

@@ -42,7 +42,14 @@
 except ImportError:
     PTY_SUPPORT = False
 
-__all__ = ("TelnetServer", "Server", "create_server", "run_server", "parse_server_args")
+__all__ = (
+    "TelnetServer",
+    "LinemodeServer",
+    "Server",
+    "create_server",
+    "run_server",
+    "parse_server_args",
+)
 
 
 class CONFIG(NamedTuple):
@@ -806,6 +813,42 @@ def connection_lost(self, exc: Optional[Exception]) -> None:
         _ = exc
 
 
+class LinemodeServer(TelnetServer):
+    """
+    :class:`TelnetServer` subclass that negotiates LINEMODE EDIT.
+
+    In addition to the standard options negotiated by :class:`TelnetServer`,
+    this server sends ``DO LINEMODE`` during advanced negotiation, proposes
+    LINEMODE EDIT (local line editing by the client), and suppresses
+    ``WILL ECHO`` so the client performs local echoing via its LINEMODE buffer.
+
+    Use with :func:`create_server` to enable RFC 1184 LINEMODE EDIT on a
+    :func:`~.telnet_server_shell` session or any custom shell.
+    """
+
+    from . import slc as _slc_module
+
+    #: Propose LINEMODE EDIT (local line editing) instead of remote mode.
+    default_linemode = _slc_module.Linemode(_slc_module.LMODE_MODE_LOCAL)
+
+    def begin_advanced_negotiation(self) -> None:
+        """Negotiate standard options plus ``DO LINEMODE``."""
+        from .telopt import DO, LINEMODE
+
+        super().begin_advanced_negotiation()
+        # Propagate the protocol-level default_linemode to the writer so that
+        # TelnetWriter.handle_will(LINEMODE) proposes the correct mode (LOCAL/EDIT)
+        # rather than the TelnetWriter class default (REMOTE).
+        self.writer.default_linemode = self.default_linemode
+        self.writer.iac(DO, LINEMODE)
+
+    def _negotiate_echo(self) -> None:
+        """Skip ``WILL ECHO`` — LINEMODE EDIT client handles local echo."""
+        if self._echo_negotiated:
+            return
+        self._echo_negotiated = True
+
+
 class Server:
     """
     Telnet server that tracks connected clients.