"""Shared helpers: polite HTTP, snapshot storage with sha256 manifest, numeric parsing.""" import datetime, gzip, hashlib, http.client, json, os, pathlib, time, urllib.error, urllib.parse, urllib.request ROOT = pathlib.Path(__file__).resolve().parent.parent RAW = ROOT / "raw" / "latest" MANIFEST = ROOT / "raw" / "MANIFEST.jsonl" UA = "near-toll-audit/1.0 (+https://code.swiftbotlab.workers.dev/near-toll-audit)" MAX_BYTES = 40 * 1024 * 1024 RPC = os.environ.get("NEAR_RPC", "https://free.rpc.fastnear.com") RPC_FALLBACK = "https://rpc.mainnet.near.org" ARCHIVAL = os.environ.get("NEAR_ARCHIVAL_RPC", "https://archival-rpc.mainnet.fastnear.com") def now_utc(): return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%MZ") # A redirect may only land on a host we already trust. Anything else could point a fetcher at a local service and get # that service's reply committed to a public repo. ALLOWED_HOSTS = {"api.llama.fi", "stablecoins.llama.fi", "coins.llama.fi", "api.coingecko.com", "api.exchange.coinbase.com", "api.nearblocks.io", "1click.chaindefuser.com", "revenue.near.org", "api.github.com", "nrretf.com", "www.nrretf.com", "free.rpc.fastnear.com", "rpc.mainnet.near.org", "archival-rpc.mainnet.fastnear.com", "explorer.near-intents.org", "oauth2.googleapis.com", "bigquery.googleapis.com"} class StopSource(Exception): """The server said stop (403/429) or the source is unusable. Keep the previous snapshot.""" class SafeRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): target = urllib.parse.urlsplit(newurl) if target.scheme != "https" or target.hostname not in ALLOWED_HOSTS: raise StopSource(f"redirect refused: {target.hostname}") new = super().redirect_request(req, fp, code, msg, headers, newurl) if new is not None and target.hostname != urllib.parse.urlsplit(req.full_url).hostname: new.remove_header("Authorization") return new OPENER = urllib.request.build_opener(SafeRedirect) def http_get(url, headers=None, timeout=40, retries=1): head = {"User-Agent": UA, "Accept": "application/json, text/html;q=0.9"} head.update(headers or {}) for attempt in range(retries + 1): try: req = urllib.request.Request(url, headers=head) with OPENER.open(req, timeout=timeout) as resp: body = resp.read(MAX_BYTES + 1) if len(body) > MAX_BYTES: raise StopSource(f"response over {MAX_BYTES} bytes") return body except urllib.error.HTTPError as exc: if exc.code in (402, 403, 429): raise StopSource(f"HTTP {exc.code}") from None if attempt == retries: raise StopSource(f"HTTP {exc.code}") from None except (urllib.error.URLError, http.client.HTTPException, TimeoutError, OSError) as exc: if attempt == retries: raise StopSource(f"network: {type(exc).__name__}") from None time.sleep(2 + attempt * 3) raise StopSource("unreachable") def http_json(url, **kw): return json.loads(http_get(url, **kw)) def rpc_call(method, params, url=None, timeout=40, retries=2): payload = json.dumps({"jsonrpc": "2.0", "id": "1", "method": method, "params": params}).encode() target = url or RPC for attempt in range(retries + 1): try: req = urllib.request.Request(target, data=payload, headers={"content-type": "application/json", "User-Agent": UA}) with OPENER.open(req, timeout=timeout) as resp: out = json.loads(resp.read(MAX_BYTES)) if "error" in out: if attempt == retries: raise StopSource(f"rpc error: {str(out['error'])[:160]}") else: return out["result"] except urllib.error.HTTPError as exc: if exc.code in (403, 429) and attempt == retries: raise StopSource(f"rpc HTTP {exc.code}") from None except (urllib.error.URLError, http.client.HTTPException, TimeoutError, OSError, ValueError): if attempt == retries: if target == RPC and url is None: return rpc_call(method, params, url=RPC_FALLBACK, timeout=timeout, retries=1) raise StopSource("rpc network failure") from None time.sleep(1.5 + attempt * 2) raise StopSource("rpc retries exhausted") def rpc_view(account, method, args=None, url=None, block_id=None): import base64 params = {"request_type": "call_function", "account_id": account, "method_name": method, "args_base64": base64.b64encode(json.dumps(args or {}).encode()).decode()} if block_id is None: params["finality"] = "final" else: params["block_id"] = block_id res = rpc_call("query", params, url=url) if "result" not in res: raise StopSource(f"view {account}.{method}: {str(res.get('error'))[:120]}") return json.loads(bytes(res["result"]).decode()) def canonical(obj): return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False).encode() def save(name, obj, source, manifest=True): """Store one snapshot as gzip JSON and append date + sha256 to the manifest.""" RAW.mkdir(parents=True, exist_ok=True) raw = canonical(obj) with gzip.GzipFile(RAW / f"{name}.json.gz", "wb", mtime=0) as fh: fh.write(raw) rec = {"name": name, "pulled_at": now_utc(), "source": source, "sha256": hashlib.sha256(raw).hexdigest(), "bytes": len(raw)} if manifest: with MANIFEST.open("a", encoding="utf8") as fh: fh.write(json.dumps(rec) + "\n") return rec def load(name, default=None): path = RAW / f"{name}.json.gz" if not path.exists(): return default with gzip.open(path, "rb") as fh: return json.loads(fh.read()) def sha_of_snapshot(name): with gzip.open(RAW / f"{name}.json.gz", "rb") as fh: return hashlib.sha256(fh.read()).hexdigest()