Bugfix MCCP2 decompression, very large NEW_ENVIRON, and --tls-auto (jquast#138)

  * bugfix: MCCP2 decompression failed on MUD servers using raw deflate or gzip-wrapped compression,
    producing garbled banners.  The client now auto-detects zlib/gzip format and falls back to raw
    deflate when needed.
  * bugfix: ``NEW_ENVIRON SEND`` requests that exceed the 256-byte subnegotiation buffer of some
    telnet clients (e.g. GNU inetutils) are now automatically split into multiple SB frames.
  * bugfix: ``telnetlib3-server`` argument ``--tls-auto`` deadlocked with plain telnet clients.
    Detection now uses a non-blocking with a configurable timeout.
  * enhancement: ``telnetlib3-fingerprint-server`` integrates with the optional ``tv-detect``
    package for terminal vulnerability probing.

Author: jquast

.github/FUNDING.yml

@@ -0,0 +1 @@
+github: jquast

bin/moderate_fingerprints.py

@@ -127,7 +127,9 @@ def _alarm_handler(signum, frame):
 
 
 def _format_banner(banner_data):
-    """Return (clean_text, raw_display) from a banner dict."""
+    """Return (clean_text, raw_display) from a banner dict or string."""
+    if isinstance(banner_data, str):
+        return banner_data, ""
     text = banner_data.get("text", "")
     raw_hex = banner_data.get("raw_hex", "")
     if _HAS_WCWIDTH and text:

docs/guidebook.rst

@@ -369,6 +369,28 @@ Or programmatically::
 
 For production, use certificates from Let's Encrypt or another trusted CA.
 
+**Mixed TLS / plain telnet (auto-detect)**
+
+Add ``--tls-auto`` to accept both TLS and plain telnet clients on the same
+port.  The server peeks at the first byte of each connection: a TLS
+ClientHello (``0x16``) upgrades to TLS; any other byte is processed as plain telnet.  An optional
+timeout::
+
+    telnetlib3-server --ssl-certfile cert.pem --ssl-keyfile key.pem \
+        --tls-auto --line-mode --shell bin.server_mud.shell 0.0.0.0 6023
+
+    # custom timeout (seconds to wait for TLS ClientHello)
+    telnetlib3-server --ssl-certfile cert.pem --ssl-keyfile key.pem \
+        --tls-auto=2.0 0.0.0.0 6023
+
+TLS clients send ClientHello immediately on connect, so they should be detected
+quickly under the default timeout of 500ms, though very slow networks might suggest to increase the
+default timeout value by specifying time in seconds as float, eg. ``--tls-auto=2.0``.
+
+RFC-compliant telnet clients wait for the server to send the first bytes, so they will appear to
+stall during this timeout unless they initiate telnet negotiation or any client input (other than
+Ctrl+V!) is received.
+
 **Client-side**
 
 Connect to a server with a CA-signed certificate (e.g. ``dunemud.net``)::

docs/history.rst

@@ -1,5 +1,16 @@
 History
 =======
+4.0.2 (unreleased)
+  * bugfix: MCCP2 decompression failed on MUD servers using raw deflate or gzip-wrapped compression,
+    producing garbled banners.  The client now auto-detects zlib/gzip format and falls back to raw
+    deflate when needed.
+  * bugfix: ``NEW_ENVIRON SEND`` requests that exceed the 256-byte subnegotiation buffer of some
+    telnet clients (e.g. GNU inetutils) are now automatically split into multiple SB frames.
+  * bugfix: ``telnetlib3-server`` argument ``--tls-auto`` deadlocked with plain telnet clients.
+    Detection now uses a non-blocking with a configurable timeout.
+  * enhancement: ``telnetlib3-fingerprint-server`` integrates with the optional ``tv-detect``
+    package for terminal vulnerability probing.
+
 4.0.1
   * new: ``--encoding=big5bbs``, BBS 半形字 (half-width characters) encoding, matching PCMan/PttBBS
     terminal clients, popular with Taiwanese BBS culture.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "telnetlib3"
-version = "4.0.1"  # Keep in sync with telnetlib3/accessories.py::get_version !
+version = "4.0.2"  # Keep in sync with telnetlib3/accessories.py::get_version !
 description = " Python Telnet server and client CLI and Protocol library"
 readme = "README.rst"
 license = "ISC"
@@ -55,7 +55,7 @@ docs = [
     "sphinx-autodoc-typehints",
 ]
 extras = [
-    "prettytable",
+    "prettytable>=3.17,<4",
     "ucs-detect>=2,<3",
 ]
 

telnetlib3/accessories.py

@@ -42,7 +42,7 @@
 
 def get_version() -> str:
     """Return the current version of telnetlib3."""
-    return "4.0.1"  # keep in sync with pyproject.toml !
+    return "4.0.2"  # keep in sync with pyproject.toml !
 
 
 def encoding_from_lang(lang: str) -> Optional[str]:
@@ -124,9 +124,7 @@ def hexdump(data: bytes, prefix: str = "") -> str:
     return "\n".join(lines)
 
 
-_DEFAULT_LOGFMT = " ".join(
-    ("%(asctime)s", "%(levelname)s", "%(filename)s:%(lineno)d", "%(message)s")
-)
+_DEFAULT_LOGFMT = " ".join(("%(levelname)s", "%(filename)s:%(lineno)d", "%(message)s"))
 
 
 def make_logger(

telnetlib3/client_base.py

@@ -69,6 +69,7 @@ def __init__(
 
         # MCCP2: server→client decompression
         self._mccp2_decompressor: Optional[zlib._Decompress] = None
+        self._mccp2_wbits_fallback: bool = False
         # MCCP3: client→server compression
         self._mccp3_compressor: Optional[zlib._Compress] = None
         self._mccp3_orig_write: Any = None
@@ -102,6 +103,7 @@ def connection_lost(self, exc: Optional[Exception]) -> None:
 
         # Clean up MCCP compressors/decompressors
         self._mccp2_decompressor = None
+        self._mccp2_wbits_fallback = False
         self._mccp3_compressor = None
         self._mccp3_orig_write = None
 
@@ -360,9 +362,20 @@ def _process_chunk(self, data: bytes) -> bool:
             try:
                 data = self._mccp2_decompressor.decompress(data)
             except zlib.error:
-                self.log.warning("MCCP2 decompression error, disabling")
-                self._mccp2_end()
-                return False
+                if not self._mccp2_wbits_fallback:
+                    self.log.debug("MCCP2 auto-detect failed, retrying raw deflate")
+                    self._mccp2_wbits_fallback = True
+                    self._mccp2_decompressor = zlib.decompressobj(wbits=-zlib.MAX_WBITS)
+                    try:
+                        data = self._mccp2_decompressor.decompress(data)
+                    except zlib.error:
+                        self.log.warning("MCCP2 decompression error, disabling")
+                        self._mccp2_end()
+                        return False
+                else:
+                    self.log.warning("MCCP2 decompression error, disabling")
+                    self._mccp2_end()
+                    return False
             if self._mccp2_decompressor.eof:
                 unused = self._mccp2_decompressor.unused_data
                 self._mccp2_end()
@@ -444,7 +457,8 @@ async def _process_rx(self) -> None:
 
     def _mccp2_start(self) -> None:
         """Start MCCP2 decompression of server→client data."""
-        self._mccp2_decompressor = zlib.decompressobj()
+        self._mccp2_decompressor = zlib.decompressobj(wbits=zlib.MAX_WBITS | 32)
+        self._mccp2_wbits_fallback = False
         self.log.debug("MCCP2 decompression started (server→client)")
 
     def _mccp2_end(self) -> None: