Add --always-dont and --always-wont=TEL,OPTS (jquast#134)

This mirrors --always-do and --always-will. If you're making custom
clients and servers or automation's you might find issues or bugs in our
implementation or the remote's, common ones like TTYPE, NAWS etc, that
can be worked around by rejecting them entirely.

Author: jquast

telnetlib3/client.py

@@ -655,9 +655,11 @@ async def run_client() -> None:
 
     always_will: set[bytes] = args["always_will"]
     always_do: set[bytes] = args["always_do"]
+    always_wont: set[bytes] = args["always_wont"]
+    always_dont: set[bytes] = args["always_dont"]
 
-    # Wrap client factory to inject always_will/always_do and encoding
-    # flags before negotiation starts.
+    # Wrap client factory to inject always_will/always_do/always_wont/always_dont
+    # and encoding flags before negotiation starts.
     encoding_explicit = args["encoding"] not in ("utf8", "utf-8", False)
     gmcp_modules: Optional[List[str]] = args.get("gmcp_modules")
 
@@ -675,6 +677,10 @@ def _patched_connection_made(transport: asyncio.BaseTransport) -> None:
             if always_will:
                 client.writer.always_will = always_will
             client.writer.always_do = always_do
+            if always_wont:
+                client.writer.always_wont = always_wont
+            if always_dont:
+                client.writer.always_dont = always_dont
             from .telopt import GMCP as _GMCP
 
             client.writer.passive_do = {_GMCP}
@@ -779,14 +785,32 @@ def _get_argument_parser() -> argparse.ArgumentParser:
         action="append",
         default=[],
         metavar="OPT",
-        help="always send DO for this option (name like GMCP or number, repeatable)",
+        help="always send DO for this option (comma-separated, named like GMCP"
+        " or numeric like 201, repeatable)",
+    )
+    parser.add_argument(
+        "--always-dont",
+        action="append",
+        default=[],
+        metavar="OPT",
+        help="always send DONT for this option, refusing even natively supported"
+        " options (comma-separated, named or numeric, repeatable)",
     )
     parser.add_argument(
         "--always-will",
         action="append",
         default=[],
         metavar="OPT",
-        help="always send WILL for this option (name like MXP or number, repeatable)",
+        help="always send WILL for this option (comma-separated, named like MXP"
+        " or numeric like 91, repeatable)",
+    )
+    parser.add_argument(
+        "--always-wont",
+        action="append",
+        default=[],
+        metavar="OPT",
+        help="always send WONT for this option, refusing even natively supported"
+        " options (comma-separated, named or numeric, repeatable)",
     )
     parser.add_argument(
         "--ansi-keys",
@@ -925,6 +949,22 @@ def _parse_option_arg(value: str) -> bytes:
         return bytes([int(value)])
 
 
+def _parse_option_list(values: List[str]) -> set[bytes]:
+    """
+    Parse a list of option arguments, splitting comma-separated values.
+
+    :param values: List of option strings, each may be comma-separated.
+    :returns: Set of parsed option bytes.
+    """
+    result: set[bytes] = set()
+    for v in values:
+        for item in v.split(","):
+            item = item.strip()
+            if item:
+                result.add(_parse_option_arg(item))
+    return result
+
+
 def _transform_args(args: argparse.Namespace) -> Dict[str, Any]:
     # Auto-enable force_binary for any non-ASCII encoding that uses high-bit bytes.
     from .encodings import FORCE_BINARY_ENCODINGS
@@ -971,8 +1011,10 @@ def _transform_args(args: argparse.Namespace) -> Dict[str, Any]:
         "connect_minwait": args.connect_minwait,
         "connect_timeout": args.connect_timeout or None,
         "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},
+        "always_will": _parse_option_list(args.always_will),
+        "always_do": _parse_option_list(args.always_do),
+        "always_wont": _parse_option_list(args.always_wont),
+        "always_dont": _parse_option_list(args.always_dont),
         "raw_mode": raw_mode,
         "ascii_eol": args.ascii_eol,
         "ansi_keys": args.ansi_keys,
@@ -1012,14 +1054,32 @@ def _get_fingerprint_argument_parser() -> argparse.ArgumentParser:
         action="append",
         default=[],
         metavar="OPT",
-        help="always send DO for this option (name like GMCP or number, repeatable)",
+        help="always send DO for this option (comma-separated, named like GMCP"
+        " or numeric like 201, repeatable)",
+    )
+    parser.add_argument(
+        "--always-dont",
+        action="append",
+        default=[],
+        metavar="OPT",
+        help="always send DONT for this option, refusing even natively supported"
+        " options (comma-separated, named or numeric, repeatable)",
     )
     parser.add_argument(
         "--always-will",
         action="append",
         default=[],
         metavar="OPT",
-        help="always send WILL for this option (name like MXP or number, repeatable)",
+        help="always send WILL for this option (comma-separated, named like MXP"
+        " or numeric like 91, repeatable)",
+    )
+    parser.add_argument(
+        "--always-wont",
+        action="append",
+        default=[],
+        metavar="OPT",
+        help="always send WONT for this option, refusing even natively supported"
+        " options (comma-separated, named or numeric, repeatable)",
     )
     parser.add_argument(
         "--banner-max-bytes", default=65536, type=int, help="max bytes per banner read call"
@@ -1141,9 +1201,11 @@ async def run_fingerprint_client() -> None:
         banner_max_bytes=args.banner_max_bytes,
     )
 
-    # Parse --always-will/--always-do option names/numbers
-    fp_always_will = {_parse_option_arg(v) for v in args.always_will}
-    fp_always_do = {_parse_option_arg(v) for v in args.always_do}
+    # Parse --always-will/--always-do/--always-wont/--always-dont option names/numbers
+    fp_always_will = _parse_option_list(args.always_will)
+    fp_always_do = _parse_option_list(args.always_do)
+    fp_always_wont = _parse_option_list(args.always_wont)
+    fp_always_dont = _parse_option_list(args.always_dont)
 
     # Parse --send-env KEY=VALUE pairs
     extra_env: Dict[str, str] = {}
@@ -1178,6 +1240,10 @@ def patched_connection_made(transport: asyncio.BaseTransport) -> None:
             mud_opts = {opt for opt, _, _ in fingerprinting.EXTENDED_OPTIONS}
             client.writer.always_will = fp_always_will | mud_opts
             client.writer.always_do = fp_always_do | mud_opts
+            if fp_always_wont:
+                client.writer.always_wont = fp_always_wont
+            if fp_always_dont:
+                client.writer.always_dont = fp_always_dont
 
         def patched_send_env(keys: Sequence[str]) -> Dict[str, Any]:
             result = orig_send_env(keys)

telnetlib3/stream_writer.py

@@ -234,6 +234,16 @@ def __init__(
         #: DONT rejection in :meth:`handle_will`.
         self.always_do: set[bytes] = set()
 
+        #: Set of option byte(s) for which the client always sends WONT
+        #: in response to DO, refusing the option even when natively
+        #: supported.  Overrides the default WILL in :meth:`handle_do`.
+        self.always_wont: set[bytes] = set()
+
+        #: Set of option byte(s) for which the client always sends DONT
+        #: in response to WILL, refusing the option even when natively
+        #: supported.  Overrides the default DO in :meth:`handle_will`.
+        self.always_dont: set[bytes] = set()
+
         #: Set of option byte(s) for which the client sends DO only
         #: in response to a server WILL (passive negotiation).
         self.passive_do: set[bytes] = set()
@@ -1824,6 +1834,11 @@ def handle_do(self, opt: bytes) -> bool:
         # False for unsupported option, or an option invalid in that context,
         # such as LOGOUT.
         self.log.debug("handle_do(%s)", name_command(opt))
+        if opt in self.always_wont:
+            self.log.debug("DO %s: always-wont, declining.", name_command(opt))
+            if not self.local_option.enabled(opt):
+                self.iac(WONT, opt)
+            return False
         if opt == ECHO and self.client:
             # What do we have here? A Telnet Server attempting to
             # fingerprint us as a broken 4.4BSD Telnet Client, which
@@ -1978,6 +1993,12 @@ def handle_will(self, opt: bytes) -> None:
         """
         self.log.debug("handle_will(%s)", name_command(opt))
 
+        if opt in self.always_dont:
+            self.log.debug("WILL %s: always-dont, refusing.", name_command(opt))
+            self.iac(DONT, opt)
+            self.remote_option[opt] = False
+            return
+
         if opt in (
             BINARY,
             SGA,

telnetlib3/tests/test_stream_writer_full.py

@@ -1111,6 +1111,38 @@ def test_handle_will_always_do_sends_do():
     assert AUTHENTICATION not in w.rejected_will
 
 
+def test_handle_will_always_dont_refuses():
+    w, t, p = new_writer(server=True)
+    w.always_dont.add(SGA)
+    w.handle_will(SGA)
+    assert t.writes[-1] == IAC + DONT + SGA
+    assert not w.remote_option.enabled(SGA)
+
+
+def test_handle_do_always_wont_refuses():
+    w, t, p = new_writer(server=False, client=True)
+    w.always_wont.add(TTYPE)
+    result = w.handle_do(TTYPE)
+    assert result is False
+    assert t.writes[-1] == IAC + WONT + TTYPE
+
+
+def test_always_dont_overrides_native_support():
+    w, t, p = new_writer(server=True)
+    w.always_dont.add(BINARY)
+    w.handle_will(BINARY)
+    assert t.writes[-1] == IAC + DONT + BINARY
+    assert not w.remote_option.enabled(BINARY)
+
+
+def test_always_wont_overrides_native_support():
+    w, t, p = new_writer(server=False, client=True)
+    w.always_wont.add(NAWS)
+    result = w.handle_do(NAWS)
+    assert result is False
+    assert t.writes[-1] == IAC + WONT + NAWS
+
+
 def test_write_non_bytes_raises_type_error():
     w, t, p = new_writer(server=True)
     with pytest.raises(TypeError, match="buf expected bytes"):

telnetlib3/tests/test_tls.py

@@ -406,6 +406,22 @@ async def test_parse_option_arg(input_val, expected):
     assert _parse_option_arg(input_val) == expected
 
 
+@pytest.mark.parametrize(
+    "values, expected",
+    [
+        pytest.param(["TTYPE,NAWS"], {bytes([24]), bytes([31])}, id="comma-separated"),
+        pytest.param(["91,TTYPE"], {bytes([91]), bytes([24])}, id="mixed-comma"),
+        pytest.param(["TTYPE", "91"], {bytes([24]), bytes([91])}, id="repeated"),
+        pytest.param(["TTYPE, 91"], {bytes([24]), bytes([91])}, id="comma-with-spaces"),
+        pytest.param([], set(), id="empty"),
+    ],
+)
+async def test_parse_option_list(values, expected):
+    from telnetlib3.client import _parse_option_list
+
+    assert _parse_option_list(values) == expected
+
+
 _MAX_SUBPROC_SECONDS = 8