#!/usr/bin/env python3
"""Hostile raw-9P2000 client aimed at the 9proc *core* (stdlib only).

Complements adv_9proc_hostile.py in this directory (framing, tags, scratch, floods)
with attacks on the freestanding engine's own paths: the /vars tree and its
comptime renderers, snapshot slots, the static tree, the fid table at its
configured maximum, directory-read offsets, the msize floor, the ctl staging rule,
and the demo's debug providers driven as black boxes.

Usage:
    adv_core_hostile.py --server zig-out/bin/9proc-demo   # spawns it on a temp unix socket
    adv_core_hostile.py --socket PATH                      # attacks a running server

Exit status is non-zero if any check fails or the server dies.
"""
import argparse
import os
import signal
import struct
import subprocess
import sys
import tempfile
import threading
import time

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import adv_9proc_hostile as base  # noqa: E402
from adv_9proc_hostile import (  # noqa: E402
    NOTAG, Tversion, Tflush, Rflush, Twalk, Rwalk, Topen, Ropen, Rcreate,
    Tread, Rread, Twrite, Rwrite, Tclunk, Rclunk, Tremove, Rremove, Tstat, Rstat, Twstat, Rwstat,
    Rerror, OREAD, OWRITE, ORDWR, OEXEC, OTRUNC, ORCLOSE, DMDIR,
    Nine, frame, s16, mkstat, parse_stat, ok, healthy, expect_dead,
    MSIZE_MIN, E_UNKNOWN_FID, E_FID_IN_USE, E_TOO_MANY_FIDS, E_BAD_USE, E_ALREADY_OPEN, E_BAD_OFFSET,
    E_PERM, E_WSTAT, E_INVAL,
)

MAX_FIDS = 32768          # demo/main.zig cfg.max_fids
SNAPSHOT_SLOTS = 8        # demo/main.zig cfg.snapshot_slots (per connection)
SCRATCH_BUDGET = 512 << 20
SCRATCH_MAX_FILE = 64 << 20


def records(d):
    """Splits a directory read into (name, raw-record) pairs."""
    out = []
    while d:
        n, = struct.unpack_from("<H", d)
        out.append((parse_stat(d[:n + 2])["name"], d[:n + 2]))
        d = d[n + 2:]
    return out


def qid_of_walk(rb):
    n, = struct.unpack_from("<H", rb)
    return [struct.unpack_from("<BIQ", rb, 2 + 13 * i) for i in range(n)]


def worker_tid(c):
    """The tid of the demo's worker thread, via /threads/<tid>/name."""
    c.walk_ok(0, 40, [b"threads"])
    c.open(40, OREAD)
    d = c.read_all(40)
    c.clunk(40)
    for name, _ in records(d):
        if c.path_read([b"threads", name, b"name"], fid=41) == b"worker":
            return name
    return None


# --------------------------------------------------------------------------- /vars


def attack_vars(path):
    print("# /vars: deep walks, hostile names, renderer edge cases, hostile writes")
    c = Nine(path)
    c.session(1 << 20)
    deep = [b"vars", b"state", b"f", b"last_job", b"f", b"id", b".", b"..", b"id", b".", b"..", b"id", b".", b"..", b"id", b"value"]
    assert len(deep) == 16
    ok("16-element walk deep into /vars/state/f/... succeeds", c.walk_ok(0, 1, deep) == 16)
    rt, _, _ = c.open(1, OREAD)
    ok("deep walk lands on a readable value file", rt == Ropen, rt)
    c.clunk(1)
    up = [b"vars", b"state", b"f", b"inner"] if False else [b"vars", b"state", b"f", b"last_job"] + [b".."] * 12
    n = c.walk_ok(0, 1, up)
    ok("12 x '..' from inside /vars climbs to the root and stays there", n == 16, n)
    rt, st = c.stat(1)
    ok("fid after the climb is the root directory", rt == Rstat and st["qid"][2] == 0xFF << 56, st)
    c.clunk(1)
    # names that are hex/decimal edge cases or otherwise hostile: never anything but Rerror/partial walk
    for nm in (b"0", b"-1", b"0x", b"0x0", b"state\x00", b"State", b" state", b"state ", b"a" * 255, b"a" * 65535, b"\xff\xfe", b"..\x00", b"f", b"value"):
        n = c.walk_ok(0, 1, [b"vars", nm])
        ok(f"walk /vars/{nm[:12]!r}{'...' if len(nm) > 12 else ''} is a partial walk (1)", n == 1, n)
        ok("  and newfid stays unbound", c.err(Tclunk, struct.pack("<I", 1)) == E_UNKNOWN_FID)
    for nm in (b"0", b"F", b"f\x00", b"ticks", b"value ", b"raw\x00"):
        n = c.walk_ok(0, 1, [b"vars", b"state", nm])
        ok(f"walk /vars/state/{nm!r} is a partial walk (2)", n == 2, n)
    # . and .. on var files and directories
    ok("walk '.' from a var file is 'not a directory'", c.walk_ok(0, 1, [b"vars", b"state", b"value"]) == 3 and c.err(Twalk, struct.pack("<IIH", 1, 2, 1) + s16(b".")) == "not a directory")
    ok("walk '..' from a var file is 'not a directory'", c.err(Twalk, struct.pack("<IIH", 1, 2, 1) + s16(b"..")) == "not a directory")
    c.clunk(1)
    c.walk_ok(0, 1, [b"vars", b"state", b"f"])
    rt, _, rb = c.walk(1, 2, [b".", b"..", b"..", b".."])
    q = qid_of_walk(rb) if rt == Rwalk else []
    ok("'.' and '..' through the var tree: f -> state -> /vars -> /", len(q) == 4 and q[3][2] == 0xFF << 56 and (q[1][0] & 0x80), q)
    c.clunk(1)
    c.clunk(2)
    # every file under /vars/state reads; raw reads beyond @sizeOf are empty
    size = int(c.path_read([b"vars", b"state", b"size"]))
    ok("/vars/state/size is a number", size > 0, size)
    raw = c.path_read([b"vars", b"state", b"raw"])
    ok("/vars/state/raw has exactly @sizeOf bytes", raw is not None and len(raw) == size, (len(raw) if raw else raw, size))
    c.walk_ok(0, 1, [b"vars", b"state", b"raw"])
    c.open(1, OREAD)
    rt, d = c.read(1, size, 100)
    ok("raw read at offset @sizeOf is empty", rt == Rread and d == b"", (rt, d))
    rt, d = c.read(1, size - 1, 100)
    ok("raw read at @sizeOf-1 returns one byte", rt == Rread and len(d) == 1, (rt, d))
    rt, d = c.read(1, (1 << 64) - 1, 100)
    ok("raw read at 2^64-1 is empty", rt == Rread and d == b"")
    rt, d = c.read(1, 0, 0xFFFFFFFF)
    ok("raw read with count 2^32-1 is clamped", rt == Rread and len(d) == size, (rt, len(d) if d else d))
    rt, st = c.stat(1)
    ok("raw stat length is @sizeOf and mode 0444", rt == Rstat and st["length"] == size and st["mode"] == 0o444, st)
    ok("raw is read-only", c.err(Twrite, struct.pack("<IQI", 1, 0, 1) + b"x") is not None)
    c.clunk(1)
    for leaf in (b"type", b"size", b"addr", b"value"):
        c.walk_ok(0, 1, [b"vars", b"state", leaf])
        e = c.err(Topen, struct.pack("<IB", 1, OWRITE))
        ok(f"/vars/state/{leaf.decode()} refuses OWRITE", e == "permission denied", e)
        e = c.err(Topen, struct.pack("<IB", 1, OREAD | OTRUNC))
        ok(f"/vars/state/{leaf.decode()} refuses OTRUNC", e == "permission denied", e)
        c.clunk(1)
    v = c.path_read([b"vars", b"state", b"value"])
    ok("/vars/state/value renders every field", v is not None and all(k in v for k in (b"ticks:", b"phase:", b"last_job:", b"id:", b"cost:")), v)
    ok("nested struct is indented", b"\n  id: " in (v or b""), v)
    # dynamic file: read at offset 0 regenerates, offset 1 is the tail of the same snapshot
    c.walk_ok(0, 1, [b"vars", b"state", b"value"])
    c.open(1, OREAD)
    rt, d = c.read(1, 0, 8192)
    rt2, d2 = c.read(1, 1, 8192)
    ok("value read at offset 1 is the tail of the snapshot", rt == Rread and rt2 == Rread and d2 == d[1:], (d, d2))
    rt3, d3 = c.read(1, len(d), 8192)
    ok("value read at the end is empty", rt3 == Rread and d3 == b"")
    rt4, d4 = c.read(1, (1 << 63) + 5, 10)
    ok("value read at 2^63+5 is empty", rt4 == Rread and d4 == b"")
    rt, st = c.stat(1)
    ok("value stat reports length 0 (dynamic)", rt == Rstat and st["length"] == 0, st)
    c.clunk(1)
    # hostile writes to scalar values: garbage, huge, negative, floats with exponents, NULs, empty
    c.walk_ok(0, 1, [b"vars", b"state", b"f", b"ticks", b"value"])
    rt, _, _ = c.open(1, OWRITE | OTRUNC)
    ok("open ticks/value OWRITE|OTRUNC", rt == Ropen, rt)
    for bad in (b"abc", b"99999999999999999999999", b"-1", b"1e3", b"", b"   ", b"4\x002", b"1.5", b"0x", b"+", b"\xd9\xa1\xd9\xa2", b"12 34", b"0b102"):
        e = c.err(Twrite, struct.pack("<IQI", 1, 0, len(bad)) + bad)
        ok(f"write {bad!r} to u64 value is 'bad value'", e == "bad value", e)
    ok("read on the write-only value fid is 'file not open'", c.err(Tread, struct.pack("<IQI", 1, 0, 10)) == E_BAD_USE)
    for good, want in ((b"  4200  \n", 4200), (b"0x10", 16), (b"+7", 7), (b"0b1010", 10), (b"0o17", 15), (b"1_000", 1000), (b"18446744073709551615", (1 << 64) - 1)):
        rt, _, rb = c.write(1, (1 << 64) - 1, good)  # offset is ignored for values
        got = c.path_read([b"vars", b"state", b"f", b"ticks", b"value"])
        try:
            gv = int(got)
        except (TypeError, ValueError):
            gv = None
        # the worker keeps incrementing (wrapping), so allow a small drift
        ok(f"write {good!r} stores {want}", rt == Rwrite and gv is not None and (gv - want) % (1 << 64) < 100_000, (rt, got))
    c.write(1, 0, b"1")
    c.clunk(1)
    # enum and float and u32 leaves
    c.walk_ok(0, 1, [b"vars", b"state", b"f", b"phase", b"value"])
    c.open(1, ORDWR)
    for bad in (b"trap\x00ped", b"IDLE", b"2", b"", b"idle extra", b"\x00idle\x00x", b"idl", b"idle\x00\x00x"):
        e = c.err(Twrite, struct.pack("<IQI", 1, 0, len(bad)) + bad)
        ok(f"enum write {bad!r} is 'bad value'", e == "bad value", e)
    rt, _, _ = c.write(1, 0, b"\n idle \x00")
    rt2, d = c.read(1, 0, 100)
    ok("enum write with surrounding whitespace/NUL is accepted", rt == Rwrite and d in (b"idle", b"working", b"trapped"), (rt, d))
    rt, st = c.stat(1)
    ok("enum value is 0644", rt == Rstat and st["mode"] == 0o644, st)
    c.clunk(1)
    c.walk_ok(0, 1, [b"vars", b"state", b"f", b"last_job", b"f", b"cost", b"value"])
    c.open(1, ORDWR)
    for bad in (b"abc", b"1.5.5", b"e5", b"", b"0x", b"1e", b"--1"):
        ok(f"float write {bad!r} is 'bad value'", c.err(Twrite, struct.pack("<IQI", 1, 0, len(bad)) + bad) == "bad value")
    for good in (b"1.5e3", b"-0x1p-2", b"1e999", b"nan", b"-0", b"2.5"):
        rt, _, _ = c.write(1, 0, good)
        rt2, d = c.read(1, 0, 100)
        ok(f"float write {good!r} accepted and renders ({d!r})", rt == Rwrite and rt2 == Rread and d != b"", (rt, d))
    c.write(1, 0, b"0")
    c.clunk(1)
    c.walk_ok(0, 1, [b"vars", b"state", b"f", b"last_job", b"f", b"id", b"value"])
    c.open(1, OWRITE)
    ok("u32 write 4294967296 is 'bad value'", c.err(Twrite, struct.pack("<IQI", 1, 0, 10) + b"4294967296") == "bad value")
    ok("u32 write -0 is accepted as 0 (std.fmt.parseInt semantics)", c.write(1, 0, b"-0")[0] == Rwrite)
    ok("u32 write -1 is 'bad value'", c.err(Twrite, struct.pack("<IQI", 1, 0, 2) + b"-1") == "bad value")
    rt, _, _ = c.write(1, 0, b"4294967295")
    ok("u32 write 4294967295 accepted", rt == Rwrite)
    c.write(1, 0, b"0")
    c.clunk(1)
    # struct values and the f directory are not writable; the tree admits no create/remove/wstat
    for names in ([b"vars"], [b"vars", b"state"], [b"vars", b"state", b"f"], [b"vars", b"state", b"f", b"last_job"], [b"vars", b"state", b"f", b"last_job", b"f"]):
        c.walk_ok(0, 1, names)
        p = b"/".join(names).decode()
        ok(f"create in {p} is denied", c.err(base.Tcreate, struct.pack("<I", 1) + s16(b"x") + struct.pack("<IB", 0o644, OWRITE)) == "permission denied")
        ok(f"wstat of {p} is denied", c.err(Twstat, struct.pack("<I", 1) + s16(mkstat(name=b"y"))) == "permission denied")
        ok(f"open {p} for write is refused by the engine", c.err(Topen, struct.pack("<IB", 1, OWRITE)) == E_PERM)
        ok(f"remove {p} is denied", c.err(Tremove, struct.pack("<I", 1)) == "permission denied")
        ok(f"  and the fid was clunked", c.err(Tclunk, struct.pack("<I", 1)) == E_UNKNOWN_FID)
    for names in ([b"vars", b"state", b"value"], [b"vars", b"state", b"f", b"last_job", b"value"], [b"vars", b"state", b"type"]):
        c.walk_ok(0, 1, names)
        p = b"/".join(names).decode()
        ok(f"wstat of {p} is denied", c.err(Twstat, struct.pack("<I", 1) + s16(mkstat(length=0))) == "permission denied")
        ok(f"{p} is not writable", c.err(Topen, struct.pack("<IB", 1, ORDWR)) == "permission denied")
        ok(f"remove {p} is denied", c.err(Tremove, struct.pack("<I", 1)) == "permission denied")
    # directory listing of /vars/state and of f/, exact record boundaries
    c.walk_ok(0, 1, [b"vars", b"state"])
    c.open(1, OREAD)
    d = c.read_all(1)
    names = [n for n, _ in records(d)]
    ok("/vars/state lists value,type,size,addr,raw,f", sorted(names) == sorted([b"value", b"type", b"size", b"addr", b"raw", b"f"]), names)
    c.clunk(1)
    c.walk_ok(0, 1, [b"vars", b"state", b"f"])
    c.open(1, OREAD)
    d = c.read_all(1)
    names = [n for n, _ in records(d)]
    ok("/vars/state/f lists the three fields", sorted(names) == [b"last_job", b"phase", b"ticks"], names)
    c.clunk(1)
    c.close()
    ok("server healthy after /vars attacks", healthy(path))


# --------------------------------------------------------------------------- snapshot slots


def attack_snapshots(path):
    print("# snapshot slots: exhaustion, hold, release by clunk and by Tversion; per-connection")
    dyn = [[b"runtime", b"pid"], [b"runtime", b"ppid"], [b"runtime", b"uptime"], [b"runtime", b"fn", b"now"], [b"runtime", b"fn", b"fib30"],
           [b"vars", b"state", b"value"], [b"vars", b"state", b"addr"], [b"vars", b"state", b"f", b"ticks", b"value"], [b"vars", b"state", b"f", b"phase", b"value"], [b"runtime", b"fn", b"uname"]]
    c = Nine(path)
    c.session()
    opened = 0
    err = None
    for i, names in enumerate(dyn):
        c.walk_ok(0, 100 + i, names)
        rt, _, rb = c.open(100 + i, OREAD)
        if rt == Ropen:
            opened += 1
        else:
            err = c.err.__self__ and rb
            n, = struct.unpack_from("<H", rb)
            err = rb[2:2 + n].decode()
            break
    ok(f"exactly {SNAPSHOT_SLOTS} dynamic files open per connection", opened == SNAPSHOT_SLOTS, opened)
    ok("the next open is 'too many open dynamic files'", err == "too many open dynamic files", err)
    ok("the refused fid is still unopened (read is 'file not open')", c.err(Tread, struct.pack("<IQI", 100 + opened, 0, 10)) == E_BAD_USE)
    # every held snapshot is still readable and consistent at offset 1
    for i in range(opened):
        rt, d = c.read(100 + i, 0, 8192)
        rt2, d2 = c.read(100 + i, 1, 8192)
        ok(f"held snapshot {i} reads and its offset-1 read is the tail", rt == Rread and rt2 == Rread and d2 == d[1:], (rt, rt2))
    # static and provider files need no slot
    ok("static file opens while slots are exhausted", c.path_read([b"build", b"zig_version"]) not in (None, b""))
    ok("/vars/state/type opens while slots are exhausted", c.path_read([b"vars", b"state", b"type"]) not in (None, b""))
    ok("/vars/state/raw opens while slots are exhausted", c.path_read([b"vars", b"state", b"raw"]) not in (None, b""))
    ok("scratch root lists while slots are exhausted", c.walk_ok(0, 50, [b"scratch"]) == 1 and c.open(50, OREAD)[0] == Ropen)
    c.clunk(50)
    # a second connection has its own slots
    c2 = Nine(path)
    c2.session()
    n2 = 0
    for i, names in enumerate(dyn[:SNAPSHOT_SLOTS]):
        c2.walk_ok(0, 100 + i, names)
        n2 += c2.open(100 + i, OREAD)[0] == Ropen
    ok("a second connection opens its own 8 dynamic files", n2 == SNAPSHOT_SLOTS, n2)
    c2.close()
    # clunk one -> the refused one now opens; clunk via Tremove (denied) also frees the slot
    c.clunk(100)
    rt, _, _ = c.open(100 + opened, OREAD)
    ok("after one clunk the refused open succeeds", rt == Ropen, rt)
    ok("remove of an open dynamic file is denied", c.err(Tremove, struct.pack("<I", 101)) == "permission denied")
    c.walk_ok(0, 60, dyn[0])
    rt, _, _ = c.open(60, OREAD)
    ok("the failed-remove fid's slot was released", rt == Ropen, rt)
    # an open fid cannot be cloned; a fresh walk to the same file takes no slot and is unopened
    ok("clone of an open dynamic fid is refused", c.err(Twalk, struct.pack("<IIH", 60, 61, 0)) == E_BAD_USE)
    c.walk_ok(0, 61, dyn[0])
    ok("the fresh fid is not open", c.err(Tread, struct.pack("<IQI", 61, 0, 10)) == E_BAD_USE)
    ok("the fresh fid cannot open (slots exhausted again)", c.err(Topen, struct.pack("<IB", 61, OREAD)) == "too many open dynamic files")
    # Tversion releases everything: 8 opens succeed again
    rt, ms, _ = c.version(65536)
    ok("mid-session Tversion", rt == base.Rversion)
    c.attach()
    n3 = 0
    for i, names in enumerate(dyn[:SNAPSHOT_SLOTS]):
        c.walk_ok(0, 100 + i, names)
        n3 += c.open(100 + i, OREAD)[0] == Ropen
    ok("after Tversion all 8 slots are free again", n3 == SNAPSHOT_SLOTS, n3)
    c.close()
    ok("server healthy after snapshot attacks", healthy(path))


# --------------------------------------------------------------------------- the static tree


def attack_static(path):
    print("# static tree: create/remove/wstat everywhere, '.'/'..' on files and provider roots")
    c = Nine(path)
    c.session()
    dirs = [[], [b"build"], [b"comptime"], [b"comptime", b"types"], [b"comptime", b"types", b"Qid"], [b"runtime"], [b"runtime", b"fn"]]
    for names in dirs:
        p = "/" + b"/".join(names).decode()
        ok(f"walk {p}", c.walk_ok(0, 1, names) == len(names))
        ok(f"create in {p} is denied", c.err(base.Tcreate, struct.pack("<I", 1) + s16(b"x") + struct.pack("<IB", 0o644, OWRITE)) == "permission denied")
        ok(f"mkdir in {p} is denied", c.err(base.Tcreate, struct.pack("<I", 1) + s16(b"d") + struct.pack("<IB", DMDIR | 0o755, OREAD)) == "permission denied")
        ok(f"wstat of {p} is denied", c.err(Twstat, struct.pack("<I", 1) + s16(mkstat(mtime=1))) == "permission denied")
        ok(f"wstat of {p} with all don't-care is a no-op the engine answers itself", c.wstat(1, mkstat())[0] == Rwstat)
        ok(f"open {p} ORDWR is refused by the engine", c.err(Topen, struct.pack("<IB", 1, ORDWR)) == E_PERM)
        ok(f"open {p} OEXEC is refused by the engine", c.err(Topen, struct.pack("<IB", 1, OEXEC)) == E_PERM)
        ok(f"open {p} OREAD", c.open(1, OREAD)[0] == Ropen)
        ok(f"write to open {p} is 'is a directory'", c.err(Twrite, struct.pack("<IQI", 1, 0, 1) + b"x") is not None)
        ok(f"remove {p} is denied", c.err(Tremove, struct.pack("<I", 1)) == "permission denied")
        ok(f"  and clunked", c.err(Tclunk, struct.pack("<I", 1)) == E_UNKNOWN_FID)
    files = [[b"README"], [b"build", b"time"], [b"comptime", b"decls"], [b"comptime", b"types", b"Qid", b"fields"], [b"runtime", b"pid"], [b"runtime", b"fn", b"fib30"], [b"runtime", b"ctl"]]
    for names in files:
        p = "/" + b"/".join(names).decode()
        ok(f"walk {p}", c.walk_ok(0, 1, names) == len(names))
        ok(f"'.' from {p} is 'not a directory'", c.err(Twalk, struct.pack("<IIH", 1, 2, 1) + s16(b".")) == "not a directory")
        ok(f"'..' from {p} is 'not a directory'", c.err(Twalk, struct.pack("<IIH", 1, 2, 1) + s16(b"..")) == "not a directory")
        ok(f"wstat of {p} is denied", c.err(Twstat, struct.pack("<I", 1) + s16(mkstat(length=0))) == "permission denied")
        ok(f"remove {p} is denied", c.err(Tremove, struct.pack("<I", 1)) == "permission denied")
        ok(f"  and clunked", c.err(Tclunk, struct.pack("<I", 1)) == E_UNKNOWN_FID)
        if names[-1] != b"ctl":
            c.walk_ok(0, 1, names)
            ok(f"open {p} OWRITE is denied", c.err(Topen, struct.pack("<IB", 1, OWRITE)) == "permission denied")
            ok(f"open {p} OREAD|OTRUNC is denied", c.err(Topen, struct.pack("<IB", 1, OREAD | OTRUNC)) == "permission denied")
            ok(f"open {p} ORCLOSE alone reads (no removal on clunk)", c.open(1, OREAD | ORCLOSE)[0] == Ropen and c.read(1, 0, 10)[0] == Rread)
            c.clunk(1)
            ok(f"{p} still exists after ORCLOSE clunk", c.walk_ok(0, 1, names) == len(names))
            c.clunk(1)
    # '.' and '..' on provider roots: '.' is the same qid, '..' is the server root
    for prov in (b"scratch", b"threads", b"addr", b"mem", b"hex", b"breakpoints", b"panic"):
        n = c.walk_ok(0, 1, [prov])
        rt, _, rb = c.walk(1, 2, [b".", b".."])
        q = qid_of_walk(rb) if rt == Rwalk else []
        rt2, st = c.stat(1)
        ok(f"/{prov.decode()}: '.' keeps the qid and '..' reaches the root", n == 1 and len(q) == 2 and q[0][2] == st["qid"][2] and q[1][2] == 0xFF << 56, (n, q))
        ok(f"/{prov.decode()}: root stat name is the mount name", rt2 == Rstat and st["name"] == prov, st)
        ok(f"/{prov.decode()}: rename of the provider root is denied", c.err(Twstat, struct.pack("<I", 1) + s16(mkstat(name=b"other"))) == "permission denied")
        ok(f"/{prov.decode()}: remove of the provider root is denied", c.err(Tremove, struct.pack("<I", 1)) == "permission denied")
        c.clunk(2)
    # qid path spaces do not collide: static tag 0xFF, vars 0xFE, providers 0..n
    seen = {}
    for names in dirs + files + [[b"vars"], [b"vars", b"state"], [b"vars", b"state", b"value"], [b"scratch"], [b"threads"], [b"panic", b"message"], [b"mem", b"maps"]]:
        c.walk_ok(0, 1, names)
        rt, st = c.stat(1)
        c.clunk(1)
        key = st["qid"][2]
        ok(f"qid path of /{b'/'.join(names).decode()} is unique", key not in seen, (key, seen.get(key)))
        seen[key] = names
    # walks through the whole tree with 16 elements of '..' never leave the root
    ok("16 x '..' from root stays at root", c.walk_ok(0, 1, [b".."] * 16) == 16)
    rt, st = c.stat(1)
    ok("  and it is the root", rt == Rstat and st["qid"][2] == 0xFF << 56)
    c.clunk(1)
    c.close()
    ok("server healthy after static attacks", healthy(path))


# --------------------------------------------------------------------------- debug providers as black boxes


def attack_debug_providers(path):
    print("# debug providers: hostile names, offsets, writes; the server must answer or Rerror, never die")
    c = Nine(path, timeout=15)
    c.session()
    tid = worker_tid(c)
    ok("worker thread found under /threads", tid is not None, tid)
    hostile = [b"0", b"0x", b"0x0", b"-1", b"+1", b"00", b"ffffffffffffffff", b"0xffffffffffffffff", b"1" * 13, b"1" * 12, b"zzz", b"1e5", b" 1", b"1 ", b"0x0000000000000001", b"8", b"0x7fffffffffff", b"ffffffffffff", b"\x00", b"." * 3, b"a" * 255]
    for tree in (b"addr", b"hex", b"mem"):
        for nm in hostile:
            n = c.walk_ok(0, 1, [tree, nm])
            if n == 2:
                rt, _, rb = c.open(1, OREAD)
                if rt == Ropen:
                    rt2, d = c.read(1, 0, 4096)
                    rt3, d3 = c.read(1, (1 << 64) - 1, 4096)
                    rt4, d4 = c.read(1, (1 << 63), 4096)
                    ok(f"/{tree.decode()}/{nm[:16]!r}: reads at 0, 2^63 and 2^64-1 are answered", rt2 in (Rread, Rerror) and rt3 in (Rread, Rerror) and rt4 in (Rread, Rerror), (rt2, rt3, rt4))
                else:
                    ok(f"/{tree.decode()}/{nm[:16]!r}: open answered", rt == Rerror, rt)
                c.clunk(1)
            else:
                ok(f"/{tree.decode()}/{nm[:16]!r}: walk refused or partial", n in (1, None), n)
    for nm in hostile + [b"1", b"4294967295", b"4294967296", b"99999999999", b"0" + (tid or b"1")]:
        n = c.walk_ok(0, 1, [b"threads", nm])
        if n == 2:
            for leaf in (b"name", b"stat", b"stack", b"regs"):
                rt, _, _ = c.walk(1, 2, [leaf])
                if rt == Rwalk:
                    rt, _, _ = c.open(2, OREAD)
                    if rt == Ropen:
                        c.read(2, 0, 8192)
                    c.clunk(2)
            c.clunk(1)
            ok(f"/threads/{nm!r} walked (a live tid) and its files answered", True)
        else:
            ok(f"/threads/{nm!r}: walk refused or partial", n in (1, None), n)
        n = c.walk_ok(0, 1, [b"breakpoints", nm])
        ok(f"/breakpoints/{nm!r}: walk refused (nothing paused)", n in (1, None), n)
    # files under a thread: '.'/'..' and walking through them
    if tid:
        ok("/threads/<tid>/stack/x is 'not a directory'", c.walk_ok(0, 1, [b"threads", tid, b"stack"]) == 3 and c.err(Twalk, struct.pack("<IIH", 1, 2, 1) + s16(b"x")) == "not a directory")
        ok("'..' from /threads/<tid>/stack is 'not a directory'", c.err(Twalk, struct.pack("<IIH", 1, 2, 1) + s16(b"..")) == "not a directory")
        c.clunk(1)
        ok("/threads/<tid>/../../<tid>/name walks", c.walk_ok(0, 1, [b"threads", tid, b"..", b"..", b"threads", tid, b"name"]) == 7)
        c.clunk(1)
        c.walk_ok(0, 1, [b"threads", tid])
        ok("create under /threads/<tid> is denied", c.err(base.Tcreate, struct.pack("<I", 1) + s16(b"x") + struct.pack("<IB", 0o644, OWRITE)) == "permission denied")
        ok("wstat of /threads/<tid> is denied", c.err(Twstat, struct.pack("<I", 1) + s16(mkstat(name=b"y"))) == "permission denied")
        ok("remove of /threads/<tid> is denied", c.err(Tremove, struct.pack("<I", 1)) == "permission denied")
        stack = c.path_read([b"threads", tid, b"stack"])
        ok("worker stack reads", stack is not None and b"workerLoop" in stack, stack)
        # thread stack at offset 1 is the tail (dynamic snapshot in the provider)
        c.walk_ok(0, 1, [b"threads", tid, b"stack"])
        c.open(1, OREAD)
        rt, d = c.read(1, 0, 65000)
        rt2, d2 = c.read(1, 1, 65000)
        ok("thread stack offset-1 read is the tail of the same snapshot", rt == Rread and rt2 == Rread and d2 == d[1:], (len(d), len(d2)))
        c.clunk(1)
    # /mem: read of unmapped memory is an error, write of unmapped memory is an error; writes at wild offsets too
    c.walk_ok(0, 1, [b"mem", b"8"])
    rt, _, _ = c.open(1, ORDWR)
    ok("/mem/8 opens", rt == Ropen, rt)
    rt, d = c.read(1, 0, 16)
    ok("read of unmapped memory is an Rerror", rt == Rerror, rt)
    rt, _, _ = c.write(1, 0, b"x")
    ok("write to unmapped memory is an Rerror", rt == Rerror, rt)
    rt, _, _ = c.write(1, (1 << 64) - 9, b"x")
    ok("write at a wrapping offset is answered", rt in (Rerror, Rwrite), rt)
    rt, _, _ = c.write(1, 0, b"")
    ok("empty write to /mem is answered", rt in (Rerror, Rwrite), rt)
    c.clunk(1)
    addr = c.path_read([b"vars", b"state", b"addr"])
    ok("/vars/state/addr reads", addr is not None and addr.startswith(b"0x"), addr)
    if addr:
        hx = addr[2:]
        size = int(c.path_read([b"vars", b"state", b"size"]))
        c.walk_ok(0, 1, [b"mem", hx])
        c.open(1, OREAD)
        rt, d = c.read(1, 0, size)
        ok("/mem/<state addr> reads @sizeOf bytes", rt == Rread and len(d) == size, (rt, len(d) if d else d))
        rt, d = c.read(1, 0, 0xFFFFFFFF)
        ok("/mem read with count 2^32-1 is clamped and answered", rt in (Rread, Rerror), rt)
        c.clunk(1)
        hexd = c.path_read([b"hex", hx])
        ok("/hex/<state addr> is a hexdump", hexd is not None and len(hexd) > 64, hexd[:40] if hexd else hexd)
        # a value written through /mem must render, not trap: corrupt the phase enum and read /vars/state/value
        phase_addr = int(c.path_read([b"vars", b"state", b"f", b"phase", b"addr"]), 16)
        c.walk_ok(0, 1, [b"mem", b"%x" % phase_addr])
        c.open(1, OWRITE)
        rt, _, _ = c.write(1, 0, b"\xee")
        ok("write a corrupt enum byte through /mem", rt == Rwrite, rt)
        c.clunk(1)
        v = c.path_read([b"vars", b"state", b"value"])
        ok("/vars/state/value renders the corrupt enum as a number instead of trapping", v is not None and b"phase: 238" in v, v)
        pv = c.path_read([b"vars", b"state", b"f", b"phase", b"value"])
        ok("/vars/state/f/phase/value renders 238", pv == b"238", pv)
        c.walk_ok(0, 1, [b"vars", b"state", b"f", b"phase", b"value"])
        c.open(1, OWRITE)
        rt, _, _ = c.write(1, 0, b"idle")
        ok("the enum can be repaired through /vars", rt == Rwrite, rt)
        c.clunk(1)
    # /panic: ctl refuses reads and garbage; message/stack read
    ok("/panic/message reads (empty, no panic)", c.path_read([b"panic", b"message"]) == b"")
    ok("/panic/stack reads", c.path_read([b"panic", b"stack"]) is not None)
    c.walk_ok(0, 1, [b"panic", b"ctl"])
    ok("/panic/ctl refuses OREAD", c.err(Topen, struct.pack("<IB", 1, OREAD)) == "permission denied")
    c.clunk(1)
    c.walk_ok(0, 1, [b"panic", b"ctl"])
    rt, _, _ = c.open(1, OWRITE)
    ok("/panic/ctl opens OWRITE", rt == Ropen, rt)
    for bad in (b"garbage", b"", b"continue please", b"\x00continue"):
        rt, _, _ = c.write(1, 0, bad)
        ok(f"/panic/ctl write {bad!r} without a panic is an Rerror", rt == Rerror, rt)
    c.clunk(1)
    # /breakpoints is empty; the debug providers refuse create/wstat/remove
    for prov in (b"threads", b"addr", b"mem", b"hex", b"breakpoints", b"panic"):
        c.walk_ok(0, 1, [prov])
        ok(f"create in /{prov.decode()} is denied", c.err(base.Tcreate, struct.pack("<I", 1) + s16(b"x") + struct.pack("<IB", 0o644, OWRITE)) == "permission denied")
        ok(f"wstat of /{prov.decode()} is denied", c.err(Twstat, struct.pack("<I", 1) + s16(mkstat(mtime=5))) == "permission denied")
        c.open(1, OREAD)
        d = c.read_all(1)
        ok(f"/{prov.decode()} lists", d is not None)
        c.clunk(1)
    # the provider's snapshot pool (shared by every connection) recovers after exhaustion
    held = []
    err = None
    for i in range(16):
        names = [b"addr", b"%x" % (0x1000 + i)]
        c.walk_ok(0, 200 + i, names)
        rt, _, rb = c.open(200 + i, OREAD)
        if rt == Ropen:
            held.append(200 + i)
        else:
            n, = struct.unpack_from("<H", rb)
            err = rb[2:2 + n].decode()
            break
    ok("debug provider snapshot pool exhausts with an Rerror", err is not None and len(held) >= 1, (len(held), err))
    for f in held:
        c.clunk(f)
    ok("after clunking, /addr opens again", c.path_read([b"addr", b"1000"]) is not None)
    # Tversion with debug files open (hexdumps of the exposed state: mapped memory)
    base_addr = int(addr, 16) if addr else 0
    opened = 0
    for i in range(4):
        c.walk_ok(0, 300 + i, [b"hex", b"%x" % (base_addr + i)])
        opened += c.open(300 + i, OREAD)[0] == Ropen
    ok("four /hex snapshots open", opened == 4, opened)
    rt, _, _ = c.version(65536)
    ok("Tversion with debug snapshots open", rt == base.Rversion)
    c.attach()
    ok("/hex still opens after the reset", c.path_read([b"hex", b"%x" % base_addr]) is not None)
    ok("/hex of unmapped memory is an Rerror at open, not a crash", c.walk_ok(0, 1, [b"hex", b"3000"]) == 2 and c.open(1, OREAD)[0] == Rerror)
    c.clunk(1)
    c.close()
    ok("server healthy after debug provider attacks", healthy(path))


# --------------------------------------------------------------------------- fids at the maximum


def flood(c, ids, names):
    """Pipelines one Twalk per id and counts the Rwalk replies; returns (ok_count, error_count, seconds)."""
    got = [0, 0]
    dead = [False]

    def reader():
        try:
            for _ in ids:
                rt, _, _ = c.recv_frame()
                if rt == Rwalk:
                    got[0] += 1
                else:
                    got[1] += 1
        except (EOFError, OSError):
            dead[0] = True

    t = threading.Thread(target=reader)
    t.start()
    t0 = time.time()
    body = b"".join(frame(Twalk, i & 0xFFFE, struct.pack("<IIH", 0, fid, len(names)) + b"".join(s16(n) for n in names)) for i, fid in enumerate(ids))
    c.raw(body)
    t.join(120)
    return got[0], got[1], time.time() - t0, dead[0] or t.is_alive()


def attack_fid_table(path):
    print(f"# fid table: {MAX_FIDS} fids, adversarial ids, clunk all, reuse")
    c = Nine(path, timeout=60)
    c.session()
    n = MAX_FIDS - 1  # fid 0 is the attach
    ids = []
    for i in range(n):
        k = i % 3
        ids.append(i * 8192 + 1 if k == 0 else (0x80000000 | i) if k == 1 else 0xFFFFFFFE - i)
    assert len(set(ids)) == n and 0 not in ids
    good, bad, dt, dead = flood(c, ids, [b"scratch"])
    ok(f"{n} walks with adversarial fid numbers all succeed", good == n and bad == 0 and not dead, (good, bad, dead))
    print(f"      {n} clones (provider handles) in {dt:.2f}s")
    ok("the next fid is 'too many fids'", c.err(Twalk, struct.pack("<IIH", 0, 7, 0)) == E_TOO_MANY_FIDS)
    ok("attach at the limit is 'too many fids'", c.err(base.Tattach, struct.pack("<II", 7, base.NOFID) + s16(b"u") + s16(b"")) == E_TOO_MANY_FIDS)
    ok("an existing id is 'fid in use'", c.err(Twalk, struct.pack("<IIH", 0, ids[12345], 0)) == E_FID_IN_USE)
    ok("a self-walk at the limit works", c.walk_ok(ids[5], ids[5], [b".."]) == 1)
    ok("an unknown fid at the limit is 'unknown fid'", c.err(Tstat, struct.pack("<I", 7)) == E_UNKNOWN_FID)
    # clunk all, pipelined, in a hostile order (every third first, then the rest reversed)
    order = ids[::3] + ids[1::3][::-1] + ids[2::3][::-1]
    got = [0]
    dead = [False]

    def reader():
        try:
            for _ in order:
                rt, _, _ = c.recv_frame()
                got[0] += rt == Rclunk
        except (EOFError, OSError):
            dead[0] = True

    t = threading.Thread(target=reader)
    t.start()
    t0 = time.time()
    c.raw(b"".join(frame(Tclunk, i & 0xFFFE, struct.pack("<I", fid)) for i, fid in enumerate(order)))
    t.join(120)
    ok(f"{n} clunks all answered", got[0] == n and not dead[0] and not t.is_alive(), (got[0], dead[0]))
    print(f"      {n} clunks in {time.time() - t0:.2f}s")
    ok("clunk of a clunked fid is 'unknown fid'", c.err(Tclunk, struct.pack("<I", ids[100])) == E_UNKNOWN_FID)
    # reuse: the whole table is available again with dense ids
    good, bad, dt, dead = flood(c, list(range(1, n + 1)), [])
    ok(f"{n} clones with dense ids after the churn all succeed", good == n and bad == 0 and not dead, (good, bad, dead))
    print(f"      {n} clones (reuse) in {dt:.2f}s")
    ok("still 'too many fids' at the limit", c.err(Twalk, struct.pack("<IIH", 0, n + 1, 0)) == E_TOO_MANY_FIDS)
    rt, _, _ = c.version(65536)
    ok("Tversion after the fid churn", rt == base.Rversion)
    c.attach()
    good, bad, dt, dead = flood(c, list(range(1, 1001)), [b"scratch"])
    ok("1000 clones after Tversion", good == 1000 and bad == 0, (good, bad))
    c.close()
    ok("server healthy after the fid table attacks", healthy(path))


# --------------------------------------------------------------------------- flush storm


def attack_flush(path):
    print("# Tflush storm")
    c = Nine(path, timeout=30)
    c.session()
    n = 2000
    blob = b"".join(frame(Tflush, i & 0xFFFE, struct.pack("<H", (i * 7919) & 0xFFFF)) for i in range(n))
    got = [0]

    def reader():
        try:
            for _ in range(n):
                rt, _, _ = c.recv_frame()
                got[0] += rt == Rflush
        except (EOFError, OSError):
            pass

    t = threading.Thread(target=reader)
    t.start()
    c.raw(blob)
    t.join(60)
    ok(f"{n} pipelined Tflush (random oldtags, including own tag) all Rflush", got[0] == n, got[0])
    rt, tag, _ = c.call(Tflush, struct.pack("<H", 5), 5)
    ok("Tflush of its own tag is Rflush", rt == Rflush and tag == 5, (rt, tag))
    rt, st = c.stat(0)
    ok("a normal request after the storm works", rt == Rstat, rt)
    c.close()
    ok("server healthy after the flush storm", healthy(path))


# --------------------------------------------------------------------------- msize floor


def attack_msize_floor(path):
    print(f"# msize floor: below {MSIZE_MIN} the connection dies; at {MSIZE_MIN} everything that fits is served")
    for ms in (24, 64, MSIZE_MIN - 1):
        c = Nine(path)
        rt, _, _ = c.version(ms)
        ok(f"Tversion msize {ms}: closed", rt is None or expect_dead(c), rt)
        c.close()
    c = Nine(path)
    rt, ms, ver = c.version(MSIZE_MIN)
    ok(f"Tversion {MSIZE_MIN}", rt == base.Rversion and ms == MSIZE_MIN, (rt, ms))
    rt, _, _ = c.attach(uname=b"u")
    ok("Tattach at the floor", rt == base.Rattach, rt)
    rt, st = c.stat(0)
    ok("Tstat of the root fits", rt == Rstat and st["name"] == b"/", (rt, st))
    rt, _, rb = c.walk(0, 1, [b"."] * 16)
    ok("a full 16-element Rwalk is exactly the floor", rt == Rwalk and struct.unpack_from("<H", rb)[0] == 16, rt)
    c.clunk(1)
    # an Rstat that cannot fit is an Rerror, not a dead connection: a long name in /scratch
    long_name = b"m" * 180  # Tcreate (18 + 180 bytes) fits; the Rstat (61 + 180) does not
    c.walk_ok(0, 1, [b"scratch"])
    rt, _, _ = c.create(1, long_name, 0o644, OREAD)
    ok("create a long name at the floor", rt == Rcreate, rt)
    e = c.err(Tstat, struct.pack("<I", 1))
    ok("Tstat that does not fit is 'Invalid argument'", e == E_INVAL, e)
    ok("remove it", c.remove(1)[0] == Rremove)
    rt, _, _ = c.walk(0, 1, [b"build"])
    rt, _, _ = c.open(1, OREAD)
    ok("Topen at the floor", rt == Ropen, rt)
    rt, d = c.read(1, 0, 4096)  # count clamped to msize - 11
    ok("a directory read at the floor yields whole records", rt == Rread and 0 < len(d) <= MSIZE_MIN - 11 and records(d), (rt, len(d) if d else d))
    e = c.err(Tread, struct.pack("<IQI", 1, 1, 4096))
    ok("dir read at offset 1 is a bad offset", e == E_BAD_OFFSET, e)
    rt, _, _ = c.clunk(1)
    ok("Tclunk at the floor", rt == Rclunk, rt)
    for nm in (b"vars", b"state", b"f", b"ticks", b"value"):
        rt, _, _ = c.walk(0 if nm == b"vars" else 1, 1, [nm])
    ok("walk to /vars/state/f/ticks/value in 5 self-walks", rt == Rwalk, rt)
    rt, _, _ = c.open(1, OWRITE)
    ok("open a writable value at the floor", rt == Ropen, rt)
    rt, _, _ = c.write(1, 0, b"5")
    ok("Twrite of one byte at the floor", rt == Rwrite, rt)
    e = c.err(Twrite, struct.pack("<IQI", 1, 0, 0) + b"")
    ok("empty write to a value at the floor is 'bad value'", e == "bad value", e)
    rt, _, _ = c.clunk(1)
    ok("clunk at the floor", rt == Rclunk, rt)
    rt, ms, _ = c.version(65536)
    ok("renegotiate a big msize on the same connection", rt == base.Rversion and ms == 65536, (rt, ms))
    c.attach()
    ok("normal service resumes", c.path_read([b"build", b"zig_version"]) not in (None, b""))
    c.close()
    # frames larger than the negotiated msize kill the connection
    c = Nine(path)
    c.version(MSIZE_MIN)
    c.raw(frame(base.Tattach, 1, struct.pack("<II", 0, base.NOFID) + s16(b"u" * 210) + s16(b"")))
    ok("an over-long Tattach at the floor: connection closed", expect_dead(c))
    c.close()
    ok("server healthy after msize-floor attacks", healthy(path))


# --------------------------------------------------------------------------- directory offsets


def attack_dir_offsets(path):
    print("# directory reads: exact record boundaries vs off by one")
    c = Nine(path)
    c.session()
    tag = os.urandom(3).hex().encode()
    root = b"do-" + tag
    c.walk_ok(0, 1, [b"scratch"])
    c.create(1, root, DMDIR | 0o755, OREAD)
    c.clunk(1)
    for nm in (b"alpha", b"beta-with-a-longer-name", b"g"):
        c.walk_ok(0, 1, [b"scratch", root])
        c.create(1, nm, 0o644, OWRITE)
        c.clunk(1)
    for names in ([], [b"vars", b"state"], [b"comptime", b"types"], [b"scratch", root], [b"threads"], [b"panic"]):
        p = "/" + b"/".join(names).decode()
        c.walk_ok(0, 1, names)
        c.open(1, OREAD)
        rt, d = c.read(1, 0, 65000)
        recs = records(d)
        if len(recs) < 2:
            ok(f"{p}: at least two entries", False, len(recs))
            c.clunk(1)
            continue
        r0 = len(recs[0][1])
        r1 = len(recs[1][1])
        rt, d0 = c.read(1, 0, r0)
        ok(f"{p}: count = first record length returns exactly that record", rt == Rread and d0 == recs[0][1], (rt, len(d0) if d0 else d0, r0))
        rt, d1 = c.read(1, r0, r1)
        ok(f"{p}: read at the record boundary returns the next record", rt == Rread and d1 == recs[1][1], (rt, len(d1) if d1 else d1))
        e = c.err(Tread, struct.pack("<IQI", 1, r0 + r1 + 1, 65000))
        ok(f"{p}: offset boundary+1 is 'bad offset'", e == E_BAD_OFFSET, e)
        e = c.err(Tread, struct.pack("<IQI", 1, r0 + r1 - 1, 65000))
        ok(f"{p}: offset boundary-1 is 'bad offset'", e == E_BAD_OFFSET, e)
        e = c.err(Tread, struct.pack("<IQI", 1, r0, 65000))
        ok(f"{p}: re-reading an earlier boundary is 'bad offset'", e == E_BAD_OFFSET, e)
        rt, rest = c.read(1, r0 + r1, 65000)
        ok(f"{p}: after a bad offset the good boundary still continues", rt == Rread and rest == d[r0 + r1:], rt)
        e = c.err(Tread, struct.pack("<IQI", 1, 0, r0 - 1))
        ok(f"{p}: count one short of a record is refused (no split)", e == E_INVAL, e)
        rt, dd = c.read(1, 0, 65000)
        ok(f"{p}: offset 0 restarts and yields the same bytes", rt == Rread and dd == d)
        rt, dd = c.read(1, len(d), 65000)
        ok(f"{p}: read at the end is empty", rt == Rread and dd == b"")
        rt, dd = c.read(1, len(d), 65000)
        ok(f"{p}: read at the end twice is empty twice", rt == Rread and dd == b"")
        c.clunk(1)
    for nm in (b"alpha", b"beta-with-a-longer-name", b"g"):
        c.walk_ok(0, 1, [b"scratch", root, nm])
        c.remove(1)
    c.walk_ok(0, 1, [b"scratch", root])
    ok("cleanup of the directory-offset test root", c.remove(1)[0] == Rremove)
    c.close()
    ok("server healthy after directory offset attacks", healthy(path))


# --------------------------------------------------------------------------- ctl staging


def attack_ctl(path):
    print("# ctl: a failed command leaves the previous result, length and qid version untouched")
    c = Nine(path)
    c.session(1 << 20)
    c.walk_ok(0, 1, [b"runtime", b"ctl"])
    c.open(1, ORDWR)
    rt, _, _ = c.write(1, 0, b"echo persist")
    rt, st = c.stat(1)
    v = st["qid"][1]
    ok("echo persist", st["length"] == 7, st)
    for bad in (b"nope", b"fib 94", b"add 1", b"", b"\x00", b"echo\x00hidden"):
        e = c.err(Twrite, struct.pack("<IQI", 1, 0, len(bad)) + bad)
        rt, d = c.read(1, 0, 100)
        rt2, st2 = c.stat(1)
        ok(f"after failed {bad!r}: result still 'persist'", d == b"persist", d)
        ok(f"after failed {bad!r}: length 7 and qid version unchanged", st2["length"] == 7 and st2["qid"][1] == v, (st2["length"], st2["qid"][1], v))
    # a second connection sees the same result and version
    c2 = Nine(path)
    c2.session()
    ok("other connection reads the surviving result", c2.path_read([b"runtime", b"ctl"]) == b"persist")
    c2.walk_ok(0, 1, [b"runtime", b"ctl"])
    rt, st3 = c2.stat(1)
    ok("other connection sees the same version", st3["qid"][1] == v, (st3["qid"][1], v))
    c2.close()
    # a successful command bumps the version and replaces the result; an empty result is a result
    rt, _, _ = c.write(1, 0, b"echo")
    rt, st4 = c.stat(1)
    rt, d = c.read(1, 0, 100)
    ok("echo with no argument yields an empty result with a new version", d == b"" and st4["length"] == 0 and st4["qid"][1] != v, (d, st4))
    e = c.err(Twrite, struct.pack("<IQI", 1, 0, 4) + b"nope")
    rt, d = c.read(1, 0, 100)
    ok("a failure after an empty result keeps it empty", d == b"", d)
    # the largest result: echo of a 60000-byte line
    big = b"echo " + b"y" * 60000
    rt, _, _ = c.write(1, 0, big)
    d = c.read_all(1)
    ok("60000-byte ctl result round-trips", rt == Rwrite and d == b"y" * 60000, (rt, len(d) if d else d))
    rt, _, _ = c.write(1, 0, b"echo " + b"z" * 70000)  # exceeds ctl_bytes (64 KiB) -> the handler's writer fails
    rt2, d = c.read(1, 0, 100)
    rt3, st5 = c.stat(1)
    ok("an over-long result is an Rerror and the previous result survives", rt == Rerror and d == b"y" * 100 and st5["length"] == 60000, (rt, d[:10] if d else d, st5["length"]))
    c.clunk(1)
    c.close()
    ok("server healthy after ctl attacks", healthy(path))


# --------------------------------------------------------------------------- fid state machine on scratch


def attack_fid_states(path):
    print("# fid state machine on /scratch")
    c = Nine(path)
    c.session()
    tag = os.urandom(3).hex().encode()
    root = b"fs-" + tag
    c.walk_ok(0, 1, [b"scratch"])
    c.create(1, root, DMDIR | 0o755, OREAD)
    c.clunk(1)
    S = [b"scratch", root]
    c.walk_ok(0, 1, S)
    rt, _, _ = c.create(1, b"f", 0o644, ORDWR)
    ok("create f", rt == Rcreate)
    ok("open of an open fid is 'file already open for I/O'", c.err(Topen, struct.pack("<IB", 1, OREAD)) == E_ALREADY_OPEN)
    ok("walk with names from an open fid is 'bad use of fid'", c.err(Twalk, struct.pack("<IIH", 1, 2, 1) + s16(b".")) == E_BAD_USE)
    ok("create on an open fid is 'file already open for I/O'", c.err(base.Tcreate, struct.pack("<I", 1) + s16(b"g") + struct.pack("<IB", 0o644, OWRITE)) == E_ALREADY_OPEN)
    ok("clone of an open fid is refused", c.err(Twalk, struct.pack("<IIH", 1, 2, 0)) == E_BAD_USE)
    ok("a fresh walk reaches the open file", c.walk_ok(0, 2, S + [b"f"]) == 3)
    ok("the fresh fid is not open", c.err(Tread, struct.pack("<IQI", 2, 0, 10)) == E_BAD_USE)
    rt, _, _ = c.open(2, OREAD)
    ok("the fresh fid opens independently", rt == Ropen)
    c.write(1, 0, b"data")
    rt, d = c.read(2, 0, 10)
    ok("the second fid sees the write", rt == Rread and d == b"data", d)
    rt, _, _ = c.wstat(2, mkstat(name=b"renamed"))
    ok("wstat through an open fid works", rt == Rwstat)
    rt, _, _ = c.remove(1)
    ok("remove through the open writer fid", rt == Rremove)
    rt, d = c.read(2, 0, 10)
    ok("the other open fid still reads the removed file", rt == Rread and d == b"data", d)
    ok("stat of the removed file still answers", c.stat(2)[0] == Rstat)
    ok("open of a removed file through a new walk is impossible (not found)", c.walk_ok(0, 3, S + [b"renamed"]) == 2)
    c.clunk(2)
    # newfid == fid walks on unopened fids rebind; on the same fid with a failing later element they do nothing
    c.walk_ok(0, 3, S)
    rt, _, rb = c.walk(3, 3, [b"..", root, b"nope"])
    n = struct.unpack_from("<H", rb)[0] if rt == Rwalk else None
    ok("partial self-walk returns 2 and leaves the fid where it was", n == 2 and c.stat(3)[1]["name"] == root, (n,))
    rt, _, _ = c.walk(3, 3, [b"..", b".."])
    rt, st = c.stat(3)
    ok("self-walk with names rebinds the fid", rt == Rstat and st["qid"][2] == 0xFF << 56, st)
    c.clunk(3)
    # Tversion while a removed-but-held file exists: nothing leaks, the server keeps serving
    c.walk_ok(0, 4, S)
    c.create(4, b"held", 0o644, OWRITE)
    c.write(4, 0, b"x" * 1000)
    c.walk_ok(0, 5, S + [b"held"])
    c.remove(5)
    c.version(65536)
    c.attach()
    ok("after Tversion the removed file is gone", c.walk_ok(0, 1, S + [b"held"]) == 2)
    c.clunk(1)
    c.walk_ok(0, 1, S)
    rt, _, _ = c.remove(1)
    ok("cleanup", rt == Rremove, rt)
    c.close()
    ok("server healthy after fid state attacks", healthy(path))


# --------------------------------------------------------------------------- scratch budget


def attack_scratch_budget(path):
    print("# scratch: the global budget after grow/truncate/rename/remove/failed writes")
    c = Nine(path, timeout=120)
    c.session()
    tag = os.urandom(3).hex().encode()
    root = b"bg-" + tag
    c.walk_ok(0, 1, [b"scratch"])
    c.create(1, root, DMDIR | 0o755, OREAD)
    c.clunk(1)
    S = [b"scratch", root]
    per = SCRATCH_MAX_FILE
    count = SCRATCH_BUDGET // per
    t0 = time.time()
    made = 0
    for i in range(count):
        c.walk_ok(0, 1, S)
        rt, _, _ = c.create(1, b"big%d" % i, 0o644, OWRITE)
        rt, _, _ = c.wstat(1, mkstat(length=per))
        c.clunk(1)
        if rt != Rwstat:
            break
        made += 1
    ok(f"{count} files of {per >> 20} MiB fill the {SCRATCH_BUDGET >> 20} MiB budget exactly", made == count, made)
    print(f"      filled the budget in {time.time() - t0:.1f}s")
    c.walk_ok(0, 1, S)
    c.create(1, b"one-more", 0o644, OWRITE)
    ok("one more byte is 'no space left on device'", c.err(Twrite, struct.pack("<IQI", 1, 0, 1) + b"x") == "no space left on device")
    ok("a failed write leaves the file empty", c.stat(1)[1]["length"] == 0)
    ok("a zero-length write at a huge offset is still fine", c.write(1, (1 << 60), b"")[0] == Rwrite)
    ok("wstat length 1 is 'no space left on device'", c.err(Twstat, struct.pack("<I", 1) + s16(mkstat(length=1))) == "no space left on device")
    # rename does not charge; truncate releases exactly its size
    c.walk_ok(0, 2, S + [b"big0"])
    ok("rename of a full file is fine", c.wstat(2, mkstat(name=b"big0-r"))[0] == Rwstat)
    ok("still no space after the rename", c.err(Twrite, struct.pack("<IQI", 1, 0, 1) + b"x") == "no space left on device")
    ok("truncate big0-r to 1 MiB", c.wstat(2, mkstat(length=1 << 20))[0] == Rwstat)
    c.clunk(2)
    rt, _, _ = c.wstat(1, mkstat(length=per - (1 << 20)))
    ok("exactly the released amount is writable again", rt == Rwstat, rt)
    ok("and not one byte more", c.err(Twrite, struct.pack("<IQI", 1, per - (1 << 20), 1) + b"x") == "no space left on device")
    ok("nor via wstat", c.err(Twstat, struct.pack("<I", 1) + s16(mkstat(length=per - (1 << 20) + 1))) == "no space left on device")
    # overwriting existing bytes costs nothing
    ok("overwrite inside the file is fine", c.write(1, 0, b"y" * 4096)[0] == Rwrite)
    ok("overwrite at the very end is fine", c.write(1, per - (1 << 20) - 4096, b"y" * 4096)[0] == Rwrite)
    # OTRUNC of a full file releases; remove of a held file releases only on the last clunk
    c.walk_ok(0, 3, S + [b"big1"])
    ok("OTRUNC releases", c.open(3, OWRITE | OTRUNC)[0] == Ropen and c.stat(3)[1]["length"] == 0)
    ok("the released space is writable", c.wstat(1, mkstat(length=per))[0] == Rwstat)
    ok("a full file writes into itself", c.write(1, per - 10, b"0123456789")[0] == Rwrite)
    ok("but not past the per-file cap", c.err(Twrite, struct.pack("<IQI", 1, per - 1, 2) + b"xy") == "no space left on device")
    c.clunk(3)
    c.clunk(1)
    # refill the remaining 63 MiB so the budget is exactly full again
    c.walk_ok(0, 7, S)
    c.create(7, b"fill", 0o644, OWRITE)
    ok("the rest of the budget fills exactly", c.wstat(7, mkstat(length=per - (1 << 20)))[0] == Rwstat)
    ok("and is full again", c.err(Twrite, struct.pack("<IQI", 7, per - (1 << 20), 1) + b"x") == "no space left on device")
    c.clunk(7)
    c.walk_ok(0, 4, S + [b"big2"])
    c.walk_ok(0, 5, S + [b"big2"])
    c.open(5, OREAD)
    c.remove(4)
    c.walk_ok(0, 6, S)
    c.create(6, b"after-remove", 0o644, OWRITE)
    ok("space of a removed-but-held file is not released yet", c.err(Twrite, struct.pack("<IQI", 6, 0, 1) + b"x") == "no space left on device")
    c.clunk(5)
    ok("the last clunk releases it", c.write(6, 0, b"x")[0] == Rwrite)
    c.clunk(6)
    # cleanup
    c.walk_ok(0, 1, S)
    c.open(1, OREAD)
    names = [n for n, _ in records(c.read_all(1))]
    c.clunk(1)
    for nm in names:
        if c.walk_ok(0, 1, S + [nm]) == 3:
            c.remove(1)
    c.walk_ok(0, 1, S)
    ok("cleanup removed the budget test root", c.remove(1)[0] == Rremove, names)
    c.walk_ok(0, 1, [b"scratch"])
    c.create(1, b"post-" + tag, 0o644, OWRITE)
    ok("the whole budget is back: a 64 MiB file fits", c.wstat(1, mkstat(length=per))[0] == Rwstat)
    c.remove(1)
    c.close()
    ok("server healthy after budget attacks", healthy(path))


# --------------------------------------------------------------------------- main


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--server")
    ap.add_argument("--socket")
    ap.add_argument("--fast", action="store_true", help="skip the 512 MiB scratch budget fill")
    args = ap.parse_args()
    proc = None
    tmp = None
    if args.server:
        tmp = tempfile.mkdtemp(prefix="adv9pcore.")
        path = os.path.join(tmp, "sock")
        proc = subprocess.Popen([os.path.abspath(args.server), "--unix", path], stderr=subprocess.PIPE)
        for _ in range(200):
            if os.path.exists(path):
                break
            time.sleep(0.02)
    elif args.socket:
        path = args.socket
    else:
        ap.error("--server or --socket")
    try:
        attack_vars(path)
        attack_snapshots(path)
        attack_static(path)
        attack_debug_providers(path)
        attack_dir_offsets(path)
        attack_ctl(path)
        attack_fid_states(path)
        attack_msize_floor(path)
        attack_flush(path)
        attack_fid_table(path)
        if not args.fast:
            attack_scratch_budget(path)
        if proc is not None:
            ok("server process still running", proc.poll() is None, proc.poll())
    finally:
        if proc is not None:
            proc.send_signal(signal.SIGTERM)
            try:
                _, err = proc.communicate(timeout=5)
            except subprocess.TimeoutExpired:
                proc.kill()
                _, err = proc.communicate()
            lines = [ln for ln in err.decode("utf-8", "replace").splitlines() if "connection ended" not in ln and "read: " not in ln]
            if lines:
                print("# server stderr (filtered):")
                for ln in lines[:40]:
                    print("      " + ln)
            if tmp:
                try:
                    os.unlink(path)
                    os.rmdir(tmp)
                except OSError:
                    pass
    print(f"# {base.PASSES} passed, {len(base.FAILS)} failed")
    for f in base.FAILS:
        print("#   FAIL " + f)
    sys.exit(1 if base.FAILS else 0)


if __name__ == "__main__":
    main()