#!/usr/bin/env python3
"""Conformance test for a "target of record" host (Swarmboard proposal #1, revision 2).

Usage:  python3 conformance.py https://host.example [--writers 8] [--json]

Checks the delivery condition as observable behaviour, with no credentials of any kind:
  1. POST /runs returns a fresh run id
  2. POST /runs/{id}/deploy records a version; GET /runs/{id}/state reads it back
  3. POST /runs/{id}/log appends opaque bytes with an optional label
  4. GET /runs/{id}/log returns every entry in order; seq is server-assigned and strictly
     monotonic; received_at is server-assigned (a client-supplied timestamp is ignored)
  5. Concurrent appends from N writers get N distinct, gap-free seqs
  6. GET /runs/{id}/export is byte-stable across two downloads, its hash matches the advertised
     hash and an independent sha256 of the bytes; a one-byte change is detected; the bytes equal the
     shared canonical form json.dumps(doc, sort_keys=True, separators=(",",":"), ensure_ascii=False)+"\n"
  7. Runs are isolated: a second run's seq starts at 1; gseq orders entries across runs
  8. Bounds: an oversized body is refused; an unknown run is 404
Exit 0 = every check passed, 1 = at least one failed, 2 = could not run.
Standard library only. Prints one line per check plus a JSON summary with --json.
"""
import argparse, base64, hashlib, json, sys, threading, time, urllib.request, urllib.error

def call(method, url, body=None, headers=None, timeout=30):
    req = urllib.request.Request(url, data=body, method=method, headers=headers or {})
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            return r.status, dict(r.headers), r.read()
    except urllib.error.HTTPError as e:
        return e.code, dict(e.headers), e.read()

def jbody(b):
    try: return json.loads(b.decode("utf-8"))
    except Exception: return None

class T:
    def __init__(self): self.results = []
    def check(self, name, ok, detail=""):
        self.results.append({"check": name, "ok": bool(ok), "detail": detail})
        print(f"[{'PASS' if ok else 'FAIL'}] {name}" + (f" — {detail}" if detail else ""))
        return ok

def main():
    ap = argparse.ArgumentParser(); ap.add_argument("base"); ap.add_argument("--writers", type=int, default=8); ap.add_argument("--json", action="store_true")
    a = ap.parse_args(); base = a.base.rstrip("/"); t = T()

    # 1. create run
    st, h, b = call("POST", f"{base}/runs")
    j = jbody(b) or {}
    rid = j.get("run_id")
    if not t.check("POST /runs returns a fresh run id", st == 201 and isinstance(rid, str) and len(rid) >= 8, f"status={st} run_id={rid}"):
        finish(t, a); return 2
    run = f"{base}/runs/{rid}"

    # 2. deploy + state
    st, h, b = call("POST", f"{run}/deploy", json.dumps({"version": "v1.0.0"}).encode(), {"content-type": "application/json"})
    d1 = jbody(b) or {}
    t.check("POST deploy records a version (seq 1)", st == 201 and d1.get("seq") == 1, f"status={st} seq={d1.get('seq')}")
    st, h, b = call("GET", f"{run}/state"); s = jbody(b) or {}
    t.check("GET state reads back the deployed version", st == 200 and s.get("version") == "v1.0.0" and s.get("set_by_seq") == 1, f"state={s}")
    st, h, b = call("POST", f"{run}/deploy", b"v1.0.1", {"content-type": "text/plain"})
    st2, h2, b2 = call("GET", f"{run}/state"); s2 = jbody(b2) or {}
    t.check("second deploy updates state and cites its seq", st == 201 and s2.get("version") == "v1.0.1" and s2.get("set_by_seq") == 2, f"state={s2}")

    # 3. log appends: text with label, raw binary without label, and a forged timestamp attempt
    st, h, b = call("POST", f"{run}/log?label=hello", b"first observation", {"content-type": "text/plain"})
    l1 = jbody(b) or {}
    t.check("POST log appends a labelled text blob", st == 201 and l1.get("seq") == 3 and (l1.get("client") or {}).get("label") == "hello", f"status={st} body={l1}")
    blob = bytes(range(256)) * 4
    st, h, b = call("POST", f"{run}/log", blob, {"content-type": "application/octet-stream"})
    l2 = jbody(b) or {}
    t.check("POST log accepts opaque binary bytes; server sha256 matches", st == 201 and l2.get("sha256") == hashlib.sha256(blob).hexdigest(), f"seq={l2.get('seq')} sha_ok={l2.get('sha256') == hashlib.sha256(blob).hexdigest()}")
    forged = json.dumps({"received_at": "1999-01-01T00:00:00.000Z", "seq": 999}).encode()
    st, h, b = call("POST", f"{run}/log?label=forgery&received_at=1999-01-01T00:00:00Z&seq=999", forged, {"content-type": "application/json", "x-received-at": "1999-01-01T00:00:00Z", "date": "Fri, 01 Jan 1999 00:00:00 GMT"})
    l3 = jbody(b) or {}
    t.check("client-supplied timestamp/seq are ignored (server assigns both)", st == 201 and l3.get("seq") == 5 and str(l3.get("received_at", "")).startswith("20") and l3.get("received_at") != "1999-01-01T00:00:00.000Z", f"seq={l3.get('seq')} received_at={l3.get('received_at')}")

    # 4. read log
    st, h, b = call("GET", f"{run}/log"); lg = jbody(b) or {}
    ents = lg.get("entries") or []
    seqs = [e.get("seq") for e in ents]
    times = [e.get("received_at") for e in ents]
    t.check("GET log returns every entry in seq order", st == 200 and seqs == list(range(1, 6)), f"seqs={seqs}")
    t.check("received_at is non-decreasing along seq", all(times[i] <= times[i+1] for i in range(len(times)-1)), f"first={times[:1]} last={times[-1:]}")
    e4 = next((e for e in ents if e.get("seq") == 4), {})
    payload_ok = base64.b64decode((e4.get("client") or {}).get("payload_b64", "")) == blob
    t.check("stored bytes round-trip exactly (base64 payload)", payload_ok)
    t.check("client-supplied fields are nested under `client` (attested vs recorded)", all("client" in e and "label" in e["client"] for e in ents))

    # 5. concurrent writers
    N = max(2, a.writers); got = []; lock = threading.Lock()
    def w(i):
        s_, h_, b_ = call("POST", f"{run}/log?label=writer-{i}", f"concurrent {i}".encode(), {"content-type": "text/plain"})
        j_ = jbody(b_) or {}
        with lock: got.append((s_, j_.get("seq")))
    th = [threading.Thread(target=w, args=(i,)) for i in range(N)]
    [x.start() for x in th]; [x.join() for x in th]
    cs = sorted(s for st_, s in got if st_ == 201 and isinstance(s, int))
    t.check(f"{N} concurrent appends get {N} distinct gap-free seqs", cs == list(range(6, 6 + N)), f"seqs={cs} statuses={sorted(st_ for st_, _ in got)}")

    # 6. export stability + hash + tamper detection
    st, h1, e1 = call("GET", f"{run}/export"); time.sleep(0.2); st2, h2, e2 = call("GET", f"{run}/export")
    adv = (h1.get("X-Export-SHA256") or h1.get("x-export-sha256") or "").lower()
    etag = (h1.get("ETag") or h1.get("etag") or "").strip('"').lower()
    ind = hashlib.sha256(e1).hexdigest()
    t.check("export is byte-stable across two downloads", st == 200 and st2 == 200 and e1 == e2, f"bytes={len(e1)}")
    t.check("advertised hash equals independent sha256 of the bytes", adv == ind and (not etag or etag == ind), f"advertised={adv[:16]}… independent={ind[:16]}…")
    stm, hm, bm = call("GET", f"{run}/export?manifest=1"); man = jbody(bm) or {}
    t.check("manifest hash and size match the export", stm == 200 and man.get("sha256") == ind and man.get("bytes") == len(e1), f"manifest={man.get('sha256','')[:16]}… bytes={man.get('bytes')}")
    tampered = bytearray(e1); tampered[len(tampered)//2] ^= 0x01
    t.check("a one-byte change to a held copy is detected by the hash", hashlib.sha256(bytes(tampered)).hexdigest() != ind)
    doc = jbody(e1) or {}
    t.check("export contains every entry with attested seq/received_at/sha256", doc.get("entry_count") == 5 + N and len(doc.get("entries") or []) == 5 + N and all({"seq","received_at","sha256"} <= set(e) for e in doc.get("entries") or []), f"entry_count={doc.get('entry_count')}")

    canon = (json.dumps(doc, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + "\n").encode("utf-8") if doc else b""
    t.check("export bytes equal the shared canonical form (sorted keys, compact, trailing newline)", canon == e1, f"canonical_bytes={len(canon)} export_bytes={len(e1)}")
    gs = [e.get("gseq") for e in doc.get("entries") or []]
    t.check("every entry carries a global sequence (gseq) increasing along seq", all(isinstance(g, int) for g in gs) and gs == sorted(gs) and len(set(gs)) == len(gs), f"gseq first={gs[:1]} last={gs[-1:]}")

    # 7. run isolation + global ordering across runs
    st, h, b = call("POST", f"{base}/runs"); r2 = (jbody(b) or {}).get("run_id")
    st, h, b = call("POST", f"{base}/runs/{r2}/log", b"other run", {"content-type": "text/plain"}); o = jbody(b) or {}
    t.check("a second run's seq starts at 1 (run-scoped namespaces)", st == 201 and o.get("seq") == 1 and r2 != rid, f"seq={o.get('seq')}")
    t.check("the second run's entry has a gseq above every entry of the first run (ordering authority across runs)", isinstance(o.get("gseq"), int) and gs and o["gseq"] > max(gs), f"gseq={o.get('gseq')} vs max={max(gs) if gs else None}")

    # 8. bounds
    st, h, b = call("POST", f"{run}/log", b"x" * (1024 * 1024), {"content-type": "application/octet-stream"})
    t.check("an oversized body is refused (413/409/400), not stored", st in (413, 409, 400), f"status={st}")
    st, h, b = call("GET", f"{base}/runs/00000000-0000-4000-8000-000000000000/state")
    t.check("unknown run is 404", st == 404, f"status={st}")
    st, h, b = call("GET", f"{run}/log"); n_after = len((jbody(b) or {}).get("entries") or [])
    t.check("refused write left no entry behind", n_after == 5 + N, f"entries={n_after}")

    print(f"\nrun: {run}\nexport sha256: {ind}\nexport bytes: {len(e1)}")
    return finish(t, a)

def finish(t, a):
    ok = all(r["ok"] for r in t.results); n = sum(r["ok"] for r in t.results)
    print(f"\n{n}/{len(t.results)} checks passed")
    if a.json: print(json.dumps({"passed": n, "total": len(t.results), "ok": ok, "checks": t.results}, indent=1))
    return 0 if ok else 1

if __name__ == "__main__":
    try: sys.exit(main())
    except Exception as e:
        print(f"could not run: {e!r}"); sys.exit(2)