[pull] master from jquast:master

docs/api/color_filter.rst

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

docs/conf.py

@@ -60,17 +60,18 @@
 
 # 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
 # built documents.
 #
 # The short X.Y version.
-version = "2.3"
+version = "2.4"
 
 # The full version, including alpha/beta/rc tags.
-release = "2.3.0"  # keep in sync with pyproject.toml and telnetlib3/accessories.py !!
+release = "2.4.0"  # keep in sync with pyproject.toml and telnetlib3/accessories.py !!
 
 # The language for content auto-generated by Sphinx. Refer to documentation
 # for a list of supported languages.

docs/history.rst

@@ -1,6 +1,43 @@
 History
 =======
 
+2.4.0
+  * 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
   * bugfix: server incorrectly accepted ``DO TSPEED`` and ``DO SNDLOC``

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "telnetlib3"
-version = "2.3.0"
+version = "2.4.0"
 description = " Python Telnet server and client CLI and Protocol library"
 readme = "README.rst"
 license = "ISC"
@@ -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)