#!/usr/bin/env python3 """Re-check the NETTEN merchant-control record from the public XRP Ledger. Read-only; no keys, no accounts. python3 recheck.py [--ledger 107713543] [--out result.json] At one pinned validated ledger, on two independent public servers (xrplcluster.com and Ripple's s1): 1. every Payment received by NETTEN's fee account up to that ledger (who paid, how much); 2. for NETTEN's five accounts and every outside account that paid the fee account: master key enabled?, RegularKey, SignerList (signers + quorum), Delegate objects it owns (permissions it has given to another account); 3. whether any of those controls points to a NETTEN account. Both servers must report the same ledger hash and the same results, otherwise it says MISMATCH. Python 3.8+, standard library only. """ import json, sys, time, urllib.request SERVERS = {"xrplcluster": "https://xrplcluster.com", "s1": "https://s1.ripple.com:51234"} NETTEN = {"Platform/Rewards": "rDQzhMQbsNn6yefNMTj34rMHnHokYV6otH", "Fee Collection": "rLHet8tyiokdM65HEcU4fF32pgzPngoJek", "Subscription": "rG4NUbWKhNiuuD7wfdZQ4n29sxmSCH7Sdy", "NETTEN TREASURY": "rasQbdfaTXGy7SUMbDQY7qpPwe8JMdwc5h", "NETTEN": "rnRiHKeqcPtfaZYidxrMFd4Truz7SgYFBh"} FEE = NETTEN["Fee Collection"] LSF_DISABLE_MASTER = 0x00100000 def rpc(url, method, params): body = json.dumps({"method": method, "params": [params]}).encode() for i in range(4): try: req = urllib.request.Request(url, body, {"content-type": "application/json", "user-agent": "kvt-recheck/1"}) r = json.load(urllib.request.urlopen(req, timeout=60))["result"] if r.get("error") in ("slowDown", "tooBusy"): raise RuntimeError(r["error"]) return r except Exception: if i == 3: raise time.sleep(2 * (i + 1)) def payments_to_fee(url, L): out, marker = [], None while True: p = {"account": FEE, "ledger_index_min": -1, "ledger_index_max": L, "limit": 400, "forward": True} if marker: p["marker"] = marker r = rpc(url, "account_tx", p) for t in r["transactions"]: tx = t.get("tx") or t.get("tx_json"); m = t["meta"] if tx["TransactionType"] != "Payment" or tx.get("Destination") != FEE or m["TransactionResult"] != "tesSUCCESS": continue amt = m.get("delivered_amount", tx.get("Amount")) amount = {"currency": "XRP", "value": str(int(amt) / 1e6)} if isinstance(amt, str) else {"currency": amt["currency"], "issuer": amt["issuer"], "value": amt["value"]} out.append({"from": tx["Account"], "ledger": t.get("ledger_index") or tx.get("ledger_index"), "hash": t.get("hash") or tx.get("hash"), "amount": amount}) marker = r.get("marker") if not marker: return sorted(out, key=lambda x: (x["ledger"], x["hash"])) def controls(url, a, L): r = rpc(url, "account_info", {"account": a, "ledger_index": L, "signer_lists": True}) d = r["account_data"] sl = (r.get("signer_lists") or d.get("signer_lists") or []) sl = sl[0] if sl else None dels, marker = [], None while True: p = {"account": a, "ledger_index": L, "type": "delegate", "limit": 400} if marker: p["marker"] = marker o = rpc(url, "account_objects", p) dels += [{"authorize": x.get("Authorize"), "permissions": [y["Permission"]["PermissionValue"] for y in x.get("Permissions", [])]} for x in o.get("account_objects", []) if x.get("LedgerEntryType") == "Delegate" and x.get("Account") == a] marker = o.get("marker") if not marker: break return {"master_enabled": not bool(d.get("Flags", 0) & LSF_DISABLE_MASTER), "regular_key": d.get("RegularKey"), "signer_list": None if not sl else {"quorum": sl["SignerQuorum"], "signers": sorted(e["SignerEntry"]["Account"] for e in sl["SignerEntries"])}, "delegates_given": sorted(dels, key=lambda x: x["authorize"] or "")} def read(url, L): h = rpc(url, "ledger", {"ledger_index": L})["ledger"]["ledger_hash"] pays = payments_to_fee(url, L) outside = sorted({p["from"] for p in pays} - set(NETTEN.values())) accts = {**{v: k for k, v in NETTEN.items()}, **{a: "paid the fee account" for a in outside}} ctl = {a: {"role": r, **controls(url, a, L)} for a, r in accts.items()} return {"ledger_hash": h, "fee_account_payments": pays, "accounts": ctl} def main(): L = int(sys.argv[sys.argv.index("--ledger") + 1]) if "--ledger" in sys.argv else 107713543 res = {n: read(u, L) for n, u in SERVERS.items()} a, b = res["xrplcluster"], res["s1"] agree = a == b netten = set(NETTEN.values()) pointing = [] for acct, c in a["accounts"].items(): tgt = ([c["regular_key"]] if c["regular_key"] else []) + (c["signer_list"]["signers"] if c["signer_list"] else []) + [d["authorize"] for d in c["delegates_given"]] if c["role"] == "paid the fee account" and any(t in netten for t in tgt): pointing.append(acct) summary = {"ledger": L, "servers": list(SERVERS), "servers_agree": agree, "ledger_hash": a["ledger_hash"], "fee_account_payments": len(a["fee_account_payments"]), "outside_payers": sorted(k for k, v in a["accounts"].items() if v["role"] == "paid the fee account"), "outside_payers_with_any_control_set": sorted(k for k, v in a["accounts"].items() if v["role"] == "paid the fee account" and (v["regular_key"] or v["signer_list"] or v["delegates_given"] or not v["master_enabled"])), "netten_accounts_with_any_control_set": sorted(k for k, v in a["accounts"].items() if v["role"] != "paid the fee account" and (v["regular_key"] or v["signer_list"] or v["delegates_given"] or not v["master_enabled"])), "outside_payers_controlled_by_netten": pointing, "result": ("MISMATCH between servers" if not agree else "HOLDS: no outside payer's signing control points to a NETTEN account" if not pointing else "DOES NOT HOLD")} out = {"summary": summary, "detail": a} if "--out" in sys.argv: json.dump(out, open(sys.argv[sys.argv.index("--out") + 1], "w"), indent=1, sort_keys=True) print(json.dumps(summary, indent=1)) return 0 if agree and not pointing else 1 if __name__ == "__main__": sys.exit(main())