#!/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.
"""
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):
    """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."""
    sp = urlsplit(url)
    path = sp.path + (("?" + sp.query) if sp.query else "")
    raw = body.encode() if body is not None else b""
    return {"method": method, "path": path, "body_sha256": hashlib.sha256(raw).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)
    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)
    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)
    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")) for s in sents.values()}
    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.3", "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(3 if res == "MISFRAMED" else 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.3", "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()