momme-world/k8s/site/telnet.py

#!/usr/bin/env python3
"""momme.world telnet shell.

Serves the same commands as the website, to real terminals:
    telnet momme.world 23

Content lives in commands.json next to this file (generated from commands/*.md
by build/build.mjs) — this server is only a REPL. Standard library only. It
answers enough telnet negotiation to keep classic clients happy; it is not a
login shell, there is nothing here to log into.

Inside the cluster it listens on TELNET_PORT (2323); the public port 23
arrives via Envoy Gateway's TCPRoute on the tailscale funnel.
"""

import json
import os
import re
import select
import socketserver
import sys
import time

HERE = os.path.dirname(os.path.abspath(__file__))
COMMANDS_FILE = os.environ.get("COMMANDS_FILE", os.path.join(HERE, "commands.json"))
PORT = int(os.environ.get("TELNET_PORT", "2323"))
IDLE_TIMEOUT = 300  # seconds; five idle minutes is a long enough hello

# telnet negotiation: answer WILL/WONT so clients stop asking
IAC, WONT, ECHO, SGA = 255, 252, 1, 3

ALIASES = {
    "whois": "whoami", "about": "whoami", "openme": "whoami",
    "tech": "stack", "tools": "stack",
    "links": "social", "socials": "social",
    "mail": "contact", "email": "contact",
    "repos": "project", "projects": "project", "rm": "sudo", "quit": "exit",
}

with open(COMMANDS_FILE, encoding="utf-8") as fh:
    SITE = json.load(fh)
COMMANDS = {c["name"]: c for c in SITE["commands"]}
BANNER = SITE["banner"]


def usages():
    return [
        f"  {c['name']:<22} {c['help']}"
        for c in SITE["commands"]
        if not c.get("hidden") and c.get("help")
    ]


def substitute(body, arg):
    def rep(match):
        key = match.group(1)
        if key == "args":
            return arg
        if key == "date":
            return time.strftime("%a %b %d %Y")
        if key == "time":
            return time.strftime("%H:%M:%S")
        if key == "help":
            return "\n".join(["usage: <command>", ""] + usages())
        if key == "ls":
            return "  ".join(n.split(" ", 1)[1] for n in COMMANDS if " " in n)
        return ""
    return re.sub(r"\$\{(\w+)\}", rep, body)


def lookup(words):
    for length in range(len(words), 0, -1):
        joined = " ".join(words[:length])
        cmd = COMMANDS.get(ALIASES.get(joined, joined))
        if cmd:
            return cmd, " ".join(words[length:])
    return None, ""



def run(line):
    """Return (text, quit_after). Empty text means print nothing."""
    words = [w for w in line.strip().lower().split() if w]
    if not words:
        return "", False
    cmd, arg = lookup(words)
    if not cmd:
        near = [n for n in COMMANDS if n != words[0] and n.startswith(words[0])][:3]
        hint = "did you mean: " + ", ".join(near) if near else "see: help"
        return f"command not found: {words[0]}\n{hint}", False
    if cmd["name"] == "clear":
        return "\033[2J\033[H", False
    if cmd["name"] == "exit":
        return "bye — https://momme.world/", True
    if arg and not cmd.get("params") and not cmd.get("prefix"):
        kids = [n[len(cmd["name"]) + 1:] for n in COMMANDS if n.startswith(cmd["name"] + " ")]
        if kids:
            kid = COMMANDS.get(cmd["name"] + " " + arg)
            if kid:
                return substitute(kid["body"], ""), False
            names = ", ".join(kids)
            return f'no {cmd["name"]} called "{arg}"\nnames: {names}', False
    return substitute(cmd["body"], arg), cmd["name"] == "quit"


class Shell(socketserver.BaseRequestHandler):
    def send(self, text):
        try:
            self.request.sendall(text.replace("\n", "\r\n").encode("utf-8"))
        except OSError:
            raise ConnectionError

    def setup(self):
        try:
            self.request.sendall(bytes([IAC, WONT, ECHO, IAC, WONT, SGA]))
            self.send(BANNER + "\n\nhello — type help for the rest of it.\n")
            self.loop()
        except (ConnectionError, ConnectionResetError, BrokenPipeError):
            pass

    def loop(self):
        line = ""
        self.send("momme@world:~$ ")
        while True:
            ready, _, _ = select.select([self.request], [], [], IDLE_TIMEOUT)
            if not ready:
                return
            data = self.request.recv(1024)
            if not data:
                return
            i = 0
            while i < len(data):
                byte = data[i]
                if byte == IAC:
                    i += 3  # skip a negotiation triple wholesale
                    continue
                i += 1
                if byte in (13, 10):
                    self.send("\n")
                    text, goodbye = run(line)
                    if text:
                        self.send(text + "\n")
                    if goodbye:
                        self.send("closing connection\n")
                        return
                    line = ""
                    self.send("momme@world:~$ ")
                elif byte in (127, 8):
                    if line:
                        line = line[:-1]
                        self.send("\b \b")
                elif 32 <= byte < 127:
                    line += chr(byte)
                    self.send(chr(byte))


class Server(socketserver.ThreadingTCPServer):
    allow_reuse_address = True
    daemon_threads = True


if __name__ == "__main__":
    with Server(("0.0.0.0", PORT), Shell) as srv:
        print(f"momme.world telnet shell on :{PORT}", file=sys.stderr, flush=True)
        srv.serve_forever()