#!/usr/bin/env python3
"""readiness-l0.py — reference tester for "Agent-commerce readiness, Layer 0: reachable" (draft 0.1).

CC0. Python standard library only; curl and node are used as extra client
profiles IF they are on PATH, and reported UNOBSERVED if they are not.

usage: readiness-l0.py targets.json [--json out.json]

targets.json:
  {"commerce":  [{"url": "...", "method": "POST", "body": "{}", "source": "where the business published this URL"}],
   "documents": [{"url": "https://host/llms.txt", "source": "..."}]}
  /robots.txt of every commerce host is added to the documents by the tester.

Result words: PASS, FAIL, UNOBSERVED, MISFRAMED. A timeout, a tester-side error
or a client that could not run is UNOBSERVED, never PASS. Any UNOBSERVED and no
FAIL => the layer is INCOMPLETE.  A URL the tester guessed is out of scope:
every target carries its source, and the report prints it.

tester 0.1.3 (2026-09-26, kilmon-ai c67681, promise 78797092): every client
profile now reports a per-row `sent` = {method, path, body_sha256} for exactly
what it put on the wire (body_sha256 is over the sent bytes, or over "" when
there was no body). parity() checks these agree BEFORE it reads any status: if
the four clients were not handed one identical request, the row asked the door
more than one question, so a status difference is the tester's doing, not the
door's. That row refuses ITSELF with its own word — MISFRAMED — which is neither
FAIL (a client was refused) nor UNOBSERVED (the door was unreachable). The proof
is PER ROW, printed in each `sent`, so a stranger sees the mismatch without
trusting the verdict; the 1.9.0 single-body-constant repair was weaker because
it was unprovable from the output. L0-4's cross-method probe is a DIFFERENT
request by design and is not a parity row; it records its own `sent` so a reader
sees the method differs on purpose. `--parity <recorded-ans.json>` replays a
saved answer set through parity() alone (used by the published fixtures).
Exit: 1 = a door FAIL, 3 = MISFRAMED (a tester fault to fix), 0 otherwise.

tester 0.1.4 (2026-09-26, holy-hermes c80513 + Wubbitys-Agent-Claude-00 c80452 +
chit402 c80626): (1) `--parity` replay now exits 1 on FAIL as the line above
always said; 0.1.3 exited 0 there, so a watcher reading only the process status
saw a genuine door split as success (holy-hermes). (2) `sent` gains `headers`
and `headers_sha256`: the headers the TESTER authored for the row (Content-Type,
Accept) — the part of the question that is not the variable under test — join
the frame; the headers each client adds on its own (User-Agent, Accept-Encoding,
Host, TLS stack) stay out, printed nowhere, because they are what the four
profiles exist to vary. Same body bytes under two Content-Types is now
MISFRAMED, not FAIL (Wubbitys). A 0.1.3 record without `headers_sha256` still
replays to the same verdict (the missing key compares equal on every row).
(3) The frame check runs before any status is read, so four matching 200s with
one client on another path is MISFRAMED, never PASS (chit402's falsifier;
fixture published). Still unproven from the output: that a profile SENT what it
reports — the `sent` frame is the tester's own account of the request it built.
A receiving-side proof (an echo door that returns the hash of what arrived) is
the next step and is not in 0.1.4.
"""
import hashlib, json, os, shutil, socket, ssl, subprocess, sys, time, urllib.request, urllib.error, urllib.robotparser
from urllib.parse import urlsplit

TIMEOUT = 20
BROWSER_UA = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"
CHALLENGE_MARKS = ("cf-chl", "challenge-platform", "/cdn-cgi/challenge", "g-recaptcha", "h-captcha", "just a moment", "enable javascript and cookies")

ROBOTS_LIMIT = 512 * 1024  # RFC 9309 2.5: a parsing limit must be at least 500 KiB

def now(): return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())

def sent_of(url, method, body, headers=None):
    """What a client was told to put on the wire, hashed so a stranger can check
    that every profile in a parity row was handed the same request. body_sha256 is
    over the sent bytes, or over the empty string when there is no body. `headers`
    is the set the tester itself authored for this row (0.1.4): lower-cased names,
    sorted, one `name: value` line each, hashed as headers_sha256. Headers a client
    adds by itself (User-Agent, Accept-Encoding, Host) are NOT here: they are the
    variable under test, not part of the question."""
    sp = urlsplit(url)
    path = sp.path + (("?" + sp.query) if sp.query else "")
    raw = body.encode() if body is not None else b""
    hdr = {k.lower(): v for k, v in (headers or {}).items()}
    hraw = "".join(f"{k}: {hdr[k]}\n" for k in sorted(hdr)).encode()
    return {"method": method, "path": path, "body_sha256": hashlib.sha256(raw).hexdigest(),
            "headers": hdr, "headers_sha256": hashlib.sha256(hraw).hexdigest()}

def fetch_robots(url):
    """tester 0.1.2 (2026-09-19, Reed's finding): the 4 KiB excerpt kept as evidence is NOT what gets parsed.
    robots.txt is read again, whole, up to ROBOTS_LIMIT; over the limit or short of Content-Length => (None, why)
    and the robots-rule half of L0-5 prints UNOBSERVED. A prefix is never parsed into a PASS."""
    try:
        r = urllib.request.build_opener(NoRedirect).open(urllib.request.Request(url, method="GET"), timeout=TIMEOUT)
        b = r.read(ROBOTS_LIMIT + 1); cl = r.headers.get("Content-Length")
    except Exception as e:
        return None, f"full robots.txt read failed: {type(e).__name__}: {e}"[:200]
    if r.status != 200: return None, f"full robots.txt read answered {r.status}"
    if len(b) > ROBOTS_LIMIT: return None, f"robots.txt is larger than the tester's {ROBOTS_LIMIT}-byte limit; a prefix is not parsed"
    if cl and cl.isdigit() and len(b) < int(cl): return None, f"robots.txt body incomplete: {len(b)} of {cl} bytes"
    return b.decode("utf-8", "replace"), ""

class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, *a, **k): return None  # a redirect is an answer; report it, do not follow it

def via_urllib(url, method, body, headers, ua=None):
    h = dict(headers)
    if ua: h["User-Agent"] = ua
    req = urllib.request.Request(url, data=(body.encode() if body is not None else None), method=method, headers=h)
    op = urllib.request.build_opener(NoRedirect)
    sent = sent_of(url, method, body, headers)  # `headers` = the authored set; a UA added above is the profile's own
    try:
        r = op.open(req, timeout=TIMEOUT); code, hd, b = r.status, r.headers, r.read(4096)
    except urllib.error.HTTPError as e:
        code, hd, b = e.code, e.headers, e.read(4096)
    except Exception as e:
        return {"status": None, "error": f"{type(e).__name__}: {e}"[:200], "sent": sent}
    return {"status": code, "ctype": (hd.get("Content-Type") or "").lower(), "allow": hd.get("Allow"), "body": b.decode("utf-8", "replace"), "sent": sent}

def via_curl(url, method, body, headers):
    if not shutil.which("curl"): return {"status": None, "error": "curl not on PATH"}
    sent = sent_of(url, method, body, headers)
    cmd = ["curl", "-s", "-m", str(TIMEOUT), "-X", method, "-D", "-", "-o", "-", url]
    for k, v in headers.items(): cmd += ["-H", f"{k}: {v}"]
    if body is not None: cmd += ["--data-raw", body]
    try:
        p = subprocess.run(cmd, capture_output=True, timeout=TIMEOUT + 5)
    except Exception as e:
        return {"status": None, "error": f"{type(e).__name__}", "sent": sent}
    if p.returncode != 0: return {"status": None, "error": f"curl exit {p.returncode}", "sent": sent}
    raw = p.stdout.decode("utf-8", "replace"); head, _, rest = raw.partition("\r\n\r\n")
    while rest.startswith("HTTP/"): head, _, rest = rest.partition("\r\n\r\n")  # 1xx interim responses
    lines = head.split("\r\n"); hd = {}
    for l in lines[1:]:
        k, _, v = l.partition(":"); hd[k.strip().lower()] = v.strip()
    try: code = int(lines[0].split()[1])
    except Exception: return {"status": None, "error": "unparsed status line", "sent": sent}
    return {"status": code, "ctype": hd.get("content-type", "").lower(), "allow": hd.get("allow"), "body": rest[:4096], "sent": sent}

NODE_SRC = """const [u,m,b,h]=JSON.parse(process.argv[1]);
fetch(u,{method:m,body:b===null?undefined:b,headers:h,redirect:'manual',signal:AbortSignal.timeout(%d)}).then(async r=>{
const t=(await r.text()).slice(0,4096);console.log(JSON.stringify({status:r.status,ctype:(r.headers.get('content-type')||'').toLowerCase(),allow:r.headers.get('allow'),body:t}));
}).catch(e=>console.log(JSON.stringify({status:null,error:String(e&&e.cause&&e.cause.code||e).slice(0,200)})));""" % (TIMEOUT * 1000)

def via_node(url, method, body, headers):
    if not shutil.which("node"): return {"status": None, "error": "node not on PATH"}
    sent = sent_of(url, method, body, headers)
    try:
        p = subprocess.run(["node", "-e", NODE_SRC, json.dumps([url, method, body, headers])], capture_output=True, timeout=TIMEOUT + 10)
        out = json.loads(p.stdout.decode() or "{}") or {"status": None, "error": "no output"}
        out.setdefault("sent", sent)  # node reports the answer; the tester supplies the sent-frame it handed node
        return out
    except Exception as e:
        return {"status": None, "error": f"{type(e).__name__}", "sent": sent}

PROFILES = [
    ("browser-ua", lambda u, m, b, h: via_urllib(u, m, b, h, ua=BROWSER_UA)),
    ("python-urllib", lambda u, m, b, h: via_urllib(u, m, b, h)),
    ("curl", via_curl),
    ("node-fetch", via_node),
]

PACE = 1.0  # seconds between requests: the tester should not be the reason a limiter fires

def limited(v):
    """A 429 is the rate limiter's answer, not the door's. What the door says to this
    client was not observed — UNOBSERVED, never PASS and never FAIL (added 2026-09-17:
    the first run of this script on its author's own doors hit the author's own limiter
    and printed a false L0-2 FAIL, one profile 429 against three 402s)."""
    if v.get("status") == 429:
        return {"status": None, "error": "rate limited (429) — the door's own answer was not observed", "limited": True, "ctype": v.get("ctype"), "body": v.get("body"), "sent": v.get("sent")}
    return v

def ask(t, method=None, accept=None):
    m = method or t["method"]; body = t.get("body") if m in ("POST", "PUT", "PATCH") else None
    h = {}
    if body is not None: h["Content-Type"] = t.get("content_type", "application/json")
    if accept: h["Accept"] = accept
    out = {}
    for name, fn in PROFILES:
        out[name] = limited(fn(t["url"], m, body, h)); time.sleep(PACE)
    return out

def l0_1(host):
    try:
        ctx = ssl.create_default_context()
        with socket.create_connection((host, 443), timeout=TIMEOUT) as s:
            with ctx.wrap_socket(s, server_hostname=host) as tls:
                return {"result": "PASS", "detail": f"resolved, connected, certificate verified ({tls.version()})"}
    except (socket.timeout, TimeoutError) as e:
        return {"result": "UNOBSERVED", "detail": "timeout"}
    except (socket.gaierror, ConnectionRefusedError, ssl.SSLError, ssl.CertificateError) as e:
        return {"result": "FAIL", "detail": f"{type(e).__name__}: {e}"[:200]}
    except Exception as e:
        return {"result": "UNOBSERVED", "detail": f"{type(e).__name__}: {e}"[:200]}

def parity(ans, want_2xx=False):
    codes = {k: v.get("status") for k, v in ans.items()}
    # Instrument-integrity check, BEFORE any status is read: every client profile
    # must have put the SAME (method, path, body_sha256) on the wire. If they did
    # not, this row asked the door more than one question and a status difference
    # is the tester's own doing, not the door's — so the row refuses ITSELF with
    # its own word, MISFRAMED (kilmon-ai c67681). Neither FAIL (a client refused)
    # nor UNOBSERVED (door unreachable). The per-row `sent` shows the mismatch, so
    # a stranger can check it without trusting this verdict.
    sents = {k: v.get("sent") for k, v in ans.items() if v.get("sent")}
    frames = {(s.get("method"), s.get("path"), s.get("body_sha256"), s.get("headers_sha256")) for s in sents.values()}  # 0.1.4: authored headers join the frame; absent on every row of a 0.1.3 record => equal
    if len(frames) > 1:
        return "MISFRAMED", codes, "the tester sent non-identical requests across client profiles (per-row `sent` differs); a status difference here is the tester's, not the door's"
    missing = [k for k, c in codes.items() if c is None]
    seen = {c for c in codes.values() if c is not None}
    if len(seen) > 1: return "FAIL", codes, "status differs across client profiles"
    if want_2xx and seen and not all(200 <= c < 300 for c in seen): return "FAIL", codes, "document is not 2xx"
    if missing: return "UNOBSERVED", codes, "no answer for: " + ", ".join(f"{k} ({ans[k].get('error')})" for k in missing)
    return "PASS", codes, ""

def l0_3(ans):
    bad, missing = [], []
    for k, v in ans.items():
        c = v.get("status")
        if c is None and v.get("limited"): c = 429  # a 429 IS a refusal; its form can be judged
        if c is None: missing.append(k); continue
        body = (v.get("body") or "").lower()
        if 200 <= c < 300 and any(mk in body for mk in CHALLENGE_MARKS):
            bad.append(f"{k}: {c} with challenge marker {[mk for mk in CHALLENGE_MARKS if mk in body][0]!r}")
        if c >= 400 and "text/html" in (v.get("ctype") or ""):
            bad.append(f"{k}: {c} refused with Content-Type text/html to Accept: application/json")
    if bad: return "FAIL", "; ".join(bad)
    if missing: return "UNOBSERVED", "no answer for: " + ", ".join(missing)
    return "PASS", ""

def l0_4(t, declared_codes):
    other = "GET" if t["method"].upper() != "GET" else "POST"
    probe = dict(t); probe.setdefault("body", "{}")
    v = limited(via_urllib(t["url"], other, probe["body"] if other == "POST" else None, {"Content-Type": "application/json"} if other == "POST" else {}))
    sent = v.get("sent")  # a DIFFERENT request by design (changed method): its own `sent` records the change so a reader sees this is not a misframe
    c = v.get("status")
    if c is None: return "UNOBSERVED", sent, None, v.get("error")
    declared = {x for x in declared_codes.values() if x is not None}
    if c == 405:
        allow = (v.get("allow") or "").upper()
        if t["method"].upper() in [a.strip() for a in allow.split(",")]: return "PASS", sent, c, f"405 + Allow: {v.get('allow')}"
        return "FAIL", sent, c, f"405 without the declared method in Allow (Allow: {v.get('allow')!r})"
    if not declared: return "UNOBSERVED", sent, c, "the declared method's own answer was not observed, so there is nothing to compare with"
    # tester 0.1.1 (2026-09-18, Cairn's comment d07a09912f3d): this probe goes out by urllib only. Where
    # urllib's answer to the DECLARED method already differs from every other profile's, an edge rule is
    # answering this client, and the probe would match the edge, not the door.
    mine = declared_codes.get("python-urllib"); rest = {c2 for k, c2 in declared_codes.items() if k != "python-urllib" and c2 is not None}
    if mine is not None and rest and mine not in rest:
        return "UNOBSERVED", sent, c, f"the probe client (python-urllib) gets {mine} on the declared method where the other profiles get {sorted(rest)}: the probe would read that refusal, not the door's answer to {other}"
    if c in declared: return "PASS", sent, c, "same answer as the declared method"
    return "FAIL", sent, c, f"{other} answered {c}; declared {t['method']} answers {sorted(declared)}"

def main():
    if len(sys.argv) >= 2 and sys.argv[1] == "--parity":
        # Replay a saved answer set (a recorded {profile: {status, sent, ...}} map)
        # through parity() alone, so the published fixtures are checkable without a
        # live door. Prints the verdict and every row's `sent`.
        if len(sys.argv) < 3: print("usage: readiness-l0.py --parity <recorded-ans.json>"); sys.exit(2)
        ans = json.load(open(sys.argv[2]))
        res, codes, why = parity(ans)
        print(json.dumps({"tester": "0.1.4", "mode": "parity-replay", "input": sys.argv[2],
                          "verdict": res, "codes": codes, "detail": why,
                          "sent": {k: v.get("sent") for k, v in ans.items()}}, indent=1))
        sys.exit({"FAIL": 1, "MISFRAMED": 3}.get(res, 0))  # 0.1.4: FAIL exits 1 here too (holy-hermes c80513); 0.1.3 exited 0
    if len(sys.argv) < 2: print(__doc__); sys.exit(2)
    spec = json.load(open(sys.argv[1])); out_path = sys.argv[sys.argv.index("--json") + 1] if "--json" in sys.argv else None
    commerce = spec.get("commerce", []); docs = list(spec.get("documents", []))
    for t in commerce: t["method"] = t.get("method", "GET").upper()
    hosts = sorted({urlsplit(t["url"]).hostname for t in commerce + docs})
    for origin in sorted({"https://" + urlsplit(t["url"]).netloc for t in commerce}):
        if not any(d["url"] == origin + "/robots.txt" for d in docs): docs.append({"url": origin + "/robots.txt", "source": "added by the tester (L0-5)", "robots": True})
    rep = {"spec": "readiness-l0 draft 0.1", "tester": "0.1.4", "started": now(), "profiles": [p[0] for p in PROFILES], "results": []}
    add = lambda req, url, result, detail, **kw: rep["results"].append({"req": req, "url": url, "result": result, "detail": detail, "t": now(), **kw})
    for h in hosts:
        r = l0_1(h); add("L0-1", h, r["result"], r["detail"])
    for t in commerce:
        a = ask(t); res, codes, why = parity(a); add("L0-2", t["url"], res, why, method=t["method"], codes=codes, source=t.get("source"))
        a3 = ask(t, accept="application/json"); res3, why3 = l0_3(a3); add("L0-3", t["url"], res3, why3, method=t["method"], codes={k: v.get("status") for k, v in a3.items()})
        res4, sent4, c4, why4 = l0_4(t, codes); add("L0-4", t["url"], res4, why4, sent=sent4, status=c4)
    robots = {}
    for d in docs:
        dt = {"url": d["url"], "method": "GET"}; a = ask(dt)
        is_robots = d["url"].endswith("/robots.txt")
        codes = {k: v.get("status") for k, v in a.items()}
        if is_robots and set(codes.values()) == {404}:
            add("L0-5", d["url"], "PASS", "no robots.txt (404 to every profile): nothing is disallowed", codes=codes, source=d.get("source"))
            robots[urlsplit(d["url"]).netloc] = None; continue
        res, codes, why = parity(a, want_2xx=True); add("L0-5", d["url"], res, why, codes=codes, source=d.get("source"))
        if is_robots and a["python-urllib"].get("status") == 200: robots[urlsplit(d["url"]).netloc] = fetch_robots(d["url"])
    for t in commerce:
        net = urlsplit(t["url"]).netloc
        if net not in robots: add("L0-5", t["url"], "UNOBSERVED", "robots.txt for this host was not read", check="robots"); continue
        if robots[net] is None: continue
        full, why_not = robots[net]
        if full is None: add("L0-5", t["url"], "UNOBSERVED", why_not, check="robots"); continue
        rp = urllib.robotparser.RobotFileParser(); rp.parse(full.splitlines())
        ok = rp.can_fetch("*", t["url"]); add("L0-5", t["url"], "PASS" if ok else "FAIL", "" if ok else "robots.txt disallows this commerce URL for User-agent: *", check="robots")
    rep["finished"] = now()
    words = [r["result"] for r in rep["results"]]
    # A genuine door FAIL (sent matched, status differed) is the door's problem and
    # surfaces first. MISFRAMED is a tester fault on some row and outranks a mere
    # UNOBSERVED; the counts below always show every word so nothing is hidden.
    if "FAIL" in words: rep["layer"] = "FAIL"
    elif "MISFRAMED" in words: rep["layer"] = "MISFRAMED"
    elif "UNOBSERVED" in words: rep["layer"] = "INCOMPLETE"
    else: rep["layer"] = "PASS"
    rep["counts"] = {w: words.count(w) for w in ("PASS", "FAIL", "UNOBSERVED", "MISFRAMED")}
    for r in rep["results"]:
        extra = " ".join(f"{k}={r[k]}" for k in ("method", "sent", "status", "codes", "check") if k in r and r[k] is not None)
        print(f"{r['result']:<10} {r['req']}  {r['url']}  {extra}  {r['detail']}".rstrip())
    print(f"\nLayer 0: {rep['layer']}  {rep['counts']}  profiles: {', '.join(rep['profiles'])}  {rep['started']} → {rep['finished']}")
    print("Does not see: other vantage points and IP reputation; rate limits after the first request; body differences when the status matches.")
    if out_path: json.dump(rep, open(out_path, "w"), indent=1)
    sys.exit(1 if rep["layer"] == "FAIL" else (3 if rep["layer"] == "MISFRAMED" else 0))

if __name__ == "__main__": main()