Kludge mode fix (WILL SGA/DO SGA) (jquast#130)

First Linemode fix, that kludge was not correctly detected.

WILL SGA and DO SGA do not really specify direction. It is a server's responsibility (WILL SGA) to suppress a GA.

But RFC wording is unclear about this, and so it can be implemented [either way and often is! 50/50](https://muds.modem.xyz/statistics.html#telnet-option-negotiation). I believe either inetutils-telnetd or bsd4.4-derived telnetd sends both WILL and DO SGA and one of them has a long explanatory comment about it, that if i recall some popular telnet made a mistake with SGA, and so the workaround is to do the same as they do -- to support both directions to mean the same.

Author: jquast

docs/history.rst

@@ -1,5 +1,21 @@
 History
 =======
+3.0.3 (unreleased)
+  * new: :func:`~telnetlib3.accessories.make_logger` accepts a ``filemode``
+    parameter (``"a"`` append or ``"w"`` overwrite) so callers can control
+    whether log files are truncated on each run.  The default remains
+    ``"a"`` (append), matching historical behaviour.
+  * new: ``--logfile-mode {append,rewrite}`` CLI flag controls whether the
+    log file is appended to (default) or overwritten on each connection.
+  * new: ``--typescript-mode {append,rewrite}`` CLI flag controls whether the
+    typescript file is appended to (default) or overwritten on each connection.
+  * bugfix: kludge mode not detected when server negotiates WILL ECHO + DO SGA
+    (requesting client WILL SGA) instead of WILL ECHO + WILL SGA.
+    :attr:`~telnetlib3.stream_writer.TelnetWriter.mode` and
+    ``_server_will_sga()`` now check SGA in either direction
+    (``remote_option`` or ``local_option``), so clients correctly switch from
+    line mode to character-at-a-time mode with these servers.
+
 3.0.2
   * bugfix: :meth:`~telnetlib3.stream_writer.TelnetWriter.request_charset` raised :exc:`TypeError`,
     :ghissue:`128`.  Offer callbacks (no-arg, returning a list of items to propose) are now

telnetlib3/accessories.py

@@ -130,7 +130,11 @@ def hexdump(data: bytes, prefix: str = "") -> str:
 
 
 def make_logger(
-    name: str, loglevel: str = "info", logfile: Optional[str] = None, logfmt: str = _DEFAULT_LOGFMT
+    name: str,
+    loglevel: str = "info",
+    logfile: Optional[str] = None,
+    logfmt: str = _DEFAULT_LOGFMT,
+    filemode: str = "a",
 ) -> logging.Logger:
     """Create and return simple logger for given arguments."""
     lvl = getattr(logging, loglevel.upper(), None)
@@ -140,6 +144,7 @@ def make_logger(
     _cfg: Dict[str, Any] = {"format": logfmt}
     if logfile:
         _cfg["filename"] = logfile
+        _cfg["filemode"] = filemode
     logging.basicConfig(**_cfg)
     for handler in logging.getLogger().handlers:
         if isinstance(handler, logging.StreamHandler) and not isinstance(

telnetlib3/client.py

@@ -645,7 +645,11 @@ async def run_client() -> None:
     config_msg = f"Client configuration: {accessories.repr_mapping(args)}"
 
     log = accessories.make_logger(
-        name=__name__, loglevel=args["loglevel"], logfile=args["logfile"], logfmt=args["logfmt"]
+        name=__name__,
+        loglevel=args["loglevel"],
+        logfile=args["logfile"],
+        logfmt=args["logfmt"],
+        filemode="w" if args.get("logfile_mode") == "rewrite" else "a",
     )
     log.debug(config_msg)
 
@@ -774,7 +778,11 @@ async def _typescript_shell(
         ) -> None:
             ctx = writer_arg.ctx
             assert typescript_path is not None
-            ts_file = open(typescript_path, "a", encoding="utf-8")  # noqa: SIM115
+            ts_file = open(  # noqa: SIM115
+                typescript_path,
+                "w" if args.get("typescript_mode") == "rewrite" else "a",
+                encoding="utf-8",
+            )
             ctx.typescript_file = ts_file
             try:
                 await _ts_inner(reader, writer_arg)
@@ -819,6 +827,13 @@ def _get_argument_parser() -> argparse.ArgumentParser:
     parser.add_argument("--loglevel", default="warn", help="log level")
     parser.add_argument("--logfmt", default=accessories._DEFAULT_LOGFMT, help="log format")
     parser.add_argument("--logfile", help="filepath")
+    parser.add_argument(
+        "--logfile-mode",
+        default="append",
+        choices=["append", "rewrite"],
+        dest="logfile_mode",
+        help="Log file write mode: append (default) or rewrite.",
+    )
     parser.add_argument(
         "--shell", default="telnetlib3.telnet_client_shell", help="module.function_name"
     )
@@ -972,6 +987,13 @@ def _get_argument_parser() -> argparse.ArgumentParser:
         metavar="FILE",
         help="record session to FILE (like Unix script(1))",
     )
+    parser.add_argument(
+        "--typescript-mode",
+        default="append",
+        choices=["append", "rewrite"],
+        dest="typescript_mode",
+        help="Typescript write mode: append (default) or rewrite.",
+    )
     return parser
 
 
@@ -1040,6 +1062,7 @@ def _transform_args(args: argparse.Namespace) -> Dict[str, Any]:
         "port": args.port,
         "loglevel": args.loglevel,
         "logfile": args.logfile,
+        "logfile_mode": args.logfile_mode,
         "logfmt": args.logfmt,
         "encoding": args.encoding,
         "tspeed": (args.speed, args.speed),
@@ -1068,6 +1091,7 @@ def _transform_args(args: argparse.Namespace) -> Dict[str, Any]:
         ),
         "compression": args.compression,
         "typescript": args.typescript,
+        "typescript_mode": args.typescript_mode,
     }
 
 

telnetlib3/client_shell.py

@@ -310,10 +310,14 @@ def _make_raw(
             )
 
         def _server_will_sga(self) -> bool:
-            """Whether server has negotiated WILL SGA."""
+            """Whether SGA has been negotiated (either direction)."""
             from .telopt import SGA
 
-            return bool(self.telnet_writer.client and self.telnet_writer.remote_option.enabled(SGA))
+            w = self.telnet_writer
+            return bool(
+                w.client
+                and (w.remote_option.enabled(SGA) or w.local_option.enabled(SGA))
+            )
 
         def check_auto_mode(
             self, switched_to_raw: bool, last_will_echo: bool

telnetlib3/stream_writer.py

@@ -911,7 +911,9 @@ def mode(self) -> str:
             if self.local_option.enabled(ECHO) and self.local_option.enabled(SGA):
                 return "kludge"
             return "local"
-        if self.remote_option.enabled(ECHO) and self.remote_option.enabled(SGA):
+        if self.remote_option.enabled(ECHO) and (
+            self.remote_option.enabled(SGA) or self.local_option.enabled(SGA)
+        ):
             return "kludge"
         return "local"
 

telnetlib3/tests/test_client_shell.py

@@ -41,7 +41,10 @@ def enabled(self, key: bytes) -> bool:
 
 
 def _make_writer(
-    will_echo: bool = False, raw_mode: "bool | None" = False, will_sga: bool = False
+    will_echo: bool = False,
+    raw_mode: "bool | None" = False,
+    will_sga: bool = False,
+    local_sga: bool = False,
 ) -> object:
     """Build a minimal mock writer with the attributes Terminal needs."""
     from telnetlib3.telopt import SGA
@@ -52,6 +55,7 @@ def _make_writer(
         will_echo=will_echo,
         client=True,
         remote_option=_MockOption({SGA: will_sga}),
+        local_option=_MockOption({SGA: local_sga}),
         log=types.SimpleNamespace(debug=lambda *a, **kw: None),
         ctx=ctx,
     )
@@ -190,6 +194,32 @@ def test_make_raw_toggle_echo_flag() -> None:
     assert not restored.lflag & termios.ICANON
 
 
+def test_server_will_sga_local_option() -> None:
+    """DO SGA from server (client WILL SGA) is detected as SGA active."""
+    term = _make_term(_make_writer(local_sga=True, raw_mode=None))
+    assert term._server_will_sga() is True
+
+
+def test_determine_mode_local_sga_goes_raw() -> None:
+    """WILL ECHO + DO SGA (local_option) triggers kludge raw mode."""
+    term = _make_term(_make_writer(will_echo=True, raw_mode=None, local_sga=True))
+    mode = _cooked_mode()
+    result = term.determine_mode(mode)
+    assert result is not mode
+    assert not result.lflag & termios.ICANON
+    assert not result.lflag & termios.ECHO
+
+
+def test_determine_mode_local_sga_without_echo() -> None:
+    """DO SGA alone (no WILL ECHO) triggers character-at-a-time with software echo."""
+    term = _make_term(_make_writer(raw_mode=None, local_sga=True))
+    mode = _cooked_mode()
+    result = term.determine_mode(mode)
+    assert result is not mode
+    assert not result.lflag & termios.ICANON
+    assert term.software_echo is True
+
+
 @pytest.mark.parametrize(
     "encoding,key,expected",
     [

telnetlib3/tests/test_server_shell_unit.py

@@ -137,6 +137,7 @@ async def test_terminal_determine_mode(monkeypatch):
         will_echo=False,
         client=True,
         remote_option=_mock_opt,
+        local_option=_mock_opt,
         log=types.SimpleNamespace(debug=lambda *a, **k: None),
         ctx=TelnetSessionContext(),
     )

telnetlib3/tests/test_stream_writer_full.py

@@ -820,6 +820,14 @@ def test_mode_client_kludge_and_server_kludge_and_remote_local():
     assert wc.mode == "remote"
 
 
+def test_mode_client_kludge_local_sga():
+    """Client mode is kludge when server WILL ECHO + client WILL SGA (DO SGA from server)."""
+    wc, _, _ = new_writer(server=False, client=True)
+    wc.remote_option[ECHO] = True
+    wc.local_option[SGA] = True
+    assert wc.mode == "kludge"
+
+
 def test_handle_send_server_and_client_charset_returns():
     ws, ts, ps = new_writer(server=True)
     assert ws.handle_send_server_charset() == ["UTF-8"]