Update the telnet server to provide the client terminal type to the prompt-toolkit output object

Author: vxgmichel

prompt_toolkit/contrib/telnet/protocol.py

@@ -36,6 +36,10 @@ def int2byte(number: int) -> bytes:
 LINEMODE = int2byte(34)
 SUPPRESS_GO_AHEAD = int2byte(3)
 
+TTYPE = int2byte(24)
+SEND = int2byte(1)
+IS = int2byte(0)
+
 DM = int2byte(242)
 BRK = int2byte(243)
 IP = int2byte(244)
@@ -65,10 +69,12 @@ def __init__(
  self,
  data_received_callback: Callable[[bytes], None],
  size_received_callback: Callable[[int, int], None],
+ ttype_received_callback: Callable[[str], None],
  ) -> None:
 
  self.data_received_callback = data_received_callback
  self.size_received_callback = size_received_callback
+ self.ttype_received_callback = ttype_received_callback
 
  self._parser = self._parse_coroutine()
  self._parser.send(None) # type: ignore
@@ -121,6 +127,17 @@ def naws(self, data: bytes) -> None:
  else:
  logger.warning("Wrong number of NAWS bytes")
 
+ def ttype(self, data: bytes) -> None:
+ """
+ Received terminal type.
+ """
+ subcmd, ttype = data[0:1], data[1:]
+ if subcmd == IS:
+ ttype = ttype.decode("ascii")
+ self.ttype_received_callback(ttype)
+ else:
+ logger.warning("Received a non-IS terminal type Subnegotiation")
+
  def negotiate(self, data: bytes) -> None:
  """
  Got negotiate data.
@@ -129,6 +146,8 @@ def negotiate(self, data: bytes) -> None:
 
  if command == NAWS:
  self.naws(payload)
+ elif command == TTYPE:
+ self.ttype(payload)
  else:
  logger.info("Negotiate (%r got bytes)", len(data))
 

prompt_toolkit/contrib/telnet/server.py

@@ -11,7 +11,7 @@
 from prompt_toolkit.application.run_in_terminal import run_in_terminal
 from prompt_toolkit.data_structures import Size
 from prompt_toolkit.formatted_text import AnyFormattedText, to_formatted_text
-from prompt_toolkit.input.posix_pipe import PosixPipeInput
+from prompt_toolkit.input import create_pipe_input
 from prompt_toolkit.output.vt100 import Vt100_Output
 from prompt_toolkit.renderer import print_formatted_text as print_formatted_text
 from prompt_toolkit.styles import BaseStyle, DummyStyle
@@ -28,6 +28,8 @@
  SE,
  SUPPRESS_GO_AHEAD,
  WILL,
+ TTYPE,
+ SEND,
  TelnetProtocolParser,
 )
 
@@ -59,6 +61,12 @@ def _initialize_telnet(connection: socket.socket) -> None:
  # Negotiate window size
  connection.send(IAC + DO + NAWS)
 
+ # Negotiate terminal type
+ connection.send(IAC + DO + TTYPE)
+
+ # Negotiate terminal type
+ connection.send(IAC + SB + TTYPE + SEND + IAC + SE)
+
 
 class _ConnectionStdout:
  """
@@ -115,6 +123,7 @@ def __init__(
  self.encoding = encoding
  self.style = style
  self._closed = False
+ self._ready = asyncio.Event()
 
  # Create "Output" object.
  self.size = Size(rows=40, columns=79)
@@ -123,14 +132,14 @@ def __init__(
  _initialize_telnet(conn)
 
  # Create input.
- self.vt100_input = PosixPipeInput()
+ self.vt100_output = None
+ self.vt100_input = create_pipe_input()
 
  # Create output.
  def get_size() -> Size:
  return self.size
 
  self.stdout = cast(TextIO, _ConnectionStdout(conn, encoding=encoding))
- self.vt100_output = Vt100_Output(self.stdout, get_size, write_binary=False)
 
  def data_received(data: bytes) -> None:
  """ TelnetProtocolParser 'data_received' callback """
@@ -139,9 +148,17 @@ def data_received(data: bytes) -> None:
  def size_received(rows: int, columns: int) -> None:
  """ TelnetProtocolParser 'size_received' callback """
  self.size = Size(rows=rows, columns=columns)
- get_app()._on_resize()
+ if self.vt100_output is not None:
+ get_app()._on_resize()
+
+ def ttype_received(ttype: str) -> None:
+ """ TelnetProtocolParser 'ttype_received' callback """
+ self.vt100_output = Vt100_Output(
+ self.stdout, get_size, term=ttype, write_binary=False
+ )
+ self._ready.set()
 
- self.parser = TelnetProtocolParser(data_received, size_received)
+ self.parser = TelnetProtocolParser(data_received, size_received, ttype_received)
  self.context: Optional[contextvars.Context] = None
 
  async def run_application(self) -> None:
@@ -158,25 +175,24 @@ def handle_incoming_data() -> None:
  logger.info("Connection closed by client. %r %r" % self.addr)
  self.close()
 
- async def run() -> None:
- # Add reader.
- loop = get_event_loop()
- loop.add_reader(self.conn, handle_incoming_data)
+ # Add reader.
+ loop = get_event_loop()
+ loop.add_reader(self.conn, handle_incoming_data)
 
- try:
+ try:
+ # Wait for v100_output to be properly instantiated
+ await self._ready.wait()
+ with create_app_session(input=self.vt100_input, output=self.vt100_output):
+ self.context = contextvars.copy_context()
  await self.interact(self)
- except Exception as e:
- print("Got %s" % type(e).__name__, e)
- import traceback
-
- traceback.print_exc()
- raise
- finally:
- self.close()
-
- with create_app_session(input=self.vt100_input, output=self.vt100_output):
- self.context = contextvars.copy_context()
- await run()
+ except Exception as e:
+ print("Got %s" % type(e).__name__, e)
+ import traceback
+
+ traceback.print_exc()
+ raise
+ finally:
+ self.close()
 
  def feed(self, data: bytes) -> None:
  """