esh-pfi-infrastructure/scripts/oneoff/albok-browse-withdrawal/run-trials.py

#!/usr/bin/env python3
"""Pathname-withdrawal test for an albok browse view shared over sshfs.

Runs on the CLIENT (an ephemeral container, see run.sh) against the replica built
by setup-replica.py. The gate (albok unit 4.3, exposure ruling): rename `.browse/`
away on the server, then a FRESH open of a child that was never opened, through a
listing the client has cached, must FAIL.

Per trial (fresh mount each time), three distinct children:
  A  opened and closed before the rename   -> re-opened after it
  B  opened before the rename, fd HELD     -> read at a far offset after it
  C  listed and stat'ed, never opened      -> opened after it (the gate)

Conditions (N trials each):
  null           no rename                      C must OPEN (null control)
  default        rename ok, sshfs defaults      C must FAIL (the gate)
  aggressive     rename ok, hour-long caches    C must FAIL (gate, worst-case client caching)
  failed_rename  rename refused (ENOTEMPTY)     C OPENS: the path remains and is served
B is the positive control: a retained handle must still read after a rename, which
proves the instrument can see access that survives a rename.

Usage: run-trials.py <manifest.json> <trials-per-condition>   (JSON to stdout)
"""
import errno
import json
import os
import shlex
import subprocess
import sys
import time

# Overridable so the same harness runs on a workstation (e.g. macOS) without the container.
SRV = os.environ.get("ALBOK_TEST_SRV", "infra-ops@10.100.50.44")
KEY = os.environ.get("ALBOK_TEST_KEY", "/keys/id")
KH = os.environ.get("ALBOK_TEST_KNOWN_HOSTS", "/keys/known_hosts")
MNT = os.environ.get("ALBOK_TEST_MNT", "/mnt/browse")
SSH_OPTS = ["-i", KEY, "-o", "BatchMode=yes", "-o", f"UserKnownHostsFile={KH}",
            "-o", "StrictHostKeyChecking=yes", "-o", "ControlMaster=auto",
            "-o", "ControlPath=/tmp/cm-%r@%h", "-o", "ControlPersist=900"]
SSHFS_BASE = ["ro", "follow_symlinks", f"IdentityFile={KEY}", f"UserKnownHostsFile={KH}",
              "StrictHostKeyChecking=yes", "BatchMode=yes"]
AGGRESSIVE = ["kernel_cache", "entry_timeout=3600", "attr_timeout=3600",
              "dir_cache=yes", "dcache_timeout=3600"]
CONDITIONS = [("null", [], None), ("default", [], "ok"),
              ("aggressive", AGGRESSIVE, "ok"), ("failed_rename", [], "fail")]
LATE_LIST_DELAY = 25  # past sshfs's default 20 s cache timeout

manifest = json.load(open(sys.argv[1]))
N = int(sys.argv[2])
ROOT = manifest["root"]
SIZE = manifest["size"]
OBJS = manifest["objects"]
VIEW, WITHHELD = f"{ROOT}/.browse", f"{ROOT}/.browse.withheld"


def ename(e: OSError) -> str:
    return errno.errorcode.get(e.errno or 0, str(e.errno))


def content(o) -> bytes:
    line = f"replica object {o['i']} {o['id']}\n".encode()
    return (line * (SIZE // len(line) + 1))[:SIZE]


def remote_py(code: str, *args: str):
    cmd = "python3 -c " + shlex.quote(code) + "".join(" " + shlex.quote(a) for a in args)
    t0 = time.monotonic()
    r = subprocess.run(["ssh", *SSH_OPTS, SRV, cmd], capture_output=True, text=True, timeout=60)
    return {"rc": r.returncode, "out": r.stdout.strip(), "err": r.stderr.strip()[-300:],
            "secs": round(time.monotonic() - t0, 3)}


RENAME = ("import os,sys,errno\n"
          "try:\n os.rename(sys.argv[1], sys.argv[2]); print('renamed')\n"
          "except OSError as e:\n print('refused', errno.errorcode.get(e.errno)); sys.exit(3)")
STATE = "import os,sys; print(os.path.isdir(sys.argv[1]), os.path.lexists(sys.argv[2]))"


def open_read(path, expect, offset=0, n=4096):
    try:
        fd = os.open(path, os.O_RDONLY)
    except OSError as e:
        return {"opened": False, "errno": ename(e)}
    try:
        os.lseek(fd, offset, os.SEEK_SET)
        b = os.read(fd, n)
        return {"opened": True, "read": len(b), "content_ok": b == expect[offset:offset + n]}
    except OSError as e:
        return {"opened": True, "read_errno": ename(e)}
    finally:
        os.close(fd)


def listing(d):
    try:
        return {"n": len(os.listdir(d))}
    except OSError as e:
        return {"errno": ename(e)}


def stat(p):
    try:
        st = os.stat(p)
        return {"ok": True, "size": st.st_size}
    except OSError as e:
        return {"ok": False, "errno": ename(e)}


def trial(cond, extra, rename, k, slot):
    A, B, C = OBJS[3 * slot], OBJS[3 * slot + 1], OBJS[3 * slot + 2]
    rec = {"cond": cond, "trial": k, "A": A["name"], "B": B["name"], "C": C["name"]}
    rec["server_before"] = remote_py(STATE, VIEW, WITHHELD)["out"]
    if rec["server_before"] != "True False":
        rec["aborted"] = "server not in the clean state"
        return rec
    if rename == "fail":  # a non-empty target makes rename(2) refuse
        remote_py("import os,sys; os.makedirs(sys.argv[1])", f"{WITHHELD}/blocker")
    os.makedirs(MNT, exist_ok=True)
    m = subprocess.run(["sshfs", f"{SRV}:{VIEW}", MNT, "-o", ",".join(SSHFS_BASE + extra)],
                       capture_output=True, text=True, timeout=60)
    if m.returncode != 0:
        rec["aborted"] = f"mount failed: {m.stderr.strip()}"
        return rec
    renamed = False
    fd_b = None
    try:
        st_dir = f"{MNT}/stash"
        names = os.listdir(st_dir)
        rec["pre_list"] = len(names)
        for nm in names:  # cache every child's attrs: the client "knows" C exists
            os.stat(f"{st_dir}/{nm}")
        rec["pre_A"] = open_read(f"{st_dir}/{A['name']}", content(A))
        fd_b = os.open(f"{st_dir}/{B['name']}", os.O_RDONLY)
        rec["pre_B_ok"] = os.read(fd_b, 4096) == content(B)[:4096]

        if rename:
            rec["rename"] = remote_py(RENAME, VIEW, WITHHELD)
            renamed = rec["rename"]["out"] == "renamed"
        t0 = time.monotonic()

        rec["post_list"] = listing(st_dir)
        rec["post_stat_C"] = stat(f"{st_dir}/{C['name']}")
        rec["post_open_C"] = open_read(f"{st_dir}/{C['name']}", content(C))
        rec["post_open_C"]["secs_after_rename"] = round(time.monotonic() - t0, 3)
        far = SIZE - 8192  # well past readahead: this read must go to the server
        try:
            os.lseek(fd_b, far, os.SEEK_SET)
            b = os.read(fd_b, 4096)
            rec["post_held_B"] = {"read": len(b), "content_ok": b == content(B)[far:far + 4096]}
        except OSError as e:
            rec["post_held_B"] = {"read_errno": ename(e)}
        rec["post_reopen_A"] = open_read(f"{st_dir}/{A['name']}", content(A))
        if cond == "default":
            time.sleep(max(0.0, LATE_LIST_DELAY - (time.monotonic() - t0)))
            rec["late_list"] = listing(st_dir)
            rec["late_list"]["secs_after_rename"] = round(time.monotonic() - t0, 3)
    finally:
        if fd_b is not None:
            os.close(fd_b)
        for um in (["fusermount3", "-u", MNT], ["fusermount", "-u", MNT], ["umount", MNT]):
            try:
                if subprocess.run(um, capture_output=True).returncode == 0:
                    break
            except FileNotFoundError:
                continue
        if renamed:
            rec["restore"] = remote_py(RENAME, WITHHELD, VIEW)["out"]
        if rename == "fail":
            remote_py("import os,sys; os.rmdir(sys.argv[1]); os.rmdir(sys.argv[2])",
                      f"{WITHHELD}/blocker", WITHHELD)
        rec["server_after"] = remote_py(STATE, VIEW, WITHHELD)["out"]
    return rec


results = []
slot = 0
for cond, extra, rename in CONDITIONS:
    for k in range(N):
        results.append(trial(cond, extra, rename, k, slot))
        slot += 1
json.dump({"started_note": "times are monotonic seconds; see run.sh for wall clock",
           "sshfs_base": SSHFS_BASE, "aggressive": AGGRESSIVE, "trials": results},
          sys.stdout, indent=1)