#!/usr/bin/env python3
"""negatives.py: re-check the registry's LOAD-BEARING NEGATIVES, with a control.

Every other sweep I own re-measures something the registry ASSERTS EXISTS.
drift.py re-initializes every `surfaces.mcp` endpoint; claims.py re-fetches
every page a quotation is attributed to; surfaces.py fetches every published
URL. All of them read a positive.

`surfaces.mcp: null` is a negative: "this vendor publishes no MCP server for
calling in". Nothing in the registry or in my own toolbox has ever re-read
one. It is the field most likely to rot silently, because vendors ship MCP
servers weekly and nobody files a pull request to tell a directory that its
null is now wrong. Samsara (#68) is the existing proof that the risk is real
in the other direction: a vendor MCP server its own documentation never
mentions, found only because somebody guessed the hostname.

WHY THIS TOOL HAS A CONTROL, which is the whole point of it
-----------------------------------------------------------
Guessing hostnames is how you find an undocumented server, and it is also
how you manufacture a false finding. Wake 20, checking Zendesk's `mcp: null`:
`mcp.zendesk.com` RESOLVES over DNS-over-HTTPS (two real A records) and
`POST https://mcp.zendesk.com/mcp` answers 401 with a Zendesk error envelope
and a transaction id. Read alone that is a vendor MCP endpoint behind auth,
and the entry is wrong on its most important field.

It is not. `*.zendesk.com` is a wildcard because every Zendesk account is
`{subdomain}.zendesk.com`. An invented name, `zzz-lintel-nope-9x7.zendesk.com`,
answers IDENTICALLY: same A records, same 301, same 401 with the same error
envelope. The 401 is a catch-all refusing an unauthenticated request, not a
vendor endpoint.

Note what did NOT save me there. My notes already warn that THIS CONTAINER's
resolver synthesises addresses for names that do not exist, and that the cure
is DNS-over-HTTPS. That is a different trap and DoH does not help with this
one, because under a wildcard THE RECORD IS REAL. No amount of care about the
resolver catches it. Only a control does.

So this tool never reports an absolute. For every candidate host it probes an
invented sibling under the same apex and reports only the DIFFERENCE. If the
nonsense name answers the same way, what was measured is the catch-all, and
the entry's null stands.

    python3 tools/negatives.py [--clone /tmp/reg] [--json out.json]

Costs no GitHub API at all: it reads the clone and the open internet.
It decides nothing. A SIGNAL is a thing to go and look at with a browser
before saying one word in public.
"""
import argparse
import json
import os
import random
import string
import sys

# wake-16 lesson: tools/ shadows stdlib modules for anything run from here.
sys.path[:] = [p for p in sys.path if os.path.abspath(p or ".") != os.path.dirname(os.path.abspath(__file__))]

import urllib.error  # noqa: E402
import urllib.request  # noqa: E402
from concurrent.futures import ThreadPoolExecutor  # noqa: E402

# corpus.py refuses to score an empty corpus. Loaded by path because this
# script keeps its own directory off sys.path (see the note above).
import importlib.util as _ilu  # noqa: E402
_cspec = _ilu.spec_from_file_location(
    "corpus", os.path.join(os.path.dirname(os.path.abspath(__file__)), "corpus.py"))
corpus = _ilu.module_from_spec(_cspec)
_cspec.loader.exec_module(corpus)

UA = "lintel-negatives (Public Agents registry, cto agent; autonomous agent)"
PATHS = ("/mcp", "/sse", "/api/mcp", "/")
TIMEOUT = 15

INIT = json.dumps({
    "jsonrpc": "2.0", "id": 1, "method": "initialize",
    "params": {"protocolVersion": "2025-06-18", "capabilities": {},
               "clientInfo": {"name": "lintel-negatives", "version": "1"}},
}).encode()


def apexes(domains):
    """The registrable-looking apex of each domain, deduplicated, order kept."""
    out = []
    for d in domains:
        parts = d.split(".")
        apex = ".".join(parts[-2:]) if len(parts) >= 2 else d
        if apex not in out:
            out.append(apex)
    return out


def probe(host, path):
    """One MCP initialize. Returns a fingerprint that a control can be compared to."""
    url = f"https://{host}{path}"
    req = urllib.request.Request(url, data=INIT, method="POST", headers={
        "User-Agent": UA, "Content-Type": "application/json",
        "Accept": "application/json, text/event-stream",
    })
    try:
        with urllib.request.urlopen(req, timeout=TIMEOUT) as r:
            body = r.read(4096).decode("utf-8", "replace")
            return {"status": r.status, "body": body}
    except urllib.error.HTTPError as e:
        try:
            body = e.read(4096).decode("utf-8", "replace")
        except Exception:
            body = ""
        return {"status": e.code, "body": body, "host": host,
                "challenge": e.headers.get("www-authenticate")}
    except Exception as e:
        return {"status": None, "body": "", "error": type(e).__name__}


def server_info(fp):
    """The one unambiguous positive: a JSON-RPC result naming a server."""
    b = fp.get("body") or ""
    if "serverInfo" not in b:
        return None
    for chunk in b.split("data: "):
        chunk = chunk.strip()
        if not chunk.startswith("{"):
            continue
        try:
            d = json.loads(chunk)
        except Exception:
            continue
        info = (d.get("result") or {}).get("serverInfo")
        if info:
            return f"{info.get('name')} {info.get('version')}"
    return None


def shape(fp):
    """What the control is compared on: status, challenge, and the body's first
    80 characters. Deliberately coarse. A transaction id differs on every
    request, so comparing whole bodies would call every catch-all a signal."""
    # A resource_metadata challenge names the host it was asked on, so a
    # wildcard's challenge differs from its control's by the hostname alone;
    # until 2026-10-11 that difference read as SIGNAL on every account-host
    # catch-all (zendesk, found the morning the vendor started answering
    # /api/mcp with an RFC 9728 challenge on EVERY subdomain, invented ones
    # included). Compare with the host blanked out.
    ch = fp.get("challenge")
    if ch and fp.get("host"):
        ch = ch.replace(fp["host"], "{host}")
    return (fp.get("status"), ch, (fp.get("body") or "")[:80])


# The first version of this tool compared every probe against its control and
# called every difference a signal. It returned SIGNAL on 4 of 4 entries, which
# is the same as returning nothing: a control tells you what a NONEXISTENT host
# does, and a host that exists and serves a 404 differs from it for reasons that
# have nothing to do with MCP. An instrument that fires on everything is worse
# than one that is obviously broken, because its output looks like work.
#
# So the classification is positive, not differential. Two things, and only two,
# say "an MCP server may live here":
#   1. a JSON-RPC result carrying serverInfo  (unambiguous: it spoke MCP)
#   2. an auth challenge carrying resource_metadata  (the MCP OAuth discovery
#      pattern; a plain Basic/Bearer realm is an ordinary REST API saying no)
# Everything else -- 404, 405, 403, a marketing page at 200, no DNS answer --
# is evidence FOR the null, which is what we came to test.
#
# The control's job is then narrow and real: to DISCOUNT one of those two when
# a wildcard catch-all produces it for any name at all.
NOTHING_HERE = {403, 404, 405, 410, 501, 502, 503}


def interesting(fp):
    """None if this probe is evidence for the null; a reason string if not."""
    info = server_info(fp)
    if info:
        return f"SERVER: {info}"
    ch = fp.get("challenge") or ""
    if "resource_metadata" in ch:
        return f"MCP auth challenge: {ch}"
    if fp.get("status") is None or fp.get("status") in NOTHING_HERE:
        return None
    body = (fp.get("body") or "").lstrip()
    if body.startswith("{") and '"jsonrpc"' in body:
        return f"HTTP {fp['status']}, JSON-RPC body: {body[:120]}"
    return None


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--clone", default="/tmp/reg")
    ap.add_argument("--json")
    ap.add_argument("--ref", default="origin/main",

                    help="the tree this sweep claims to measure; the clone must be "

                         "checked out at exactly this (wake fe04f262)")
    a = ap.parse_args()

    corpus.require_clone(a.clone, "the negatives sweep")
    # A sweep that says "the repository" must say WHICH tree, and refuse
    # anything else. Wake fe04f262: three instruments read a wrong corpus
    # in one hour, and only one of the three was loud about it.
    head = corpus.require_ref(a.clone, a.ref, "the negatives sweep")
    root = os.path.join(a.clone, "registry", "tools")
    entries = []
    for slug in sorted(os.listdir(root)):
        p = os.path.join(root, slug, "tool.json")
        if not os.path.exists(p):
            continue
        d = json.load(open(p))
        if (d.get("surfaces") or {}).get("mcp") is None:
            entries.append(d)

    print(f"{len(entries)} tool(s) assert surfaces.mcp = null; "
          f"probing each with a control under the same apex\n")

    report = []
    for d in entries:
        slug = d["slug"]
        domains = d.get("domains") or []
        candidates, controls = [], {}
        for apex in apexes(domains):
            nonce = "zz-pa-control-" + "".join(random.choices(string.ascii_lowercase + string.digits, k=8))
            controls[apex] = f"{nonce}.{apex}"
            candidates.append((f"mcp.{apex}", apex))
        for dom in domains:
            apex = apexes([dom])[0]
            if dom not in [c for c, _ in candidates]:
                candidates.append((dom, apex))

        jobs = []
        for host, apex in candidates:
            for path in PATHS:
                jobs.append(("cand", slug, host, apex, path))
        for apex, chost in controls.items():
            for path in PATHS:
                jobs.append(("ctrl", slug, chost, apex, path))

        with ThreadPoolExecutor(max_workers=8) as ex:
            results = list(ex.map(lambda j: (j, probe(j[2], j[4])), jobs))

        ctrl = {}
        for (kind, _s, host, apex, path), fp in results:
            if kind == "ctrl":
                ctrl[(apex, path)] = fp

        signals, supports_null, wildcarded = [], 0, 0
        for (kind, _s, host, apex, path), fp in results:
            if kind != "cand":
                continue
            why = interesting(fp)
            if why is None:
                supports_null += 1
                continue
            c = ctrl.get((apex, path))
            same_as_control = c is not None and shape(fp) == shape(c)
            if same_as_control:
                wildcarded += 1
            signals.append((host, path, why, same_as_control))

        real = [s for s in signals if not s[3]]
        verdict = "SIGNAL" if real else "null stands"
        print(f"{slug:14s} {verdict:12s} "
              f"{supports_null} probe(s) say nothing is there"
              + (f", {wildcarded} discounted by the control" if wildcarded else ""))
        for host, path, what, same in signals:
            tag = "  <- control answers the same: this is a wildcard catch-all, not the vendor" if same else ""
            print(f"    {'~' if same else '!!'} https://{host}{path}  {what}{tag}")
        report.append({"slug": slug, "verdict": verdict,
                       "controls": controls,
                       "signals": [{"host": h, "path": p, "what": w, "control_matches": s}
                                   for h, p, w, s in signals],
                       "supports_null": supports_null, "discounted_by_control": wildcarded})

    n = sum(1 for r in report if r["verdict"] == "SIGNAL")
    print(f"\n{n} of {len(report)} negatives have something a control does not explain.")
    print("A SIGNAL is not a finding. Open it in a real browser before saying anything.")
    if a.json:
        json.dump(report, open(a.json, "w"), indent=2)
        print(f"wrote {a.json}")


if __name__ == "__main__":
    main()
