site generator: commands/*.md with front matter -> k8s/site/
site / generated-is-current (push) Failing after 36s
site / generated-is-current (push) Failing after 36s
- build/build.mjs: one-file-per-command content (filename = command,
underscores = spaces; help/suggestion/params/prefix/text/hidden in
front matter; ${args} ${date} ${time} ${ls} ${help} resolved at runtime)
- generates index.html (banner + chips), commands.json (the single data
source), hello.txt for curl, style.css via tailwind v4
- src/site.js is now dispatch-only over commands.json; typewriter kept
- src/telnet.py: stdlib telnet shell sharing commands.json
- k8s: python telnet sidecar in the pod + LoadBalancer service 23->2323
- .gitea/workflows/site.yml fails CI when k8s/site/ is not the built output
This commit is contained in:
@@ -0,0 +1,165 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user