more mud scanning improvements (jquast#121)

- Many terminals have custom palettes for colors 0-16 and that's great, but, when connecting to a remote machine we must assume the "usual" colors intended by the remote author, "solarized" and similar color schemes especially screw up artwork on bbs's, so we fix this by remapping those to their exact RGB color, new arguments:
  - telnetlib3-client --colormatch --color-brightness --color-contrast --background-color and --reverse-video.
- support COM-PORT-OPTION, should help with bbs fingerprinting, many of them use serial bridges for their crappy old DOS programs.
- better support misbehaving LINEMODE negotiation and IAC breakdown/freakouts of remote servers (caused usually by using wrong encoding).
- bugfix LINEMODE FORWARDMASK and TTYPE IS exchange in client.
- Experimental new "color palette", to help connect to legacy BBS's, it interprets the colors 0-16 to the likely author-intended* vga-style color codes. Many terminals have "custom" pallettes, and that's grea
- new ``encoding=petscii`` and ``--encoding=atarist``

Author: jquast

docs/api/color_filter.rst

@@ -0,0 +1,7 @@
+color_filter
+------------
+
+.. automodule:: telnetlib3.color_filter
+   :members:
+   :undoc-members:
+   :show-inheritance:

docs/conf.py

@@ -60,7 +60,8 @@
 
 # General information about the project.
 project = "telnetlib3"
-copyright = "2013 Jeff Quast"
+import datetime
+copyright = f"2013-{datetime.datetime.now().year} Jeff Quast"
 
 # The version info for the project you're documenting, acts as replacement for
 # |version| and |release|, also used in various other places throughout the

docs/history.rst

@@ -1,5 +1,41 @@
 History
 =======
+2.4.0 *unreleased*
+  * new: :mod:`telnetlib3.color_filter` module — translates 16-color ANSI SGR
+    codes to 24-bit RGB from hardware palettes (EGA, CGA, VGA, Amiga, xterm).
+    Enabled by default. New client CLI options: ``--colormatch``,
+    ``--color-brightness``, ``--color-contrast``, ``--background-color``,
+    ``--reverse-video``.
+  * new: :func:`~telnetlib3.mud.zmp_decode`,
+    :func:`~telnetlib3.mud.atcp_decode`, and
+    :func:`~telnetlib3.mud.aardwolf_decode` decode functions for ZMP (option
+    93), ATCP (option 200), and Aardwolf (option 102) MUD protocols.
+  * new: :meth:`~telnetlib3.stream_writer.TelnetWriter.handle_zmp`,
+    :meth:`~telnetlib3.stream_writer.TelnetWriter.handle_atcp`,
+    :meth:`~telnetlib3.stream_writer.TelnetWriter.handle_aardwolf`,
+    :meth:`~telnetlib3.stream_writer.TelnetWriter.handle_msp`, and
+    :meth:`~telnetlib3.stream_writer.TelnetWriter.handle_mxp` callbacks for
+    receiving MUD extended protocol subnegotiations, with accumulated data
+    stored in ``zmp_data``, ``atcp_data``, and ``aardwolf_data`` attributes.
+  * new: COM-PORT-OPTION (:rfc:`2217`) subnegotiation parsing with
+    ``comport_data`` attribute and
+    :meth:`~telnetlib3.stream_writer.TelnetWriter.request_comport_signature`.
+  * enhancement: ``telnetlib3-fingerprint`` now always probes extended MUD
+    options (MSP, MXP, ZMP, AARDWOLF, ATCP) during server scans and captures
+    ZMP, ATCP, Aardwolf, MXP, and COM-PORT data in session output.
+  * enhancement: ``telnetlib3-fingerprint`` smart prompt detection —
+    auto-answers yes/no, color, UTF-8 menu, ``who``, and ``help`` prompts.
+  * enhancement: ``--banner-max-bytes`` option for ``telnetlib3-fingerprint``;
+    default raised from 1024 to 65536.
+  * enhancement: new ``--encoding=petscii`` and ``--encoding=atarist``
+  * bugfix: rare LINEMODE ACK loop with misbehaving servers that re-send
+    unchanged MODE without ACK.
+  * bugfix: unknown IAC commands no longer raise ``ValueError``; treated as
+    data.
+  * bugfix: client no longer asserts on ``TTYPE IS`` from server.
+  * bugfix: ``request_forwardmask()`` only called on server side.
+  * change: ``wcwidth`` is now a required dependency.
+
 
 2.3.0
   * bugfix: repeat "socket.send() raised exception." exceptions

pyproject.toml

@@ -43,6 +43,9 @@ classifiers = [
     "Topic :: Terminals :: Telnet",
 ]
 requires-python = ">=3.9"
+dependencies = [
+    "wcwidth>=0.2.13",
+]
 
 [project.optional-dependencies]
 docs = [

telnetlib3/__init__.py

@@ -16,6 +16,7 @@
 from . import stream_reader
 from . import client_base
 from . import client_shell
+from . import color_filter
 from . import client
 from . import telopt
 from . import mud
@@ -26,6 +27,7 @@
 from . import server_fingerprinting
 if sys.platform != "win32":
     from . import fingerprinting_display  # noqa: F401
+from . import encodings  # noqa: F401 - registers custom codecs (petscii, atarist)
 from . import sync
 from .server_base import *  # noqa
 from .server import *  # noqa

telnetlib3/client.py

@@ -528,11 +528,48 @@ def _patched_connection_made(transport: asyncio.BaseTransport) -> None:
 
         client_factory = _client_factory
 
+    # Wrap the shell callback to inject color filter when enabled
+    colormatch: str = args["colormatch"]
+    shell_callback = args["shell"]
+    if colormatch.lower() != "none":
+        # local
+        from .color_filter import (  # pylint: disable=import-outside-toplevel
+            PALETTES,
+            ColorConfig,
+            ColorFilter,
+        )
+
+        if colormatch not in PALETTES:
+            print(
+                f"Unknown palette {colormatch!r}," f" available: {', '.join(sorted(PALETTES))}",
+                file=sys.stderr,
+            )
+            sys.exit(1)
+        color_config = ColorConfig(
+            palette_name=colormatch,
+            brightness=args["color_brightness"],
+            contrast=args["color_contrast"],
+            background_color=args["background_color"],
+            reverse_video=args["reverse_video"],
+        )
+        color_filter = ColorFilter(color_config)
+        original_shell = shell_callback
+
+        async def _color_shell(
+            reader: Union[TelnetReader, TelnetReaderUnicode],
+            writer_arg: Union[TelnetWriter, TelnetWriterUnicode],
+        ) -> None:
+            # pylint: disable-next=protected-access
+            writer_arg._color_filter = color_filter  # type: ignore[union-attr]
+            await original_shell(reader, writer_arg)
+
+        shell_callback = _color_shell
+
     # Build connection kwargs explicitly to avoid pylint false positive
     connection_kwargs: Dict[str, Any] = {
         "encoding": args["encoding"],
         "tspeed": args["tspeed"],
-        "shell": args["shell"],
+        "shell": shell_callback,
         "term": args["term"],
         "force_binary": args["force_binary"],
         "encoding_errors": args["encoding_errors"],
@@ -607,6 +644,43 @@ def _get_argument_parser() -> argparse.ArgumentParser:
         metavar="OPT",
         help="always send DO for this option (name like GMCP or number, repeatable)",
     )
+    parser.add_argument(
+        "--colormatch",
+        default="ega",
+        metavar="PALETTE",
+        help=(
+            "translate basic 16-color ANSI codes to exact 24-bit RGB values"
+            " from a named hardware palette, bypassing the terminal's custom"
+            " palette to preserve intended MUD/BBS artwork colors"
+            " (ega, cga, vga, amiga, xterm, none)"
+        ),
+    )
+    parser.add_argument(
+        "--color-brightness",
+        default=0.9,
+        type=float,
+        metavar="FLOAT",
+        help="color brightness scale [0.0..1.0], where 1.0 is original",
+    )
+    parser.add_argument(
+        "--color-contrast",
+        default=0.8,
+        type=float,
+        metavar="FLOAT",
+        help="color contrast scale [0.0..1.0], where 1.0 is original",
+    )
+    parser.add_argument(
+        "--background-color",
+        default="#101010",
+        metavar="#RRGGBB",
+        help="forced background color as hex RGB (near-black by default)",
+    )
+    parser.add_argument(
+        "--reverse-video",
+        action="store_true",
+        default=False,
+        help="swap foreground/background for light-background terminals",
+    )
     return parser
 
 
@@ -627,6 +701,20 @@ def _parse_option_arg(value: str) -> bytes:
         return bytes([int(value)])
 
 
+def _parse_background_color(value: str) -> Tuple[int, int, int]:
+    """
+    Parse hex color string to RGB tuple.
+
+    :param value: Color string like ``"#RRGGBB"`` or ``"RRGGBB"``.
+    :returns: (R, G, B) tuple with values 0-255.
+    :raises ValueError: When *value* is not a valid hex color.
+    """
+    h = value.lstrip("#")
+    if len(h) != 6:
+        raise ValueError(f"invalid hex color: {value!r}")
+    return (int(h[0:2], 16), int(h[2:4], 16), int(h[4:6], 16))
+
+
 def _transform_args(args: argparse.Namespace) -> Dict[str, Any]:
     return {
         "host": args.host,
@@ -645,6 +733,11 @@ def _transform_args(args: argparse.Namespace) -> Dict[str, Any]:
         "send_environ": tuple(v.strip() for v in args.send_environ.split(",") if v.strip()),
         "always_will": {_parse_option_arg(v) for v in args.always_will},
         "always_do": {_parse_option_arg(v) for v in args.always_do},
+        "colormatch": args.colormatch,
+        "color_brightness": args.color_brightness,
+        "color_contrast": args.color_contrast,
+        "background_color": _parse_background_color(args.background_color),
+        "reverse_video": args.reverse_video,
     }
 
 
@@ -741,6 +834,9 @@ def _get_fingerprint_argument_parser() -> argparse.ArgumentParser:
     parser.add_argument(
         "--banner-max-wait", default=8.0, type=float, help="max seconds to wait for banner data"
     )
+    parser.add_argument(
+        "--banner-max-bytes", default=65536, type=int, help="max bytes per banner read call"
+    )
     return parser
 
 
@@ -778,6 +874,7 @@ async def run_fingerprint_client() -> None:
         mssp_wait=args.mssp_wait,
         banner_quiet_time=args.banner_quiet_time,
         banner_max_wait=args.banner_max_wait,
+        banner_max_bytes=args.banner_max_bytes,
     )
 
     # Parse --always-will/--always-do option names/numbers

telnetlib3/client_base.py

@@ -123,7 +123,7 @@ def connection_lost(self, exc: Optional[Exception]) -> None:
             # the StreamReader will receive eof.
             self._waiter_connected.set_result(None)
 
-        if self.shell is None:
+        if self.shell is None and not self.waiter_closed.done():
             # when a shell is defined, we allow the completion of the coroutine
             # to set the result of waiter_closed.
             self.waiter_closed.set_result(weakref.proxy(self))
@@ -200,6 +200,7 @@ def begin_shell(self, future: asyncio.Future[None]) -> None:
                     lambda fut_obj: (
                         self.waiter_closed.set_result(weakref.proxy(self))
                         if self.waiter_closed is not None
+                        and not self.waiter_closed.done()
                         else None
                     )
                 )

telnetlib3/client_shell.py

@@ -242,6 +242,11 @@ def _on_winch() -> None:
                             if telnet_task in wait_for:
                                 telnet_task.cancel()
                                 wait_for.remove(telnet_task)
+                            _cf = getattr(telnet_writer, "_color_filter", None)
+                            if _cf is not None:
+                                _flush = _cf.flush()
+                                if _flush:
+                                    stdout.write(_flush.encode())
                             stdout.write(f"\033[m{linesep}Connection closed.{linesep}".encode())
                             # Cleanup resize handler on local escape close
                             if term._istty and remove_winch:  # pylint: disable=protected-access
@@ -273,6 +278,11 @@ def _on_winch() -> None:
                         if stdin_task in wait_for:
                             stdin_task.cancel()
                             wait_for.remove(stdin_task)
+                        _cf = getattr(telnet_writer, "_color_filter", None)
+                        if _cf is not None:
+                            _flush = _cf.flush()
+                            if _flush:
+                                stdout.write(_flush.encode())
                         stdout.write(
                             f"\033[m{linesep}Connection closed by foreign host.{linesep}".encode()
                         )
@@ -289,6 +299,9 @@ def _on_winch() -> None:
                             except Exception:  # pylint: disable=broad-exception-caught
                                 pass
                     else:
+                        _cf = getattr(telnet_writer, "_color_filter", None)
+                        if _cf is not None:
+                            out = _cf.filter(out)
                         stdout.write(out.encode() or b":?!?:")
                         telnet_task = accessories.make_reader_task(telnet_reader, size=2**24)
                         wait_for.add(telnet_task)