conformance.py

#!/usr/bin/env python3
"""XCP v1 conformance checker.

Verifies that a server implements the Xanadu Content Protocol v1:

  1. GET /.well-known/xanadu-server.json  (identity)
  2. GET /api/public/work/{id}             (content + hash)
  3. BLAKE3 hash verification of the served text
  4. Tumbler format validation
  5. Optional: range retrieval, backlink notify, search

Exit code 0 = conformant (required checks pass); 1 = failures.
Optional-feature results are reported but never fail the run.

Usage:
  ./conformance.py http://server:port [work_id]

If work_id is omitted, the script tries to discover one via the
optional search endpoint, else fails with instructions.
"""

import hashlib  # noqa: F401 (blake3 via external lib; see below)
import json
import re
import sys
import urllib.request
import urllib.error

try:
    import blake3
    def _b3(data: bytes) -> str:
        return blake3.blake3(data).hexdigest()
except ImportError:
    # Pure-python fallback is impractical; refuse clearly.
    def _b3(data: bytes) -> str:
        print("ERROR: `pip install blake3` is required for hash verification")
        sys.exit(2)

PASS, FAIL, WARN, SKIP = "PASS", "FAIL", "WARN", "SKIP"
results = []


def check(name, status, detail=""):
    if status is True:
        status = PASS
    elif status is False:
        status = FAIL
    results.append((name, status, detail))
    print(f"[{status}] {name}" + (f" — {detail}" if detail else ""))
    return status == PASS


def get(base, path, timeout=15):
    req = urllib.request.Request(base.rstrip("/") + path,
                                 headers={"Accept": "application/json"})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.loads(r.read().decode("utf-8"))


TUMBLER_RE = re.compile(
    r'^"?[^".]+"?\.[0-9a-fx]+[0-9a-zA-Z.:-]*$'
)


def valid_tumbler(t: str) -> bool:
    """`"server".work.revision[.start-end]` per spec section 4."""
    if not TUMBLER_RE.match(t or ""):
        return False
    parts = t.split(".")
    # last range component, if present: start-end digits
    if "-" in parts[-1]:
        rng = parts[-1].split("-")
        if len(rng) == 2 and rng[0].isdigit() and rng[1].isdigit():
            return int(rng[0]) <= int(rng[1])
    return True


def main():
    if len(sys.argv) < 2:
        print(__doc__)
        sys.exit(2)
    base = sys.argv[1]

    # ── 1. Identity ─────────────────────────────────────────────
    try:
        ident = get(base, "/.well-known/xanadu-server.json")
    except Exception as e:
        check("identity endpoint reachable", FAIL, str(e))
        report()
        return
    ok = check("identity endpoint reachable", PASS)

    ok &= check("protocol field present",
                ident.get("protocol") in ("xcp", "xcgp"),
                f"got {ident.get('protocol')!r}")
    ok &= check("protocol_version == 1",
                ident.get("protocol_version") == 1,
                f"got {ident.get('protocol_version')!r}")
    ok &= check("hash_algorithm == blake3",
                ident.get("hash_algorithm") == "blake3",
                f"got {ident.get('hash_algorithm')!r}")
    for f in ("implementation", "public_address", "server_id",
              "tumbler_prefix", "content_api"):
        ok &= check(f"required field: {f}", bool(ident.get(f)))

    # server_id: hex ed25519 key (32 bytes)
    sid = ident.get("server_id", "")
    ok &= check("server_id is 64-hex",
                bool(re.fullmatch(r"[0-9a-fA-F]{64}", sid or "")),
                f"len={len(sid or '')}")

    # ── 2. Discover a work id ───────────────────────────────────
    work_id = sys.argv[2] if len(sys.argv) > 2 else None
    if not work_id:
        # optional search endpoint
        try:
            s = get(base, "/api/public/work/_xcp_list")
            work_id = None  # placeholder; real servers may not have this
        except Exception:
            pass
    if not work_id:
        try:
            search = get(base, "/api/search?q=")
            works = search.get("works") or search.get("results") or []
            if works:
                w = works[0]
                work_id = w.get("work_id") or w.get("id")
        except Exception:
            pass
    if not work_id:
        check("work discovery", FAIL,
              "no work_id given and discovery endpoints unavailable; "
              "rerun with: conformance.py BASE WORK_ID")
        report()
        return
    check("work under test", PASS, str(work_id))

    # ── 3. Content retrieval ────────────────────────────────────
    try:
        work = get(base, f"/api/public/work/{work_id}")
    except urllib.error.HTTPError as e:
        check("content endpoint", FAIL, f"HTTP {e.code}")
        report()
        return
    except Exception as e:
        check("content endpoint", FAIL, str(e))
        report()
        return
    ok &= check("content endpoint", PASS)

    text = work.get("text")
    ok &= check("text present", isinstance(text, str) and len(text) > 0,
                f"{len(text) if isinstance(text, str) else 0} chars")
    ok &= check("api_version == 1", work.get("api_version") == 1)
    ok &= check("content_hash_blake3 present",
                bool(work.get("content_hash_blake3")))
    ok &= check("hash_algorithm == blake3",
                work.get("hash_algorithm") == "blake3")

    # ── 4. Hash verification (the core trust step) ──────────────
    if isinstance(text, str):
        actual = _b3(text.encode("utf-8"))
        claimed = work.get("content_hash_blake3", "")
        ok &= check("BLAKE3(text) == content_hash_blake3",
                    actual == claimed,
                    f"claimed={claimed[:16]}… actual={actual[:16]}…"
                    if actual != claimed else f"{actual[:16]}… verified")
        ok &= check("char_count correct",
                    work.get("char_count") == len(text),
                    f"claimed={work.get('char_count')} actual={len(text)}")

    # ── 5. Tumbler ──────────────────────────────────────────────
    tumbler = work.get("tumbler", "")
    ok &= check("tumbler present + format", valid_tumbler(tumbler),
                tumbler or "(missing)")
    prefix = ident.get("tumbler_prefix", "")
    if tumbler and prefix:
        ok &= check("tumbler carries server prefix",
                    tumbler.startswith(prefix),
                    f"prefix={prefix!r}")

    # ── 6. Optional features (never fail the run) ───────────────
    start, end = 0, min(10, len(text) - 1) if text else (0, 1)
    try:
        rng = get(base, f"/api/public/work/{work_id}/range/{start}/{end}")
        rt = rng.get("text", "")
        exp = text[start:end] if text else ""
        check("[optional] range endpoint",
              rt == exp, f"got {len(rt)} chars, expected slice {start}:{end}")
        rh = _b3(rt.encode("utf-8"))
        check("[optional] range hash matches range text",
              rh == rng.get("content_hash_blake3"))
    except Exception as e:
        check("[optional] range endpoint", SKIP, str(e)[:60])

    report()


def report():
    print()
    fails = sum(1 for _, s, _ in results if s == FAIL)
    total = len(results)
    print(f"{total} checks: {total - fails} passed, {fails} failed")
    print("CONFORMANT ✓" if fails == 0 else "NOT CONFORMANT ✗")
    sys.exit(0 if fails == 0 else 1)


if __name__ == "__main__":
    main()