mirror of
https://git.k-space.ee/arti/vt1200.git
synced 2026-10-31 12:05:05 +02:00
SDD
264d5330ad
The terminal broadcasts an RFC 951 BOOTP request at power-up. Add an opt-in responder that answers it with yiaddr 192.168.99.2 and siaddr 192.168.99.1, echoing htype/hlen/xid/chaddr. It offers no boot file: the request's sname and file are empty, so the reply's are too. Off by default -- a bare 'serve.py 192.168.99.1' behaves exactly as before (TFTP on port 69, nothing else). Enable with --bootp. Delivery note: the host IP stack never receives the source-0.0.0.0 broadcast, so the responder attaches to the MAME tap's L2 (AF_PACKET) and writes the reply frame back. See harness/BOOTP_REPORT.md for the evidence and the still-open 'User settings invalid' / 'Disk full' symptoms.
327 lines
13 KiB
Python
327 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""Serve VT1200 font files over TFTP, optionally with a BOOTP responder.
|
|
|
|
The terminal's "TFTP Font Path" is ``<host>:<filename>``; it requests that exact
|
|
filename over TFTP and (if it parses) loads it as a font. This serves several
|
|
plausible names so we catch whatever it asks for.
|
|
|
|
sudo python3 -u harness/serve.py [host] [port]
|
|
|
|
The VT1200 also broadcasts a BOOTP request (RFC 951) at power-up so it can learn
|
|
its own IP address. ``--bootp`` adds a minimal responder for that, so the
|
|
terminal no longer needs its address typed in by hand. It is **off by default**:
|
|
without the flag the server behaves exactly as before. The responder answers
|
|
*any* MAC that asks (the known terminal MAC is only used to label the log).
|
|
|
|
sudo python3 -u harness/serve.py 192.168.99.1 --bootp
|
|
"""
|
|
import argparse
|
|
import fcntl
|
|
import os
|
|
import socket
|
|
import struct
|
|
import sys
|
|
import threading
|
|
import time
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
ROOT = os.path.dirname(HERE)
|
|
sys.path.insert(0, os.path.join(ROOT, "tools"))
|
|
|
|
import fontgen
|
|
|
|
FONTS = os.path.join(ROOT, "reference", "fonts")
|
|
|
|
# The terminal's known Ethernet address (DEC OUI 08:00:2b). Used only to label
|
|
# the BOOTP log; the responder answers whatever MAC sends a request.
|
|
VT1200_MAC = "08:00:2b:16:4b:5a"
|
|
|
|
# --------------------------------------------------------------------------- BOOTP
|
|
#
|
|
# The terminal broadcasts a BOOTP request (op=1) at power-up to learn its own IP
|
|
# address. It is sent from source 0.0.0.0 to 255.255.255.255:67 into the MAME
|
|
# tap (``tap-mess-<uid>-0``, enslaved to the host bridge). A UDP socket bound to
|
|
# 0.0.0.0:67 on the host never receives it (verified, with and without
|
|
# rp_filter) -- but the tap's L2 does carry it, which is where tcpdump sees it.
|
|
# So the responder reads the tap at the L2 layer (AF_PACKET) and writes the reply
|
|
# frame back. This is a delivery-layer choice, not a protocol one: it is still
|
|
# an RFC 951 reply.
|
|
|
|
BOOTP_REQUEST = 1
|
|
BOOTP_REPLY = 2
|
|
BOOTP_SERVER_PORT = 67
|
|
BOOTP_CLIENT_PORT = 68
|
|
ETH_P_ALL = 0x0003 # ETH_P_IP misses frames bridged in from the tap
|
|
BROADCAST_MAC = b"\xff" * 6
|
|
BROADCAST_IP = "255.255.255.255"
|
|
# Linux ifreq ioctls; the IPv4 address / netmask sit at bytes 20..24 and the
|
|
# hardware address at 18..24 of the returned 16-byte-name + sockaddr buffer.
|
|
SIOCGIFADDR = 0x8915
|
|
SIOCGIFNETMASK = 0x891B
|
|
SIOCGIFHWADDR = 0x8927
|
|
|
|
|
|
def _ifreq(ifname, request):
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
try:
|
|
return fcntl.ioctl(s.fileno(), request,
|
|
struct.pack("256s", ifname.encode()[:15]))
|
|
except OSError:
|
|
return None
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
def _iface_ipv4(ip):
|
|
"""Return (ifname, ip, netmask) for the interface carrying *ip*, or None."""
|
|
want = socket.inet_aton(ip)
|
|
for _, name in socket.if_nameindex(): # (if_index, if_name)
|
|
r = _ifreq(name, SIOCGIFADDR)
|
|
if r is None or r[20:24] != want:
|
|
continue
|
|
m = _ifreq(name, SIOCGIFNETMASK)
|
|
return name, r[20:24], (m[20:24] if m else b"\xff\xff\xff\x00")
|
|
return None
|
|
|
|
|
|
def _iface_hwaddr(ifname):
|
|
r = _ifreq(ifname, SIOCGIFHWADDR) if ifname else None
|
|
return r[18:24] if r is not None else None
|
|
|
|
|
|
def _find_tap():
|
|
"""The MAME tap is named ``tap-mess-<uid>-0``; find any tap-mess device."""
|
|
names = [n for _, n in socket.if_nameindex() if n.startswith("tap-mess-")]
|
|
for n in names:
|
|
if n.endswith("-0"):
|
|
return n
|
|
return names[0] if names else None
|
|
|
|
|
|
def _ip_checksum(data):
|
|
if len(data) & 1:
|
|
data += b"\x00"
|
|
total = sum(struct.unpack(">%dH" % (len(data) // 2), data))
|
|
while total >> 16:
|
|
total = (total & 0xFFFF) + (total >> 16)
|
|
return (~total) & 0xFFFF
|
|
|
|
|
|
def MAC_BYTES(b):
|
|
return ":".join(f"{x:02x}" for x in b)
|
|
|
|
|
|
class BootpResponder:
|
|
"""Minimal RFC 951 BOOTP responder, attached to the MAME tap's L2.
|
|
|
|
Answers only ``op=1`` requests, filling ``yiaddr`` (the client) and
|
|
``siaddr`` (this server). It serves **no boot file**: the VT1200 request
|
|
carries an empty ``sname``/``file``, so the reply's are empty too -- inventing
|
|
a filename it never asked for would be a protocol fiction. It answers any
|
|
MAC that asks; the known terminal MAC is only used to label the log.
|
|
|
|
The VT1200 sends to 255.255.255.255:67 from port 68, ``flags=0``, and no DHCP
|
|
magic cookie, so the reply is a plain 300-byte RFC 951 broadcast on port 68.
|
|
"""
|
|
|
|
def __init__(self, server_ip, client_ip, iface=None, log=None):
|
|
self.server_ip = server_ip
|
|
self.client_ip = client_ip
|
|
self.log = log if log is not None else []
|
|
self._server_ip_b = socket.inet_aton(server_ip)
|
|
self._client_ip_b = socket.inet_aton(client_ip)
|
|
info = _iface_ipv4(server_ip)
|
|
self.server_iface = info[0] if info else None
|
|
self.iface = iface or _find_tap()
|
|
if self.iface is None:
|
|
raise SystemExit(
|
|
"bootp: no tap-mess-* interface found (is MAME's tap created?); "
|
|
"pass --bootp-iface to name it")
|
|
self.src_mac = (_iface_hwaddr(self.server_iface)
|
|
or _iface_hwaddr(self.iface)
|
|
or b"\x02\x00\x00\x00\x00\x01")
|
|
self._stop = threading.Event()
|
|
self._sock = None
|
|
self._thread = None
|
|
|
|
def start(self):
|
|
self._sock = socket.socket(socket.AF_PACKET, socket.SOCK_RAW,
|
|
socket.htons(ETH_P_ALL))
|
|
self._sock.bind((self.iface, 0))
|
|
self._sock.settimeout(0.5)
|
|
self._thread = threading.Thread(target=self._serve, daemon=True)
|
|
self._thread.start()
|
|
return self
|
|
|
|
def stop(self):
|
|
self._stop.set()
|
|
if self._sock:
|
|
try:
|
|
self._sock.close()
|
|
except OSError:
|
|
pass
|
|
if self._thread:
|
|
self._thread.join(timeout=1)
|
|
|
|
def _serve(self):
|
|
while not self._stop.is_set():
|
|
try:
|
|
frame, _addr = self._sock.recvfrom(2048)
|
|
except socket.timeout:
|
|
continue
|
|
except OSError as exc:
|
|
print(f"bootp: receive socket error: {exc!r}",
|
|
file=sys.stderr, flush=True)
|
|
return
|
|
try:
|
|
self._handle(frame)
|
|
except Exception as exc: # never let one request kill it
|
|
print(f"bootp: request error: {exc!r}", file=sys.stderr, flush=True)
|
|
|
|
def _handle(self, frame):
|
|
if len(frame) < 14 + 20 + 8 + 236 or frame[12:14] != b"\x08\x00":
|
|
return
|
|
ip = frame[14:]
|
|
if (ip[0] >> 4) != 4 or ip[9] != 17:
|
|
return
|
|
ihl = (ip[0] & 0x0F) * 4
|
|
udp = ip[ihl:]
|
|
sport, dport, ulen = struct.unpack(">HHH", udp[:6])
|
|
if dport != BOOTP_SERVER_PORT:
|
|
return
|
|
payload = udp[8:ulen]
|
|
if len(payload) < 236 or payload[0] != BOOTP_REQUEST:
|
|
return
|
|
mac = MAC_BYTES(payload[28:34])
|
|
self._sock.send(self._frame(payload))
|
|
self.log.append((time.time(), mac, payload[4:8].hex()))
|
|
note = "" if mac == VT1200_MAC else " (not the known VT1200 MAC)"
|
|
print(f"bootp: request from {mac} xid={payload[4:8].hex()} "
|
|
f"-> {self.client_ip} (siaddr {self.server_ip}, "
|
|
f"tap {self.iface}){note}", file=sys.stderr, flush=True)
|
|
|
|
def _frame(self, req):
|
|
# RFC 951 fixed 236-byte header plus a zero 64-byte vend area = 300.
|
|
bootp = bytearray(300)
|
|
bootp[0] = BOOTP_REPLY
|
|
bootp[1] = req[1] # htype, echoed
|
|
bootp[2] = req[2] # hlen, echoed
|
|
bootp[3] = 0 # hops
|
|
bootp[4:8] = req[4:8] # xid, echoed
|
|
bootp[8:10] = req[8:10] # secs, echoed
|
|
bootp[10:12] = req[10:12] # flags, echoed
|
|
# ciaddr and giaddr stay zero; this is not a relay.
|
|
bootp[16:20] = self._client_ip_b # yiaddr
|
|
bootp[20:24] = self._server_ip_b # siaddr
|
|
bootp[28:44] = req[28:44] # chaddr, echoed verbatim
|
|
# sname[44:108] and file[108:236] stay empty: the client asked for no
|
|
# boot file, so none is offered.
|
|
udp = struct.pack(">HHHH", BOOTP_SERVER_PORT, BOOTP_CLIENT_PORT,
|
|
8 + len(bootp), 0) + bytes(bootp)
|
|
ip = bytearray(struct.pack(">BBHHHBBH4s4s", 0x45, 0,
|
|
20 + len(udp), 0, 0, 64, 17, 0,
|
|
self._server_ip_b,
|
|
socket.inet_aton(BROADCAST_IP)))
|
|
struct.pack_into(">H", ip, 10, _ip_checksum(bytes(ip)))
|
|
return BROADCAST_MAC + self.src_mac + b"\x08\x00" + bytes(ip) + udp
|
|
|
|
|
|
def load(name):
|
|
with open(os.path.join(FONTS, name), "rb") as fh:
|
|
return fh.read()
|
|
|
|
|
|
def main(argv=None):
|
|
ap = argparse.ArgumentParser(
|
|
description="Serve VT1200 fonts over TFTP (optionally answer BOOTP).")
|
|
ap.add_argument("host", nargs="?", default="100.103.255.254",
|
|
help="address to bind the TFTP server to (default: %(default)s)")
|
|
ap.add_argument("port", nargs="?", type=int, default=69,
|
|
help="TFTP port (default: %(default)s)")
|
|
ap.add_argument("--bootp", action="store_true",
|
|
help="also answer BOOTP requests so the terminal learns its "
|
|
"IP address at power-up (off by default; needs root)")
|
|
ap.add_argument("--bootp-client", default="192.168.99.2",
|
|
help="address to hand the client (default: %(default)s)")
|
|
ap.add_argument("--bootp-server", default=None,
|
|
help="siaddr for the reply (default: the TFTP host)")
|
|
ap.add_argument("--bootp-iface", default=None,
|
|
help="MAME tap interface to attach to on L2 "
|
|
"(default: auto-detect tap-mess-*)")
|
|
args = ap.parse_args(argv)
|
|
|
|
baseline = load("baseline.pcf")
|
|
evil = load("evil-real.pcf")
|
|
probe = load("probe-uniq.pcf")
|
|
probe_snf = load("probe.snf")
|
|
probe_dec = load("probe-dec.snf")
|
|
probe_dec2 = load("probe-dec2.snf")
|
|
real = load("system-10x20.pcf")
|
|
fonts_dir = load("fonts.dir")
|
|
files = {}
|
|
# The font path value is <host>:<fontdir>. The terminal fetches
|
|
# "<fontdir>/fonts.dir" (the directory index), then the font files it names.
|
|
for d in ("69", "font", "fonts", "misc", "X11"):
|
|
files[f"{d}/fonts.dir"] = fonts_dir
|
|
files[f"{d}/baseline.pcf"] = baseline
|
|
files[f"{d}/v1200a.pcf"] = baseline
|
|
files[f"{d}/evil.pcf"] = evil
|
|
# also serve the plain names, in case it asks for those first
|
|
for n in ("69", "fonts.dir", "font", "fonts", "ROM", "rom", "real.pcf"):
|
|
files[n] = real
|
|
files["baseline.pcf"] = baseline
|
|
files["evil.pcf"] = evil
|
|
if os.path.exists(os.path.join(FONTS, "fonts.dir")):
|
|
files["fonts.dir"] = load("fonts.dir")
|
|
|
|
host = args.host
|
|
port = args.port
|
|
def resolve(name):
|
|
# The terminal first fetches the bare path, then "<dir>/fonts.dir".
|
|
# Accept ANY name: the index for fonts.dir, the payload otherwise.
|
|
if name == "fonts.dir" or name.endswith("/fonts.dir"):
|
|
return fonts_dir
|
|
if name.endswith(".snf"):
|
|
if "dec2" in name:
|
|
return probe_dec2
|
|
return probe_dec if "dec" in name else probe_snf
|
|
if name.endswith((".pcf", ".DECW$FONT")):
|
|
if "evil" in name:
|
|
return evil
|
|
if "probe" in name:
|
|
return probe
|
|
return baseline
|
|
return baseline # bare directory name: give it data, not an error
|
|
|
|
srv = fontgen.TFTPServer((host, port), files, timeout=60.0, retries=10,
|
|
resolver=resolve).start()
|
|
print(f"serving {len(files)} names on {srv.address}", flush=True)
|
|
print(" names: " + ", ".join(sorted(files)), flush=True)
|
|
|
|
bootp = None
|
|
if args.bootp:
|
|
bootp = BootpResponder(
|
|
server_ip=args.bootp_server or host,
|
|
client_ip=args.bootp_client,
|
|
iface=args.bootp_iface).start()
|
|
print(f"bootp: L2 responder on {bootp.iface}: handing out "
|
|
f"{bootp.client_ip}, siaddr {bootp.server_ip} "
|
|
f"(src mac {MAC_BYTES(bootp.src_mac)})", flush=True)
|
|
try:
|
|
while True:
|
|
time.sleep(0.5)
|
|
while srv.log:
|
|
ts, name, addr = srv.log.pop(0)
|
|
stamp = time.strftime("%H:%M:%S", time.localtime(ts))
|
|
print(f"{stamp} {addr[0]}:{addr[1]} requested {name!r}", flush=True)
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
srv.stop()
|
|
if bootp:
|
|
bootp.stop()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|