jq/telnetlib3 fingerprint server (jquast#117)

* new: ``telnetlib3-fingerprint-server`` CLI with extended ``NEW_ENVIRON``
  for client fingerprinting (uses ``FingerprintingServer`` protocol factory).

Author: jquast

bin/server_mud.py

@@ -6,6 +6,7 @@
     $ python bin/server_mud.py
     $ telnet localhost 6023
 """
+
 from __future__ import annotations
 
 # std imports
@@ -858,8 +859,6 @@ async def main(argv: list[str] | None = None) -> None:
         f"{SERVER_NAME} running on"
         f" {args.host}:{args.port}\n"
         f"Connect with: telnet {args.host} {args.port}\n"
-        "Or use GMCP client:"
-        " python bin/client_gmcp.py\n"
         "Press Ctrl+C to stop"
     )
     await server.wait_closed()

docs/guidebook.rst

@@ -471,17 +471,21 @@ Install with optional dependencies for full fingerprinting support
 
     pip install telnetlib3[extras]
 
+A dedicated CLI entry point is provided::
 
-::
+    telnetlib3-fingerprint-server --data-dir data
 
-    TELNETLIB3_DATA_DIR=data telnetlib3-server --shell telnetlib3.fingerprinting_server_shell
+This uses :class:`~telnetlib3.fingerprinting.FingerprintingServer` as the
+protocol factory and :func:`~telnetlib3.fingerprinting.fingerprinting_server_shell`
+as the default shell. All ``telnetlib3-server`` options (``--host``, ``--port``,
+etc.) are accepted.
 
 Storage
 -------
 
 Results are saved as JSON files organized by fingerprint hash::
 
-    $TELNETLIB3_DATA_DIR/client/<telnet-hash>/<terminal-hash>/
+    <data-dir>/client/<telnet-hash>/<terminal-hash>/
 
 Moderating
 ----------

docs/history.rst

@@ -9,6 +9,10 @@ History
   * new: ``connect_timeout`` arguments for client and ``--connect-timeout``
     Client CLI argument, :ghissue:`30`.
   * bugfix: missing LICENSE.txt in sdist file.
+  * new: ``telnetlib3-fingerprint-server`` CLI with extended ``NEW_ENVIRON``
+    for client fingerprinting (uses ``FingerprintingServer`` protocol factory).
+  * note: fingerprint hashes for MUD clients detected via GMCP/MSDP
+    may change due to improved client classification.
 
 2.2.0
   * bugfix: workaround for Microsoft Telnet client crash on

pyproject.toml

@@ -58,6 +58,7 @@ extras = [
 [project.scripts]
 telnetlib3-server = "telnetlib3.server:main"
 telnetlib3-client = "telnetlib3.client:main"
+telnetlib3-fingerprint-server = "telnetlib3.fingerprinting:fingerprint_server_main"
 
 [project.urls]
 Homepage = "https://github.com/jquast/telnetlib3"

telnetlib3/fingerprinting.py

@@ -16,11 +16,13 @@
 import asyncio
 import hashlib
 import logging
+import argparse
 import datetime
 from typing import Any, Dict, List, Tuple, Union, Callable, Optional, cast
 
 # local
 from . import slc
+from .server import TelnetServer  # pylint: disable=cyclic-import
 from .telopt import (
     BM,
     DO,
@@ -120,7 +122,9 @@
 
 __all__ = (
     "ENVIRON_EXTENDED",
+    "FingerprintingServer",
     "FingerprintingTelnetServer",
+    "fingerprint_server_main",
     "fingerprinting_server_shell",
     "fingerprinting_post_script",
     "get_client_fingerprint",
@@ -190,6 +194,19 @@ def on_request_environ(self) -> list[Union[str, bytes]]:
         return base[:insert_at] + extra + base[insert_at:]
 
 
+class FingerprintingServer(FingerprintingTelnetServer, TelnetServer):
+    """
+    :class:`~telnetlib3.server.TelnetServer` with extended ``NEW_ENVIRON``.
+
+    Combines :class:`FingerprintingTelnetServer` with :class:`~telnetlib3.server.TelnetServer`
+    so that :func:`fingerprinting_server_shell` receives the full set of
+    environment variables needed for stable fingerprint hashes.
+
+    Used as the default ``protocol_factory`` by
+    :func:`fingerprint_server_main` / ``telnetlib3-fingerprint-server`` CLI.
+    """
+
+
 # Timeout for probe_client_capabilities in _run_probe (seconds)
 _PROBE_TIMEOUT = 0.5
 
@@ -993,6 +1010,44 @@ def fingerprinting_post_script(filepath: str) -> None:
     _fps(filepath)
 
 
+def fingerprint_server_main() -> None:
+    """
+    Entry point for ``telnetlib3-fingerprint-server`` CLI.
+
+    Reuses :func:`~telnetlib3.server.parse_server_args` and
+    :func:`~telnetlib3.server.run_server` with
+    :class:`FingerprintingServer` as the default protocol factory
+    and :func:`fingerprinting_server_shell` as the default shell.
+
+    Accepts ``--data-dir`` to set the fingerprint data directory.
+    Falls back to the ``TELNETLIB3_DATA_DIR`` environment variable.
+    """
+    # pylint: disable=import-outside-toplevel,global-statement
+    # local import is required to prevent circular imports
+    # local
+    from .server import _config, run_server, parse_server_args  # noqa: PLC0415
+
+    global DATA_DIR
+    # Extract --data-dir before parse_server_args() sees argv.
+    pre = argparse.ArgumentParser(add_help=False)
+    pre.add_argument(
+        "--data-dir",
+        default=None,
+        help="directory for fingerprint data" " (default: $TELNETLIB3_DATA_DIR)",
+    )
+    pre_args, remaining = pre.parse_known_args()
+    sys.argv[1:] = remaining
+
+    if pre_args.data_dir is not None:
+        DATA_DIR = pre_args.data_dir
+
+    args = parse_server_args()
+    if args["shell"] is _config.shell:
+        args["shell"] = fingerprinting_server_shell
+    args["protocol_factory"] = FingerprintingServer
+    asyncio.run(run_server(**args))
+
+
 def main() -> None:
     """CLI entry point for fingerprinting post-processing."""
     if len(sys.argv) != 2:

telnetlib3/server.py

@@ -1046,6 +1046,7 @@ async def run_server(  # pylint: disable=too-many-positional-arguments,too-many-
     pty_fork_limit: int = _config.pty_fork_limit,
     status_interval: int = _config.status_interval,
     never_send_ga: bool = _config.never_send_ga,
+    protocol_factory: Optional[Type[asyncio.Protocol]] = None,
 ) -> None:
     """
     Program entry point for server daemon.
@@ -1125,6 +1126,7 @@ async def guarded_shell(
         host,
         port,
         shell=shell,
+        protocol_factory=protocol_factory,
         encoding=encoding,
         force_binary=force_binary,
         never_send_ga=never_send_ga,

telnetlib3/tests/test_encoding.py

@@ -109,9 +109,7 @@ async def test_telnet_client_and_server_encoding_bidirectional(bind_host, unused
 
 async def test_telnet_server_encoding_by_LANG(bind_host, unused_tcp_port):
     """Server's encoding negotiated by LANG value."""
-    async with create_server(
-        host=bind_host, port=unused_tcp_port, connect_maxwait=0.5
-    ) as server:
+    async with create_server(host=bind_host, port=unused_tcp_port, connect_maxwait=0.5) as server:
         async with asyncio_connection(bind_host, unused_tcp_port) as (reader, writer):
             writer.write(IAC + DO + BINARY)
             writer.write(IAC + WILL + BINARY)
@@ -136,9 +134,7 @@ async def test_telnet_server_encoding_by_LANG(bind_host, unused_tcp_port):
 
 async def test_telnet_server_encoding_LANG_no_encoding_suffix(bind_host, unused_tcp_port):
     """Server falls back to default when LANG has no encoding suffix."""
-    async with create_server(
-        host=bind_host, port=unused_tcp_port, connect_maxwait=0.5
-    ) as server:
+    async with create_server(host=bind_host, port=unused_tcp_port, connect_maxwait=0.5) as server:
         async with asyncio_connection(bind_host, unused_tcp_port) as (reader, writer):
             writer.write(IAC + DO + BINARY)
             writer.write(IAC + WILL + BINARY)
@@ -161,9 +157,7 @@ async def test_telnet_server_encoding_LANG_no_encoding_suffix(bind_host, unused_
 
 async def test_telnet_server_encoding_LANG_invalid_encoding(bind_host, unused_tcp_port):
     """Server falls back to default when LANG has unknown encoding."""
-    async with create_server(
-        host=bind_host, port=unused_tcp_port, connect_maxwait=0.5
-    ) as server:
+    async with create_server(host=bind_host, port=unused_tcp_port, connect_maxwait=0.5) as server:
         async with asyncio_connection(bind_host, unused_tcp_port) as (reader, writer):
             writer.write(IAC + DO + BINARY)
             writer.write(IAC + WILL + BINARY)

telnetlib3/tests/test_fingerprinting.py

@@ -1006,4 +1006,87 @@ def test_process_client_fingerprint_skips_ucs_detect_for_mud(monkeypatch, tmp_pa
         fpd._process_client_fingerprint(filepath, data)
     except (ImportError, AttributeError, TypeError):
         pass
+    capsys.readouterr()
     assert not ucs_called
+
+
+def test_protocol_fingerprint_hash_stability():
+    """Hash must not change across releases for the same probe data."""
+    w = MockWriter(
+        extra={"TERM": "xterm", "HOME": "/home/user", "USER": "alice", "SHELL": "/bin/bash"}
+    )
+    w._protocol = MockProtocol({"HOME": "/home/user", "USER": "alice", "SHELL": "/bin/bash"})
+    probe = {
+        "BINARY": {"status": "WILL", "opt": fps.BINARY},
+        "TTYPE": {"status": "WILL", "opt": fps.TTYPE},
+        "SGA": {"status": "WONT", "opt": fps.SGA},
+    }
+    fp = fps._create_protocol_fingerprint(w, probe)
+    assert fps._hash_fingerprint(fp) == "426327fe80f38c2c"
+
+
+def test_fingerprinting_server_on_request_environ():
+    """FingerprintingServer includes HOME and SHELL in environ request."""
+    srv = fps.FingerprintingServer.__new__(fps.FingerprintingServer)
+    env = srv.on_request_environ()
+    assert "HOME" in env
+    assert "SHELL" in env
+    assert "USER" in env
+
+
+def test_fingerprint_server_shell_has_no_protocol_factory():
+    """Shell is a plain callback, not annotated with protocol_factory."""
+    assert not hasattr(fps.fingerprinting_server_shell, "protocol_factory")
+
+
+def test_fingerprint_server_main_exists():
+    """Entry point function is importable."""
+    assert callable(fps.fingerprint_server_main)
+
+
+def _noop_asyncio_run(coro):
+    """Discard a coroutine without running it (avoids RuntimeWarning)."""
+    coro.close()
+
+
+def test_fingerprint_server_main_data_dir_flag(tmp_path, monkeypatch):
+    """--data-dir sets DATA_DIR and passes remaining args through."""
+    data_dir = str(tmp_path / "fp-data")
+    monkeypatch.setattr(sys, "argv", ["prog", "--data-dir", data_dir, "127.0.0.1", "9999"])
+    monkeypatch.setattr("telnetlib3.fingerprinting.asyncio.run", _noop_asyncio_run)
+
+    captured: dict = {}
+    # local
+    from telnetlib3.server import parse_server_args
+
+    original_parse = parse_server_args
+
+    def patched_parse() -> dict:
+        result = original_parse()
+        captured.update(result)
+        return result
+
+    monkeypatch.setattr("telnetlib3.server.parse_server_args", patched_parse)
+
+    old_data_dir = fps.DATA_DIR
+    try:
+        fps.fingerprint_server_main()
+        assert fps.DATA_DIR == data_dir
+        assert captured["host"] == "127.0.0.1"
+        assert captured["port"] == 9999
+    finally:
+        fps.DATA_DIR = old_data_dir
+
+
+def test_fingerprint_server_main_env_fallback(monkeypatch):
+    """DATA_DIR unchanged when --data-dir is not provided."""
+    monkeypatch.setattr(sys, "argv", ["prog"])
+    monkeypatch.setattr("telnetlib3.fingerprinting.asyncio.run", _noop_asyncio_run)
+
+    old_data_dir = fps.DATA_DIR
+    try:
+        fps.DATA_DIR = "/original"
+        fps.fingerprint_server_main()
+        assert fps.DATA_DIR == "/original"
+    finally:
+        fps.DATA_DIR = old_data_dir