First git commit

bbs/term.py

@@ -0,0 +1,163 @@
+"""Telnet terminal I/O for the txt3 BBS.
+
+Deliberately conservative: real telnet clients, netcat, PuTTY in raw mode and
+Windows telnet all have to work, so we negotiate the bare minimum (suppress
+go-ahead, refuse everything else) and strip IAC sequences from the byte stream
+rather than trying to be a full RFC 854 implementation.
+"""
+import socket
+
+IAC  = 255
+DONT = 254
+DO   = 253
+WONT = 252
+WILL = 251
+SB   = 250
+SE   = 240
+ECHO = 1
+SGA  = 3
+
+CRLF = b"\r\n"
+
+
+class Hangup(Exception):
+    """Raised when the peer disconnects or times out."""
+
+
+class Term:
+    def __init__(self, sock, addr, idle_timeout=600):
+        self.s = sock
+        self.addr = addr
+        self.buf = b""
+        self.s.settimeout(idle_timeout)
+        # We will echo input ourselves only for passwords (as masking);
+        # otherwise let the client echo locally, which is what most do.
+        self._send_raw(bytes([IAC, WILL, SGA]))
+        self._send_raw(bytes([IAC, DO, SGA]))
+
+    # ------------------------------------------------------------- low level
+    def _send_raw(self, data):
+        try:
+            self.s.sendall(data)
+        except (BrokenPipeError, ConnectionResetError, OSError):
+            raise Hangup()
+
+    def write(self, text=""):
+        if isinstance(text, str):
+            # Telnet wants CRLF; also escape a literal IAC byte.
+            text = text.replace("\n", "\r\n").encode("utf-8", "replace")
+        self._send_raw(text.replace(bytes([IAC]), bytes([IAC, IAC])))
+
+    def line(self, text=""):
+        self.write(text + "\n")
+
+    # ------------------------------------------------------------- input
+    def _fill(self):
+        try:
+            chunk = self.s.recv(1024)
+        except socket.timeout:
+            raise Hangup()
+        except (ConnectionResetError, OSError):
+            raise Hangup()
+        if not chunk:
+            raise Hangup()
+        self.buf += chunk
+
+    def _pop_byte(self):
+        while not self.buf:
+            self._fill()
+        b = self.buf[0]
+        self.buf = self.buf[1:]
+        return b
+
+    def _read_char(self):
+        """Return the next data byte, transparently consuming telnet commands."""
+        while True:
+            b = self._pop_byte()
+            if b != IAC:
+                return b
+            # IAC ...
+            c = self._pop_byte()
+            if c == IAC:
+                return IAC              # escaped literal 255
+            if c in (DO, DONT, WILL, WONT):
+                opt = self._pop_byte()
+                # Refuse everything except SGA, which we already agreed.
+                if c == DO:
+                    resp = WILL if opt == SGA else WONT
+                elif c == WILL:
+                    resp = DO if opt == SGA else DONT
+                else:
+                    resp = WONT if c == DO else DONT
+                self._send_raw(bytes([IAC, resp, opt]))
+                continue
+            if c == SB:
+                # swallow the subnegotiation up to IAC SE
+                prev = None
+                while True:
+                    x = self._pop_byte()
+                    if prev == IAC and x == SE:
+                        break
+                    prev = x
+                continue
+            # any other 2-byte command: ignore
+            continue
+
+    def read(self, prompt="", mask=False, maxlen=200):
+        """Read one line of input. Handles backspace and bare CR or LF."""
+        if prompt:
+            self.write(prompt)
+        out = bytearray()
+        while True:
+            b = self._read_char()
+            if b in (13, 10):           # CR or LF
+                # A CR is often followed by LF or NUL; peek and drop it.
+                if self.buf[:1] in (b"\n", b"\x00"):
+                    self.buf = self.buf[1:]
+                self.write("\n")
+                break
+            if b in (8, 127):           # backspace / delete
+                if out:
+                    out.pop()
+                    if mask:
+                        self.write("\b \b")
+                    else:
+                        self.write("\b \b")
+                continue
+            if b == 3:                  # ^C
+                raise Hangup()
+            if b == 4 and not out:      # ^D on an empty line
+                raise Hangup()
+            if b < 32 or b > 126:
+                continue                # ignore other control/non-ascii
+            if len(out) >= maxlen:
+                continue
+            out.append(b)
+            if mask:
+                self.write("*")
+        return out.decode("utf-8", "replace").strip()
+
+    def ask(self, prompt, maxlen=200):
+        return self.read(prompt, mask=False, maxlen=maxlen)
+
+    def secret(self, prompt="Password: "):
+        return self.read(prompt, mask=True, maxlen=60)
+
+    # ------------------------------------------------------------- screen
+    def rule(self, ch="-", n=60):
+        self.line(ch * n)
+
+    def header(self, title):
+        self.line()
+        self.rule("=")
+        self.line(" " + title)
+        self.rule("=")
+
+    def pause(self):
+        self.read("\n[enter] ")
+
+    def close(self):
+        try:
+            self.s.close()
+        except OSError:
+            pass