MTTS fallback for CHARSET negotiation (jquast#116)

* Add MTTS fallback for CHARSET negotiation

When a client sends CHARSET REJECTED but has advertised UTF-8 support
via MTTS (MUD Terminal Type Standard), resolve encoding to UTF-8
immediately instead of waiting for connect_maxwait timeout.

MTTS can be sent via NEW_ENVIRON (MTTS=2825) or TTYPE round 3
(MTTS 2825). Bit 3 (value 8) indicates UTF-8 capability.

* Add tests for MTTS charset fallback

Test that when CHARSET REJECTED is received:
- UTF-8 is resolved if MTTS bit 3 is set
- No resolution occurs if MTTS lacks UTF-8 bit
- Both NEW_ENVIRON and TTYPE sources work

* Add deferred MTTS fallback for CHARSET rejection timing

CHARSET REJECTED can arrive before NEW_ENVIRON or TTYPE round 3
data is available. Set _charset_rejected flag in stream_writer's
REJECTED handler, then retry _check_mtts_for_utf8() in two
deferred sites: server.on_environ() when MTTS key arrives, and
server.on_ttype() when ttype3 contains MTTS data.

Also improves _check_mtts_for_utf8 docstring with spec URL and
adds debug logging for malformed MTTS values.

* Fix MTTS charset fallback tests and add edge cases

Fix protocol_factory usage (class not lambda), waiter pattern
(set_result(self) for protocol instance), client method names
(send_env/send_ttype), and negative test pattern (pytest.raises
TimeoutError instead of asyncio.sleep). Remove debug prints.

Add unit tests for _check_mtts_for_utf8 edge cases: environ
precedence over ttype3, MTTS=0, and malformed MTTS values.

* Resolve encoding wait when MTTS confirms UTF-8

The deferred MTTS checks were calling on_charset() but not unblocking
the encoding or negotiation waiters. check_negotiation() requires
_check_encoding() to return True, which gates on BINARY mode — but
MUD clients like tintin++ reject BINARY while advertising UTF-8 via
MTTS bit 3.

Set force_binary=True when MTTS confirms UTF-8 (the MUD world's
replacement for RFC 856 BINARY negotiation) and resolve
waiter_encoding directly. This lets check_negotiation() complete
immediately instead of waiting for connect_maxwait timeout.

Author: shmup

telnetlib3/server.py

@@ -509,6 +509,22 @@ def on_environ(self, mapping: Dict[str, str]) -> None:
 
         self._extra.update(u_mapping)
 
+        # Deferred MTTS fallback: CHARSET REJECTED may arrive before
+        # NEW_ENVIRON IS, so the immediate check in stream_writer finds
+        # nothing. Now that environ data is available, retry.
+        if "MTTS" in u_mapping and getattr(self.writer, "_charset_rejected", False):
+            charset = self.writer._check_mtts_for_utf8()
+            if charset:
+                logger.debug("MTTS indicates UTF-8 (deferred), resolving to UTF-8")
+                self.on_charset(charset)
+                self.writer._charset_rejected = False
+                # MTTS bit 3 is the MUD world's replacement for BINARY
+                # negotiation — the client is declaring UTF-8 capability
+                # regardless of RFC 856 BINARY mode.
+                self.force_binary = True
+                if not self.waiter_encoding.done():
+                    self.waiter_encoding.set_result(True)
+
     def on_request_charset(self) -> List[str]:
         """
         Definition for CHARSET request by client, :rfc:`2066`.
@@ -600,6 +616,20 @@ def on_ttype(self, ttype: str) -> None:
             self._extra["TERM"] = val
             self._negotiate_environ()
 
+            # Deferred MTTS fallback: CHARSET REJECTED may arrive before
+            # TTYPE round 3 completes, so the immediate check in
+            # stream_writer finds nothing. Now that ttype3 is stored,
+            # retry the MTTS check.
+            if getattr(self.writer, "_charset_rejected", False):
+                charset = self.writer._check_mtts_for_utf8()
+                if charset:
+                    logger.debug("MTTS indicates UTF-8 (deferred from ttype3)")
+                    self.on_charset(charset)
+                    self.writer._charset_rejected = False
+                    self.force_binary = True
+                    if not self.waiter_encoding.done():
+                        self.waiter_encoding.set_result(True)
+
         elif ttype == _lastval:
             logger.debug("ttype cycle stop at %s: %s, repeated.", key, ttype)
             self._negotiate_environ()

telnetlib3/stream_writer.py

@@ -1866,6 +1866,35 @@ def _write(self, buf: bytes, escape_iac: bool = True) -> None:
 
     # Private sub-negotiation (SB) routines
 
+    def _check_mtts_for_utf8(self) -> str | None:
+        """
+        Check MTTS (MUD Terminal Type Standard) data for UTF-8 support.
+
+        Called as a fallback when CHARSET negotiation is rejected by the client.
+        MTTS is a bitmask where bit 3 (value 8) indicates UTF-8 capability.
+        Returns "UTF-8" if supported, None otherwise.
+
+        https://tintin.sourceforge.io/protocols/mtts/
+        """
+        # Try to get MTTS from NEW_ENVIRON (e.g., "MTTS=2825")
+        mtts_str = self._protocol.get_extra_info("MTTS")
+        if not mtts_str:
+            # Try TTYPE round 3 (e.g., "MTTS 2825")
+            ttype3 = self._protocol.get_extra_info("ttype3")
+            if ttype3 and ttype3.upper().startswith("MTTS "):
+                mtts_str = ttype3[5:].strip()
+
+        if mtts_str:
+            try:
+                mtts_value = int(mtts_str)
+                # Bit 3 (value 8) indicates UTF-8 support
+                if mtts_value & 8:
+                    return "UTF-8"
+            except (ValueError, TypeError):
+                self.log.debug("Failed to parse MTTS value: %r", mtts_str)
+
+        return None
+
     def _handle_sb_charset(self, buf: collections.deque[bytes]) -> None:
         cmd = buf.popleft()
         assert cmd == CHARSET
@@ -1894,6 +1923,18 @@ def _handle_sb_charset(self, buf: collections.deque[bytes]) -> None:
             self._ext_callback[CHARSET](charset)
         elif opt == REJECTED:
             self.log.warning("recv IAC SB CHARSET REJECTED IAC SE")
+            # After CHARSET rejection, check MTTS for UTF-8 support.
+            # TTYPE round 3 data (ttype3) is available here since TTYPE
+            # cycling completes before CHARSET REQUEST is sent. However,
+            # NEW_ENVIRON data may not have arrived yet — server.on_environ()
+            # handles that case with a deferred check.
+            if self.server and CHARSET in self._ext_callback:
+                charset_from_mtts = self._check_mtts_for_utf8()
+                if charset_from_mtts:
+                    self.log.debug("MTTS indicates UTF-8 support, resolving to UTF-8")
+                    self._ext_callback[CHARSET](charset_from_mtts)
+                else:
+                    self._charset_rejected = True
         elif opt in (TTABLE_IS, TTABLE_ACK, TTABLE_NAK, TTABLE_REJECTED):
             raise NotImplementedError(
                 f"Translation table command received but not supported: {opt!r}"

telnetlib3/tests/test_charset.py

@@ -4,6 +4,9 @@
 import asyncio
 import collections
 
+# 3rd party
+import pytest
+
 # local
 import telnetlib3
 import telnetlib3.stream_writer
@@ -451,3 +454,278 @@ async def test_charset_explicit_non_latin1_encoding(bind_host, unused_tcp_port):
             connect_maxwait=0.25,
         ) as (reader, writer):
             assert writer.protocol.encoding(incoming=True) == "US-ASCII"
+
+
+async def test_charset_rejected_with_mtts_utf8(bind_host, unused_tcp_port):
+    """Test CHARSET REJECTED fallback to UTF-8 when MTTS indicates support."""
+    _waiter = asyncio.Future()
+
+    class ServerTestCharsetMTTS(telnetlib3.TelnetServer):
+        def on_charset(self, charset):
+            super().on_charset(charset)
+            if not _waiter.done():
+                _waiter.set_result(self)
+
+    # Create a client that rejects CHARSET but advertises UTF-8 via MTTS
+    class ClientWithMTTS(telnetlib3.TelnetClient):
+        def send_charset(self, offered):
+            # Reject CHARSET negotiation
+            return None
+
+        def send_env(self, keys):
+            # Include MTTS=2825 which has bit 3 (UTF-8) set
+            # 2825 = 2048 + 512 + 256 + 8 + 1
+            env = super().send_env(keys)
+            env["MTTS"] = "2825"
+            return env
+
+    async with create_server(
+        protocol_factory=ServerTestCharsetMTTS,
+        host=bind_host,
+        port=unused_tcp_port,
+        connect_maxwait=0.25,
+    ) as server:
+        async with open_connection(
+            client_factory=ClientWithMTTS,
+            host=bind_host,
+            port=unused_tcp_port,
+            connect_minwait=0.05,
+        ) as (reader, writer):
+            srv_instance = await asyncio.wait_for(_waiter, timeout=1.0)
+            val = srv_instance.get_extra_info("charset")
+            assert val == "UTF-8", f"Expected 'UTF-8', got {val!r}"
+
+
+async def test_charset_rejected_with_mtts_ttype3(bind_host, unused_tcp_port):
+    """Test CHARSET REJECTED fallback to UTF-8 when MTTS in TTYPE round 3."""
+    _waiter = asyncio.Future()
+
+    class ServerTestCharsetMTTS(telnetlib3.TelnetServer):
+        def on_charset(self, charset):
+            super().on_charset(charset)
+            _waiter.set_result(self)
+
+    # Create a client that rejects CHARSET but sends MTTS via TTYPE
+    class ClientWithMTTSTtype(telnetlib3.TelnetClient):
+        def __init__(self, *args, **kwargs):
+            super().__init__(*args, **kwargs)
+            self._ttype_round = 0
+
+        def send_charset(self, offered):
+            # Reject CHARSET negotiation
+            return None
+
+        def send_ttype(self):
+            # Cycle through TTYPE responses, ending with MTTS
+            responses = ["TINTIN++", "xterm-256color", "MTTS 2825"]
+            self._ttype_round += 1
+            if self._ttype_round <= len(responses):
+                return responses[self._ttype_round - 1]
+            return responses[-1]
+
+    async with create_server(
+        protocol_factory=ServerTestCharsetMTTS,
+        host=bind_host,
+        port=unused_tcp_port,
+        connect_maxwait=0.25,
+    ) as server:
+        async with open_connection(
+            client_factory=ClientWithMTTSTtype,
+            host=bind_host,
+            port=unused_tcp_port,
+            connect_minwait=0.05,
+        ) as (reader, writer):
+            srv_instance = await asyncio.wait_for(_waiter, timeout=1.0)
+            assert srv_instance.get_extra_info("charset") == "UTF-8"
+
+
+async def test_charset_rejected_without_mtts_utf8(bind_host, unused_tcp_port):
+    """Test CHARSET REJECTED without UTF-8 in MTTS does not resolve."""
+    _waiter = asyncio.Future()
+    charset_called = []
+
+    class ServerTestCharsetNoMTTS(telnetlib3.TelnetServer):
+        def on_charset(self, charset):
+            super().on_charset(charset)
+            charset_called.append(charset)
+            _waiter.set_result(charset)
+
+    # Create a client that rejects CHARSET and has MTTS without UTF-8 bit
+    class ClientNoUTF8MTTS(telnetlib3.TelnetClient):
+        def send_charset(self, offered):
+            # Reject CHARSET negotiation
+            return None
+
+        def send_env(self, keys):
+            # MTTS=257 = 256 + 1 (no bit 3 for UTF-8)
+            env = super().send_env(keys)
+            env["MTTS"] = "257"
+            return env
+
+    async with create_server(
+        protocol_factory=ServerTestCharsetNoMTTS,
+        host=bind_host,
+        port=unused_tcp_port,
+        connect_maxwait=0.25,
+    ) as server:
+        async with open_connection(
+            client_factory=ClientNoUTF8MTTS,
+            host=bind_host,
+            port=unused_tcp_port,
+            connect_minwait=0.05,
+        ) as (reader, writer):
+            # on_charset should NOT fire — wait past connect_maxwait and
+            # confirm the future never resolves
+            with pytest.raises(asyncio.TimeoutError):
+                await asyncio.wait_for(_waiter, timeout=0.5)
+
+            assert not charset_called, "on_charset should not be called without UTF-8 in MTTS"
+
+            # Server charset should remain the default (utf8), not
+            # upgraded via MTTS fallback since bit 3 is not set
+            srv_instance = server.clients[0]
+            assert srv_instance.get_extra_info("charset") == "utf8"
+
+
+def test_check_mtts_for_utf8_from_environ():
+    """Test _check_mtts_for_utf8 with MTTS from NEW_ENVIRON."""
+
+    class MockProtocolWithMTTS:
+        def __init__(self):
+            self._extra = {"MTTS": "2825"}
+
+        def get_extra_info(self, name, default=None):
+            return self._extra.get(name, default)
+
+        async def _drain_helper(self):
+            pass
+
+    w, t, _ = new_writer(server=True, client=False)
+    w._protocol = MockProtocolWithMTTS()
+
+    # MTTS=2825 has bit 3 set (UTF-8)
+    result = w._check_mtts_for_utf8()
+    assert result == "UTF-8"
+
+
+def test_check_mtts_for_utf8_from_ttype3():
+    """Test _check_mtts_for_utf8 with MTTS from TTYPE round 3."""
+
+    class MockProtocolWithTtype:
+        def __init__(self):
+            self._extra = {"ttype3": "MTTS 2825"}
+
+        def get_extra_info(self, name, default=None):
+            return self._extra.get(name, default)
+
+        async def _drain_helper(self):
+            pass
+
+    w, t, _ = new_writer(server=True, client=False)
+    w._protocol = MockProtocolWithTtype()
+
+    # MTTS 2825 has bit 3 set (UTF-8)
+    result = w._check_mtts_for_utf8()
+    assert result == "UTF-8"
+
+
+def test_check_mtts_for_utf8_without_utf8_bit():
+    """Test _check_mtts_for_utf8 when UTF-8 bit is not set."""
+
+    class MockProtocolNoUTF8:
+        def __init__(self):
+            # MTTS=257 = 256 + 1 (no bit 3)
+            self._extra = {"MTTS": "257"}
+
+        def get_extra_info(self, name, default=None):
+            return self._extra.get(name, default)
+
+        async def _drain_helper(self):
+            pass
+
+    w, t, _ = new_writer(server=True, client=False)
+    w._protocol = MockProtocolNoUTF8()
+
+    result = w._check_mtts_for_utf8()
+    assert result is None
+
+
+def test_check_mtts_for_utf8_no_mtts():
+    """Test _check_mtts_for_utf8 when MTTS is not present."""
+
+    class MockProtocolNoMTTS:
+        def get_extra_info(self, name, default=None):
+            return default
+
+        async def _drain_helper(self):
+            pass
+
+    w, t, _ = new_writer(server=True, client=False)
+    w._protocol = MockProtocolNoMTTS()
+
+    result = w._check_mtts_for_utf8()
+    assert result is None
+
+
+def test_check_mtts_for_utf8_environ_takes_precedence():
+    """Test NEW_ENVIRON MTTS is checked before TTYPE round 3."""
+
+    class MockProtocolBothSources:
+        def __init__(self):
+            self._extra = {
+                "MTTS": "257",  # no UTF-8 bit
+                "ttype3": "MTTS 2825",  # has UTF-8 bit
+            }
+
+        def get_extra_info(self, name, default=None):
+            return self._extra.get(name, default)
+
+        async def _drain_helper(self):
+            pass
+
+    w, t, _ = new_writer(server=True, client=False)
+    w._protocol = MockProtocolBothSources()
+
+    # NEW_ENVIRON (MTTS=257, no UTF-8) should win over ttype3
+    result = w._check_mtts_for_utf8()
+    assert result is None
+
+
+def test_check_mtts_for_utf8_zero():
+    """Test _check_mtts_for_utf8 with MTTS=0."""
+
+    class MockProtocolZero:
+        def __init__(self):
+            self._extra = {"MTTS": "0"}
+
+        def get_extra_info(self, name, default=None):
+            return self._extra.get(name, default)
+
+        async def _drain_helper(self):
+            pass
+
+    w, t, _ = new_writer(server=True, client=False)
+    w._protocol = MockProtocolZero()
+
+    result = w._check_mtts_for_utf8()
+    assert result is None
+
+
+def test_check_mtts_for_utf8_malformed():
+    """Test _check_mtts_for_utf8 with non-numeric MTTS value."""
+
+    class MockProtocolMalformed:
+        def __init__(self):
+            self._extra = {"MTTS": "not-a-number"}
+
+        def get_extra_info(self, name, default=None):
+            return self._extra.get(name, default)
+
+        async def _drain_helper(self):
+            pass
+
+    w, t, _ = new_writer(server=True, client=False)
+    w._protocol = MockProtocolMalformed()
+
+    result = w._check_mtts_for_utf8()
+    assert result is None