#!/usr/bin/env python3 """oracle_sentinel.py — watch-tier regression sentinel for verdict oracles. Independently verifies a URL-checking oracle's TLS verdicts against locally computed ground truth, then diffs. Catches both failure directions: FALSE-NEG oracle reports tls.valid=false on a chain that is actually valid FALSE-POS oracle reports tls.valid=true on a chain that is actually broken (the dangerous direction — a scam site could pass) NULL oracle returned no TLS verdict (fetch failed / http:// target) ORACLE-ERR oracle endpoint unreachable or non-JSON LOCAL-ERR sentinel could not reach the target itself (no accusation) Also checks the oracle's reported issuer / days_remaining against local values when both sides agree the chain is valid (VERDICT-DRIFT, informational). Stdlib only, Python 3.8+. No funds, no posting — read-only verification. Usage: python3 oracle_sentinel.py --watch watchlist.txt \ --oracle-base https://getunstuck.space/unstuck/api \ --report sample-report.md --ledger sentinel-ledger.jsonl python3 oracle_sentinel.py --self-test # hermetic, no network python3 oracle_sentinel.py --json # machine-readable stdout Watch-list format: one "URL [expect]" per line; expect in {valid,invalid,null}. '#' comments and blank lines ignored. 'expect' is the local-truth direction the operator believes correct — a mismatch flags a possible truth-source problem, not just an oracle problem. """ import argparse import datetime import json import socket import ssl import sys import urllib.parse import urllib.request UA = {"User-Agent": "oracle-sentinel/1.0 (+ARION; read-only verification)"} TIMEOUT = 20 # ---------- ground truth: local TLS handshake ---------- def local_tls(url): """Return ground-truth TLS verdict for an https:// URL. -> {"valid":bool,"days_remaining":int|None,"issuer":str|None, "error":str|None} """ host = urllib.parse.urlparse(url).hostname scheme = urllib.parse.urlparse(url).scheme if scheme != "https": return {"valid": None, "days_remaining": None, "issuer": None, "error": "no-tls-layer(%s)" % scheme} if not host: return {"valid": None, "days_remaining": None, "issuer": None, "error": "no-host"} try: ctx = ssl.create_default_context() with socket.create_connection((host, 443), timeout=TIMEOUT) as sock: with ctx.wrap_socket(sock, server_hostname=host) as ss: cert = ss.getpeercert() na = cert.get("notAfter") # e.g. 'Jan 2 23:59:59 2027 GMT' days = None if na: exp = datetime.datetime.strptime(na, "%b %d %H:%M:%S %Y %Z") days = (exp - datetime.datetime.utcnow()).days issuer = None for rdn in cert.get("issuer", ()): for k, v in rdn: if k in ("organizationName", "commonName"): issuer = issuer or v return {"valid": True, "days_remaining": days, "issuer": issuer, "error": None} except ssl.SSLCertVerificationError as e: return {"valid": False, "days_remaining": None, "issuer": None, "error": "cert-verify: %s" % e.verify_message[:80]} except (ssl.SSLError, socket.gaierror, socket.timeout, OSError) as e: return {"valid": None, "days_remaining": None, "issuer": None, "error": "local-err: %s" % str(e)[:80]} # ---------- oracle probe ---------- def oracle_check(oracle_base, url): """Query the oracle-check endpoint. -> parsed row dict.""" q = urllib.parse.quote(url, safe="") req = urllib.request.Request( "%s/v1/oracle-check?url=%s" % (oracle_base.rstrip("/"), q), headers=UA) with urllib.request.urlopen(req, timeout=TIMEOUT) as r: return json.load(r) # ---------- diff ---------- def compare(url, expect, oracle, truth): """Return (verdict, detail) given oracle row + local truth row.""" o_tls = (oracle or {}).get("tls") or {} o_valid = o_tls.get("valid") t_valid = truth.get("valid") if oracle is None: return "ORACLE-ERR", "unreachable/non-JSON" if (oracle or {}).get("error"): # An oracle whose fetcher throws on a genuinely invalid chain is # refusing to vouch — correct handling of a negative control. if t_valid is False: return "REFUSES-BAD", "oracle errors on invalid chain: %s" % str(oracle.get("error"))[:70] return "ORACLE-ERR", str(oracle.get("error"))[:80] if t_valid is None: terr = truth.get("error") or "" if terr.startswith("no-tls-layer"): # http:// target — no TLS to judge if o_valid is None: return "MATCH", "both report no TLS layer" return "NULL-DISAGREE", "oracle reports TLS verdict on non-TLS URL" return "LOCAL-ERR", terr or "no verdict" if o_tls == {} or o_valid is None: return "ORACLE-NULL", "oracle produced no TLS verdict (status=%s)" % oracle.get("final_status") if o_valid and t_valid: drift = [] o_days, t_days = o_tls.get("days_remaining"), truth.get("days_remaining") if isinstance(o_days, int) and isinstance(t_days, int) and abs(o_days - t_days) > 2: drift.append("days %s vs %s" % (o_days, t_days)) if drift: return "VERDICT-DRIFT", "; ".join(drift) return "MATCH", "both valid" if not o_valid and t_valid: return "FALSE-NEG", "oracle=invalid truth=valid (issuer=%s)" % truth.get("issuer") if o_valid and not t_valid: return "FALSE-POS", "oracle=valid truth=invalid (%s)" % truth.get("error") return "MATCH", "both invalid (%s)" % (truth.get("error") or "bad chain") def expect_check(expect, truth): """Is local truth consistent with the operator's expectation?""" if expect is None: return None tv = truth.get("valid") if expect == "valid": return tv is True if expect == "invalid": return tv is False if expect == "null": return tv is None return None def parse_watch(path): rows = [] for line in open(path): line = line.strip() if not line or line.startswith("#"): continue parts = line.split() url = parts[0] expect = parts[1] if len(parts) > 1 else None rows.append((url, expect)) return rows def run(watch, oracle_base, ledger_path=None): rows = [] counts = {"MATCH": 0, "FALSE-NEG": 0, "FALSE-POS": 0, "ORACLE-NULL": 0, "ORACLE-ERR": 0, "REFUSES-BAD": 0, "NULL-DISAGREE": 0, "VERDICT-DRIFT": 0, "LOCAL-ERR": 0} for url, expect in watch: try: oracle = oracle_check(oracle_base, url) except Exception as e: oracle = {"error": "fetch: %s" % str(e)[:80]} truth = local_tls(url) verdict, detail = compare(url, expect, oracle, truth) counts[verdict] = counts.get(verdict, 0) + 1 exp_ok = expect_check(expect, truth) o_tls = (oracle or {}).get("tls") or {} rows.append({ "url": url, "expect": expect, "expect_ok": exp_ok, "verdict": verdict, "detail": detail, "oracle": {"valid": o_tls.get("valid"), "days_remaining": o_tls.get("days_remaining"), "issuer": o_tls.get("issuer"), "final_status": (oracle or {}).get("final_status"), "score": (oracle or {}).get("score")}, "truth": truth}) ts = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ") run_row = {"ts": ts, "oracle_base": oracle_base, "checked": len(rows), "counts": counts, "rows": rows} if ledger_path: with open(ledger_path, "a") as f: f.write(json.dumps(run_row) + "\n") return run_row def render_md(run_row): L = ["# Oracle regression sentinel — run %s" % run_row["ts"], "", "Oracle under watch: `%s`" % run_row["oracle_base"], "", "| URL | expect | oracle TLS | truth TLS | verdict | detail |", "|---|---|---|---|---|---|"] for r in run_row["rows"]: o, t = r["oracle"], r["truth"] o_s = ("valid" if o.get("valid") else "invalid" if o.get("valid") is False else "null") t_s = ("valid" if t.get("valid") else "invalid" if t.get("valid") is False else "null") if o.get("valid"): o_s += " (%s, %sd)" % (o.get("issuer"), o.get("days_remaining")) if t.get("valid"): t_s += " (%s, %sd)" % (t.get("issuer"), t.get("days_remaining")) flag = " **%s**" % r["verdict"] if r["verdict"] not in ("MATCH",) else r["verdict"] L.append("| %s | %s | %s | %s | %s | %s |" % ( r["url"], r.get("expect") or "-", o_s, t_s, flag, (r["detail"] or "")[:90])) L += ["", "## Counts"] + ["- %s: %d" % (k, v) for k, v in sorted(run_row["counts"].items()) if v] bad = [k for k in ("FALSE-NEG", "FALSE-POS") if run_row["counts"].get(k)] L += ["", ("REGRESSION PRESENT: %s" % ", ".join(bad)) if bad else "NO REGRESSION: oracle verdicts match ground truth"] return "\n".join(L) + "\n" # ---------- hermetic self-test ---------- def self_test(): cases = [ # (oracle tls dict or None, truth row, expect verdict prefix) ({"valid": True, "days_remaining": 90, "issuer": "X"}, {"valid": True, "days_remaining": 90, "issuer": "X", "error": None}, "MATCH"), ({"valid": False}, {"valid": True, "days_remaining": 90, "issuer": "X", "error": None}, "FALSE-NEG"), ({"valid": True, "days_remaining": 90, "issuer": "X"}, {"valid": False, "days_remaining": None, "issuer": None, "error": "cert-verify: expired"}, "FALSE-POS"), ({"valid": False}, {"valid": False, "days_remaining": None, "issuer": None, "error": "cert-verify: expired"}, "MATCH"), ({"valid": True, "days_remaining": 40, "issuer": "X"}, {"valid": True, "days_remaining": 90, "issuer": "X", "error": None}, "VERDICT-DRIFT"), (None, {"valid": True, "days_remaining": 1, "issuer": "X", "error": None}, "ORACLE-NULL"), ({"valid": True}, {"valid": None, "days_remaining": None, "issuer": None, "error": "no-tls-layer(http)"}, "NULL-DISAGREE"), ({"valid": False}, {"valid": None, "days_remaining": None, "issuer": None, "error": "local-err: timeout"}, "LOCAL-ERR"), ("__error__", {"valid": False, "days_remaining": None, "issuer": None, "error": "cert-verify: expired"}, "REFUSES-BAD"), ("__error__", {"valid": True, "days_remaining": 10, "issuer": "X", "error": None}, "ORACLE-ERR"), ] fails = 0 for i, (o_tls, truth, want) in enumerate(cases): if o_tls == "__error__": oracle = {"error": "self-signed certificate"} else: oracle = {"tls": o_tls, "final_status": 200} if o_tls is not None else \ {"tls": None, "final_status": None} got, _ = compare("u", None, oracle, truth) ok = got == want fails += not ok print("case %d: %s (want %s) %s" % (i, got, want, "OK" if ok else "FAIL")) print("self-test: %s" % ("PASS" if fails == 0 else "%d FAIL" % fails)) sys.exit(0 if fails == 0 else 1) def main(): p = argparse.ArgumentParser() p.add_argument("--watch", default=None) p.add_argument("--oracle-base", default="https://getunstuck.space/unstuck/api") p.add_argument("--report", default=None) p.add_argument("--ledger", default=None) p.add_argument("--json", action="store_true") p.add_argument("--self-test", action="store_true") a = p.parse_args() if a.self_test: self_test() watch = parse_watch(a.watch) if a.watch else [] if not watch: print("no watch entries", file=sys.stderr) sys.exit(2) run_row = run(watch, a.oracle_base, a.ledger) if a.json: print(json.dumps(run_row, indent=1)) else: print(render_md(run_row)) if a.report: open(a.report, "w").write(render_md(run_row)) if __name__ == "__main__": main()