Bugfixes and Enhancements from mud fingerprinting (jquast#119)

* enhancement: reversed ``WILL``/``DO`` for directional options (e.g. ``WILL
  NAWS`` from server, ``DO TTYPE`` from client) now gracefully refused with
  ``DONT``/``WONT`` instead of raising ``ValueError``.
* bugfix: ``LINEMODE DO FORWARDMASK`` subnegotiation no longer raises
  ``NotImplementedError``; the mask is accepted and logged at debug level.
* enhancement: ``NEW_ENVIRON SEND`` and response logging improved --
  ``SEND (all)`` / ``env send: (empty)`` instead of raw byte dumps.
* enhancement: ``telnetlib3-fingerprint`` now probes MSDP and MSSP options
  and captures MSSP server status data in session output.
* bugfix: server incorrectly accepted ``DO TSPEED`` and ``DO SNDLOC``
    with ``WILL`` responses. These are client-only options per :rfc:`1079`
    and :rfc:`779`; the server now correctly rejects them.
* bugfix: GMCP, MSDP, and MSSP decoding now uses ``--encoding`` when set,
    falling back to latin-1 for non-UTF-8 bytes instead of lossy replacement.

Author: jquast

docs/history.rst

@@ -1,12 +1,14 @@
 History
 =======
 2.3.0 *unreleased*
-  * bugfix: missing LICENSE.txt in sdist file.
+  * bugfix: repeat "socket.send() raised exception." exceptions
   * bugfix: server incorrectly accepted ``DO TSPEED`` and ``DO SNDLOC``
     with ``WILL`` responses. These are client-only options per :rfc:`1079`
     and :rfc:`779`; the server now correctly rejects them.
-  * bugfix: repeat "socket.send() raised exception." exceptions
+  * bugfix: ``LINEMODE DO FORWARDMASK`` subnegotiation no longer raises
+    ``NotImplementedError``; the mask is accepted (logged only).
   * bugfix: echo doubling in ``--pty-exec`` without ``--pty-raw`` (linemode).
+  * bugfix: missing LICENSE.txt in sdist file.
   * new: :mod:`telnetlib3.mud` module with encode/decode functions for
     GMCP (option 201), MSDP (option 69), and MSSP (option 70) MUD telnet
     protocols.
@@ -22,6 +24,15 @@ History
     for fingerprinting of connected clients.
   * new: ``telnetlib3-fingerprint`` CLI for fingerprinting the given remote
     server, probing telnet option support and capturing banners.
+  * enhancement: reversed ``WILL``/``DO`` for directional options (e.g. ``WILL
+    NAWS`` from server, ``DO TTYPE`` from client) now gracefully refused with
+    ``DONT``/``WONT`` instead of raising ``ValueError``.
+  * enhancement: ``NEW_ENVIRON SEND`` and response logging improved --
+    ``SEND (all)`` / ``env send: (empty)`` instead of raw byte dumps.
+  * bugfix: GMCP, MSDP, and MSSP decoding now uses ``--encoding`` when set,
+    falling back to latin-1 for non-UTF-8 bytes instead of lossy replacement.
+  * enhancement: ``telnetlib3-fingerprint`` now probes MSDP and MSSP options
+    and captures MSSP server status data in session output.
 
 2.2.0
   * bugfix: workaround for Microsoft Telnet client crash on

telnetlib3/fingerprinting.py

@@ -36,6 +36,7 @@
     ECHO,
     GMCP,
     MSDP,
+    MSSP,
     NAMS,
     NAOL,
     NAOP,
@@ -249,7 +250,11 @@ class FingerprintingServer(FingerprintingTelnetServer, TelnetServer):
 # icy_term (icy_net) only accepts option bytes 0-49, 138-140, and 255,
 # returning a hard error for anything else. GMCP-capable MUD clients
 # typically self-announce via IAC WILL GMCP, so probing is unnecessary.
-EXTENDED_OPTIONS = [(GMCP, "GMCP", "Generic MUD Communication Protocol")]
+EXTENDED_OPTIONS = [
+    (GMCP, "GMCP", "Generic MUD Communication Protocol"),
+    (MSDP, "MSDP", "MUD Server Data Protocol"),
+    (MSSP, "MSSP", "MUD Server Status Protocol"),
+]
 
 LEGACY_OPTIONS = [
     (AUTHENTICATION, "AUTHENTICATION", "Telnet authentication"),

telnetlib3/mud.py

@@ -40,6 +40,20 @@
 )
 
 
+def _decode_best_effort(buf: bytes, encoding: str = "utf-8") -> str:
+    """
+    Decode bytes trying *encoding* first, falling back to latin-1.
+
+    :param buf: Raw bytes to decode.
+    :param encoding: Primary encoding to attempt.
+    :returns: Decoded string.
+    """
+    try:
+        return buf.decode(encoding)
+    except (UnicodeDecodeError, LookupError):
+        return buf.decode("latin-1")
+
+
 def gmcp_encode(package: str, data: Any = None) -> bytes:
     """
     Encode a GMCP message.
@@ -53,21 +67,22 @@ def gmcp_encode(package: str, data: Any = None) -> bytes:
     return package.encode("utf-8") + b" " + json.dumps(data, separators=(",", ":")).encode("utf-8")
 
 
-def gmcp_decode(buf: bytes) -> tuple[str, Any]:
+def gmcp_decode(buf: bytes, encoding: str = "utf-8") -> tuple[str, Any]:
     """
     Decode a GMCP payload.
 
     :param buf: GMCP payload bytes
+    :param encoding: Character encoding to try first, falls back to latin-1.
     :returns: Tuple of (package, data), where data is None if no JSON present
     :raises ValueError: If JSON is malformed
     """
     parts = buf.split(b" ", 1)
     if len(parts) == 1:
-        return (buf.decode("utf-8"), None)
+        return (_decode_best_effort(buf, encoding), None)
 
-    package = parts[0].decode("utf-8")
+    package = _decode_best_effort(parts[0], encoding)
     try:
-        data = json.loads(parts[1].decode("utf-8"))
+        data = json.loads(_decode_best_effort(parts[1], encoding))
     except json.JSONDecodeError as exc:
         raise ValueError(f"Invalid JSON in GMCP payload: {exc}") from exc
     return (package, data)
@@ -108,18 +123,19 @@ class MsdpParser:
 
     _DELIMITERS = (MSDP_VAR, MSDP_VAL, MSDP_TABLE_CLOSE, MSDP_ARRAY_CLOSE)
 
-    def __init__(self, buf: bytes) -> None:
+    def __init__(self, buf: bytes, encoding: str = "utf-8") -> None:
         """Initialize parser with raw MSDP buffer."""
         self.buf = buf
         self.idx = 0
+        self.encoding = encoding
 
     def _read_string(self) -> str:
         start = self.idx
         while (
             self.idx < len(self.buf) and self.buf[self.idx : self.idx + 1] not in self._DELIMITERS
         ):
             self.idx += 1
-        return self.buf[start : self.idx].decode("utf-8")
+        return _decode_best_effort(self.buf[start : self.idx], self.encoding)
 
     def _read_key(self) -> str:
         start = self.idx
@@ -128,7 +144,7 @@ def _read_key(self) -> str:
             MSDP_VAR,
         ):
             self.idx += 1
-        return self.buf[start : self.idx].decode("utf-8")
+        return _decode_best_effort(self.buf[start : self.idx], self.encoding)
 
     def _parse_table(self) -> dict[str, Any]:
         table: dict[str, Any] = {}
@@ -181,14 +197,15 @@ def parse(self) -> dict[str, Any]:
         return result
 
 
-def msdp_decode(buf: bytes) -> dict[str, Any]:
+def msdp_decode(buf: bytes, encoding: str = "utf-8") -> dict[str, Any]:
     """
     Decode MSDP wire bytes to dictionary.
 
     :param buf: MSDP payload bytes
+    :param encoding: Character encoding to try first, falls back to latin-1.
     :returns: Dictionary of variable names to values
     """
-    return MsdpParser(buf).parse()
+    return MsdpParser(buf, encoding=encoding).parse()
 
 
 def mssp_encode(variables: dict[str, str | list[str]]) -> bytes:
@@ -209,11 +226,12 @@ def mssp_encode(variables: dict[str, str | list[str]]) -> bytes:
     return result
 
 
-def mssp_decode(buf: bytes) -> dict[str, str | list[str]]:
+def mssp_decode(buf: bytes, encoding: str = "utf-8") -> dict[str, str | list[str]]:
     """
     Decode MSSP wire bytes to dictionary.
 
     :param buf: MSSP payload bytes
+    :param encoding: Character encoding to try first, falls back to latin-1.
     :returns: Dictionary with str values for single entries, list[str] for multiple
     """
     result: dict[str, str | list[str]] = {}
@@ -226,13 +244,13 @@ def mssp_decode(buf: bytes) -> dict[str, str | list[str]]:
             var_start = idx
             while idx < len(buf) and buf[idx : idx + 1] not in (MSSP_VAL, MSSP_VAR):
                 idx += 1
-            current_var = buf[var_start:idx].decode("utf-8")
+            current_var = _decode_best_effort(buf[var_start:idx], encoding)
         elif buf[idx : idx + 1] == MSSP_VAL:
             idx += 1
             val_start = idx
             while idx < len(buf) and buf[idx : idx + 1] not in (MSSP_VAL, MSSP_VAR):
                 idx += 1
-            value = buf[val_start:idx].decode("utf-8")
+            value = _decode_best_effort(buf[val_start:idx], encoding)
 
             if current_var is not None:
                 if current_var in result:

telnetlib3/server_fingerprinting.py

@@ -25,6 +25,7 @@
 from . import fingerprinting as _fps
 from .telopt import (
     VAR,
+    MSSP,
     NAWS,
     LFLOW,
     TTYPE,
@@ -169,6 +170,13 @@ async def _fingerprint_session(
     probe_results = await probe_server_capabilities(writer)
     probe_time = time.time() - probe_start
 
+    # 5b. If server acknowledged MSSP but data hasn't arrived yet, poll briefly
+    if writer.remote_option.enabled(MSSP) and writer.mssp_data is None:
+        for _ in range(10):
+            await asyncio.sleep(0.05)
+            if writer.mssp_data is not None:
+                break
+
     # 6. Build session dicts
     session_data: dict[str, Any] = {
         "encoding": writer.environ_encoding,
@@ -177,6 +185,8 @@ async def _fingerprint_session(
         "banner_after_return": _format_banner(banner_after, encoding=writer.environ_encoding),
         "timing": {"probe": probe_time, "total": time.time() - start_time},
     }
+    if writer.mssp_data is not None:
+        session_data["mssp"] = writer.mssp_data
     session_entry: dict[str, Any] = {
         "host": host,
         "ip": (writer.get_extra_info("peername") or (host,))[0],