Broadcast TLS information in demo banners with API

Author: jquast

bin/server_mud.py

@@ -763,7 +763,12 @@ async def shell(reader: Any, writer: Any) -> None:
     writer.iac(WILL, MSSP)
     writer.set_ext_callback(GMCP, lambda pkg, data: on_gmcp(writer, pkg, data))
     writer.set_ext_callback(MSDP, lambda variables: on_msdp(writer, variables))
-    writer.write("Welcome to the Mini-MUD!\r\n")
+    ssl_obj = writer.get_extra_info("ssl_object")
+    if ssl_obj is not None:
+        version = ssl_obj.version() or "TLS"
+        writer.write(f"Welcome to the Mini-MUD! ({version} Secured)\r\n")
+    else:
+        writer.write("Welcome to the Mini-MUD!\r\n")
 
     env = writer.get_extra_info("USER") or writer.get_extra_info("LOGNAME") or ""
     if env:

bin/server_tls.py

@@ -23,7 +23,12 @@
 
 async def shell(reader, writer):
     """Simple echo shell over TLS."""
-    writer.write("Welcome to the TLS echo server!\r\n")
+    ssl_obj = writer.get_extra_info("ssl_object")
+    if ssl_obj is not None:
+        version = ssl_obj.version() or "TLS"
+        writer.write(f"Welcome to the TLS echo server! ({version} Secured)\r\n")
+    else:
+        writer.write("Welcome to the echo server!\r\n")
     await writer.drain()
 
     while True:

docs/history.rst

@@ -8,6 +8,8 @@ History
     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: ``writer.get_extra_info("ssl_object")`` is available in shell callbacks to detect
+    TLS-secured connections and query the negotiated protocol version and cipher.
   * enhancement: ``telnetlib3-fingerprint-server`` integrates with the optional ``tv-detect``
     package for terminal vulnerability probing.
 

telnetlib3/_base.py

@@ -125,3 +125,15 @@ def get_extra_info(self, name: str, default: Any = None) -> Any:
         if self._transport:
             default = self._transport.get_extra_info(name, default)
         return self._extra.get(name, default)
+
+    def _log_tls_info(self, logger: logging.Logger) -> None:
+        """Log TLS connection details at debug level, if TLS is active."""
+        ssl_obj = self.get_extra_info("ssl_object")
+        if ssl_obj is None:
+            return
+        version = getattr(ssl_obj, "version", lambda: None)() or "TLS"
+        cipher_info = getattr(ssl_obj, "cipher", lambda: None)()
+        if cipher_info:
+            logger.debug("TLS handshake: %s cipher=%s", version, cipher_info[0])
+        else:
+            logger.debug("TLS handshake: %s", version)

telnetlib3/client_base.py

@@ -186,6 +186,7 @@ def connection_made(self, transport: asyncio.BaseTransport) -> None:
         )
 
         self.log.info("Connected to %s", self)
+        self._log_tls_info(self.log)
 
         self._waiter_connected.add_done_callback(self.begin_shell)
         asyncio.get_event_loop().call_soon(self.begin_negotiation)

telnetlib3/server.py

@@ -853,6 +853,9 @@ async def _upgrade_to_tls(self) -> None:
                 self._transport.close()
             return
         assert ssl_transport is not None
+        logger.debug(
+            "tls-auto: TLS handshake succeeded for %s", self._transport.get_extra_info("peername")
+        )
         protocol.connection_made(ssl_transport)
 
     def _handoff_plain(self) -> None:

telnetlib3/server_base.py

@@ -171,6 +171,7 @@ def connection_made(self, transport: asyncio.BaseTransport) -> None:
         )
 
         logger.info("Connection from %s", self)
+        self._log_tls_info(logger)
 
         self._waiter_connected.add_done_callback(self.begin_shell)
         asyncio.get_event_loop().call_soon(self.begin_negotiation)

telnetlib3/tests/test_server.py

@@ -459,3 +459,41 @@ def close(self):
     data = b"plain text no iac"
     server.data_received(data)
     assert bytes(server.reader._buffer).endswith(data)
+
+
+@pytest.mark.parametrize(
+    "ssl_obj,expect_log",
+    [
+        pytest.param(None, False, id="no_tls"),
+        pytest.param(
+            type(
+                "SSL",
+                (),
+                {"version": lambda self: "TLSv1.3", "cipher": lambda self: ("AES", "TLSv1.3", 256)},
+            )(),
+            True,
+            id="tls_with_cipher",
+        ),
+        pytest.param(
+            type("SSL", (), {"version": lambda self: None, "cipher": lambda self: None})(),
+            True,
+            id="tls_no_cipher",
+        ),
+    ],
+)
+def test_log_tls_info(ssl_obj, expect_log, caplog):
+    from telnetlib3._base import TelnetProtocolBase
+
+    proto = TelnetProtocolBase.__new__(TelnetProtocolBase)
+    proto._extra = {}
+    proto._transport = MagicMock()
+    proto._transport.get_extra_info = lambda name, default=None: (
+        ssl_obj if name == "ssl_object" else default
+    )
+    log = logging.getLogger("test_tls_info")
+    with caplog.at_level(logging.DEBUG, logger="test_tls_info"):
+        proto._log_tls_info(log)
+    tls_msgs = [r for r in caplog.records if "TLS handshake" in r.message]
+    assert bool(tls_msgs) == expect_log
+    if ssl_obj and ssl_obj.cipher():
+        assert "AES" in tls_msgs[0].message

telnetlib3/tests/test_tls.py

@@ -725,3 +725,32 @@ def test_cli_run_fingerprint_client_ssl(
     ca.cert_pem.write_to_path(ca_pem)
     extra_argv = extra_argv_factory(ca_pem)
     _pty_run_fingerprint(bind_host, unused_tcp_port, extra_argv, server_ssl_ctx)
+
+
+@_start_tls_xfail
+@_start_tls_timeout
+async def test_tls_handshake_logged(
+    bind_host, unused_tcp_port, server_ssl_ctx, client_ssl_ctx, caplog
+):
+    """Successful TLS handshake logs version and cipher at debug level."""
+    shell, waiter = _echo_shell("hi", "ok")
+    import logging
+
+    with caplog.at_level(logging.DEBUG):
+        async with create_server(
+            host=bind_host, port=unused_tcp_port, shell=shell, ssl=server_ssl_ctx
+        ):
+            async with open_connection(
+                bind_host,
+                unused_tcp_port,
+                ssl=client_ssl_ctx,
+                server_hostname="localhost",
+                **_FAST_CLIENT,
+            ) as (reader, writer):
+                writer.write("hi")
+                await writer.drain()
+                await asyncio.wait_for(waiter, 2.0)
+
+    tls_logs = [r for r in caplog.records if "TLS handshake" in r.message]
+    assert len(tls_logs) >= 1
+    assert any("TLSv1" in r.message for r in tls_logs)