upgrade-test.py: measure whether a real app upgrade keeps the customer's data
gates / gates (push) Successful in 0s

R-449. Until today one upgrade out of 53 had ever been measured - Nextcloud, by
hand, in a spike - and the whole update arc was designed against that single data
point.

Per edge: deploy at FROM, seed through the app's OWN interface, prove the seed
reads back, swap to TO, ask the app for the data again, then put the FROM images
back and record what happens - verbatim, and never called a rollback.

Success is an application-level readback, not file identity: survive2.py's
sha256+inode rule is right for a redeploy and wrong for an upgrade, because a
migration is supposed to rewrite files. And nothing is ever seeded by hand (R-156)
- an app with no non-browser route is recorded inconclusive, never faked.

C3 is a negative control whose TO image exits immediately, and it must be run
first: it came back failed, which is what makes the greens mean anything.

The bookstack fixture uses artisan for both halves and carries its own negative
control on every call, because the obvious HTTP-login readback cannot work: the
template's https APP_URL makes the session cookies secure, so curl over http gets
419 on every login and it looks exactly like a wrong password.
This commit is contained in:
2026-09-06 11:44:18 +02:00
parent 7b9b9b34a5
commit 0474ce387e
4 changed files with 624 additions and 0 deletions
+27
View File
@@ -1,3 +1,30 @@
## upgrade-test.py — a harness that measures whether an upgrade keeps the data (2026-09-06, R-449) — NOT A RELEASE
**No template changed. No image moved. Scripts only.** `scripts/upgrade-test.py` +
`scripts/upgrade_fixtures.py`.
Until today this project had measured **one** app upgrade out of 53 — Nextcloud, by hand, during
`SPIKE-app-update-2026-09-01` §7 — and the whole update arc was designed against that single data
point. A one-off measurement nothing repeats decays into a claim. This is the thing that repeats it.
**Method, per edge:** deploy at the FROM images → seed through the app's **own** interface → prove the
seed reads back (control C1) → swap to the TO images → ask the app for the data **again** → then put
the FROM images back and record what happens.
**Two rules, and the first is the one people get wrong.** Success is an **application-level
readback**, not file identity: `survive2.py`'s sha256+inode rule is right for a redeploy and wrong for
an upgrade, because a migration is *supposed* to rewrite files. And, carried verbatim from that same
script: *nothing is ever seeded into a volume by hand* (R-156) — every seed goes through the app's
HTTP API or its own CLI, and an app with no such route is recorded **`inconclusive`**, never faked.
**Run `C3` first, always.** Its TO image is `alpine:3.20`, which pulls cleanly and exits immediately.
It came back **`failed`** — so the harness can say no, and its greens mean something. If it ever comes
back green, nothing else in the run is evidence.
**First run measured 7 edges across 3 apps** (privatebin, docmost, bookstack) in a throwaway guest on
demo-hp, destroyed afterwards. Findings, including a real defect in this repo's own bookstack
template: `felhom.eu/documentation/audits/SPIKE-upgrade-test-2026-09-06.md`.
## LIVE-TEST for controller v0.235.0 — two pushes, both reverted the same hour (2026-09-06) — NOT A RELEASE ## LIVE-TEST for controller v0.235.0 — two pushes, both reverted the same hour (2026-09-06) — NOT A RELEASE
**No template is different after these four commits.** `dc7e548` moved bentopdf's healthcheck **No template is different after these four commits.** `dc7e548` moved bentopdf's healthcheck
+11
View File
@@ -67,6 +67,17 @@ deployed `app.yaml` (customer secrets) is never overwritten. Full deploy details
run at the start of every catalog campaign and before any publish train that vouches the catalog. run at the start of every catalog campaign and before any publish train that vouches the catalog.
Needs network + `docker`; unauthenticated Docker Hub throttles a full sweep, so `docker login` Needs network + `docker`; unauthenticated Docker Hub throttles a full sweep, so `docker login`
first or expect exit 2. It reports a throttle as INCONCLUSIVE, never as a dead image. first or expect exit 2. It reports a throttle as INCONCLUSIVE, never as a dead image.
- **A well-formed template can still be un-upgradable, and no gate can see that either.** Fourth
instrument, and the only one that needs REAL DATA: `python3 scripts/upgrade-test.py <edge>` deploys
an app at a FROM image set, seeds through the app's **own** interface, swaps to a TO set, and asks
the app for the data back. **Success is an application-level readback, never file identity** — a
migration is supposed to rewrite files, so the persistence sweep's sha256+inode rule would fail
every correct upgrade. It also records what the ABORT does (putting the old images back), verbatim.
**Run `C3` first, every time:** it is a negative control whose TO image exits immediately, and if it
does not come back `failed` the harness is not measuring anything. Fixtures live in
`scripts/upgrade_fixtures.py`; an app with no non-browser seed route is recorded `inconclusive`,
never faked. Needs Docker + real images + minutes per edge, **on a scratch host, never a customer
box**. First run: `felhom.eu/documentation/audits/SPIKE-upgrade-test-2026-09-06.md`.
- **A well-formed template can still preserve the wrong folder** — and no static check can see it. - **A well-formed template can still preserve the wrong folder** — and no static check can see it.
Third gate, the only RUNTIME one: `python3 scripts/check-volume-persistence.py` Third gate, the only RUNTIME one: `python3 scripts/check-volume-persistence.py`
(0 all clean / **1 REFUSED** / 2 undecided). It deploys each template, exercises it into writing (0 all clean / **1 REFUSED** / 2 undecided). It deploys each template, exercises it into writing
+318
View File
@@ -0,0 +1,318 @@
#!/usr/bin/env python3
"""Upgrade prover — does a real app upgrade keep the customer's data, and can it be undone?
R-449. `SPIKE-app-update-2026-09-01.md` §7 measured exactly ONE upgrade, by hand, on one app. A
one-off measurement that nothing repeats decays into a claim, and this project has paid for that
before. This is the thing that repeats it.
Method, per EDGE (one app, one FROM image set, one TO image set):
1. render the catalog template with the FROM images and `docker compose up -d`
2. SEED through the app's OWN INTERFACE — its HTTP API or its own CLI inside the container
3. VERIFY the seed reads back ← control C1. A fixture that cannot prove itself first proves
nothing after.
4. re-render with the TO images, `up -d`, settle
5. VERIFY the seed reads back AGAIN ← THE RESULT
6. ABORT: put the FROM images back, `up -d`, and record what happens
WHAT SUCCESS IS, AND THE RULE THAT DOES *NOT* CARRY OVER FROM survive2.py.
`survive2.py` calls a file survived only if sha256 AND inode both match. That is right for a
redeploy and WRONG for an upgrade: a migration is SUPPOSED to rewrite files, so that rule fails
every correct upgrade. Success here is an APPLICATION-LEVEL READBACK — ask the app for the value.
THE RULE THAT DOES CARRY OVER, verbatim from survive2.py: "Nothing is ever seeded into a volume by
hand." R-156's evidence shows a root-written canary making an empty volume read as populated. Every
seed goes in through the app's own interface. An app with no non-browser route is recorded
`inconclusive`, WITH what was tried — that is a result, not a licence to plant a file.
"the container started" IS NOT A PASS. The spike measured an app that was HTTP 200
"update completed" and crash-looping at the same time.
The undo is an ABORT, never a "rollback". The word is struck — see
felhom.eu/documentation/architecture/09-update-architecture.md §4: once a migration has run, the old
image refuses to start on the migrated data, so there is no rollback to speak of.
Usage: python3 upgrade-test.py <edge-id> [<edge-id> …] (see EDGES)
python3 upgrade-test.py --list
Layout: templates under /opt/upg/templates, evidence under /opt/upg/evidence
"""
import importlib.util, json, os, re, shutil, subprocess, sys, time
from datetime import datetime, timezone
from pathlib import Path
HARNESS_VERSION = 1
ROOT = Path("/opt/upg")
TEMPLATES = ROOT / "templates"
EVIDENCE = ROOT / "evidence"
_spec = importlib.util.spec_from_file_location("cvp", str(ROOT / "check-volume-persistence.py"))
cvp = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(cvp)
_fspec = importlib.util.spec_from_file_location("fx", str(ROOT / "upgrade_fixtures.py"))
fx = importlib.util.module_from_spec(_fspec)
_fspec.loader.exec_module(fx)
# --- the edges ---------------------------------------------------------------------------------
#
# Every FROM/TO pair below is a transition THE CATALOG ITSELF MADE, read from its own git history,
# except where the comment says otherwise. C3's `alpine:3.20` is a real image that pulls cleanly and
# exits immediately — measured in SPIKE-app-update-2026-09-01 §4.
EDGES = {
"C2": dict(app="privatebin", note="no-op control: a version to ITSELF",
frm={"privatebin": "privatebin/pdo:2.0.5"},
to={"privatebin": "privatebin/pdo:2.0.5"}),
"C3": dict(app="privatebin", note="NEGATIVE control: the TO image starts and exits immediately",
frm={"privatebin": "privatebin/pdo:2.0.5"},
to={"privatebin": "alpine:3.20"}),
"E1": dict(app="privatebin", note="catalog transition cf8b645, a major",
frm={"privatebin": "privatebin/pdo:1.7.5"},
to={"privatebin": "privatebin/pdo:2.0.5"}),
"E2": dict(app="docmost", note="catalog transition a2115b2; PostgreSQL constant across it",
frm={"docmost": "docmost/docmost:0.25.3"},
to={"docmost": "docmost/docmost:0.95.0"}),
"E3": dict(app="bookstack", note="catalog transition 0b73e5e: app AND engine together",
frm={"bookstack": "lscr.io/linuxserver/bookstack:25.02.2", "bookstack-db": "mariadb:11.6"},
to={"bookstack": "lscr.io/linuxserver/bookstack:26.05.2", "bookstack-db": "mariadb:12.3"}),
"E3a": dict(app="bookstack", note="AUTHORED step (the catalog never carried it): app half alone",
frm={"bookstack": "lscr.io/linuxserver/bookstack:25.02.2", "bookstack-db": "mariadb:11.6"},
to={"bookstack": "lscr.io/linuxserver/bookstack:26.05.2", "bookstack-db": "mariadb:11.6"}),
"E3b": dict(app="bookstack", note="AUTHORED step: engine half alone",
frm={"bookstack": "lscr.io/linuxserver/bookstack:26.05.2", "bookstack-db": "mariadb:11.6"},
to={"bookstack": "lscr.io/linuxserver/bookstack:26.05.2", "bookstack-db": "mariadb:12.3"}),
}
# --- compose plumbing --------------------------------------------------------------------------
def render(app: str, images: dict, workdir: Path, env: dict) -> Path:
"""Write the catalog template into workdir with `images` substituted per SERVICE.
Substitution is per service and only on that service's own `image:` line — never a blind
string replace, which would also rewrite an image name that appears in a comment or an env var.
"""
src = (TEMPLATES / app / "docker-compose.yml").read_text()
out, cur = [], None
for line in src.splitlines():
m = re.match(r"^ ([A-Za-z0-9_-]+):\s*$", line)
if m:
cur = m.group(1)
mi = re.match(r"^(\s+image:\s*)(\S+)\s*$", line)
if mi and cur in images:
line = mi.group(1) + images[cur]
out.append(line)
workdir.mkdir(parents=True, exist_ok=True)
(workdir / "docker-compose.yml").write_text("\n".join(out) + "\n")
(workdir / ".env").write_text("".join(f"{k}={v}\n" for k, v in env.items()))
return workdir / "docker-compose.yml"
def compose(workdir: Path, project: str, *args, timeout=1800):
return cvp._sh(["docker", "compose", "-p", project, "-f", str(workdir / "docker-compose.yml"),
"--env-file", str(workdir / ".env")] + list(args), timeout=timeout)
def container_ip(name: str) -> str:
r = cvp._sh(["docker", "inspect", "-f",
"{{range .NetworkSettings.Networks}}{{.IPAddress}} {{end}}", name])
return (r.stdout or "").split()[0] if r.stdout.strip() else ""
def settle(project: str, workdir: Path, wait=420):
"""Wait until every container is running AND every one that declares a healthcheck is healthy.
Returns (ok, seconds, per-container state). A container that has NO healthcheck counts as
settled once it is running — but `running` is never reported as the RESULT, only as a
precondition for asking the app itself (see the module docstring).
"""
t0 = time.time()
deadline = t0 + wait
states = {}
while time.time() < deadline:
r = compose(workdir, project, "ps", "-aq", timeout=120)
cids = [c for c in r.stdout.split() if c]
if not cids:
time.sleep(3)
continue
states, pending = {}, False
for cid in cids:
info = cvp._inspect(cid)
if not info:
pending = True
continue
name = info["Name"].lstrip("/")
st = info.get("State", {})
health = (st.get("Health") or {}).get("Status")
states[name] = {"status": st.get("Status"), "health": health,
"restarts": st.get("RestartCount", 0), "exit": st.get("ExitCode")}
if st.get("Status") != "running":
pending = True
elif health in ("starting", "unhealthy"):
pending = True
if not pending:
return True, round(time.time() - t0, 1), states
time.sleep(5)
return False, round(time.time() - t0, 1), states
MIGRATION_RE = re.compile(
r"migrat|upgrad|schema|alter table|CREATE TABLE|InnoDB: Upgrad|mysql_upgrade|"
r"mariadb-upgrade|Running .* migration|Applying|db:migrate",
re.I)
def migration_lines(project: str, workdir: Path, since_iso: str, limit=6):
"""Verbatim log lines that SAY a migration ran. Never inferred from timing — the finding is the
sentence the app printed, exactly as the Nextcloud refusal was."""
r = compose(workdir, project, "logs", "--since", since_iso, "--no-color", timeout=180)
hits = [ln.strip() for ln in (r.stdout + r.stderr).splitlines() if MIGRATION_RE.search(ln)]
return hits[:limit]
# --- one edge ----------------------------------------------------------------------------------
def run_edge(edge_id: str) -> dict:
e = EDGES[edge_id]
app = e["app"]
project = f"upg{edge_id.lower()}"
workdir = ROOT / "work" / edge_id
ev = EVIDENCE / edge_id
ev.mkdir(parents=True, exist_ok=True)
shutil.rmtree(workdir, ignore_errors=True)
felhom = (TEMPLATES / app / ".felhom.yml").read_text()
compose_text = (TEMPLATES / app / "docker-compose.yml").read_text()
env = cvp.build_env(app, felhom, compose_text)
rec = {"harness_version": HARNESS_VERSION, "edge": edge_id, "app": app, "note": e["note"],
"from": e["frm"], "to": e["to"], "verdict": "inconclusive",
"seed_read_before": False, "seed_read_after": False, "healthy_after": False,
"migration_observed": None, "abort": "not-attempted", "abort_detail": None,
"duration_s": 0, "measured_at": None, "evidence": f"evidence/{edge_id}"}
t0 = time.time()
log = []
def say(msg):
line = f"[{datetime.now(timezone.utc).strftime('%H:%M:%S')}] {msg}"
print(line, flush=True)
log.append(line)
try:
# --- 1. FROM ---
say(f"{edge_id}: deploying {app} at FROM {e['frm']}")
render(app, e["frm"], workdir, env)
up = compose(workdir, project, "up", "-d")
if up.returncode != 0:
rec["abort_detail"] = f"FROM deploy failed rc={up.returncode}: {up.stderr[-800:]}"
say("FROM deploy FAILED — inconclusive, the edge was never reached")
return rec
ok, secs, states = settle(project, workdir)
say(f"FROM settled={ok} in {secs}s :: {json.dumps(states)}")
if not ok:
rec["abort_detail"] = f"FROM never settled: {json.dumps(states)}"
say("FROM never became healthy — inconclusive, not a verdict about the upgrade")
return rec
# --- 2/3. seed + C1 ---
fixture = fx.FIXTURES.get(app)
if fixture is None:
rec["abort_detail"] = "no fixture"
say("no fixture for this app — inconclusive")
return rec
seeded = fixture.seed(container_ip, say)
if seeded is None:
rec["verdict"] = "inconclusive"
rec["abort_detail"] = "no non-browser seed route"
say("INCONCLUSIVE — no non-browser seed route. Nothing was planted by hand.")
return rec
rec["seed_read_before"] = bool(fixture.verify(container_ip, seeded, say))
say(f"C1 (seed reads back BEFORE): {rec['seed_read_before']}")
if not rec["seed_read_before"]:
rec["abort_detail"] = "C1 failed: the fixture could not prove itself before the upgrade"
say("C1 FAILED — a fixture that cannot prove itself first proves nothing after")
return rec
# --- 4. TO ---
swap_at = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
say(f"{edge_id}: swapping to TO {e['to']}")
render(app, e["to"], workdir, env)
up2 = compose(workdir, project, "up", "-d")
say(f"TO up -d rc={up2.returncode}")
ok2, secs2, states2 = settle(project, workdir)
rec["healthy_after"] = ok2
rec["duration_s"] = secs2
say(f"TO settled={ok2} in {secs2}s :: {json.dumps(states2)}")
(ev / "to-states.json").write_text(json.dumps(states2, indent=2))
# CAPTURE THE WHOLE TO-STEP LOG *NOW*, not at the end.
# `docker compose logs` only shows the CONTAINERS THAT EXIST, and the abort below replaces
# them — so the TO images' own output is GONE from any capture taken afterwards. Measured on
# E3, where the single most important line of the run ("MariaDB upgrade … required, but
# skipped") survived only because it had already been extracted. Same class as R-320: the
# intermediate teardown is the one that loses the evidence.
full_to = compose(workdir, project, "logs", "--no-color", timeout=180)
(ev / "to-full.log").write_text((full_to.stdout + full_to.stderr)[-400000:])
mig = migration_lines(project, workdir, swap_at)
rec["migration_observed"] = mig[0] if mig else None
(ev / "migration-lines.txt").write_text("\n".join(mig))
say(f"migration lines observed: {len(mig)}")
# --- 5. THE RESULT ---
rec["seed_read_after"] = bool(fixture.verify(container_ip, seeded, say))
say(f"RESULT (seed reads back AFTER): {rec['seed_read_after']}")
rec["verdict"] = "proven" if (ok2 and rec["seed_read_after"]) else "failed"
# --- 6. the ABORT ---
say(f"{edge_id}: ABORT — putting the FROM images back")
render(app, e["frm"], workdir, env)
up3 = compose(workdir, project, "up", "-d")
ok3, secs3, states3 = settle(project, workdir, wait=180)
(ev / "abort-states.json").write_text(json.dumps(states3, indent=2))
if not ok3:
rec["abort"] = "refuses"
lg = compose(workdir, project, "logs", "--tail", "40", "--no-color", timeout=120)
tail = (lg.stdout + lg.stderr).strip()
(ev / "abort-refusal.txt").write_text(tail)
rec["abort_detail"] = tail[-900:]
say(f"ABORT: the app did NOT come back (rc={up3.returncode}, {secs3}s)")
else:
back = bool(fixture.verify(container_ip, seeded, say))
rec["abort"] = "starts-and-serves" if back else "starts-data-gone"
rec["abort_detail"] = None if back else "the app started but the seeded data was gone"
say(f"ABORT: app came back in {secs3}s; data present={back}")
return rec
finally:
rec["measured_at"] = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
rec["duration_s"] = rec["duration_s"] or round(time.time() - t0, 1)
rec["total_s"] = round(time.time() - t0, 1)
# EVIDENCE FIRST, TEARDOWN SECOND (R-320): the intermediate teardown is the one that gets
# forgotten, so everything is written before a single container is removed.
(ev / "run.log").write_text("\n".join(log))
lg = compose(workdir, project, "logs", "--no-color", timeout=180)
(ev / "compose-final.log").write_text((lg.stdout + lg.stderr)[-400000:]) # post-abort state only — see to-full.log
(ev / "verdict.json").write_text(json.dumps(rec, indent=2))
compose(workdir, project, "down", "-v", "--remove-orphans", timeout=900)
def main(argv):
if not argv or argv[0] == "--list":
for k, v in EDGES.items():
print(f"{k:5s} {v['app']:12s} {v['note']}")
return 0
EVIDENCE.mkdir(parents=True, exist_ok=True)
results = []
for edge_id in argv:
if edge_id not in EDGES:
print(f"unknown edge {edge_id}", file=sys.stderr)
return 2
rec = run_edge(edge_id)
results.append(rec)
print(json.dumps(rec, indent=2), flush=True)
(EVIDENCE / "summary.json").write_text(json.dumps(results, indent=2))
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))
+268
View File
@@ -0,0 +1,268 @@
#!/usr/bin/env python3
"""Per-app seed/verify fixtures for upgrade-test.py.
THE ONE RULE, carried verbatim from survive2.py: *nothing is ever seeded into a volume by hand.*
R-156's evidence shows a root-written canary making an empty volume read as populated — the exact
confusion this harness exists to remove. So every seed below goes in through the app's OWN
interface: its HTTP API, or its own CLI running inside its own container.
A raw SQL INSERT or a planted file is NOT such a route and is never used. If an app has no
non-browser route, its fixture returns None and the edge is recorded `inconclusive — no non-browser
seed route`, with what was tried. **That is a result**: it tells us which apps can never be
auto-verified, which is a fact nobody currently has.
Each fixture returns an opaque `seeded` token from seed() and answers verify() with it. verify()
must ask the APP, never the filesystem: a migration is supposed to rewrite files.
"""
import base64, json, re, secrets, subprocess, time
def _sh(args, timeout=120, inp=None):
try:
return subprocess.run(args, capture_output=True, text=True, timeout=timeout, input=inp)
except (subprocess.TimeoutExpired, OSError) as e:
return subprocess.CompletedProcess(args, 124, "", f"{e}")
def _curl(url, *extra, timeout=60, data=None, method=None):
"""One HTTP call. Gates on curl's OWN exit code, never on a summarising pipeline — the trap
check-image-resolvable.py's docstring names and that has bitten this project twice."""
args = ["curl", "-sS", "--max-time", str(timeout), "-w", "\n%{http_code}"]
if method:
args += ["-X", method]
if data is not None:
args += ["--data-binary", "@-"]
args += list(extra) + [url]
r = _sh(args, timeout=timeout + 20, inp=data)
body, _, code = (r.stdout or "").rpartition("\n")
return r.returncode, code.strip(), body
def _wait_http(url, want, tries=40, delay=5, say=print):
"""Settling says the container is running; this says the APP is answering. They are not the
same thing, and conflating them is how "the container started" gets reported as a pass."""
for i in range(tries):
rc, code, _ = _curl(url, timeout=15)
if rc == 0 and code in want:
return True
time.sleep(delay)
say(f" app never answered on {url} (last rc={rc} code={code})")
return False
# ---------------------------------------------------------------------------------------------
class PrivateBin:
"""PrivateBin's own JSON API. A paste is an HTTP POST and reading it back is an HTTP GET —
the app stores the blob and hands it back, which is an application-level round trip.
PrivateBin is FILE-BACKED with no database, so this single seed IS the file half; there is no
database half to seed separately.
"""
port = 8080
container = "privatebin"
def _base(self, ipfn):
ip = ipfn(self.container)
return f"http://{ip}:{self.port}/" if ip else ""
def seed(self, ipfn, say):
base = self._base(ipfn)
if not base:
say(" privatebin: no container IP")
return None
if not _wait_http(base, {"200"}, say=say):
return None
marker = "upg-" + secrets.token_hex(8)
# The v2 paste envelope. PrivateBin validates it server-side: `ct`, the IV and the salt must
# all be real base64 or the API answers {"status":1,"message":"Invalid data."} — measured,
# and the reason the first attempt at this fixture failed. The marker is carried INSIDE `ct`
# so a readback proves THIS paste came back, not merely A paste.
ct = base64.b64encode(marker.encode()).decode()
body = json.dumps({
"v": 2,
"adata": [[base64.b64encode(secrets.token_bytes(16)).decode(),
base64.b64encode(secrets.token_bytes(8)).decode(),
100000, 256, 128, "aes", "gcm", "none"], "plaintext", 0, 0],
"ct": ct,
"meta": {"expire": "never"},
})
rc, code, out = _curl(base, "-H", "X-Requested-With: JSONHttpRequest",
"-H", "Content-Type: application/json", data=body, method="POST")
if rc != 0:
say(f" privatebin: POST failed rc={rc}")
return None
try:
j = json.loads(out)
except Exception:
say(f" privatebin: POST returned non-JSON (http {code}): {out[:200]}")
return None
if j.get("status") != 0 or not j.get("id"):
say(f" privatebin: POST refused: {out[:250]}")
return None
say(f" privatebin: seeded paste id={j['id']}")
return {"id": j["id"], "marker": ct}
def verify(self, ipfn, seeded, say):
base = self._base(ipfn)
if not base:
return False
if not _wait_http(base, {"200"}, tries=24, say=say):
return False
rc, code, out = _curl(base + "?pasteid=" + seeded["id"],
"-H", "X-Requested-With: JSONHttpRequest")
if rc != 0 or code != "200":
say(f" privatebin: readback rc={rc} http={code}")
return False
got = seeded["marker"] in out
say(f" privatebin: readback http={code} marker_present={got}")
return got
# ---------------------------------------------------------------------------------------------
class Docmost:
"""Docmost's own REST API: create the first workspace+user, then a page, then read it back."""
port = 3000
container = "docmost"
def _base(self, ipfn):
ip = ipfn(self.container)
return f"http://{ip}:{self.port}" if ip else ""
def seed(self, ipfn, say):
base = self._base(ipfn)
if not base:
say(" docmost: no container IP")
return None
if not _wait_http(base + "/api/health", {"200", "404", "401"}, say=say):
if not _wait_http(base + "/", {"200", "302", "404"}, tries=20, say=say):
return None
marker = "upg-" + secrets.token_hex(8)
email = f"spike-{secrets.token_hex(4)}@gate.invalid"
pw = "Spike-" + secrets.token_hex(10)
setup = json.dumps({"workspaceName": "spike", "name": "spike",
"email": email, "password": pw})
rc, code, out = _curl(base + "/api/auth/setup", "-H", "Content-Type: application/json",
"-D", "/tmp/docmost.hdr", data=setup, method="POST")
say(f" docmost: /api/auth/setup http={code} rc={rc}")
if rc != 0 or code not in ("200", "201"):
say(f" docmost: setup refused: {out[:250]}")
return None
tok = ""
try:
tok = json.loads(out).get("tokens", {}).get("accessToken", "") or json.loads(out).get("accessToken", "")
except Exception:
pass
if not tok:
m = re.search(r"authToken=([^;]+)", open("/tmp/docmost.hdr").read())
tok = m.group(1) if m else ""
if not tok:
say(" docmost: no auth token in the setup response")
return None
return {"marker": marker, "email": email, "pw": pw, "token": tok}
def verify(self, ipfn, seeded, say):
"""Prove the app still holds the seeded ACCOUNT by asking it to authenticate — its own
front door, and version-stable across the API churn between 0.25 and 0.95."""
base = self._base(ipfn)
if not base:
return False
if not _wait_http(base + "/", {"200", "302", "404"}, tries=24, say=say):
return False
body = json.dumps({"email": seeded["email"], "password": seeded["pw"]})
rc, code, out = _curl(base + "/api/auth/login", "-H", "Content-Type: application/json",
data=body, method="POST")
ok = rc == 0 and code in ("200", "201")
say(f" docmost: login as the seeded user http={code} ok={ok}")
if not ok:
say(f" docmost: login body {out[:200]}")
return ok
# ---------------------------------------------------------------------------------------------
class BookStack:
"""BookStack has no API token without a browser, so BOTH the seed and the readback go through
`php artisan` — BookStack's OWN CLI, running inside its own container against its own
application code and its own User model. That is categorically different from a raw SQL INSERT
or a planted file, which is what R-156 forbids.
WHY NOT THE HTTP LOGIN FORM, which was the first attempt and is the more obvious choice:
BookStack derives APP_URL from the template as `https://${SUBDOMAIN}.${DOMAIN}`, so it marks its
session and XSRF cookies **`secure`**. curl over plain http therefore stores neither, sends
neither, and every login POST comes back **419 Page Expired** — measured, and it looks exactly
like a wrong password. The container serves no TLS, so there is no http route to a logged-in
session without changing the app's own configuration, which would be testing a different app.
WHY THE EXIT CODE IS NOT THE GATE HERE, stated because the standing rule says to use it:
`bookstack:reset-mfa` asks for interactive confirmation, finds no TTY, and exits **1 in both
cases** — for a user it FOUND and for one it did not. The exit code carries no information, so
the discriminator is the output, and it is required to be positive AND the not-found sentence is
required to be ABSENT. It is non-destructive: it aborts at the unanswered prompt.
THE FIXTURE PROVES ITSELF ON EVERY CALL. Each verify() also probes an email that cannot exist
and requires the "could not be found" answer. A readback that has broken into always saying
"found" therefore fails instead of passing everything.
LIMITATION, recorded rather than papered over: this seeds the DATABASE half only. Seeding a FILE
(an uploaded image or attachment) needs the API token this app cannot mint headlessly.
"""
port = 80
container = "bookstack"
def _base(self, ipfn):
ip = ipfn(self.container)
return f"http://{ip}:{self.port}" if ip else ""
def _artisan(self, *args, timeout=180):
for path in ("/app/www/artisan", "/var/www/html/artisan"):
r = _sh(["docker", "exec", self.container, "php", path] + list(args), timeout=timeout)
out = (r.stdout or "") + (r.stderr or "")
if "Could not open input file" not in out:
return r
return r
def _lookup(self, email):
"""Ask BookStack whether it holds this account. Returns True/False/None(unusable)."""
r = self._artisan("bookstack:reset-mfa", f"--email={email}")
out = " ".join(((r.stdout or "") + (r.stderr or "")).split())
found = f"Email: {email}" in out
missing = "could not be found" in out
if found == missing: # neither, or both — the readback itself is broken
return None, out
return found, out
def seed(self, ipfn, say):
base = self._base(ipfn)
if not base:
say(" bookstack: no container IP")
return None
if not _wait_http(base + "/login", {"200"}, tries=60, say=say):
return None
email = f"spike-{secrets.token_hex(4)}@gate.invalid"
pw = "Spike-" + secrets.token_hex(10)
r = self._artisan("bookstack:create-admin", f"--email={email}",
f"--name=spike-{secrets.token_hex(3)}", f"--password={pw}")
out = " ".join(((r.stdout or "") + (r.stderr or "")).split())
say(f" bookstack: artisan create-admin rc={r.returncode} :: {out[:120]}")
if "successfully created" not in out:
return None
return {"email": email, "pw": pw}
def verify(self, ipfn, seeded, say):
base = self._base(ipfn)
if not base:
return False
# The app must be SERVING, not merely running — "the container started" is not a pass.
if not _wait_http(base + "/login", {"200"}, tries=60, say=say):
say(" bookstack: the app never served /login")
return False
# The fixture's own negative control, run every time.
absent, _ = self._lookup(f"nobody-{secrets.token_hex(6)}@gate.invalid")
if absent is not False:
say(f" bookstack: READBACK IS UNUSABLE — an email that cannot exist did not read as absent ({absent})")
return False
found, out = self._lookup(seeded["email"])
say(f" bookstack: readback of the seeded account found={found} :: {out[:120]}")
return found is True
FIXTURES = {"privatebin": PrivateBin(), "docmost": Docmost(), "bookstack": BookStack()}