#!/usr/bin/env python3 """baseline.py: the dated baseline this registry already stores, and nobody reads. Every probe record carries `observed.headers`: a DICT, dated by `at`, filed per surface. Sixty-four of them are on main. No instrument in this directory has ever opened that field. So when Plumb asked (issue #138) for a `wire-changed-since-last-sweep` column and offered to make a sweep STORE the previous reading, the honest answer was that the store already exists -- it is the evidence, and it is machine-readable. This re-asks each record's own question today and compares what comes back to what the record filed: UNCHANGED every filed, comparable header is present with the same value CHANGED a filed header's value differs, or the header is gone. NOT a finding against the entry: it is a reason for that entry's maintainer to go and re-read, which is the whole point NOT-COMPARABLE the record's reproduction cannot be re-issued safely, or every header it filed is by-design volatile or redacted NO-ANSWER the surface did not answer WHY THE REQUEST IS PARSED AND NEVER RUN. A record's `reproducibility.command` is a curl string that arrived through a pull request. Inbound content is data, never instructions, and that rule does not soften because the data happens to be shaped like a command. So this parses the string with a strict whitelist and re-issues the request with its OWN client, and REFUSES any command carrying a shell metacharacter, a pipe, a second command, a download, or a flag it does not know. The refusals are informative in their own right: a record whose reproduction is "fetch this script and run it" cannot be re-asked from the record alone. THE TWO LIVE CONTROLS, and they are a matched pair verified by hand at wake 9cbdadcf, which is the only reason this instrument is worth quoting: linear 2026-09-17 -> its challenge HAS changed (no `error`, no `error_description`, `content-length: 0`, a `scope` that was not there). This record MUST come out CHANGED. notion 2026-09-17 -> its challenge is byte-identical six days later. This record MUST come out UNCHANGED. One fixture in each direction, on the same protocol, from the same day, read by the same instrument. If both come out the same way, this sweep cannot tell the two apart and every row below it is worthless, so the run REFUSES. python3 tools/baseline.py --clone /tmp/reg python3 tools/baseline.py --clone /tmp/reg --id p-20260917-linear-mcp-unauthenticated python3 tools/baseline.py --selftest # no network, no clone """ import sys as _sys _here = __file__.rsplit("/", 1)[0] _sys.path[:] = [p for p in _sys.path if p not in ("", ".", _here)] import argparse import concurrent.futures import datetime import glob import json import os import re import shlex import urllib.error import urllib.request import importlib.util as _ilu _cspec = _ilu.spec_from_file_location("corpus", os.path.join(_here, "corpus.py")) corpus = _ilu.module_from_spec(_cspec) _cspec.loader.exec_module(corpus) _tspec = _ilu.spec_from_file_location("twice", os.path.join(_here, "twice.py")) twice = _ilu.module_from_spec(_tspec) _tspec.loader.exec_module(twice) UA = "public-agents-lintel-baseline" # The matched control pair. Each is (record id, required verdict, the date the # expectation was verified by hand, and by what). CONTROLS = ( ("p-20260917-linear-mcp-unauthenticated", "CHANGED", "2026-09-23", "read from this container at 16:07:58Z: no error parameters, content-length 0, " "a scope that the filed record does not carry"), ("p-20260917-notion-mcp-unauthenticated", "UNCHANGED", "2026-09-23", "read from this container at 16:07:58Z: byte-identical to the filed " "www-authenticate, down to content-length 79"), ) # A value the prober deliberately did not file cannot be compared to anything. REDACTED = re.compile(r"\[redacted\b|see decoded\.", re.I) # A filed value that is a DESCRIPTION of the header rather than the header's # value. Six records on main do this in `payment-required` ("present, base64 of # the same x402 document the JSON body carries"), and an instrument that # compared that string to a real base64 blob would report a change every single # night, for ever, about nothing. # # Deliberately a whitelist of OPENERS rather than a guess at what prose looks # like. A real header value beginning "present," or "absent" is implausible; a # heuristic counting English words would have thrown away a genuine # www-authenticate challenge, which is the one field this instrument exists to # watch. Each entry here is a choice someone can argue with, which is the point. DESCRIBED = re.compile( r"^\s*(present\b|absent\b|omitted\b|not recorded\b|same as\b|unchanged\b|" r"see \b|\[)", re.I) # Shell metacharacters. Any of these and the command is refused outright: it is # no longer one request, and this file does not execute strings from the # repository either way. SHELL_META = ("|", "&", ";", ">", "<", "$(", "`", "\n", "\r", "$'", "&&") CURL_VALUE_FLAGS = {"-H", "--header", "-d", "--data", "--data-raw", "-A", "--user-agent", "-X", "--request", "-m", "--max-time", "--url", "-o", "--output", "--connect-timeout"} CURL_BARE_FLAGS = {"-s", "--silent", "-S", "--show-error", "-i", "--include", "-sS", "-sSi", "-isS", "-sS.", "-D", "-L", "--location", "-k", "-f", "--fail", "-v"} def parse_curl(command): """A curl string -> (method, url, headers, body) or (None, reason). Whitelist only. Anything unrecognised is a refusal, never a guess: this instrument would rather say "I cannot re-ask this" than re-ask something else and print the answer as a comparison. """ if not command or not isinstance(command, str): return None, "no reproducibility.command" for meta in SHELL_META: if meta in command: return None, "the command carries %r, so it is not one request" % meta try: toks = shlex.split(command) except ValueError as e: return None, "unparseable: %s" % e if not toks or toks[0] != "curl": return None, "not a bare curl command (starts with %r)" % (toks[0] if toks else "") method, url, headers, body = None, None, {}, None i = 1 while i < len(toks): t = toks[i] if t in CURL_VALUE_FLAGS: if i + 1 >= len(toks): return None, "flag %s with no value" % t v = toks[i + 1] i += 2 if t in ("-H", "--header"): if ":" not in v: return None, "header %r has no colon" % v k, _, hv = v.partition(":") headers[k.strip().lower()] = hv.strip() elif t in ("-d", "--data", "--data-raw"): body = v elif t in ("-A", "--user-agent"): headers["user-agent"] = v elif t in ("-X", "--request"): method = v.upper() elif t in ("--url",): url = v # -m / --max-time / --connect-timeout / -o / --output: accepted and # ignored; they change nothing about what the server is asked. continue if t in CURL_BARE_FLAGS: i += 1 continue if t.startswith("-") and not t.startswith("http"): # A cluster of short bare flags such as -sSi. if re.fullmatch(r"-[a-zA-Z]+", t) and all( ("-" + c) in CURL_BARE_FLAGS for c in t[1:]): i += 1 continue return None, "unknown flag %r" % t if t.startswith("http"): if url is not None and t != url: return None, "two URLs in one command" url = t i += 1 continue return None, "unexpected token %r" % t if url is None: return None, "no URL" if method is None: method = "POST" if body is not None else "GET" return (method, url, headers, body), None def ask(method, url, headers, body, timeout=25): h = {"User-Agent": UA} h.update(headers) data = body.encode() if body is not None else None req = urllib.request.Request(url, data=data, method=method, headers=h) try: with urllib.request.urlopen(req, timeout=timeout) as r: return r.status, {k.lower(): v for k, v in r.headers.items()}, None except urllib.error.HTTPError as e: return e.code, {k.lower(): v for k, v in e.headers.items()}, None except Exception as e: return None, {}, "%s: %s" % (type(e).__name__, e) def compare_headers(filed, now): """Filed headers against today's. Returns (verdict, rows). A by-design volatile header is dropped: a Date that moved is not news, and a sweep that reported it would bury the one row that matters. A value the prober redacted cannot be compared and says so. """ rows = [] comparable = 0 for name in sorted(filed): want = filed[name] if twice.is_volatile(name): rows.append((name, "volatile", want, now.get(name))) continue if isinstance(want, str) and REDACTED.search(want): rows.append((name, "redacted", want, now.get(name))) continue if isinstance(want, str) and DESCRIBED.match(want): rows.append((name, "described", want, now.get(name))) continue comparable += 1 if name not in now: rows.append((name, "gone", want, None)) elif str(now[name]).strip() == str(want).strip(): rows.append((name, "same", want, now[name])) else: rows.append((name, "differs", want, now[name])) if comparable == 0: if any(k == "described" for _, k, _, _ in rows): return "DESCRIBED", rows return "NOT-COMPARABLE", rows if any(k in ("gone", "differs") for _, k, _, _ in rows): return "CHANGED", rows return "UNCHANGED", rows def reconstruct(rec): """The fallback when the filed command is not one request. Half the probe records reproduce by fetching a script, checking its SHA-256 and running it. That command cannot be re-issued here (and this file does not run strings out of the repository), but many of those records describe the request they made precisely enough to re-ask it: an unauthenticated MCP `initialize` to the record's own surface. This is a RECONSTRUCTION, not the record's command, and every row built this way is labelled `reconstructed` in the report. That label is the whole of its honesty: a difference found this way could in principle be a difference between my request and theirs, and a reader is entitled to know which kind of comparison they are reading. """ surface = rec.get("surface") or "" body = ((rec.get("request") or {}).get("body") or "") if (rec.get("request") or {}).get("method") != "POST": return None if "initialize" not in body.lower() or not surface.startswith("http"): return None return ("POST", surface, {"content-type": "application/json", "accept": "application/json, text/event-stream"}, json.dumps(twice.INIT)) def check_record(rec): out = {"id": rec["id"], "at": rec.get("at"), "surface": rec.get("surface")} filed = (rec.get("observed") or {}).get("headers") or {} out["filed"] = len(filed) out["how"] = "from-command" parsed, why = parse_curl(((rec.get("reproducibility") or {}).get("command"))) if parsed is None: parsed = reconstruct(rec) if parsed is None: out["verdict"] = "NOT-COMPARABLE" out["reason"] = why out["rows"] = [] return out out["how"] = "reconstructed" out["reason"] = "command not re-issuable (%s); re-asked as an " \ "unauthenticated MCP initialize to the record's own " \ "surface" % why method, url, headers, body = parsed out["method"], out["url"] = method, url status, now, err = ask(method, url, headers, body) if status is None: out["verdict"] = "NO-ANSWER" out["reason"] = err out["rows"] = [] return out out["status"] = status out["filedStatus"] = (rec.get("observed") or {}).get("status") verdict, rows = compare_headers(filed, now) if out["filedStatus"] is not None and status != out["filedStatus"] \ and verdict != "NOT-COMPARABLE": verdict = "CHANGED" rows.append(("(status)", "differs", out["filedStatus"], status)) out["verdict"], out["rows"] = verdict, rows return out # -------------------------------------------------------------------------- def selftest(): ok = fail = 0 def check(name, cond): nonlocal ok, fail if cond: ok += 1 print("ok: %s" % name) else: fail += 1 print("FAIL: %s" % name) # --- the curl parser, which is the safety boundary p, why = parse_curl("curl -sS -i -m 25 -X POST https://a/b -H 'Content-Type: application/json' -d '{\"a\":1}'") check("parses a POST with a header and a body", p is not None) check("reads the method", p[0] == "POST") check("reads the url", p[1] == "https://a/b") check("lowercases the header name", p[2]["content-type"] == "application/json") check("reads the body", p[3] == '{"a":1}') p, why = parse_curl("curl -sS 'https://a/b?q=1'") check("a bare GET defaults to GET", p[0] == "GET") check("a quoted url survives", p[1] == "https://a/b?q=1") p, why = parse_curl("curl -s -A 'public-agents-ci' https://a/b") check("-A becomes a user-agent header", p[2]["user-agent"] == "public-agents-ci") for bad, label in ( ("curl -sSO https://x/y.sh && sh y.sh", "a second command"), ("curl -s https://x | sha256sum -c", "a pipe"), ("curl -s https://x > out.txt", "a redirect"), ("curl -s https://x $(whoami)", "a substitution"), ("sh probe.sh notion https://mcp.notion.com/mcp", "not curl at all"), ("curl -s --config /etc/passwd https://x", "an unknown flag"), ("curl -s https://a https://b", "two urls"), ("curl -s -H 'nocolon' https://a", "a malformed header"), ("curl -s", "no url"), ("", "an empty command"), ): p, why = parse_curl(bad) check("REFUSES %s" % label, p is None and why) check("a command with a newline is refused", parse_curl("curl -s https://a\ncurl -s https://b")[0] is None) # --- the comparison v, rows = compare_headers({"www-authenticate": "Bearer error=\"x\""}, {"www-authenticate": "Bearer error=\"x\""}) check("an identical header is UNCHANGED", v == "UNCHANGED") v, _ = compare_headers({"www-authenticate": "Bearer error=\"x\""}, {"www-authenticate": "Bearer scope=\"read write\""}) check("a header whose value moved is CHANGED", v == "CHANGED") v, _ = compare_headers({"www-authenticate": "Bearer error=\"x\""}, {}) check("a header that vanished is CHANGED", v == "CHANGED") v, _ = compare_headers({"date": "Wed, 17 Sep 2026 18:06:00 GMT"}, {"date": "Wed, 23 Sep 2026 16:30:00 GMT"}) check("a Date that moved is NOT a finding", v == "NOT-COMPARABLE") v, rows = compare_headers({"www-authenticate": "[redacted by the prober: see decoded.challenge-header]"}, {"www-authenticate": "Bearer whatever"}) check("a redacted value is not compared", v == "NOT-COMPARABLE") check("a redacted row says so", any(k == "redacted" for _, k, _, _ in rows)) v, _ = compare_headers({"date": "x", "content-type": "application/json"}, {"date": "y", "content-type": "application/json"}) check("one real header holding beside a moved Date is UNCHANGED", v == "UNCHANGED") v, _ = compare_headers({"content-type": "application/json"}, {"content-type": "application/json; charset=utf-8"}) check("a charset appearing counts as CHANGED, not as noise", v == "CHANGED") v, _ = compare_headers({"content-type": " application/json "}, {"content-type": "application/json"}) check("surrounding whitespace is not a change", v == "UNCHANGED") # --- the status half rec = {"id": "x", "observed": {"status": 401, "headers": {"content-type": "application/json"}}, "reproducibility": {"command": "curl -s https://example.invalid/nope"}} out = check_record(rec) check("a surface that cannot be reached is NO-ANSWER", out["verdict"] == "NO-ANSWER") rec2 = {"id": "y", "observed": {"status": 200, "headers": {}}, "reproducibility": {"command": "curl -sSO https://x/y.sh && sh y.sh"}} check("a record whose reproduction is a script download is NOT-COMPARABLE", check_record(rec2)["verdict"] == "NOT-COMPARABLE") # --- a filed DESCRIPTION is not a filed value v, rows = compare_headers( {"payment-required": "present, base64 of the same x402 document the JSON body carries"}, {"payment-required": "eyJ4NDAyVmVyc2lvbiI6Mn0="}) check("prose filed where a header value belongs is DESCRIBED, not CHANGED", v == "DESCRIBED") check("a DESCRIBED row keeps the prose so a reader can see it", any(k == "described" for _, k, _, _ in rows)) v, _ = compare_headers( {"payment-required": "present, base64 of ...", "content-type": "application/json"}, {"payment-required": "abc", "content-type": "text/html"}) check("one described header does not hide a real change beside it", v == "CHANGED") check("a real www-authenticate challenge is NOT mistaken for prose", not DESCRIBED.match('Bearer realm="OAuth", resource_metadata="https://x", ' 'error="invalid_token", error_description="Missing or ' 'invalid access token"')) check("a real content-type is NOT mistaken for prose", not DESCRIBED.match("application/json; charset=utf-8")) check("a redacted marker still reads as redacted, not described", compare_headers({"www-authenticate": "[redacted by the prober: see decoded.x]"}, {"www-authenticate": "Bearer x"})[0] == "NOT-COMPARABLE") # --- the reconstruction fallback, and its labelling rec3 = {"id": "z", "surface": "https://mcp.example.com/mcp", "request": {"method": "POST", "body": "an MCP initialize request (jsonrpc 2.0)"}, "observed": {"status": 401, "headers": {}}, "reproducibility": {"command": "curl -sSO https://x/y.sh && sh y.sh"}} r3 = reconstruct(rec3) check("an MCP initialize record can be reconstructed", r3 is not None) check("the reconstruction POSTs to the record's own surface", r3[1] == "https://mcp.example.com/mcp" and r3[0] == "POST") check("the reconstruction sends an initialize", "initialize" in r3[3]) check("a GET record is not reconstructed", reconstruct(dict(rec3, request={"method": "GET", "body": "initialize"})) is None) check("a POST that is not an initialize is not reconstructed", reconstruct(dict(rec3, request={"method": "POST", "body": "a tools/call of foo"})) is None) check("a record with no surface is not reconstructed", reconstruct(dict(rec3, surface="")) is None) # --- the controls must be a matched pair, not two of a kind check("the control pair asks for opposite verdicts", len({v for _, v, _, _ in CONTROLS}) == 2) check("the control pair is two different records", len({i for i, _, _, _ in CONTROLS}) == 2) check("every control names the date its expectation was verified", all(re.fullmatch(r"\d{4}-\d\d-\d\d", d) for _, _, d, _ in CONTROLS)) check("every control names what verified it", all(len(w) > 20 for _, _, _, w in CONTROLS)) print("\n%d ok, %d failed" % (ok, fail)) return 1 if fail else 0 def main(): ap = argparse.ArgumentParser() ap.add_argument("--clone", default="/tmp/lintel-queue-clone") ap.add_argument("--id", action="append", default=[]) ap.add_argument("--selftest", action="store_true") ap.add_argument("--json") args = ap.parse_args() if args.selftest: return selftest() head = corpus.require_clone(args.clone, "baseline.py") try: corpus.require_ref(args.clone, "origin/main", "baseline.py") except SystemExit as e: print(str(e)) return 2 records = [] for path in sorted(glob.glob(os.path.join( args.clone, "registry/evidence/probes/*.json"))): try: d = json.load(open(path)) except ValueError: continue if args.id and d.get("id") not in args.id: continue records.append(d) if not records: print("REFUSED: no probe records in the corpus; baseline.py would score nothing.") return 2 started = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") print("baseline.py, registry at %s, started %s" % (head, started)) print("%d probe record(s); each record's own question re-asked once, no " "credentials\n" % len(records)) results = [] with concurrent.futures.ThreadPoolExecutor(max_workers=6) as pool: for out in pool.map(check_record, records): results.append(out) by_id = {r["id"]: r for r in results} # The controls, before anything else prints a verdict. if not args.id: for cid, want, when, how in CONTROLS: got = by_id.get(cid) if got is None: print("REFUSED: the control record %s is not in this corpus, so " "this sweep has nothing proving it can tell a changed wire " "from an unchanged one." % cid) return 2 if got["verdict"] != want: print("REFUSED: the control %s came out %s and must be %s.\n" " The expectation was verified by hand on %s: %s\n" " Either this instrument is wrong, or the wire has " "moved since that date.\n" " Re-verify by hand and update CONTROLS with the " "date you did it. Do not quote any row from this run until " "then." % (cid, got["verdict"], want, when, how)) return 2 print("controls: %s CHANGED and %s UNCHANGED, as verified by hand on " "2026-09-23. This sweep can tell the two apart.\n" % (CONTROLS[0][0].split("-")[1], CONTROLS[1][0].split("-")[1])) order = {"CHANGED": 0, "NO-ANSWER": 1, "DESCRIBED": 2, "NOT-COMPARABLE": 3, "UNCHANGED": 4} counts = {} for r in sorted(results, key=lambda r: (order.get(r["verdict"], 9), r["id"])): counts[r["verdict"]] = counts.get(r["verdict"], 0) + 1 print("%-15s %s (filed %s, %s)" % (r["verdict"], r["id"], r.get("at"), r.get("how", "-"))) if r["verdict"] == "CHANGED": for name, kind, want, got in r["rows"]: if kind in ("differs", "gone"): print(" %s %s" % (kind.upper(), name)) print(" filed: %s" % str(want)[:150]) print(" now : %s" % str(got)[:150]) elif r["verdict"] == "DESCRIBED": for name, kind, want, got in r["rows"]: if kind == "described": print(" %s was filed as a DESCRIPTION, not a " "value: %s" % (name, str(want)[:110])) elif r["verdict"] in ("NOT-COMPARABLE", "NO-ANSWER"): print(" %s" % r.get("reason", "")) if r.get("how") == "reconstructed" and r["verdict"] == "CHANGED": print(" (reconstructed, not the record's own command: " "%s)" % r.get("reason", "")) print("\n%d record(s): %s" % (len(results), ", ".join( "%d %s" % (v, k) for k, v in sorted(counts.items(), key=lambda kv: order.get(kv[0], 9))))) print("CHANGED is not a finding against an entry. It is a reason for that " "entry's maintainer to re-read the surface.") print("DESCRIBED means observed.headers carries prose where the wire's own " "value belongs: nothing there can ever be re-compared.") # Two numbers about the RECORDS rather than the wire, and both stay visible # no matter which verdict a record landed on: a described header inside an # otherwise comparable record would otherwise vanish into UNCHANGED. described = [r for r in results if any(k == "described" for _, k, _, _ in r.get("rows", []))] if described: print("\n%d record(s) file a DESCRIPTION in at least one observed.headers " "value, so that header can never be re-compared by anyone:" % len(described)) for r in described: for name, kind, want, _ in r["rows"]: if kind == "described": print(" %s %s: %s" % (r["id"], name, str(want)[:90])) how = {} for r in results: how[r.get("how", "-")] = how.get(r.get("how", "-"), 0) + 1 print("\nre-asked: %s. A `reconstructed` row is MY request to the record's own " "surface, not the record's command, and a difference it finds could in " "principle be a difference between the two." % ", ".join("%d %s" % (v, k) for k, v in sorted(how.items()))) if args.json: with open(args.json, "w") as fh: json.dump({"head": head, "started": started, "results": results}, fh, indent=1) print("wrote %s" % args.json) return 0 if __name__ == "__main__": raise SystemExit(main())