mirror of
https://git.k-space.ee/arti/vt1200.git
synced 2026-10-30 12:03:54 +02:00
The firmware's tftp_open hangs in its resolver from the rdebug process, so the payload speaks RFC 783 over the firmware socket API: socket(2,2,0), sendto(RRQ), recvfrom(DATA)/ACK into a malloc'd buffer, close. Verified live: 600 B fetched from 192.168.99.1 into the malloc buffer, server RRQ logged, and rdebug peek matches probe.bin byte-for-byte. Notes: dwt/notes/TCPIP.md (function table, ABI, limits) and the loader/transport hand-off in dwt/notes/LOADER_CROSS.md.
76 lines
2.8 KiB
Python
76 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
|
"""TFTP + BOOTP server for the dwt/tcpip fetch experiment.
|
|
|
|
Serves ``dwt/tcpip/probe.bin`` for *any* requested name (the terminal's TFTP
|
|
client asks for the exact filename we put in the payload) and logs every RRQ.
|
|
``--bootp`` reuses the L2 BOOTP responder from ``harness/serve.py`` so the
|
|
terminal learns 192.168.99.2 at power-up; that file is imported, not modified.
|
|
|
|
sudo python3 -u dwt/tcpip/serve_probe.py 192.168.99.1 --bootp
|
|
|
|
Run this instead of ``harness/serve.py`` when the file to serve is the probe:
|
|
both want UDP 69, so only one can be up.
|
|
"""
|
|
import argparse
|
|
import os
|
|
import sys
|
|
import time
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
ROOT = os.path.dirname(os.path.dirname(HERE))
|
|
sys.path.insert(0, os.path.join(ROOT, "tools"))
|
|
sys.path.insert(0, os.path.join(ROOT, "harness"))
|
|
|
|
import fontgen # noqa: E402
|
|
import serve as hserve # noqa: E402
|
|
|
|
PROBE = os.path.join(HERE, "probe.bin")
|
|
|
|
|
|
def main(argv=None):
|
|
ap = argparse.ArgumentParser(description=__doc__,
|
|
formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
ap.add_argument("host", nargs="?", default="192.168.99.1")
|
|
ap.add_argument("port", nargs="?", type=int, default=69)
|
|
ap.add_argument("--bootp", action="store_true")
|
|
ap.add_argument("--bootp-client", default="192.168.99.2")
|
|
ap.add_argument("--bootp-iface", default=None)
|
|
ap.add_argument("--file", default=PROBE)
|
|
args = ap.parse_args(argv)
|
|
|
|
content = open(args.file, "rb").read()
|
|
# Answer any name with the probe: the payload chooses the name, and the
|
|
# log records exactly what the terminal asked for.
|
|
srv = fontgen.TFTPServer((args.host, args.port), {}, timeout=60.0,
|
|
retries=10, resolver=lambda name: content).start()
|
|
print(f"tftp: serving {len(content)} bytes ({os.path.basename(args.file)}) "
|
|
f"for any name on {srv.address}", flush=True)
|
|
|
|
bootp = None
|
|
if args.bootp:
|
|
bootp = hserve.BootpResponder(server_ip=args.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}", flush=True)
|
|
|
|
start = time.time()
|
|
try:
|
|
while True:
|
|
time.sleep(0.2)
|
|
while srv.log:
|
|
ts, name, addr = srv.log.pop(0)
|
|
stamp = time.strftime("%H:%M:%S", time.localtime(ts))
|
|
print(f"{stamp} RRQ {addr[0]}:{addr[1]} requested {name!r} "
|
|
f"({content.__len__()} bytes)", flush=True)
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
srv.stop()
|
|
if bootp:
|
|
bootp.stop()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|