rebbarb/scripts/test-fsp-server.py

#!/usr/bin/env python3
"""Self-test for fsp-server.py, using a client that packs requests the way
Swiss's fsplib/tcpip.c do (path\\npassword\\0, 2-byte xlen, v2 checksum)."""
import os, socket, struct, subprocess, sys, tempfile, time

HSIZE = 12
CC_VERSION, CC_ERR, CC_GET_DIR, CC_GET_FILE, CC_STAT = 0x10, 0x40, 0x41, 0x42, 0x4D
PW = "alltheroms"
fails = []


def csum(pkt):
    """fsp_pkt_write / tcpip.c: INCLUDES the length.  For requests."""
    t = len(pkt) + sum(pkt)
    t += t >> 8
    return t & 0xFF


def fsplib_verify(pkt):
    """fsp_pkt_read, transcribed literally -- this is what Swiss's menu client
    accepts, and it does NOT add the length:

        mysum = -ptr[SUM]; for(i..recv_len) mysum += ptr[i];
        mysum = (mysum + (mysum >> 8)) & 0xff;
        ok if mysum == ptr[SUM]

    ⚠️ Verify replies with THIS, never with the server's own function: an
    earlier version of this test used the server's symmetric checksum, passed
    every assertion, and the real Swiss client rejected every reply."""
    mysum = -pkt[1]
    for b in pkt:
        mysum += b
    mysum = (mysum + (mysum >> 8)) & 0xFF
    return mysum == pkt[1]


def req(cmd, seq, pos, path, pw=PW, extra=b""):
    data = path.encode() + (b"\n" + pw.encode() if pw else b"") + b"\0"
    body = struct.pack("!BBHHHI", cmd, 0, 0x1234, seq, len(data), pos) + data + extra
    return body[:1] + bytes([csum(body)]) + body[2:]


def parse(pkt):
    cmd, cs, key, seq, ln, pos = struct.unpack("!BBHHHI", pkt[:HSIZE])
    # Replies are checked with FSPLIB's verifier, which is the only opinion
    # that matters -- not with csum() above, which is the request formula.
    return dict(cmd=cmd, seq=seq, pos=pos, data=pkt[HSIZE:HSIZE + ln],
                csum_ok=fsplib_verify(pkt))


tmp = tempfile.mkdtemp()
PAYLOAD = bytes((i * 7 + 3) & 0xFF for i in range(8192))
open(os.path.join(tmp, "dvd.iso"), "wb").write(PAYLOAD)
os.mkdir(os.path.join(tmp, "subdir"))

srv = subprocess.Popen(
    [sys.executable, "-I", "scripts/fsp-server.py", "--root", tmp,
     "--password", PW, "--bind", "127.0.0.1", "--port", "2121", "--quiet-reads"],
    stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)
time.sleep(1.0)
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.settimeout(3.0)
A = ("127.0.0.1", 2121)


def rt(pkt):
    s.sendto(pkt, A)
    raw = s.recv(2048)
    if not fsplib_verify(raw):
        fails.append(f"reply to cmd {pkt[0]:#04x} FAILS fsplib's verifier "
                     f"-- Swiss would drop it and retry")
    return parse(raw)

try:
    # T1 checksum both ways
    r = rt(req(CC_VERSION, 1, 0, ""))
    if r["cmd"] != CC_VERSION or not r["csum_ok"]:
        fails.append(f"T1 VERSION: cmd {r['cmd']:#04x} csum_ok {r['csum_ok']}")
    else:
        print("T1 VERSION answered, reply checksum verifies")

    # T2 STAT returns the real size
    r = rt(req(CC_STAT, 2, 0, "/dvd.iso"))
    t, sz, ty = struct.unpack("!IIB", r["data"][:9])
    if sz != len(PAYLOAD) or ty != 0x01:
        fails.append(f"T2 STAT size {sz} type {ty}, want {len(PAYLOAD)} / 1")
    else:
        print(f"T2 STAT -> size {sz}, type FILE")

    # T3 GET_DIR lists the file and the subdir, and parses as fsplib does
    r = rt(req(CC_GET_DIR, 3, 0, "/"))
    d, i, found = r["data"], 0, {}
    while i + 9 <= len(d):
        ty = d[i + 8]
        if ty == 0x00:
            break
        sz = struct.unpack("!I", d[i + 4:i + 8])[0]
        end = d.index(b"\0", i + 9)
        found[d[i + 9:end].decode()] = (ty, sz)
        i = end + 1
        i += (-i) % 4
    if found.get("dvd.iso") != (0x01, len(PAYLOAD)) or found.get("subdir", (0,))[0] != 0x02:
        fails.append(f"T3 GET_DIR entries wrong: {found}")
    else:
        print(f"T3 GET_DIR -> {found}")
    if (len(r['data'])) % 4:
        fails.append("T3 dir block not 4-byte aligned (fsplib rejects dirpos%4)")

    # T4 second GET_DIR page must be EMPTY so the client stops looping
    r2 = rt(req(CC_GET_DIR, 4, len(r["data"]), "/"))
    if len(r2["data"]) != 0:
        fails.append(f"T4 second dir page returned {len(r2['data'])} B, want 0")
    else:
        print("T4 second GET_DIR page empty -> client terminates")

    # T5 GET_FILE: right bytes, and sequence+position ECHOED (the in-game
    # client discards a reply whose seq or pos does not match)
    for pos, want in ((0, 1024), (4096, 1024), (8000, 1024)):
        r = rt(req(CC_GET_FILE, 10 + pos, pos, "/dvd.iso",
                   extra=struct.pack("!H", want)))
        exp = PAYLOAD[pos:pos + want]
        if r["data"] != exp:
            fails.append(f"T5 GET_FILE pos={pos}: {len(r['data'])} B, "
                         f"want {len(exp)}, match={r['data']==exp}")
        elif r["seq"] != 10 + pos or r["pos"] != pos:
            fails.append(f"T5 seq/pos not echoed: seq {r['seq']} pos {r['pos']}")
    print("T5 GET_FILE at 0 / 4096 / EOF-straddle: bytes exact, seq+pos echoed")

    # T6 a 32-aligned request must give a 32-aligned reply, or the in-game
    # client's OSRoundDown32B silently truncates and the transfer stalls
    r = rt(req(CC_GET_FILE, 99, 0, "/dvd.iso", extra=struct.pack("!H", 1024)))
    if len(r["data"]) % 32:
        fails.append(f"T6 reply {len(r['data'])} B is not a multiple of 32")
    else:
        print(f"T6 1024 B request -> {len(r['data'])} B, 32-aligned")

    # T7 wrong password must be refused
    r = rt(req(CC_STAT, 7, 0, "/dvd.iso", pw="wrong"))
    if r["cmd"] != CC_ERR:
        fails.append(f"T7 wrong password accepted (cmd {r['cmd']:#04x})")
    else:
        print("T7 wrong password -> CC_ERR")

    # T8 path traversal must not escape the root
    r = rt(req(CC_STAT, 8, 0, "/../../etc/passwd"))
    t, sz, ty = struct.unpack("!IIB", (r["data"] + b"\0" * 9)[:9])
    if r["cmd"] != CC_ERR and ty != 0:
        fails.append(f"T8 traversal escaped the root: type {ty} size {sz}")
    else:
        print("T8 ../../etc/passwd refused")
    # T9 GET_PRO: flags go in the EXTRA region at buf[len], and pos must
    # equal FSP_PRO_BYTES (1) or fsplib returns ENOMSG.
    CC_GET_PRO = 0x47
    r = rt(req(CC_GET_PRO, 20, 0, ""))
    if r["cmd"] != CC_GET_PRO:
        fails.append(f"T9 GET_PRO cmd {r['cmd']:#04x}")
    elif r["pos"] != 1:
        fails.append(f"T9 GET_PRO pos {r['pos']}, want 1 (FSP_PRO_BYTES) "
                     f"- fsplib returns ENOMSG otherwise")
    else:
        s.sendto(req(CC_GET_PRO, 21, 0, ""), A)
        raw = s.recv(2048)
        ln = struct.unpack("!H", raw[8:10])[0]
        pro = raw[HSIZE + ln]          # buf[len] -- the first extra byte
        if not (pro & 0x10) or not (pro & 0x40):
            fails.append(f"T9 GET_PRO flags {pro:#04x} lack GET|LIST")
        elif pro & 0x01:
            fails.append(f"T9 GET_PRO claims OWNER without --claim-owner")
        else:
            print(f"T9 GET_PRO -> pos=1, flags {pro:#04x} (GET|LIST) at buf[len]")

    # T10 a missing directory must be CC_ERR, so fsplib reports ENOENT and
    # Swiss falls back from "swiss/patches" to "swiss" to "".
    r = rt(req(CC_GET_PRO, 22, 0, "swiss/patches"))
    if r["cmd"] != CC_ERR:
        fails.append(f"T10 GET_PRO on a missing dir returned {r['cmd']:#04x}, "
                     f"want CC_ERR -- Swiss's fallback chain needs ENOENT")
    else:
        print("T10 GET_PRO 'swiss/patches' -> CC_ERR (Swiss falls back to root)")

finally:
    srv.terminate()

print("\nFAILURES:\n  " + "\n  ".join(fails) if fails else
      "\nAll fsp-server self-tests passed.")
sys.exit(1 if fails else 0)