#!/usr/bin/env python3 """proof_of_done: verify that a claimed-live URL actually responds as claimed. Stdlib only, Python 3.8+. The result is derived only from observed HTTP(S) responses. There is no flag to bypass TLS verification. Exit codes: 0 = LIVE_VERIFIED, 1 = FAIL / CODE_ONLY, 2 = usage error / refused target. Status labels: LIVE_VERIFIED every requested check passed against observed responses FAIL a response was observed but did not meet an expectation CODE_ONLY no valid response observed (DNS/connect/timeout/TLS failure) ERROR usage error or refused target (exit 2) """ import argparse import datetime import hashlib import http.client import ipaddress import json import socket import ssl import sys import urllib.parse MAX_HOPS = 5 MAX_BODY = 5 * 1024 * 1024 class Refused(Exception): pass class NoResponse(Exception): pass def _utc_now(): return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def parse_target(url, allow_local): """Validate scheme/credentials/host; return (parts, ip, port). Raises Refused.""" p = urllib.parse.urlsplit(url) if p.scheme not in ("http", "https"): raise Refused("scheme %r not allowed (http/https only)" % p.scheme) if p.username is not None or p.password is not None: raise Refused("URLs with credentials are refused") if not p.hostname: raise Refused("URL has no host") try: port = p.port or (443 if p.scheme == "https" else 80) except ValueError: raise Refused("invalid port") try: infos = socket.getaddrinfo(p.hostname, port, type=socket.SOCK_STREAM) except socket.gaierror as e: raise NoResponse("DNS resolution failed: %s" % e) if not infos: raise NoResponse("DNS returned no addresses") if not allow_local: for info in infos: ip = ipaddress.ip_address(info[4][0].split("%")[0]) if (ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved or ip.is_multicast or ip.is_unspecified): raise Refused("target resolves to non-public address %s (use --allow-local for tests only)" % ip) return p, infos[0], port def fetch(url, timeout, allow_local, want_body): """One request, no redirect following. Connects to the vetted IP. Returns dict.""" p, info, port = parse_target(url, allow_local) family, _, _, _, addr = info ctx = ssl.create_default_context() if p.scheme == "https" else None try: sock = socket.socket(family, socket.SOCK_STREAM) sock.settimeout(timeout) sock.connect(addr) if ctx is not None: sock = ctx.wrap_socket(sock, server_hostname=p.hostname) host_hdr = p.hostname if ":" not in p.hostname else "[%s]" % p.hostname if p.port: host_hdr += ":%d" % p.port conn = http.client.HTTPConnection(p.hostname, port, timeout=timeout) conn.sock = sock path = urllib.parse.urlunsplit(("", "", p.path or "/", p.query, "")) conn.putrequest("GET", path, skip_host=True, skip_accept_encoding=True) conn.putheader("Host", host_hdr) conn.putheader("User-Agent", "proof-of-done/0.1.0") conn.putheader("Accept-Encoding", "identity") conn.putheader("Connection", "close") conn.endheaders() resp = conn.getresponse() body = b"" truncated = False if want_body: body = resp.read(MAX_BODY + 1) if len(body) > MAX_BODY: truncated = True body = body[:MAX_BODY] out = {"status": resp.status, "location": resp.getheader("Location"), "body": body, "truncated": truncated} conn.close() return out except ssl.SSLError as e: raise NoResponse("TLS error (certificate not trusted or handshake failed): %s" % e) except socket.timeout: raise NoResponse("timeout after %ss" % timeout) except (OSError, http.client.HTTPException) as e: raise NoResponse("connection error: %s" % e) def _norm(u): p = urllib.parse.urlsplit(u) return urllib.parse.urlunsplit((p.scheme, p.netloc.lower(), p.path or "/", p.query, "")) def verify(url, expect_status=200, expect_redirect_to=None, sha256=None, expect_text=None, timeout=10.0, allow_local=False): """Run all checks. Returns result dict (never raises for network problems).""" result = {"tool": "proof_of_done", "version": "0.1.0", "target": url, "checked_at_utc": _utc_now(), "hops": [], "checks": [], "status": None} checks = result["checks"] def add(name, ok, evidence): checks.append({"name": name, "ok": bool(ok), "evidence": evidence}) want_body = sha256 is not None or expect_text is not None current = url final = None try: for hop in range(MAX_HOPS + 1): r = fetch(current, timeout, allow_local, want_body) result["hops"].append({"url": current, "status": r["status"], "location": r["location"]}) follow = (r["status"] in (301, 302, 303, 307, 308) and r["location"] and expect_redirect_to is not None) if not follow: final = (current, r) break current = urllib.parse.urljoin(current, r["location"]) else: add("redirect_hops", False, "more than %d redirects" % MAX_HOPS) except Refused as e: result["status"] = "ERROR" result["error"] = str(e) return result except NoResponse as e: add("response_received", False, str(e)) result["status"] = "CODE_ONLY" return result if final is None: result["status"] = "FAIL" return result final_url, r = final result["final_url"] = final_url add("response_received", True, "HTTP %d from %s" % (r["status"], final_url)) if expect_redirect_to is not None: n = len(result["hops"]) - 1 add("redirect_target", n >= 1 and _norm(final_url) == _norm(expect_redirect_to), "%d redirect hop(s); landed on %s; expected %s" % (n, final_url, expect_redirect_to)) elif r["status"] in (301, 302, 303, 307, 308) and expect_status not in (301, 302, 303, 307, 308): add("unexpected_redirect", False, "target redirects to %s but no --expect-redirect-to was declared" % r["location"]) add("status", r["status"] == expect_status, "observed %d, expected %d" % (r["status"], expect_status)) if sha256 is not None: if r["truncated"]: add("sha256", False, "body larger than %d bytes; not hashed" % MAX_BODY) else: got = hashlib.sha256(r["body"]).hexdigest() add("sha256", got == sha256.lower(), "observed %s over %d bytes, expected %s" % (got, len(r["body"]), sha256.lower())) if expect_text is not None: text = r["body"].decode("utf-8", "replace") add("expect_text", expect_text in text, "text %s in first %d bytes of body" % ("found" if expect_text in text else "NOT found", len(r["body"]))) result["status"] = "LIVE_VERIFIED" if all(c["ok"] for c in checks) else "FAIL" return result def build_parser(): ap = argparse.ArgumentParser(description="Verify a claimed-live URL from observed responses only.") ap.add_argument("--url", required=True) ap.add_argument("--expect-status", type=int, default=200) ap.add_argument("--expect-redirect-to") ap.add_argument("--sha256", help="expected sha256 (hex) of the fetched body") ap.add_argument("--sha256-file", help="read expected sha256 (first token) from this local file") ap.add_argument("--expect-text") ap.add_argument("--timeout", type=float, default=10.0, help="per-request timeout in seconds") ap.add_argument("--allow-local", action="store_true", help="TEST ONLY: allow private/loopback/link-local targets (disables SSRF guard)") ap.add_argument("--json", action="store_true") return ap def main(argv=None): args = build_parser().parse_args(argv) sha = args.sha256 if args.sha256_file: if sha: print("error: use only one of --sha256 / --sha256-file", file=sys.stderr) return 2 try: with open(args.sha256_file) as f: toks = f.read().split() sha = toks[0] if toks else "" except OSError as e: print("error: cannot read --sha256-file: %s" % e, file=sys.stderr) return 2 if sha is not None: try: if len(sha) != 64: raise ValueError int(sha, 16) except ValueError: print("error: sha256 must be 64 hex characters", file=sys.stderr) return 2 if args.timeout <= 0: print("error: --timeout must be > 0", file=sys.stderr) return 2 res = verify(args.url, args.expect_status, args.expect_redirect_to, sha, args.expect_text, args.timeout, args.allow_local) if args.json: print(json.dumps(res, indent=2)) else: print("%s %s (%s)" % (res["status"], res["target"], res["checked_at_utc"])) for h in res["hops"]: print(" hop: %s -> %s %s" % (h["url"], h["status"], h["location"] or "")) for c in res["checks"]: print(" [%s] %s: %s" % ("ok" if c["ok"] else "XX", c["name"], c["evidence"])) if res.get("error"): print(" refused: %s" % res["error"]) return {"LIVE_VERIFIED": 0, "ERROR": 2}.get(res["status"], 1) if __name__ == "__main__": sys.exit(main())