From af4d3113be25eca09ca45eeff7571cd035c073ee Mon Sep 17 00:00:00 2001 From: rplusq Date: Tue, 14 Jul 2026 20:32:35 +0100 Subject: [PATCH 1/4] tools: add safe-tx-verify (independent Safe multisig tx verifier) A uv script (PEP 723 inline deps) to verify a Safe transaction before signing on a Ledger: recomputes the EIP-712 domainHash / messageHash / safeTxHash from the raw parameters (never trusting the backend), decodes the calldata via this repo's compiled ABIs, and labels addresses against DEPLOYMENT_ADDRESSES.md. Supports the unified Safe Transaction Service endpoint, multi-chain prefixes, --onchain nonce/version checks, and --json. Run: uv run tools/safe-tx-verify/verify_safe_tx.py "oeth:0x..." --nonce N EIP-712 hashing validated against real Optimism Safe txs (recomputed safeTxHash matched the backend for every sampled tx, incl. delegatecall multiSend batches). Co-Authored-By: Claude Opus 4.8 (1M context) --- tools/safe-tx-verify/README.md | 66 ++++ tools/safe-tx-verify/verify_safe_tx.py | 413 +++++++++++++++++++++++++ 2 files changed, 479 insertions(+) create mode 100644 tools/safe-tx-verify/README.md create mode 100644 tools/safe-tx-verify/verify_safe_tx.py diff --git a/tools/safe-tx-verify/README.md b/tools/safe-tx-verify/README.md new file mode 100644 index 0000000..d0a2402 --- /dev/null +++ b/tools/safe-tx-verify/README.md @@ -0,0 +1,66 @@ +# safe-tx-verify + +Independently verify a Safe (Gnosis Safe) multisig transaction **before signing it on a Ledger**. + +The script recomputes the Safe EIP-712 hashes — `domainHash`, `messageHash`, `safeTxHash` — **from the raw +transaction parameters**. It never trusts the hash the Safe backend returns: it fetches the queued transaction's +parameters, recomputes the hash locally, and tells you whether they agree. It then decodes the calldata using this +repo's compiled ABIs and labels every address against [`DEPLOYMENT_ADDRESSES.md`](../../DEPLOYMENT_ADDRESSES.md). + +When you sign a Safe transaction, the Ledger shows the pair **(domain hash, message hash)**. The message hash is +the one that encodes the actual transaction, so it is the value to check against the device. + +## Requirements + +- [`uv`](https://docs.astral.sh/uv/) — the only prerequisite. Dependencies (`eth-utils`, `eth-abi`) are declared + inline (PEP 723) and resolved automatically by `uv run`; nothing to install manually. +- Optional: this repo's compiled artifacts (`evm/out`, via `forge build`) for richer calldata decoding, and network + access to `api.safe.global` (the Safe Transaction Service; no API key required for reads). + +## Usage + +```bash +uv run tools/safe-tx-verify/verify_safe_tx.py "" --nonce N \ + [--chain-id N] [--version V] [--onchain] [--rpc URL] [--json] +``` + +- `target` — a Safe app URL (`https://app.safe.global/...?safe=oeth:0x…`) or a `prefix:0xADDRESS` (e.g. `oeth:0x398A…`). +- `--nonce N` — the queued transaction's nonce to verify (required). +- `--chain-id N` — override the chain id (otherwise derived from the prefix). +- `--version V` — Safe contract version (default `1.3.0`; auto-read with `--onchain`). Governs the EIP-712 domain + (chainId is only bound for ≥ 1.3.0) and the `SafeTx` typehash. +- `--onchain` — also read the Safe's live `nonce` / `threshold` / `VERSION()` via RPC and flag a stale nonce. +- `--rpc URL` — RPC endpoint (otherwise a public default for the chain). +- `--json` — machine-readable output. Exit code `0` = backend hash matches, `2` = mismatch. + +### Examples + +```bash +# Verify nonce 48 on the Optimism admin Safe +uv run tools/safe-tx-verify/verify_safe_tx.py "oeth:0x398A2749487B2a91f2f543C01F7afD19AEE4b6b0" --nonce 48 + +# From a Safe app URL, cross-checking the on-chain nonce/version too +uv run tools/safe-tx-verify/verify_safe_tx.py \ + "https://app.safe.global/transactions/queue?safe=oeth:0x398A2749487B2a91f2f543C01F7afD19AEE4b6b0" \ + --nonce 48 --onchain +``` + +## What to check before signing + +1. **`MATCH`** — the parameters the backend gave and the hash it returned are consistent (the script recomputed the + same `safeTxHash` from the params). A `MISMATCH` means **do not sign**. +2. The **message hash** printed here equals what the Ledger displays. +3. The **decoded action** (`to`, `operation`, function + args) is what you intend — `operation: 1` is a + `delegatecall`; batched `multiSend` calls are expanded and each inner call decoded. +4. Every address is the one you expect — known addresses are labeled from `DEPLOYMENT_ADDRESSES.md`. + +## Notes + +- Supported chains (prefix → chainId): `eth`(1), `oeth`(10), `arb1`(42161), `base`(8453), `matic`(137), `gno`(100), + `bnb`(56), `avax`(43114), `linea`(59144), `scr`(534352), `celo`(42220), `blast`(81457), `zksync`(324), `mnt`(5000), + plus testnets `sep`, `basesep`, `oeth-sep`, `arb-sep`. Others: pass `--chain-id` + `--rpc`. +- The Safe Transaction Service moved to the unified `https://api.safe.global/tx-service//api/v1/…` endpoint; + this tool targets it. +- The EIP-712 hashing was validated against real Optimism Safe transactions (recomputed `safeTxHash` matched the + backend for every sampled tx, including delegatecall `multiSend` batches). Calldata decoding is best-effort and + never affects the hash verification. diff --git a/tools/safe-tx-verify/verify_safe_tx.py b/tools/safe-tx-verify/verify_safe_tx.py new file mode 100644 index 0000000..1a7eb91 --- /dev/null +++ b/tools/safe-tx-verify/verify_safe_tx.py @@ -0,0 +1,413 @@ +#!/usr/bin/env -S uv run --script +# /// script +# requires-python = ">=3.10" +# dependencies = [ +# "eth-utils>=4.1", # keccak +# "eth-abi>=5.0", # best-effort calldata decoding +# ] +# /// +""" +Verify a Safe (Gnosis Safe) multisig transaction before signing on a Ledger. + +Independently recomputes the Safe EIP-712 hashes (domain hash, message hash, safeTxHash) from the raw +transaction PARAMETERS -- it never trusts the hash the Safe backend returns -- then decodes the calldata into a +human-readable action using THIS repo's compiled ABIs and cross-references every address against +DEPLOYMENT_ADDRESSES.md. + +What your Ledger shows when signing a Safe tx is the pair (domain hash, message hash); the message hash is the one +that encodes the actual transaction, so it is the headline value here. + +Run it with uv (dependencies are resolved automatically from the inline metadata above): + + uv run tools/safe-tx-verify/verify_safe_tx.py "" [--nonce N] + [--chain-id N] [--version V] [--onchain] [--rpc URL] [--json] + +Examples: + uv run tools/safe-tx-verify/verify_safe_tx.py \ + "https://app.safe.global/transactions/queue?safe=oeth:0x398A2749487B2a91f2f543C01F7afD19AEE4b6b0" --nonce 48 + uv run tools/safe-tx-verify/verify_safe_tx.py "oeth:0x398A2749487B2a91f2f543C01F7afD19AEE4b6b0" --nonce 48 --onchain +""" +from __future__ import annotations + +import argparse +import json +import os +import re +import sys +import urllib.request +import urllib.error +from pathlib import Path + +from eth_utils import keccak, to_checksum_address + +try: + from eth_abi import decode as abi_decode # eth-abi >= 4 +except Exception: # pragma: no cover + abi_decode = None + +# --- EIP-712 type hashes (computed from the canonical strings so they are self-verifying) ------------------------ +DOMAIN_TYPEHASH = keccak(text="EIP712Domain(uint256 chainId,address verifyingContract)") +DOMAIN_TYPEHASH_OLD = keccak(text="EIP712Domain(address verifyingContract)") # Safe < 1.3.0 (no chainId) +SAFE_TX_TYPEHASH = keccak( + text="SafeTx(address to,uint256 value,bytes data,uint8 operation,uint256 safeTxGas,uint256 baseGas," + "uint256 gasPrice,address gasToken,address refundReceiver,uint256 nonce)" +) +SAFE_TX_TYPEHASH_OLD = keccak( # Safe < 1.0.0 used `dataGas` instead of `baseGas` + text="SafeTx(address to,uint256 value,bytes data,uint8 operation,uint256 safeTxGas,uint256 dataGas," + "uint256 gasPrice,address gasToken,address refundReceiver,uint256 nonce)" +) + +MULTISEND_SELECTOR = "0x8d80ff0a" # multiSend(bytes) + +# Safe app chain short-name (used verbatim in the tx-service path) -> (chainId, default public RPC). +# Safe deprecated the per-chain `safe-transaction-.safe.global` hosts; the unified service is +# `https://api.safe.global/tx-service//api/v1/...` (no API key required for reads as of 2026-07). +TX_SERVICE = "https://api.safe.global/tx-service/{short}/api/v1" +CHAINS = { + "eth": (1, "https://eth.llamarpc.com"), + "oeth": (10, "https://mainnet.optimism.io"), + "arb1": (42161, "https://arb1.arbitrum.io/rpc"), + "base": (8453, "https://mainnet.base.org"), + "matic": (137, "https://polygon-rpc.com"), + "gno": (100, "https://rpc.gnosischain.com"), + "bnb": (56, "https://bsc-dataseed.binance.org"), + "avax": (43114, "https://api.avax.network/ext/bc/C/rpc"), + "linea": (59144, "https://rpc.linea.build"), + "scr": (534352, "https://rpc.scroll.io"), + "celo": (42220, "https://forno.celo.org"), + "blast": (81457, "https://rpc.blast.io"), + "zksync": (324, "https://mainnet.era.zksync.io"), + "mnt": (5000, "https://rpc.mantle.xyz"), + "sep": (11155111, "https://ethereum-sepolia-rpc.publicnode.com"), + "basesep": (84532, "https://sepolia.base.org"), + "oeth-sep": (11155420, "https://sepolia.optimism.io"), + "arb-sep": (421614, "https://sepolia-rollup.arbitrum.io/rpc"), +} +CHAINID_TO_SHORT = {cid: short for short, (cid, _rpc) in CHAINS.items()} +CHAINID_TO_RPC = {cid: rpc for _, (cid, rpc) in CHAINS.items()} + +_USE_COLOR = sys.stdout.isatty() + + +def c(text: str, code: str) -> str: + return f"\033[{code}m{text}\033[0m" if _USE_COLOR else text + + +def die(msg: str) -> "None": + print(c(f"Error: {msg}", "31"), file=sys.stderr) + sys.exit(1) + + +# --- hashing helpers ------------------------------------------------------------------------------------------- +def word(value) -> bytes: + """Left-pad an address/bytes32/int to a 32-byte EVM word.""" + if isinstance(value, str): + h = value.lower().removeprefix("0x") + b = bytes.fromhex(h) + if len(b) > 32: + die(f"value does not fit in 32 bytes: {value}") + return b.rjust(32, b"\x00") + if isinstance(value, int): + return value.to_bytes(32, "big") + raise TypeError(type(value)) + + +def cmp_versions(a: str, b: str) -> int: + def parts(v): + return [int(x) for x in re.findall(r"\d+", v)] + pa, pb = parts(a), parts(b) + pa += [0] * (len(pb) - len(pa)) + pb += [0] * (len(pa) - len(pb)) + return (pa > pb) - (pa < pb) + + +def compute_hashes(tx: dict, safe: str, chain_id: int, version: str) -> dict: + """Recompute (domainHash, messageHash, safeTxHash) from raw params. This is the trust anchor.""" + safe = to_checksum_address(safe) + data_bytes = bytes.fromhex(tx["data"].lower().removeprefix("0x")) if tx.get("data") else b"" + + # Domain separator: >= 1.3.0 binds chainId; older versions do not. + if cmp_versions(version, "1.3.0") >= 0: + domain_hash = keccak(word(DOMAIN_TYPEHASH.hex()) + word(chain_id) + word(safe)) + else: + domain_hash = keccak(word(DOMAIN_TYPEHASH_OLD.hex()) + word(safe)) + + tx_typehash = SAFE_TX_TYPEHASH_OLD if cmp_versions(version, "1.0.0") < 0 else SAFE_TX_TYPEHASH + struct_hash = keccak( + word(tx_typehash.hex()) + + word(to_checksum_address(tx["to"])) + + word(int(tx["value"])) + + word(keccak(data_bytes).hex()) + + word(int(tx["operation"])) + + word(int(tx["safeTxGas"])) + + word(int(tx["baseGas"])) + + word(int(tx["gasPrice"])) + + word(to_checksum_address(tx["gasToken"])) + + word(to_checksum_address(tx["refundReceiver"])) + + word(int(tx["nonce"])) + ) + safe_tx_hash = keccak(b"\x19\x01" + domain_hash + struct_hash) + return { + "domainHash": "0x" + domain_hash.hex(), + "messageHash": "0x" + struct_hash.hex(), # the EIP-712 struct hash = the Ledger "message hash" + "safeTxHash": "0x" + safe_tx_hash.hex(), + } + + +# --- IO helpers ------------------------------------------------------------------------------------------------ +def http_get_json(url: str, timeout: int = 20) -> dict: + req = urllib.request.Request(url, headers={"Accept": "application/json", "User-Agent": "verify-safe-tx"}) + try: + with urllib.request.urlopen(req, timeout=timeout) as r: + return json.loads(r.read().decode()) + except urllib.error.HTTPError as e: + die(f"HTTP {e.code} for {url}") + except Exception as e: # noqa: BLE001 + die(f"request failed for {url}: {e}") + + +def rpc_call(rpc: str, to: str, data: str, timeout: int = 20) -> str: + payload = json.dumps( + {"jsonrpc": "2.0", "id": 1, "method": "eth_call", "params": [{"to": to, "data": data}, "latest"]} + ).encode() + req = urllib.request.Request(rpc, data=payload, headers={"Content-Type": "application/json"}) + with urllib.request.urlopen(req, timeout=timeout) as r: + return json.loads(r.read().decode()).get("result", "0x") + + +def parse_target(target: str) -> tuple[str, str]: + """Return (chain_prefix, checksummed_address) from 'oeth:0x..' or a Safe app URL.""" + m = re.search(r"(?:safe=)?([a-z0-9-]+):(0x[0-9a-fA-F]{40})", target) + if not m: + die(f"could not parse a 'chain:0xaddress' from: {target}") + return m.group(1), to_checksum_address(m.group(2)) + + +def repo_root() -> Path: + p = Path(__file__).resolve() + for parent in [p, *p.parents]: + if (parent / "DEPLOYMENT_ADDRESSES.md").exists(): + return parent + if (parent / "evm" / "DEPLOYMENT_ADDRESSES.md").exists(): + return parent + return Path.cwd() + + +def load_address_book(root: Path) -> dict[str, str]: + """address(lower) -> label, parsed from DEPLOYMENT_ADDRESSES.md.""" + book: dict[str, str] = {} + for cand in (root / "DEPLOYMENT_ADDRESSES.md", root / "evm" / "DEPLOYMENT_ADDRESSES.md"): + if not cand.exists(): + continue + section = "" + for line in cand.read_text().splitlines(): + h = re.match(r"##+\s+(.*)", line) + if h: + section = h.group(1).strip() + continue + cells = [x.strip() for x in line.split("|")] + if len(cells) < 3 or cells[1] in ("Contract", "--------", ""): + continue + name = cells[1] + for addr in re.findall(r"0x[0-9a-fA-F]{40}", line): + book.setdefault(addr.lower(), f"{name} [{section}]") + break + return book + + +def load_selectors(root: Path) -> dict[str, tuple[str, list[str], list[str]]]: + """4-byte selector -> (name, arg_types, arg_names) from this repo's compiled ABIs (evm/out).""" + sels: dict[str, tuple[str, list[str], list[str]]] = {} + out = root / "evm" / "out" + if not out.exists(): + out = root / "out" + if not out.exists(): + return sels + for jf in out.rglob("*.json"): + try: + abi = json.loads(jf.read_text()).get("abi", []) + except Exception: # noqa: BLE001 + continue + for item in abi: + if item.get("type") != "function": + continue + types = [_canonical_type(i) for i in item.get("inputs", [])] + names = [i.get("name", "") for i in item.get("inputs", [])] + sig = f"{item['name']}({','.join(types)})" + sel = "0x" + keccak(text=sig)[:4].hex() + sels.setdefault(sel, (item["name"], types, names)) + return sels + + +def _canonical_type(inp: dict) -> str: + t = inp["type"] + if t.startswith("tuple"): + inner = ",".join(_canonical_type(c) for c in inp.get("components", [])) + return f"({inner}){t[len('tuple'):]}" + return t + + +def decode_calldata(data: str, sels: dict, book: dict, depth: int = 0) -> list[str]: + lines: list[str] = [] + ind = " " * depth + data = data.lower() + if not data or data == "0x" or len(data) < 10: + lines.append(f"{ind}(no calldata / plain ETH transfer)") + return lines + selector = data[:10] + if selector == MULTISEND_SELECTOR: + lines.append(f"{ind}multiSend(bytes) -> batched transactions:") + lines += decode_multisend(data, sels, book, depth + 1) + return lines + match = sels.get(selector) + if not match: + lines.append(f"{ind}selector {selector} (unknown -- not in repo ABIs)") + return lines + name, types, names = match + lines.append(f"{ind}{name}({','.join(types)})") + if abi_decode is not None: + try: + values = abi_decode(types, bytes.fromhex(data[10:])) + for t, n, v in zip(types, names, values): + lines.append(f"{ind} {n or '_'}: {_fmt_value(t, v, book)}") + except Exception as e: # noqa: BLE001 + lines.append(f"{ind} (could not decode args: {e})") + return lines + + +def decode_multisend(data: str, sels: dict, book: dict, depth: int) -> list[str]: + lines: list[str] = [] + # multiSend(bytes): after the selector, a standard ABI-encoded single `bytes` arg. + raw = bytes.fromhex(data[10:]) + if len(raw) < 64: + return [f"{' ' * depth}(malformed multiSend)"] + length = int.from_bytes(raw[32:64], "big") + payload = raw[64:64 + length] + i, n = 0, 0 + while i + 85 <= len(payload): + op = payload[i] + to = to_checksum_address(payload[i + 1:i + 21].hex()) + val = int.from_bytes(payload[i + 21:i + 53], "big") + dlen = int.from_bytes(payload[i + 53:i + 85], "big") + inner = payload[i + 85:i + 85 + dlen] + i += 85 + dlen + lines.append( + f"{' ' * depth}[{n}] op={op} to={_fmt_addr(to, book)} value={val}" + ) + lines += decode_calldata("0x" + inner.hex(), sels, book, depth + 1) + n += 1 + return lines + + +def _fmt_addr(addr: str, book: dict) -> str: + label = book.get(addr.lower()) + return f"{addr} ({c(label, '36')})" if label else c(addr, "33") + + +def _fmt_value(t: str, v, book: dict) -> str: + if t == "address": + return _fmt_addr(to_checksum_address(v), book) + if t.endswith("[]") and isinstance(v, (list, tuple)): + return "[" + ", ".join(_fmt_value(t[:-2], x, book) for x in v) + "]" + if isinstance(v, bytes): + return "0x" + v.hex() + return str(v) + + +# --- main ------------------------------------------------------------------------------------------------------ +def main() -> None: + ap = argparse.ArgumentParser(description="Independently verify a Safe multisig tx before signing.") + ap.add_argument("target", help="Safe app URL or 'prefix:0xADDRESS' (e.g. oeth:0x398A...)") + ap.add_argument("--nonce", type=int, help="Safe nonce of the queued tx to verify") + ap.add_argument("--chain-id", type=int, help="override chain id (else derived from the prefix)") + ap.add_argument("--version", default=None, help="Safe contract version (default 1.3.0 or on-chain with --onchain)") + ap.add_argument("--onchain", action="store_true", help="also read nonce/threshold/owners/version via RPC") + ap.add_argument("--rpc", help="RPC URL (else a public default for the chain)") + ap.add_argument("--json", action="store_true", dest="as_json", help="emit machine-readable JSON") + args = ap.parse_args() + + prefix, safe = parse_target(args.target) + if prefix not in CHAINS and args.chain_id is None: + die(f"unknown chain prefix '{prefix}'; pass --chain-id") + chain_id = args.chain_id or CHAINS[prefix][0] + short = prefix if prefix in CHAINS else CHAINID_TO_SHORT.get(chain_id) + rpc = args.rpc or CHAINID_TO_RPC.get(chain_id) + root = repo_root() + book = load_address_book(root) + + version = args.version or "1.3.0" + onchain: dict = {} + if args.onchain: + if not rpc: + die("no RPC for --onchain; pass --rpc") + try: + onchain["nonce"] = int(rpc_call(rpc, safe, "0xaffed0e0") or "0x0", 16) + onchain["threshold"] = int(rpc_call(rpc, safe, "0xe75235b8") or "0x0", 16) + ver_raw = rpc_call(rpc, safe, "0xffa1ad74") # VERSION() + if ver_raw and len(ver_raw) > 130: + strlen = int(ver_raw[66:130], 16) + version = bytes.fromhex(ver_raw[130:130 + strlen * 2]).decode() or version + if args.version: + version = args.version + except Exception as e: # noqa: BLE001 + die(f"--onchain RPC call failed: {e}") + + if args.nonce is None: + die("pass --nonce N (the queued tx nonce to verify)") + + if not short: + die(f"no Safe tx-service short-name known for chain {chain_id}; use a known chain prefix") + api = f"{TX_SERVICE.format(short=short)}/safes/{safe}/multisig-transactions/?nonce={args.nonce}" + results = http_get_json(api).get("results", []) + if not results: + die(f"no queued tx at nonce {args.nonce} for {safe} on chain {chain_id}") + if len(results) > 1: + print(c(f"WARNING: {len(results)} proposals share nonce {args.nonce}; verifying the first.", "33"), + file=sys.stderr) + tx = results[0] + tx.setdefault("baseGas", tx.get("baseGas", 0)) + + computed = compute_hashes(tx, safe, chain_id, version) + backend_hash = (tx.get("safeTxHash") or "").lower() + match = backend_hash == computed["safeTxHash"].lower() + + if args.as_json: + print(json.dumps({ + "safe": safe, "chainId": chain_id, "version": version, "nonce": args.nonce, + "params": {k: tx.get(k) for k in + ("to", "value", "data", "operation", "safeTxGas", "baseGas", "gasPrice", + "gasToken", "refundReceiver", "nonce")}, + "computed": computed, "backendSafeTxHash": backend_hash, + "backendMatches": match, "onchain": onchain, + }, indent=2)) + sys.exit(0 if match else 2) + + print(c("=== Safe transaction verification ===", "1")) + print(f"Safe: {_fmt_addr(safe, book)}") + print(f"Chain: {chain_id} ({short}) Safe version: {version}") + print(f"Nonce: {args.nonce}") + if onchain: + print(f"On-chain: nonce={onchain.get('nonce')} threshold={onchain.get('threshold')}") + if onchain.get("nonce", args.nonce) > args.nonce: + print(c(" ! on-chain nonce is already past this tx's nonce", "31")) + print() + print(c("Ledger will show:", "1")) + print(f" domain hash: {computed['domainHash']}") + print(f" message hash: {c(computed['messageHash'], '1;36')} <- verify THIS on the device") + print(f" safeTxHash: {computed['safeTxHash']}") + print() + print(f"Backend safeTxHash: {backend_hash}") + print(" " + (c("MATCH — backend params are consistent with the hash", "1;32") if match + else c("MISMATCH — do NOT sign; params/hash disagree", "1;31"))) + print() + print(c("Decoded action:", "1")) + print(f" to: {_fmt_addr(to_checksum_address(tx['to']), book)}") + print(f" value: {tx.get('value')} operation: {tx.get('operation')} ({'delegatecall' if int(tx.get('operation', 0)) == 1 else 'call'})") + sels = load_selectors(root) + for line in decode_calldata(tx.get("data") or "0x", sels, book): + print(" " + line) + sys.exit(0 if match else 2) + + +if __name__ == "__main__": + main() From 76ce5bb9d86e82f2516a3d5de2c23e3a51be1e05 Mon Sep 17 00:00:00 2001 From: rplusq Date: Tue, 14 Jul 2026 20:48:22 +0100 Subject: [PATCH 2/4] tools: add safe-tx-verify skill wrapping the committed script MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A thin Claude Code skill (.claude/skills/safe-tx-verify) so "verify this safe tx" runs the committed tools/safe-tx-verify script — no duplicated hashing logic, the script stays the single trust anchor. Also gitignore .claude/ runtime artifacts while tracking shared skills. Co-Authored-By: Claude Opus 4.8 (1M context) --- .claude/skills/safe-tx-verify/SKILL.md | 40 ++++++++++++++++++++++++++ .gitignore | 6 ++++ 2 files changed, 46 insertions(+) create mode 100644 .claude/skills/safe-tx-verify/SKILL.md diff --git a/.claude/skills/safe-tx-verify/SKILL.md b/.claude/skills/safe-tx-verify/SKILL.md new file mode 100644 index 0000000..35e49cd --- /dev/null +++ b/.claude/skills/safe-tx-verify/SKILL.md @@ -0,0 +1,40 @@ +--- +name: safe-tx-verify +description: Independently verify a Safe (Gnosis Safe) multisig transaction before signing on a Ledger — recomputes the EIP-712 domain/message/safeTxHash from raw params, decodes calldata, labels addresses. Use when the user wants to check a queued Safe tx, confirm a hash before signing, or asks "verify this safe tx". +--- + +# safe-tx-verify + +Verify a queued Safe transaction **before it is signed on a Ledger**. This skill is a thin wrapper around the +committed, version-controlled script at `tools/safe-tx-verify/verify_safe_tx.py` — that script is the trust anchor +(it recomputes the Safe EIP-712 hashes from raw parameters and never trusts the backend's hash). Do not reimplement +the hashing here; always shell out to the script so the verification stays reproducible and reviewable. + +## How to run + +Only prerequisite is [`uv`](https://docs.astral.sh/uv/) (deps are declared inline via PEP 723 and resolved by +`uv run`). From the repo root: + +```bash +uv run tools/safe-tx-verify/verify_safe_tx.py "" --nonce N \ + [--chain-id N] [--version V] [--onchain] [--rpc URL] [--json] +``` + +- `` — a Safe app URL (`https://app.safe.global/...?safe=oeth:0x…`) or `prefix:0xADDRESS` (e.g. `oeth:0x398A…`). +- `--nonce N` — the queued transaction's nonce to verify (required). +- `--onchain` — also read the Safe's live nonce/threshold/VERSION() via RPC and flag a stale nonce. +- `--json` — machine-readable; exit `0` = backend hash matches, `2` = mismatch. + +If `uv` is not installed, do NOT install it silently — ask the user (this repo forbids unprompted package installs). + +## What to do with the output + +Report to the user, in this order: +1. **MATCH vs MISMATCH** — a `MISMATCH` means the backend's hash does not equal the hash recomputed from the raw + params. Tell the user **not to sign**. +2. The **message hash** — this is what the Ledger displays; the user should confirm it matches the device. +3. The **decoded action** (`to`, `operation`, function + args). Call out `operation: 1` (delegatecall) and expand + `multiSend` batches. Flag anything that doesn't match the user's stated intent. +4. **Address labels** — resolved from `DEPLOYMENT_ADDRESSES.md`; flag any unknown/unexpected address. + +For full flag/endpoint details see `tools/safe-tx-verify/README.md`. diff --git a/.gitignore b/.gitignore index eb7497c..6010f21 100644 --- a/.gitignore +++ b/.gitignore @@ -27,3 +27,9 @@ airdrop_data.json # broadcasts */broadcast/* evm/.common.env + +# Claude Code: track shared skills only; ignore local runtime artifacts +.claude/* +!.claude/skills/ +.claude/skills/* +!.claude/skills/*/ From 0fabe0e2fa6863dc792474cec7bb1618c6e20724 Mon Sep 17 00:00:00 2001 From: rplusq Date: Tue, 14 Jul 2026 20:55:47 +0100 Subject: [PATCH 3/4] tools(safe-tx-verify): require --index on competing proposals; simplify baseGas default MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address auto-review: (1) when multiple proposals share a nonce, refuse to silently verify results[0] — list every candidate's safeTxHash and exit non-zero until the signer picks with --index (prevents verifying a different tx than is being signed, esp. under --json/redirected stdout). (2) tx.setdefault("baseGas", 0) — the inner .get() was a no-op. Co-Authored-By: Claude Opus 4.8 (1M context) --- .claude/skills/safe-tx-verify/SKILL.md | 4 +++- tools/safe-tx-verify/README.md | 4 +++- tools/safe-tx-verify/verify_safe_tx.py | 19 ++++++++++++++----- 3 files changed, 20 insertions(+), 7 deletions(-) diff --git a/.claude/skills/safe-tx-verify/SKILL.md b/.claude/skills/safe-tx-verify/SKILL.md index 35e49cd..343c5bb 100644 --- a/.claude/skills/safe-tx-verify/SKILL.md +++ b/.claude/skills/safe-tx-verify/SKILL.md @@ -17,11 +17,13 @@ Only prerequisite is [`uv`](https://docs.astral.sh/uv/) (deps are declared inlin ```bash uv run tools/safe-tx-verify/verify_safe_tx.py "" --nonce N \ - [--chain-id N] [--version V] [--onchain] [--rpc URL] [--json] + [--index N] [--chain-id N] [--version V] [--onchain] [--rpc URL] [--json] ``` - `` — a Safe app URL (`https://app.safe.global/...?safe=oeth:0x…`) or `prefix:0xADDRESS` (e.g. `oeth:0x398A…`). - `--nonce N` — the queued transaction's nonce to verify (required). +- `--index N` — if several proposals share the nonce the tool lists them and exits non-zero; re-run with the index + of the one being signed. Never guess on the user's behalf here. - `--onchain` — also read the Safe's live nonce/threshold/VERSION() via RPC and flag a stale nonce. - `--json` — machine-readable; exit `0` = backend hash matches, `2` = mismatch. diff --git a/tools/safe-tx-verify/README.md b/tools/safe-tx-verify/README.md index d0a2402..85bc9d9 100644 --- a/tools/safe-tx-verify/README.md +++ b/tools/safe-tx-verify/README.md @@ -21,11 +21,13 @@ the one that encodes the actual transaction, so it is the value to check against ```bash uv run tools/safe-tx-verify/verify_safe_tx.py "" --nonce N \ - [--chain-id N] [--version V] [--onchain] [--rpc URL] [--json] + [--index N] [--chain-id N] [--version V] [--onchain] [--rpc URL] [--json] ``` - `target` — a Safe app URL (`https://app.safe.global/...?safe=oeth:0x…`) or a `prefix:0xADDRESS` (e.g. `oeth:0x398A…`). - `--nonce N` — the queued transaction's nonce to verify (required). +- `--index N` — when several proposals share the same nonce, the tool refuses to guess: it lists each candidate + (index + `safeTxHash`) and exits non-zero. Re-run with `--index N` to pick the exact one you intend to sign. - `--chain-id N` — override the chain id (otherwise derived from the prefix). - `--version V` — Safe contract version (default `1.3.0`; auto-read with `--onchain`). Governs the EIP-712 domain (chainId is only bound for ≥ 1.3.0) and the `SafeTx` typehash. diff --git a/tools/safe-tx-verify/verify_safe_tx.py b/tools/safe-tx-verify/verify_safe_tx.py index 1a7eb91..40ffbb3 100644 --- a/tools/safe-tx-verify/verify_safe_tx.py +++ b/tools/safe-tx-verify/verify_safe_tx.py @@ -323,6 +323,8 @@ def main() -> None: ap.add_argument("--version", default=None, help="Safe contract version (default 1.3.0 or on-chain with --onchain)") ap.add_argument("--onchain", action="store_true", help="also read nonce/threshold/owners/version via RPC") ap.add_argument("--rpc", help="RPC URL (else a public default for the chain)") + ap.add_argument("--index", type=int, default=None, + help="which proposal to verify when several share the nonce (see the listing on ambiguity)") ap.add_argument("--json", action="store_true", dest="as_json", help="emit machine-readable JSON") args = ap.parse_args() @@ -361,11 +363,18 @@ def main() -> None: results = http_get_json(api).get("results", []) if not results: die(f"no queued tx at nonce {args.nonce} for {safe} on chain {chain_id}") - if len(results) > 1: - print(c(f"WARNING: {len(results)} proposals share nonce {args.nonce}; verifying the first.", "33"), - file=sys.stderr) - tx = results[0] - tx.setdefault("baseGas", tx.get("baseGas", 0)) + if len(results) > 1 and args.index is None: + # Competing proposals at one nonce: never silently pick one — the signer must choose explicitly, + # otherwise --json / redirected stdout could verify a different tx than the one being signed. + lines = [f"{len(results)} proposals share nonce {args.nonce}; re-run with --index N to pick one:"] + for i, r in enumerate(results): + lines.append(f" --index {i} safeTxHash={r.get('safeTxHash')} to={r.get('to')} op={r.get('operation')}") + die("\n".join(lines)) + idx = args.index or 0 + if idx < 0 or idx >= len(results): + die(f"--index {idx} out of range (found {len(results)} proposal(s) at nonce {args.nonce})") + tx = results[idx] + tx.setdefault("baseGas", 0) computed = compute_hashes(tx, safe, chain_id, version) backend_hash = (tx.get("safeTxHash") or "").lower() From a050e96984bc7d6e488bda2a18683c6717200ecf Mon Sep 17 00:00:00 2001 From: rplusq Date: Tue, 14 Jul 2026 21:13:00 +0100 Subject: [PATCH 4/4] tools: restore full safe-tx-verify skill (propose mode, wrapper recursion, stale-nonce), drop thin reimpl Replace the minimal tools/ reimplementation with the complete original skill at .claude/skills/safe-tx-verify/ (committed to the repo, and usable as a Claude Code skill). The original is far more capable: propose mode (verify a not-yet-queued tx you're about to create, next-free-nonce aware), recursion into multiSend batches + Timelock execute/schedule/*Batch + ProxyAdmin upgradeAndCall/upgradeTo, automatic stale-nonce detection via on-chain nonce(), MultiSend singleton (canonical vs eip155) labeling, repo-ABI selector index with 4byte fallback, and version-aware EIP-712 typehashes. Dependency-free (cast for keccak/decode); wrapped with a PEP-723 header so it runs via 'uv run' (or plain python3). Validated end-to-end: OP Safe nonce 48 -> message hash matches, safeTxHash matches the Safe API cross-check, forceWithdrawAll decoded + address labeled from DEPLOYMENT_ADDRESSES.md. Co-Authored-By: Claude Opus 4.8 (1M context) --- .claude/skills/safe-tx-verify/SKILL.md | 170 ++++- .../safe-tx-verify/scripts/verify_safe_tx.py | 670 ++++++++++++++++++ .gitignore | 2 - tools/safe-tx-verify/README.md | 68 -- tools/safe-tx-verify/verify_safe_tx.py | 422 ----------- 5 files changed, 815 insertions(+), 517 deletions(-) create mode 100644 .claude/skills/safe-tx-verify/scripts/verify_safe_tx.py delete mode 100644 tools/safe-tx-verify/README.md delete mode 100644 tools/safe-tx-verify/verify_safe_tx.py diff --git a/.claude/skills/safe-tx-verify/SKILL.md b/.claude/skills/safe-tx-verify/SKILL.md index 343c5bb..9ee3dfe 100644 --- a/.claude/skills/safe-tx-verify/SKILL.md +++ b/.claude/skills/safe-tx-verify/SKILL.md @@ -1,42 +1,162 @@ --- name: safe-tx-verify -description: Independently verify a Safe (Gnosis Safe) multisig transaction before signing on a Ledger — recomputes the EIP-712 domain/message/safeTxHash from raw params, decodes calldata, labels addresses. Use when the user wants to check a queued Safe tx, confirm a hash before signing, or asks "verify this safe tx". +description: Verify a Safe (Gnosis Safe) multisig transaction before signing on a Ledger. Independently recomputes the Safe EIP-712 hashes (message hash / domain hash / safeTxHash) from raw params WITHOUT trusting the Safe backend, then decodes the calldata into a human-readable action using THIS repo's compiled ABIs and cross-references every address against DEPLOYMENT_ADDRESSES.md. Use when someone pastes a Safe URL (app.safe.global) or a prefixed address like "oeth:0x…", asks "what will my Ledger show / what does this tx do / is this safe to sign", or wants to check a queued Safe transaction. --- # safe-tx-verify -Verify a queued Safe transaction **before it is signed on a Ledger**. This skill is a thin wrapper around the -committed, version-controlled script at `tools/safe-tx-verify/verify_safe_tx.py` — that script is the trust anchor -(it recomputes the Safe EIP-712 hashes from raw parameters and never trusts the backend's hash). Do not reimplement -the hashing here; always shell out to the script so the verification stays reproducible and reviewable. +Answers one question safely: **"what am I actually about to sign on my Ledger, and is it what I think it is?"** — for a Safe multisig transaction in the Reown contracts. + +It does three independent things: + +1. **Recomputes the EIP-712 hashes locally.** It pulls only the raw transaction + *parameters* (to, value, data, operation, nonce, gas fields) and the Safe + contract version from the Safe Transaction Service, then computes the + **message hash**, **domain hash**, and **safeTxHash** itself with `cast keccak`. + The backend's own `safeTxHash` is never trusted — it's only shown at the end + as an independent cross-check. +2. **Decodes the calldata** into a human-readable action using THIS repo's + compiled ABIs (`evm/out/**/*.json`, matched by 4-byte selector — no + collisions, unlike the public 4byte DB). It recurses into `multiSend` + batches, Timelock `execute`/`executeBatch`/`schedule`/`scheduleBatch`, and + ProxyAdmin `upgradeAndCall`, so nested actions (e.g. a proxy upgrade hidden + inside a timelock batch) are surfaced. +3. **Cross-references every address** against `DEPLOYMENT_ADDRESSES.md` + (chain-aware), labeling `to` and address arguments with the contract name + (e.g. `StakeWeight [Optimism]`, `NTT Manager`, `ProxyAdmin`). + +## When to use + +- A Safe URL is pasted: `https://app.safe.global/transactions/queue?safe=oeth:0x…` + or `…/transactions/tx?id=multisig_0x…_0x…&safe=eth:0x…` +- A prefixed address: `oeth:0x398A2749487B2a91f2f543C01F7afD19AEE4b6b0` +- "What will my Ledger show?", "what does this queued tx do?", "is this safe to sign?" ## How to run -Only prerequisite is [`uv`](https://docs.astral.sh/uv/) (deps are declared inline via PEP 723 and resolved by -`uv run`). From the repo root: +Runs via [`uv`](https://docs.astral.sh/uv/) (the script carries a PEP-723 header with +no Python dependencies, so `uv run` needs no install step); plain `python3` works too. +Requires `cast` (foundry) on PATH — already used elsewhere in this repo. No build step. + +```bash +uv run .claude/skills/safe-tx-verify/scripts/verify_safe_tx.py "" [--nonce N] +``` + +Flags: +- `--nonce N` — pick a specific pending transaction (required when several nonces are queued) +- `--onchain` — run `cast` sanity checks (code exists at `to`, resolve ERC-1967 proxy impl) +- `--rpc URL` — RPC for `--onchain` (defaults to a public RPC per network) +- `--chain-id N` — override chain id if the network short-name isn't recognized +- `--version V` — override the Safe contract version (default: fetched; else 1.3.0) +- `--json` — machine-readable output (hashes, params, flattened decode) + +If no `--nonce` is given and multiple nonces are queued, it lists them (exit 2) +so the user can pick one. + +### Propose mode — verify a tx you're about to *create* (not yet queued) + +When YOU are the proposer (the tx needs your signature and isn't in the Safe +backend yet), pass the raw params instead of relying on a queued tx. The script +computes the same Ledger hashes (message + domain), decodes the calldata, and +prints the **safeTxHash the Safe UI should display once you create the tx** — so +you can cross-check in the other direction (UI/simulation → local computation). ```bash -uv run tools/safe-tx-verify/verify_safe_tx.py "" --nonce N \ - [--index N] [--chain-id N] [--version V] [--onchain] [--rpc URL] [--json] +uv run .claude/skills/safe-tx-verify/scripts/verify_safe_tx.py "" \ + --to 0x --data 0x [--value W] [--operation 0|1] [--nonce N] ``` -- `` — a Safe app URL (`https://app.safe.global/...?safe=oeth:0x…`) or `prefix:0xADDRESS` (e.g. `oeth:0x398A…`). -- `--nonce N` — the queued transaction's nonce to verify (required). -- `--index N` — if several proposals share the nonce the tool lists them and exits non-zero; re-run with the index - of the one being signed. Never guess on the user's behalf here. -- `--onchain` — also read the Safe's live nonce/threshold/VERSION() via RPC and flag a stale nonce. -- `--json` — machine-readable; exit `0` = backend hash matches, `2` = mismatch. +Propose-mode flags: +- `--to`, `--data` — target + calldata of the tx you will create (presence of `--to` triggers propose mode) +- `--value` (default 0), `--operation` (0=CALL, 1=DELEGATECALL; default 0) +- `--safe-tx-gas`/`--base-gas`/`--gas-price`/`--gas-token`/`--refund-receiver` — default 0 / zero-address (matches how the Safe UI builds a standard tx) +- `--nonce` — if omitted, defaults to the Safe's next FREE nonce (on-chain `nonce()` reconciled against already-queued txs, so a tx queued behind others gets the right slot) + +Since there's no queued tx to cross-check against, propose mode prints the +computed safeTxHash and tells the user to confirm the Safe UI shows the same +value (and the Ledger the same message/domain hash) before signing. +Build calldata with `cast calldata "fn(types)" args`. + +#### Verifying a BATCH (Tx Builder "All actions") in propose mode + +A Safe Tx-Builder batch is a single **`DELEGATECALL` (operation 1)** into a **MultiSend** +singleton whose `data` is `multiSend(bytes)` (selector `0x8d80ff0a`) packing each inner +call as `op(1B)+to(20B)+value(32B)+len(32B)+data`. Two gotchas that will make your +computed hash disagree with the UI: + +1. **Which MultiSend singleton.** Safe v1.3.0 shipped a *canonical* MultiSendCallOnly + (`0x40A2aCCbd92BCA938b02010E17A5b8929b49130D`) AND an *eip155* one + (`0xA1dabEF33b3B82c7814B6D82A79e50F4AC44102B`). The UI picks one per network and it + becomes the tx `to`, so guessing wrong changes the message hash / safeTxHash (the + *domain* hash still matches — it's only Safe+chain). **Reown's 1.3.0+L2 Safes on + Optimism use the eip155 `0xA1dabEF3…44102B`** (validated 2026-07-10). Pass that as `--to`. +2. **Operation must be `1`.** The Ledger will show `DELEGATECALL ⚠` — expected for a batch. + +Build + verify a 2-action batch: +```bash +# pack the inner calls (op=00 CALL, to=target, value=0, len, data) then wrap in multiSend(bytes) +INNER=00 00 +MS=$(cast calldata "multiSend(bytes)" 0x$INNER) +uv run .../verify_safe_tx.py "oeth:0x" --to 0xA1dabEF33b3B82c7814B6D82A79e50F4AC44102B --data "$MS" --operation 1 --nonce +``` +The script decodes the multiSend into its inner calls (labeled via repo ABIs), so you can +confirm each action + target before signing. Known MultiSend/MultiSendCallOnly addresses +(both v1.3.0 variants + v1.4.1) are auto-labeled instead of showing "unknown contract". + +Validated: propose `forceWithdrawAll(0x5cD9…f43a)` on StakeWeight [Optimism] at +nonce 49 → message hash `0x5bf7db87f5c2c18c9990727da4fa35c60cd5c545e332ad6e73af3b7c55ba607c`, +safeTxHash `0x5b55971dcb0216a83d8bb724b3002eedd3e678ed3f3f39cb3df7256c2576863e` +(matched the ExecutionSuccess txHash in the Safe simulation). + +**Stale nonces are detected automatically.** Safe nonces execute strictly in +order, so any queued tx whose nonce is below the Safe's current on-chain nonce +can never execute (it was superseded/replaced). The script reads the on-chain +`nonce()` (best-effort, via the same per-network public RPC as `--onchain`; skipped +if unreachable) and: +- excludes stale nonces from the "live nonces" it asks you to pick from, and + labels them `[STALE — can never execute, superseded]` in the exit-2 listing + (the next executable one is labeled `[next to execute]`); +- auto-selects when exactly one *live* transaction remains, even if dead entries + are still in the queue; +- prints a loud `⚠` warning (but still verifies) if you explicitly pass a stale + `--nonce N`, and fails cleanly if *every* queued tx is stale. + +**On exit 2 (multiple live pending nonces), do NOT auto-verify all of them.** A bare +queue URL does not identify a transaction — relay the one-line summary the script +printed (nonce → `to` → method, with stale ones already flagged) and ask the user +which live nonce they want fully verified, then re-run with `--nonce N` for that +one. Only verify every nonce if the user explicitly asks for all of them. -If `uv` is not installed, do NOT install it silently — ask the user (this repo forbids unprompted package installs). +## Reading the result to the user -## What to do with the output +- The headline **Message hash** (and the **Domain hash**) are the two values a + Ledger displays when signing a Safe EIP-712 tx. Tell the user to compare them + character-for-character on the device. The **safeTxHash** is what Safe/ + Etherscan label the tx with but is NOT shown on the Ledger. +- Walk them through the decoded action(s) in plain language, using the contract + labels. Call out anything high-stakes: `DELEGATECALL ⚠`, owner changes + (`swapOwner`/`addOwner`/`changeThreshold`), proxy upgrades (`upgradeAndCall`/ + `upgradeTo` → new implementation address), `setMinter`, role grants. +- `[repo ABI: …]` = decoded from this repo's verified ABI (trustworthy). + `[public 4byte DB — UNVERIFIED]` = signature came from the network 4byte DB and + could be a selector collision; treat the decode as a hint, not proof. +- `✓ safeTxHash matches Safe API` = independent cross-check passed. A + `✗ MISMATCH` (exit code 3) means **do not sign** — investigate. -Report to the user, in this order: -1. **MATCH vs MISMATCH** — a `MISMATCH` means the backend's hash does not equal the hash recomputed from the raw - params. Tell the user **not to sign**. -2. The **message hash** — this is what the Ledger displays; the user should confirm it matches the device. -3. The **decoded action** (`to`, `operation`, function + args). Call out `operation: 1` (delegatecall) and expand - `multiSend` batches. Flag anything that doesn't match the user's stated intent. -4. **Address labels** — resolved from `DEPLOYMENT_ADDRESSES.md`; flag any unknown/unexpected address. +## Notes / limits -For full flag/endpoint details see `tools/safe-tx-verify/README.md`. +- Address labels come from `DEPLOYMENT_ADDRESSES.md`; regenerate it + (`pnpm run sync:deployments`) if a new deployment isn't recognized. +- Selector index is built from `evm/out` — run `forge build` if artifacts are stale. +- Safe singleton functions (e.g. `swapOwner`) and OZ `upgradeAndCall` resolve via + the public 4byte DB (flagged UNVERIFIED) because those ABIs aren't compiled into + `evm/out`; the decode is still correct for these well-known selectors. +- Validated: Optimism Safe `0x398A…b6b0` nonce 48 → message hash + `0x0ca4d2606a60453fc21696b7cf2965534de48e86a043bdefbf2d4d546d965059`, + safeTxHash `0xf2400894943443afd32961ffbd90a2c8a9bc07e9d7cd9be7dce7f58c37dec34a`. +- Validated (BATCH, propose mode): a 2-action `forceWithdrawAll` batch on StakeWeight [Optimism] + via Safe `0x398A…b6b0`, DELEGATECALL into **eip155** MultiSendCallOnly `0xA1dabEF3…44102B`, + reproduced the Safe UI's domain hash, message hash, and safeTxHash exactly. Key lesson: the + first attempt used the *canonical* MultiSend `0x40A2…130D` and produced a different (wrong) + message hash — for these Safes the **eip155** variant is the correct `to`. (Specific targets + omitted — token-repurchase actions are confidential per the WCT Repurchase Policy.) diff --git a/.claude/skills/safe-tx-verify/scripts/verify_safe_tx.py b/.claude/skills/safe-tx-verify/scripts/verify_safe_tx.py new file mode 100644 index 0000000..45392ac --- /dev/null +++ b/.claude/skills/safe-tx-verify/scripts/verify_safe_tx.py @@ -0,0 +1,670 @@ +#!/usr/bin/env -S uv run --script +# /// script +# requires-python = ">=3.9" +# dependencies = [] +# /// +"""Verify a Safe multisig transaction before signing on a Ledger. + +Independently recomputes the Safe EIP-712 hashes (domain hash, message hash, +safeTxHash) from the raw transaction PARAMETERS — it never trusts the hash the +Safe backend returns — then decodes the calldata into a human-readable action +using THIS repo's compiled ABIs and cross-references every address against +DEPLOYMENT_ADDRESSES.md. + +What your Ledger shows when signing a Safe tx is the pair (domain hash, message +hash); the message hash is the one that encodes the actual transaction, so it is +the headline value here. + +Two modes: + 1. QUEUE (default) — verify a tx already queued in the Safe Transaction Service. + 2. PROPOSE (--to ...) — verify a tx you are ABOUT TO CREATE that isn't queued + yet. You supply the raw params; the script computes the same Ledger hashes + locally, decodes the calldata, and prints the safeTxHash the Safe UI should + show once you create the tx. Nonce defaults to the Safe's next on-chain + nonce (override with --nonce). + +Usage (via `uv run`; plain `python3 verify_safe_tx.py …` also works): + uv run verify_safe_tx.py "" [--nonce N] + [--chain-id N] [--version V] [--onchain] [--rpc URL] + [--json] + uv run verify_safe_tx.py "" --to 0x... --data 0x... + [--value W] [--operation 0|1] [--nonce N] # propose mode + +Examples: + uv run verify_safe_tx.py "https://app.safe.global/transactions/queue?safe=oeth:0x398A2749487B2a91f2f543C01F7afD19AEE4b6b0" --nonce 48 + uv run verify_safe_tx.py "oeth:0x398A2749487B2a91f2f543C01F7afD19AEE4b6b0" --nonce 48 --onchain + # propose a forceWithdrawAll before it is queued: + uv run verify_safe_tx.py "oeth:0x398A2749487B2a91f2f543C01F7afD19AEE4b6b0" \ + --to 0x521B4C065Bbdbe3E20B3727340730936912DfA46 \ + --data 0x0f0824be0000000000000000000000005cd9e6560a4a86bba1d463e8729e4f1a2651f43a --nonce 49 + +No Python dependencies (PEP 723 header declares none); needs only `uv` (or python3) +and `cast` (foundry) on PATH. +""" +import argparse +import json +import os +import re +import subprocess +import sys +import urllib.request + +# --- EIP-712 Safe type hashes ----------------------------------------------- +DOMAIN_TYPEHASH = "0x47e79534a245952e8b16893a336b85a3d9ea9fa8c573f3d803afb92a79469218" # > 1.2.0 (chainId) +DOMAIN_TYPEHASH_OLD = "0x035aff83d86937d35b32e04f0ddc6ff469290eef2f1b692d8a815c89404d4749" # <= 1.2.0 (no chainId) +SAFE_TX_TYPEHASH = "0xbb8310d486368db6bd6f849402fdd73ad53d316b5a4b2644ad6efe0f941286d8" # >= 1.0.0 +SAFE_TX_TYPEHASH_OLD = "0x14d461bc7412367e924637b363c7bf29b8f47e2f84869f4426e5633d8af47b20" # < 1.0.0 + +MULTISEND_SELECTOR = "0x8d80ff0a" # multiSend(bytes) +ZERO = "0x0000000000000000000000000000000000000000" + +# Well-known Safe infrastructure singletons (chain-agnostic, not in DEPLOYMENT_ADDRESSES.md). +# NOTE: v1.3.0 shipped TWO deployments of each — the "canonical" (CREATE2, same addr everywhere) +# and the "eip155" (chain-specific replay-protected) variant. The Safe UI on a given network may +# pick EITHER, and the choice changes `to` → changes the message hash / safeTxHash. Reown's 1.3.0+L2 +# Safes on Optimism build Tx-Builder batches through the **eip155 MultiSendCallOnly** +# 0xA1dabEF33b3B82c7814B6D82A79e50F4AC44102B (verified 2026-07-10), NOT the canonical 0x40A2…130D. +# So in propose mode for a batch, default the MultiSend `--to` to the eip155 address for this org. +KNOWN_SINGLETONS = { + "0xa1dabef33b3b82c7814b6d82a79e50f4ac44102b": "MultiSendCallOnly v1.3.0 (eip155) [Safe]", + "0x40a2accbd92bca938b02010e17a5b8929b49130d": "MultiSendCallOnly v1.3.0 (canonical) [Safe]", + "0x998739bfdaadde7c933b942a68053933098f9eda": "MultiSend v1.3.0 (eip155) [Safe]", + "0xa238cbeb142c10ef7ad8442c6d1f9e89e07e7761": "MultiSend v1.3.0 (canonical) [Safe]", + "0x9641d764fc13c8b624c04430c7356c1c7c8102e2": "MultiSendCallOnly v1.4.1 [Safe]", + "0x38869bf66a61cf6bdb996a6ae40d5853fd43b526": "MultiSend v1.4.1 [Safe]", +} +# Default MultiSend target for this org's batch propose-mode (see note above). +DEFAULT_MULTISEND = "0xA1dabEF33b3B82c7814B6D82A79e50F4AC44102B" + +# Safe "short name" -> chainId (for the EIP-712 domain). +PREFIX_TO_CHAINID = { + "eth": "1", "oeth": "10", "matic": "137", "pol": "137", "bnb": "56", + "arb1": "42161", "avax": "43114", "gno": "100", "base": "8453", + "zksync": "324", "sep": "11155111", "gor": "5", "base-sep": "84532", + "arb-sep": "421614", "oeth-sep": "11155420", "linea": "59144", + "scr": "534352", "mantle": "5000", "celo": "42220", "blast": "81457", +} +# Default public RPCs for --onchain checks (override with --rpc). +PREFIX_TO_RPC = { + "eth": "https://eth.llamarpc.com", "oeth": "https://mainnet.optimism.io", + "base": "https://mainnet.base.org", "arb1": "https://arb1.arbitrum.io/rpc", + "matic": "https://polygon-rpc.com", "pol": "https://polygon-rpc.com", + "gno": "https://rpc.gnosischain.com", "bnb": "https://bsc-dataseed.binance.org", +} +# ERC-1967 implementation slot. +IMPL_SLOT = "0x360894a13ba1a3210667c828492db98dca3e2076cc3735a920a3ca505d382bbc" + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..", "..")) + +C = {"b": "\033[1m", "d": "\033[2m", "g": "\033[32m", "r": "\033[31m", + "c": "\033[36m", "y": "\033[33m", "x": "\033[0m"} + + +def color(s, k): + return f"{C[k]}{s}{C['x']}" if sys.stdout.isatty() else s + + +def fail(msg): + print(f"{color('Error:', 'r')} {msg}", file=sys.stderr) + sys.exit(1) + + +def cast(*args, timeout=60): + r = subprocess.run(["cast", *args], capture_output=True, text=True, timeout=timeout) + if r.returncode != 0: + raise RuntimeError(f"cast {' '.join(args[:2])}: {r.stderr.strip()[:200]}") + return r.stdout.strip() + + +# --- hex / word helpers ------------------------------------------------------ +def word(x): + """Left-pad an address/bytes32 hex string or an integer to a 32-byte hex word (no 0x).""" + if isinstance(x, str) and x.startswith("0x"): + h = x[2:] + else: + h = format(int(x), "x") + if len(h) > 64: + fail(f"value does not fit in 32 bytes: {x}") + return h.rjust(64, "0") + + +def concat(*words): + return "0x" + "".join(w[2:] if w.startswith("0x") else w for w in words) + + +def keccak(hexdata): + # cast keccak needs even-length hex; empty data hashes the empty byte string. + h = hexdata if hexdata.startswith("0x") else "0x" + hexdata + if len(h) % 2 != 0: + fail(f"odd-length hex: {h}") + return cast("keccak", h) + + +def cmp_versions(a, b): + pa = [int(p) if p.isdigit() else 0 for p in a.split("+")[0].split(".")] + pb = [int(p) if p.isdigit() else 0 for p in b.split("+")[0].split(".")] + for i in range(max(len(pa), len(pb))): + av, bv = (pa[i] if i < len(pa) else 0), (pb[i] if i < len(pb) else 0) + if av != bv: + return av - bv + return 0 + + +# --- core hash computation --------------------------------------------------- +def compute_hashes(p): + version = (p.get("version") or "1.3.0").strip() + if cmp_versions(version, "1.2.0") <= 0: + domain_enc = concat(DOMAIN_TYPEHASH_OLD, word(p["safe"])) + else: + domain_enc = concat(DOMAIN_TYPEHASH, word(p["chainId"]), word(p["safe"])) + domain_hash = keccak(domain_enc) + + tx_typehash = SAFE_TX_TYPEHASH_OLD if cmp_versions(version, "1.0.0") < 0 else SAFE_TX_TYPEHASH + data_hash = keccak(p.get("data") or "0x") + message = concat( + tx_typehash, word(p["to"]), word(p["value"]), data_hash, word(p["operation"]), + word(p["safeTxGas"]), word(p["baseGas"]), word(p["gasPrice"]), + word(p["gasToken"]), word(p["refundReceiver"]), word(p["nonce"]), + ) + message_hash = keccak(message) + safe_tx_hash = keccak(concat("0x1901", domain_hash, message_hash)) + return {"domainHash": domain_hash, "messageHash": message_hash, "safeTxHash": safe_tx_hash} + + +# --- input parsing ----------------------------------------------------------- +def parse_safe_input(s): + s = s.strip() + m = re.search(r"0x[a-fA-F0-9]{40}", s) + if not m: + return None + address = m.group(0) + prefix = "eth" + sp = re.search(r"[?&]safe=([a-z0-9-]+):0x[a-fA-F0-9]{40}", s, re.I) + bare = re.match(r"^([a-z0-9-]+):0x[a-fA-F0-9]{40}", s, re.I) + if sp: + prefix = sp.group(1).lower() + elif bare: + prefix = bare.group(1).lower() + txm = re.search(r"multisig_0x[a-fA-F0-9]{40}_(0x[a-fA-F0-9]{64})", s, re.I) + return {"address": address, "prefix": prefix, "txHash": txm.group(1) if txm else None} + + +# --- Safe Transaction Service (params only) ---------------------------------- +def fetch_json(url): + req = urllib.request.Request(url, headers={"Accept": "application/json", "User-Agent": "safe-tx-verify"}) + with urllib.request.urlopen(req, timeout=30) as resp: + return json.load(resp) + + +# --- repo cross-reference: address labels + selector index ------------------- +def load_address_labels(): + """Parse DEPLOYMENT_ADDRESSES.md -> {lower_addr: [ {name, chain, chainId} ]}. + + An address may be deployed on several chains under (usually) the same name, + so we keep every occurrence and disambiguate by chainId at lookup time. + """ + path = os.path.join(REPO_ROOT, "DEPLOYMENT_ADDRESSES.md") + labels = {} + if not os.path.exists(path): + return labels + chain, chain_id = "?", None + owners = {} # secondary labels (ProxyAdmin/Owner column) applied only if no primary + with open(path) as f: + for line in f: + hm = re.match(r"^##\s+(.+?)\s*$", line) + if hm: + chain = hm.group(1).strip() + cm = re.search(r"Chain ID:\s*(\d+)", chain) + chain_id = cm.group(1) if cm else None + continue + cells = [c.strip() for c in line.split("|") if c.strip()] + if len(cells) >= 2: + am = re.search(r"0x[a-fA-F0-9]{40}", cells[1]) + name = cells[0].strip("` ") + if am and name and name.lower() != "contract": + labels.setdefault(am.group(0).lower(), []).append( + {"name": name, "chain": chain, "chainId": chain_id}) + if len(cells) >= 3: # owner/proxy-admin column, e.g. "`0x..` (ProxyAdmin)" + om = re.search(r"0x[a-fA-F0-9]{40}", cells[2]) + roles = [r for r in re.findall(r"\(([^)]+)\)", cells[2]) if not r.startswith("http")] + if om: + owners.setdefault(om.group(0).lower(), []).append( + {"name": roles[-1] if roles else "Owner/Admin", "chain": chain, "chainId": chain_id}) + for addr, entries in owners.items(): + labels.setdefault(addr, entries) + return labels + + +def build_selector_index(): + """Scan evm/out/*.sol/*.json methodIdentifiers -> {selector: {sig: set(contracts)}}.""" + out_dir = os.path.join(REPO_ROOT, "evm", "out") + index = {} + if not os.path.isdir(out_dir): + return index + for sol in os.listdir(out_dir): + sol_path = os.path.join(out_dir, sol) + if not os.path.isdir(sol_path): + continue + for fn in os.listdir(sol_path): + if not fn.endswith(".json"): + continue + try: + with open(os.path.join(sol_path, fn)) as f: + art = json.load(f) + except Exception: + continue + for sig, sel in (art.get("methodIdentifiers") or {}).items(): + sel = "0x" + sel.lower() + index.setdefault(sel, {}).setdefault(sig, set()).add(fn[:-5]) + return index + + +def label_for(addr, labels, chain_id=None): + entries = labels.get(addr.lower()) + if entries: + e = next((x for x in entries if x["chainId"] == str(chain_id)), entries[0]) + return f"{e['name']} [{e['chain']}]" + # Fall back to well-known Safe infra singletons (MultiSend etc.) not in DEPLOYMENT_ADDRESSES.md + return KNOWN_SINGLETONS.get(addr.lower()) + + +def annotate_addresses(text, labels, chain_id=None): + """Append '(= ContractName)' after any known address appearing in text.""" + def repl(m): + lbl = label_for(m.group(0), labels, chain_id) + return f"{m.group(0)} {color('(= ' + lbl + ')', 'c')}" if lbl else m.group(0) + return re.sub(r"0x[a-fA-F0-9]{40}", repl, text) + + +def _parse_arg_lines(sig, data): + """cast calldata-decode prints one line per top-level argument; return them raw.""" + try: + return cast("calldata-decode", sig, data).splitlines() + except Exception: + return None + + +def _split_array(line): + inner = line.strip() + if inner.startswith("[") and inner.endswith("]"): + inner = inner[1:-1].strip() + return [x.strip() for x in inner.split(",")] if inner else [] + + +def expand_wrapper(sig, data): + """If sig is a Timelock/ProxyAdmin wrapper, return [(to, value, data, note)] sub-calls, else None.""" + name = sig.split("(")[0] + args = _parse_arg_lines(sig, data) + if not args: + return None + try: + if name in ("execute", "schedule"): # (target, value, payload, ...) + return [(args[0], args[1], args[2], None)] + if name in ("executeBatch", "scheduleBatch"): # (targets[], values[], payloads[], ...) + tgts, vals, plds = _split_array(args[0]), _split_array(args[1]), _split_array(args[2]) + return [(tgts[i], vals[i] if i < len(vals) else "0", plds[i], None) + for i in range(len(tgts))] + if name == "upgradeAndCall": # (proxy, newImpl, data) — data runs on the new impl + proxy, impl, payload = args[0], args[1], args[2] + note = f"upgrade proxy {proxy} → implementation {impl}" + return [(impl, "0", payload, note)] if payload not in ("0x", "") else [(impl, "0", "0x", note)] + if name in ("upgradeTo",): + return [(args[0], "0", "0x", f"upgrade to implementation {args[0]}")] + except (IndexError, ValueError): + return None + return None + + +def decode_call(to, data, index, labels, chain_id=None, safe=None, depth=0): + """Human-readable lines for this call; recurses into multiSend and Timelock/ProxyAdmin wrappers.""" + ind = " " * depth + lines = [] + data = data or "0x" + + lbl = label_for(to, labels, chain_id) + if lbl: + to_str = f"{to} {color('(= ' + lbl + ')', 'c')}" + elif safe and to.lower() == safe.lower(): + to_str = f"{to} {color('(= this Safe — account/config management)', 'c')}" + else: + to_str = f"{to}{color(' (unknown contract)', 'y')}" + + if len(data) < 10: + lines.append(f"{ind}{to_str}: plain value transfer (no calldata)") + return lines + + selector = data[:10].lower() + + if selector == MULTISEND_SELECTOR: + lines.append(f"{ind}{color('multiSend', 'b')} batch → {to_str}") + for i, inner in enumerate(unpack_multisend(data)): + op = "DELEGATECALL " + color("⚠", "r") if inner["operation"] == 1 else "CALL" + lines.append(f"{ind} [{i}] {op} value={inner['value']}") + lines += decode_call(inner["to"], inner["data"], index, labels, chain_id, safe, depth + 2) + return lines + + # Resolve signature: prefer THIS repo's compiled ABIs; else public 4byte DB (flagged). + entry = index.get(selector) + if entry: + sig = sorted(entry.keys())[0] + contracts = sorted({c for cs in entry.values() for c in cs}) + src = color("[repo ABI: " + ", ".join(contracts[:3]) + "]", "d") + verified = True + else: + sig = None + try: + sig = cast("4byte", selector).splitlines()[0].strip() + except Exception: + pass + src = color("[public 4byte DB — UNVERIFIED, possible collision]", "y") + verified = False + + lines.append(f"{ind}{to_str}") + if not sig: + lines.append(f"{ind} selector {selector} {color('— not in repo ABIs or 4byte DB', 'y')}") + lines.append(f"{ind} raw: {data[:74] + '…' if len(data) > 74 else data}") + return lines + + lines.append(f"{ind} {color(sig, 'b')} {src}") + decoded = _parse_arg_lines(sig, data) or [] + for arg_line in decoded: + lines.append(f"{ind} {annotate_addresses(arg_line.strip(), labels, chain_id)}") + + # Dive into wrapper calls (Timelock schedule/execute, ProxyAdmin upgradeAndCall). + subcalls = expand_wrapper(sig, data) + if subcalls: + lines.append(f"{ind} {color('↳ inner call(s):', 'b')}") + for j, (sto, sval, sdata, note) in enumerate(subcalls): + if note: + lines.append(f"{ind} [{j}] {annotate_addresses(note, labels, chain_id)}") + else: + lines.append(f"{ind} [{j}] value={sval}") + if (sdata or "0x") == "0x" and note: + continue # e.g. upgrade with no initializer call — nothing more to decode + lines += decode_call(sto, sdata, index, labels, chain_id, safe, depth + 3) + return lines + + +def unpack_multisend(data): + """Unpack multiSend(bytes) calldata into a list of inner transactions.""" + # data = 0x8d80ff0a + abi.encode(bytes). Decode the bytes arg, then walk the packed blob. + try: + blob = cast("calldata-decode", "multiSend(bytes)", data).strip() + except Exception: + return [] + b = bytes.fromhex(blob[2:] if blob.startswith("0x") else blob) + txs, i = [], 0 + while i + 85 <= len(b): + operation = b[i]; i += 1 + to = "0x" + b[i:i + 20].hex(); i += 20 + value = int.from_bytes(b[i:i + 32], "big"); i += 32 + dlen = int.from_bytes(b[i:i + 32], "big"); i += 32 + payload = b[i:i + dlen]; i += dlen + txs.append({"operation": operation, "to": to, "value": str(value), "data": "0x" + payload.hex()}) + return txs + + +# --- optional on-chain sanity checks ----------------------------------------- +def onchain_safe_nonce(address, rpc): + """Return the Safe's current on-chain nonce, or None if unavailable. + + Safe nonces execute strictly in order, so any queued transaction whose nonce + is BELOW this value can never execute — it's a stale leftover (superseded or + replaced by another tx that already consumed that nonce). Best-effort: returns + None if there's no RPC or the call fails, and staleness detection is skipped. + """ + if not rpc: + return None + try: + out = cast("call", address, "nonce()(uint256)", "--rpc-url", rpc) + return int(out.split()[0]) + except Exception: + return None + + +def onchain_checks(to, rpc): + lines = [] + try: + code = cast("code", to, "--rpc-url", rpc) + if code in ("0x", ""): + lines.append(color(f"⚠ {to} has NO code on-chain (EOA or wrong chain?)", "r")) + else: + lines.append(f"{to}: has code ({len(code) // 2 - 1} bytes)") + try: + impl = cast("storage", to, IMPL_SLOT, "--rpc-url", rpc) + impl_addr = "0x" + impl[-40:] + if int(impl, 16) != 0: + lines.append(f" ERC-1967 proxy → implementation {impl_addr}") + except Exception: + pass + except Exception as e: + lines.append(color(f"on-chain check skipped: {e}", "y")) + return lines + + +# --- main -------------------------------------------------------------------- +def main(): + ap = argparse.ArgumentParser(add_help=True) + ap.add_argument("input", help="Safe URL or address (optionally prefixed, e.g. oeth:0x...)") + ap.add_argument("--nonce") + ap.add_argument("--chain-id", dest="chain_id") + ap.add_argument("--version") + ap.add_argument("--onchain", action="store_true", help="run cast on-chain sanity checks") + ap.add_argument("--rpc") + ap.add_argument("--json", action="store_true") + # --- propose mode: verify a tx you are about to CREATE (not yet queued) --- + # Supply the raw params; the script computes the same Ledger hashes locally, + # decodes the calldata, and prints the safeTxHash to expect in the Safe UI. + ap.add_argument("--to", help="propose mode: target address of the tx you will create") + ap.add_argument("--data", help="propose mode: calldata (0x...) — default 0x") + ap.add_argument("--value", default="0", help="propose mode: wei value (default 0)") + ap.add_argument("--operation", default="0", help="propose mode: 0=CALL, 1=DELEGATECALL (default 0)") + ap.add_argument("--safe-tx-gas", dest="safe_tx_gas", default="0") + ap.add_argument("--base-gas", dest="base_gas", default="0") + ap.add_argument("--gas-price", dest="gas_price", default="0") + ap.add_argument("--gas-token", dest="gas_token", default=ZERO) + ap.add_argument("--refund-receiver", dest="refund_receiver", default=ZERO) + a = ap.parse_args() + + parsed = parse_safe_input(a.input) + if not parsed: + fail(f"could not find a Safe address in: {a.input}") + address, prefix, tx_hash = parsed["address"], parsed["prefix"], parsed["txHash"] + + chain_id = a.chain_id or PREFIX_TO_CHAINID.get(prefix) + if not chain_id: + fail(f'unknown network short-name "{prefix}"; pass --chain-id.') + + rpc = a.rpc or PREFIX_TO_RPC.get(prefix) + base = f"https://api.safe.global/tx-service/{prefix}/api/v1" + + version = a.version + if not version: + try: + version = fetch_json(f"{base}/safes/{address}/").get("version") or "1.3.0" + except Exception as e: + version = "1.3.0" + print(color(f"warn: could not fetch Safe version ({e}); assuming {version}", "y"), file=sys.stderr) + + propose = a.to is not None + if propose: + # No queued tx to fetch — build the params from the flags directly. + nonce = a.nonce + if nonce is None: + nonce = onchain_safe_nonce(address, rpc) + if nonce is None: + fail("propose mode: could not read the Safe's on-chain nonce; pass --nonce N " + "(the next nonce the Safe will use).") + # nonce() is the next-to-EXECUTE; a new tx queued behind already-pending + # ones takes the next FREE slot (max queued nonce + 1). Reconcile so we + # don't compute hashes for a nonce that's already taken. + try: + pend = fetch_json(f"{base}/safes/{address}/multisig-transactions/" + f"?executed=false&limit=100").get("results") or [] + queued = [int(t["nonce"]) for t in pend if t.get("nonce") is not None] + next_free = max([nonce] + [n + 1 for n in queued]) if queued else nonce + except Exception: + next_free = nonce + if next_free != nonce: + print(color(f"note: on-chain next-to-execute nonce is {nonce}, but " + f"transaction(s) are already queued ahead — a NEW tx takes " + f"nonce {next_free}. Using {next_free} (override with --nonce).", "y"), + file=sys.stderr) + nonce = next_free + else: + print(color(f"note: using the Safe's next nonce {nonce} " + f"(override with --nonce)", "d"), file=sys.stderr) + params = { + "safe": address, "chainId": chain_id, "version": version, + "to": a.to, "value": a.value, "data": a.data or "0x", + "operation": a.operation, "safeTxGas": a.safe_tx_gas, "baseGas": a.base_gas, + "gasPrice": a.gas_price, "gasToken": a.gas_token, + "refundReceiver": a.refund_receiver, "nonce": nonce, + } + api_hash = None + api_method = None + return _report(a, params, prefix, address, chain_id, version, rpc, + api_hash, api_method, propose=True) + + try: + listing = fetch_json(f"{base}/safes/{address}/multisig-transactions/?executed=false&limit=100") + except Exception as e: + fail(f"failed to fetch pending transactions: {e}") + results = listing.get("results") or [] + if not results: + fail(f"no pending transactions for {prefix}:{address}") + + # Safe nonces execute strictly in order: any queued tx with nonce < the + # Safe's current on-chain nonce can NEVER execute (stale/superseded). + current_nonce = onchain_safe_nonce(address, rpc) + + def is_stale(n): + return current_nonce is not None and n is not None and int(n) < current_nonce + + if a.nonce is not None: + tx = next((t for t in results if str(t.get("nonce")) == str(a.nonce)), None) + if not tx: + fail(f"no pending transaction with nonce {a.nonce}") + elif tx_hash: + tx = next((t for t in results if (t.get("safeTxHash") or "").lower() == tx_hash.lower()), None) + if not tx: + fail(f"tx {tx_hash} not found in pending queue") + else: + nonces = sorted({t.get("nonce") for t in results}) + # Only nonces >= the current on-chain nonce can actually execute. + live_txs = [t for t in results if not is_stale(t.get("nonce"))] + live_nonces = sorted({t.get("nonce") for t in live_txs}) + if len(live_txs) == 1: + tx = live_txs[0] + elif len(live_txs) == 0: + fail(f"all {len(results)} pending transaction(s) are stale (below the Safe's " + f"current on-chain nonce {current_nonce}) — none can execute.") + else: + stale_nonces = [n for n in nonces if is_stale(n)] + print("Multiple pending transactions queued.", file=sys.stderr) + if current_nonce is not None: + print(f"Safe current on-chain nonce: {current_nonce} " + f"(anything below it can never execute).", file=sys.stderr) + + def status(n): + if is_stale(n): + return " [STALE — can never execute, superseded]" + if current_nonce is not None and n == current_nonce: + return " [next to execute]" + return " [queued]" if current_nonce is not None else "" + + # Live transactions first, stale ones last. + for t in sorted(results, key=lambda x: (is_stale(x.get("nonce")), x.get("nonce") or 0)): + print(f" nonce {t.get('nonce')}{status(t.get('nonce'))} to {t.get('to')} " + f"method {(t.get('dataDecoded') or {}).get('method', '(raw)')}", file=sys.stderr) + pick = live_nonces if current_nonce is not None else nonces + note = f" (ignore stale {', '.join(map(str, stale_nonces))})" if stale_nonces else "" + print(f"\nRe-run with --nonce — live nonces: {', '.join(map(str, pick))}{note}.", + file=sys.stderr) + sys.exit(2) + + if is_stale(tx.get("nonce")): + print(color(f"⚠ nonce {tx.get('nonce')} is below the Safe's current on-chain nonce " + f"{current_nonce} — this transaction can NEVER execute (already superseded). " + f"Verifying anyway for inspection.", "y"), file=sys.stderr) + + params = { + "safe": address, "chainId": chain_id, "version": version, + "to": tx.get("to") or ZERO, "value": tx.get("value") or "0", + "data": tx.get("data") or "0x", "operation": tx.get("operation") or 0, + "safeTxGas": tx.get("safeTxGas") or 0, "baseGas": tx.get("baseGas") or 0, + "gasPrice": tx.get("gasPrice") or "0", "gasToken": tx.get("gasToken") or ZERO, + "refundReceiver": tx.get("refundReceiver") or ZERO, "nonce": tx.get("nonce"), + } + api_hash = tx.get("safeTxHash") + api_method = (tx.get("dataDecoded") or {}).get("method") + return _report(a, params, prefix, address, chain_id, version, rpc, + api_hash, api_method, propose=False) + + +def _report(a, params, prefix, address, chain_id, version, rpc, api_hash, api_method, propose): + """Compute hashes, decode calldata, and print the verification report. + + Shared by both the queue path (api_hash from the Safe backend, cross-checked) + and the propose path (no backend tx yet — api_hash is None and the computed + safeTxHash is what the Safe UI should show once the tx is created). + """ + computed = compute_hashes(params) + match = bool(api_hash) and api_hash.lower() == computed["safeTxHash"].lower() + + labels = load_address_labels() + index = build_selector_index() + decoded_lines = decode_call(params["to"], params["data"], index, labels, chain_id, safe=address) + + if a.json: + print(json.dumps({ + "network": prefix, "chainId": chain_id, "version": version, "mode": "propose" if propose else "queue", + "params": params, "computed": computed, "apiSafeTxHash": api_hash, "apiMatch": match, + "decoded": [re.sub(r"\033\[[0-9;]*m", "", ln) for ln in decoded_lines], + }, indent=2)) + return + + print() + if propose: + print(color("PROPOSE MODE — hashes for a tx you are about to CREATE (nothing queued yet).", "y")) + print() + print(color("👉 CHECK THIS ON YOUR LEDGER — Message hash:", "b")) + print(color(color(f" {computed['messageHash']}", "c"), "b")) + print() + print(color(" Supporting values (all computed locally — nothing trusted from the API):", "d")) + print(color(f" Domain hash: {computed['domainHash']} (also on the Ledger; = chainId {chain_id} + safe addr)", "d")) + print(color(f" SafeTxHash: {computed['safeTxHash']} (Safe/Etherscan label; not on the Ledger)", "d")) + print("─" * 78) + print(f"Safe: {prefix}:{address} · version {version} · nonce {params['nonce']}") + op = "DELEGATECALL " + color("⚠", "r") if int(params["operation"]) == 1 else "CALL" + print(f"Operation: {params['operation']} ({op}) Value: {params['value']}") + if api_method: + print(color(f"Safe backend decoded this as: {api_method}", "d")) + print() + print(color("What this transaction does (decoded from THIS repo's ABIs):", "b")) + for ln in decoded_lines: + print(" " + ln) + print("─" * 78) + if propose: + print(color("ℹ Propose mode: no queued tx to cross-check against. After you create the tx in " + "the Safe UI, confirm it shows the SafeTxHash above — and that your Ledger shows " + "the message + domain hash above — before signing.", "c")) + elif api_hash: + print(color("✓ safeTxHash matches Safe API (independent cross-check passed)", "g") if match + else color(f"✗ MISMATCH vs Safe API {api_hash} — DO NOT SIGN, investigate.", "r")) + + if a.onchain: + print() + if not rpc: + print(color(f"on-chain checks skipped: no RPC for {prefix} (pass --rpc)", "y")) + else: + print(color(f"On-chain sanity ({prefix}):", "b")) + for ln in onchain_checks(params["to"], rpc): + print(" " + ln) + + if api_hash and not match: + sys.exit(3) + + +if __name__ == "__main__": + main() diff --git a/.gitignore b/.gitignore index 6010f21..9164949 100644 --- a/.gitignore +++ b/.gitignore @@ -31,5 +31,3 @@ evm/.common.env # Claude Code: track shared skills only; ignore local runtime artifacts .claude/* !.claude/skills/ -.claude/skills/* -!.claude/skills/*/ diff --git a/tools/safe-tx-verify/README.md b/tools/safe-tx-verify/README.md deleted file mode 100644 index 85bc9d9..0000000 --- a/tools/safe-tx-verify/README.md +++ /dev/null @@ -1,68 +0,0 @@ -# safe-tx-verify - -Independently verify a Safe (Gnosis Safe) multisig transaction **before signing it on a Ledger**. - -The script recomputes the Safe EIP-712 hashes — `domainHash`, `messageHash`, `safeTxHash` — **from the raw -transaction parameters**. It never trusts the hash the Safe backend returns: it fetches the queued transaction's -parameters, recomputes the hash locally, and tells you whether they agree. It then decodes the calldata using this -repo's compiled ABIs and labels every address against [`DEPLOYMENT_ADDRESSES.md`](../../DEPLOYMENT_ADDRESSES.md). - -When you sign a Safe transaction, the Ledger shows the pair **(domain hash, message hash)**. The message hash is -the one that encodes the actual transaction, so it is the value to check against the device. - -## Requirements - -- [`uv`](https://docs.astral.sh/uv/) — the only prerequisite. Dependencies (`eth-utils`, `eth-abi`) are declared - inline (PEP 723) and resolved automatically by `uv run`; nothing to install manually. -- Optional: this repo's compiled artifacts (`evm/out`, via `forge build`) for richer calldata decoding, and network - access to `api.safe.global` (the Safe Transaction Service; no API key required for reads). - -## Usage - -```bash -uv run tools/safe-tx-verify/verify_safe_tx.py "" --nonce N \ - [--index N] [--chain-id N] [--version V] [--onchain] [--rpc URL] [--json] -``` - -- `target` — a Safe app URL (`https://app.safe.global/...?safe=oeth:0x…`) or a `prefix:0xADDRESS` (e.g. `oeth:0x398A…`). -- `--nonce N` — the queued transaction's nonce to verify (required). -- `--index N` — when several proposals share the same nonce, the tool refuses to guess: it lists each candidate - (index + `safeTxHash`) and exits non-zero. Re-run with `--index N` to pick the exact one you intend to sign. -- `--chain-id N` — override the chain id (otherwise derived from the prefix). -- `--version V` — Safe contract version (default `1.3.0`; auto-read with `--onchain`). Governs the EIP-712 domain - (chainId is only bound for ≥ 1.3.0) and the `SafeTx` typehash. -- `--onchain` — also read the Safe's live `nonce` / `threshold` / `VERSION()` via RPC and flag a stale nonce. -- `--rpc URL` — RPC endpoint (otherwise a public default for the chain). -- `--json` — machine-readable output. Exit code `0` = backend hash matches, `2` = mismatch. - -### Examples - -```bash -# Verify nonce 48 on the Optimism admin Safe -uv run tools/safe-tx-verify/verify_safe_tx.py "oeth:0x398A2749487B2a91f2f543C01F7afD19AEE4b6b0" --nonce 48 - -# From a Safe app URL, cross-checking the on-chain nonce/version too -uv run tools/safe-tx-verify/verify_safe_tx.py \ - "https://app.safe.global/transactions/queue?safe=oeth:0x398A2749487B2a91f2f543C01F7afD19AEE4b6b0" \ - --nonce 48 --onchain -``` - -## What to check before signing - -1. **`MATCH`** — the parameters the backend gave and the hash it returned are consistent (the script recomputed the - same `safeTxHash` from the params). A `MISMATCH` means **do not sign**. -2. The **message hash** printed here equals what the Ledger displays. -3. The **decoded action** (`to`, `operation`, function + args) is what you intend — `operation: 1` is a - `delegatecall`; batched `multiSend` calls are expanded and each inner call decoded. -4. Every address is the one you expect — known addresses are labeled from `DEPLOYMENT_ADDRESSES.md`. - -## Notes - -- Supported chains (prefix → chainId): `eth`(1), `oeth`(10), `arb1`(42161), `base`(8453), `matic`(137), `gno`(100), - `bnb`(56), `avax`(43114), `linea`(59144), `scr`(534352), `celo`(42220), `blast`(81457), `zksync`(324), `mnt`(5000), - plus testnets `sep`, `basesep`, `oeth-sep`, `arb-sep`. Others: pass `--chain-id` + `--rpc`. -- The Safe Transaction Service moved to the unified `https://api.safe.global/tx-service//api/v1/…` endpoint; - this tool targets it. -- The EIP-712 hashing was validated against real Optimism Safe transactions (recomputed `safeTxHash` matched the - backend for every sampled tx, including delegatecall `multiSend` batches). Calldata decoding is best-effort and - never affects the hash verification. diff --git a/tools/safe-tx-verify/verify_safe_tx.py b/tools/safe-tx-verify/verify_safe_tx.py deleted file mode 100644 index 40ffbb3..0000000 --- a/tools/safe-tx-verify/verify_safe_tx.py +++ /dev/null @@ -1,422 +0,0 @@ -#!/usr/bin/env -S uv run --script -# /// script -# requires-python = ">=3.10" -# dependencies = [ -# "eth-utils>=4.1", # keccak -# "eth-abi>=5.0", # best-effort calldata decoding -# ] -# /// -""" -Verify a Safe (Gnosis Safe) multisig transaction before signing on a Ledger. - -Independently recomputes the Safe EIP-712 hashes (domain hash, message hash, safeTxHash) from the raw -transaction PARAMETERS -- it never trusts the hash the Safe backend returns -- then decodes the calldata into a -human-readable action using THIS repo's compiled ABIs and cross-references every address against -DEPLOYMENT_ADDRESSES.md. - -What your Ledger shows when signing a Safe tx is the pair (domain hash, message hash); the message hash is the one -that encodes the actual transaction, so it is the headline value here. - -Run it with uv (dependencies are resolved automatically from the inline metadata above): - - uv run tools/safe-tx-verify/verify_safe_tx.py "" [--nonce N] - [--chain-id N] [--version V] [--onchain] [--rpc URL] [--json] - -Examples: - uv run tools/safe-tx-verify/verify_safe_tx.py \ - "https://app.safe.global/transactions/queue?safe=oeth:0x398A2749487B2a91f2f543C01F7afD19AEE4b6b0" --nonce 48 - uv run tools/safe-tx-verify/verify_safe_tx.py "oeth:0x398A2749487B2a91f2f543C01F7afD19AEE4b6b0" --nonce 48 --onchain -""" -from __future__ import annotations - -import argparse -import json -import os -import re -import sys -import urllib.request -import urllib.error -from pathlib import Path - -from eth_utils import keccak, to_checksum_address - -try: - from eth_abi import decode as abi_decode # eth-abi >= 4 -except Exception: # pragma: no cover - abi_decode = None - -# --- EIP-712 type hashes (computed from the canonical strings so they are self-verifying) ------------------------ -DOMAIN_TYPEHASH = keccak(text="EIP712Domain(uint256 chainId,address verifyingContract)") -DOMAIN_TYPEHASH_OLD = keccak(text="EIP712Domain(address verifyingContract)") # Safe < 1.3.0 (no chainId) -SAFE_TX_TYPEHASH = keccak( - text="SafeTx(address to,uint256 value,bytes data,uint8 operation,uint256 safeTxGas,uint256 baseGas," - "uint256 gasPrice,address gasToken,address refundReceiver,uint256 nonce)" -) -SAFE_TX_TYPEHASH_OLD = keccak( # Safe < 1.0.0 used `dataGas` instead of `baseGas` - text="SafeTx(address to,uint256 value,bytes data,uint8 operation,uint256 safeTxGas,uint256 dataGas," - "uint256 gasPrice,address gasToken,address refundReceiver,uint256 nonce)" -) - -MULTISEND_SELECTOR = "0x8d80ff0a" # multiSend(bytes) - -# Safe app chain short-name (used verbatim in the tx-service path) -> (chainId, default public RPC). -# Safe deprecated the per-chain `safe-transaction-.safe.global` hosts; the unified service is -# `https://api.safe.global/tx-service//api/v1/...` (no API key required for reads as of 2026-07). -TX_SERVICE = "https://api.safe.global/tx-service/{short}/api/v1" -CHAINS = { - "eth": (1, "https://eth.llamarpc.com"), - "oeth": (10, "https://mainnet.optimism.io"), - "arb1": (42161, "https://arb1.arbitrum.io/rpc"), - "base": (8453, "https://mainnet.base.org"), - "matic": (137, "https://polygon-rpc.com"), - "gno": (100, "https://rpc.gnosischain.com"), - "bnb": (56, "https://bsc-dataseed.binance.org"), - "avax": (43114, "https://api.avax.network/ext/bc/C/rpc"), - "linea": (59144, "https://rpc.linea.build"), - "scr": (534352, "https://rpc.scroll.io"), - "celo": (42220, "https://forno.celo.org"), - "blast": (81457, "https://rpc.blast.io"), - "zksync": (324, "https://mainnet.era.zksync.io"), - "mnt": (5000, "https://rpc.mantle.xyz"), - "sep": (11155111, "https://ethereum-sepolia-rpc.publicnode.com"), - "basesep": (84532, "https://sepolia.base.org"), - "oeth-sep": (11155420, "https://sepolia.optimism.io"), - "arb-sep": (421614, "https://sepolia-rollup.arbitrum.io/rpc"), -} -CHAINID_TO_SHORT = {cid: short for short, (cid, _rpc) in CHAINS.items()} -CHAINID_TO_RPC = {cid: rpc for _, (cid, rpc) in CHAINS.items()} - -_USE_COLOR = sys.stdout.isatty() - - -def c(text: str, code: str) -> str: - return f"\033[{code}m{text}\033[0m" if _USE_COLOR else text - - -def die(msg: str) -> "None": - print(c(f"Error: {msg}", "31"), file=sys.stderr) - sys.exit(1) - - -# --- hashing helpers ------------------------------------------------------------------------------------------- -def word(value) -> bytes: - """Left-pad an address/bytes32/int to a 32-byte EVM word.""" - if isinstance(value, str): - h = value.lower().removeprefix("0x") - b = bytes.fromhex(h) - if len(b) > 32: - die(f"value does not fit in 32 bytes: {value}") - return b.rjust(32, b"\x00") - if isinstance(value, int): - return value.to_bytes(32, "big") - raise TypeError(type(value)) - - -def cmp_versions(a: str, b: str) -> int: - def parts(v): - return [int(x) for x in re.findall(r"\d+", v)] - pa, pb = parts(a), parts(b) - pa += [0] * (len(pb) - len(pa)) - pb += [0] * (len(pa) - len(pb)) - return (pa > pb) - (pa < pb) - - -def compute_hashes(tx: dict, safe: str, chain_id: int, version: str) -> dict: - """Recompute (domainHash, messageHash, safeTxHash) from raw params. This is the trust anchor.""" - safe = to_checksum_address(safe) - data_bytes = bytes.fromhex(tx["data"].lower().removeprefix("0x")) if tx.get("data") else b"" - - # Domain separator: >= 1.3.0 binds chainId; older versions do not. - if cmp_versions(version, "1.3.0") >= 0: - domain_hash = keccak(word(DOMAIN_TYPEHASH.hex()) + word(chain_id) + word(safe)) - else: - domain_hash = keccak(word(DOMAIN_TYPEHASH_OLD.hex()) + word(safe)) - - tx_typehash = SAFE_TX_TYPEHASH_OLD if cmp_versions(version, "1.0.0") < 0 else SAFE_TX_TYPEHASH - struct_hash = keccak( - word(tx_typehash.hex()) - + word(to_checksum_address(tx["to"])) - + word(int(tx["value"])) - + word(keccak(data_bytes).hex()) - + word(int(tx["operation"])) - + word(int(tx["safeTxGas"])) - + word(int(tx["baseGas"])) - + word(int(tx["gasPrice"])) - + word(to_checksum_address(tx["gasToken"])) - + word(to_checksum_address(tx["refundReceiver"])) - + word(int(tx["nonce"])) - ) - safe_tx_hash = keccak(b"\x19\x01" + domain_hash + struct_hash) - return { - "domainHash": "0x" + domain_hash.hex(), - "messageHash": "0x" + struct_hash.hex(), # the EIP-712 struct hash = the Ledger "message hash" - "safeTxHash": "0x" + safe_tx_hash.hex(), - } - - -# --- IO helpers ------------------------------------------------------------------------------------------------ -def http_get_json(url: str, timeout: int = 20) -> dict: - req = urllib.request.Request(url, headers={"Accept": "application/json", "User-Agent": "verify-safe-tx"}) - try: - with urllib.request.urlopen(req, timeout=timeout) as r: - return json.loads(r.read().decode()) - except urllib.error.HTTPError as e: - die(f"HTTP {e.code} for {url}") - except Exception as e: # noqa: BLE001 - die(f"request failed for {url}: {e}") - - -def rpc_call(rpc: str, to: str, data: str, timeout: int = 20) -> str: - payload = json.dumps( - {"jsonrpc": "2.0", "id": 1, "method": "eth_call", "params": [{"to": to, "data": data}, "latest"]} - ).encode() - req = urllib.request.Request(rpc, data=payload, headers={"Content-Type": "application/json"}) - with urllib.request.urlopen(req, timeout=timeout) as r: - return json.loads(r.read().decode()).get("result", "0x") - - -def parse_target(target: str) -> tuple[str, str]: - """Return (chain_prefix, checksummed_address) from 'oeth:0x..' or a Safe app URL.""" - m = re.search(r"(?:safe=)?([a-z0-9-]+):(0x[0-9a-fA-F]{40})", target) - if not m: - die(f"could not parse a 'chain:0xaddress' from: {target}") - return m.group(1), to_checksum_address(m.group(2)) - - -def repo_root() -> Path: - p = Path(__file__).resolve() - for parent in [p, *p.parents]: - if (parent / "DEPLOYMENT_ADDRESSES.md").exists(): - return parent - if (parent / "evm" / "DEPLOYMENT_ADDRESSES.md").exists(): - return parent - return Path.cwd() - - -def load_address_book(root: Path) -> dict[str, str]: - """address(lower) -> label, parsed from DEPLOYMENT_ADDRESSES.md.""" - book: dict[str, str] = {} - for cand in (root / "DEPLOYMENT_ADDRESSES.md", root / "evm" / "DEPLOYMENT_ADDRESSES.md"): - if not cand.exists(): - continue - section = "" - for line in cand.read_text().splitlines(): - h = re.match(r"##+\s+(.*)", line) - if h: - section = h.group(1).strip() - continue - cells = [x.strip() for x in line.split("|")] - if len(cells) < 3 or cells[1] in ("Contract", "--------", ""): - continue - name = cells[1] - for addr in re.findall(r"0x[0-9a-fA-F]{40}", line): - book.setdefault(addr.lower(), f"{name} [{section}]") - break - return book - - -def load_selectors(root: Path) -> dict[str, tuple[str, list[str], list[str]]]: - """4-byte selector -> (name, arg_types, arg_names) from this repo's compiled ABIs (evm/out).""" - sels: dict[str, tuple[str, list[str], list[str]]] = {} - out = root / "evm" / "out" - if not out.exists(): - out = root / "out" - if not out.exists(): - return sels - for jf in out.rglob("*.json"): - try: - abi = json.loads(jf.read_text()).get("abi", []) - except Exception: # noqa: BLE001 - continue - for item in abi: - if item.get("type") != "function": - continue - types = [_canonical_type(i) for i in item.get("inputs", [])] - names = [i.get("name", "") for i in item.get("inputs", [])] - sig = f"{item['name']}({','.join(types)})" - sel = "0x" + keccak(text=sig)[:4].hex() - sels.setdefault(sel, (item["name"], types, names)) - return sels - - -def _canonical_type(inp: dict) -> str: - t = inp["type"] - if t.startswith("tuple"): - inner = ",".join(_canonical_type(c) for c in inp.get("components", [])) - return f"({inner}){t[len('tuple'):]}" - return t - - -def decode_calldata(data: str, sels: dict, book: dict, depth: int = 0) -> list[str]: - lines: list[str] = [] - ind = " " * depth - data = data.lower() - if not data or data == "0x" or len(data) < 10: - lines.append(f"{ind}(no calldata / plain ETH transfer)") - return lines - selector = data[:10] - if selector == MULTISEND_SELECTOR: - lines.append(f"{ind}multiSend(bytes) -> batched transactions:") - lines += decode_multisend(data, sels, book, depth + 1) - return lines - match = sels.get(selector) - if not match: - lines.append(f"{ind}selector {selector} (unknown -- not in repo ABIs)") - return lines - name, types, names = match - lines.append(f"{ind}{name}({','.join(types)})") - if abi_decode is not None: - try: - values = abi_decode(types, bytes.fromhex(data[10:])) - for t, n, v in zip(types, names, values): - lines.append(f"{ind} {n or '_'}: {_fmt_value(t, v, book)}") - except Exception as e: # noqa: BLE001 - lines.append(f"{ind} (could not decode args: {e})") - return lines - - -def decode_multisend(data: str, sels: dict, book: dict, depth: int) -> list[str]: - lines: list[str] = [] - # multiSend(bytes): after the selector, a standard ABI-encoded single `bytes` arg. - raw = bytes.fromhex(data[10:]) - if len(raw) < 64: - return [f"{' ' * depth}(malformed multiSend)"] - length = int.from_bytes(raw[32:64], "big") - payload = raw[64:64 + length] - i, n = 0, 0 - while i + 85 <= len(payload): - op = payload[i] - to = to_checksum_address(payload[i + 1:i + 21].hex()) - val = int.from_bytes(payload[i + 21:i + 53], "big") - dlen = int.from_bytes(payload[i + 53:i + 85], "big") - inner = payload[i + 85:i + 85 + dlen] - i += 85 + dlen - lines.append( - f"{' ' * depth}[{n}] op={op} to={_fmt_addr(to, book)} value={val}" - ) - lines += decode_calldata("0x" + inner.hex(), sels, book, depth + 1) - n += 1 - return lines - - -def _fmt_addr(addr: str, book: dict) -> str: - label = book.get(addr.lower()) - return f"{addr} ({c(label, '36')})" if label else c(addr, "33") - - -def _fmt_value(t: str, v, book: dict) -> str: - if t == "address": - return _fmt_addr(to_checksum_address(v), book) - if t.endswith("[]") and isinstance(v, (list, tuple)): - return "[" + ", ".join(_fmt_value(t[:-2], x, book) for x in v) + "]" - if isinstance(v, bytes): - return "0x" + v.hex() - return str(v) - - -# --- main ------------------------------------------------------------------------------------------------------ -def main() -> None: - ap = argparse.ArgumentParser(description="Independently verify a Safe multisig tx before signing.") - ap.add_argument("target", help="Safe app URL or 'prefix:0xADDRESS' (e.g. oeth:0x398A...)") - ap.add_argument("--nonce", type=int, help="Safe nonce of the queued tx to verify") - ap.add_argument("--chain-id", type=int, help="override chain id (else derived from the prefix)") - ap.add_argument("--version", default=None, help="Safe contract version (default 1.3.0 or on-chain with --onchain)") - ap.add_argument("--onchain", action="store_true", help="also read nonce/threshold/owners/version via RPC") - ap.add_argument("--rpc", help="RPC URL (else a public default for the chain)") - ap.add_argument("--index", type=int, default=None, - help="which proposal to verify when several share the nonce (see the listing on ambiguity)") - ap.add_argument("--json", action="store_true", dest="as_json", help="emit machine-readable JSON") - args = ap.parse_args() - - prefix, safe = parse_target(args.target) - if prefix not in CHAINS and args.chain_id is None: - die(f"unknown chain prefix '{prefix}'; pass --chain-id") - chain_id = args.chain_id or CHAINS[prefix][0] - short = prefix if prefix in CHAINS else CHAINID_TO_SHORT.get(chain_id) - rpc = args.rpc or CHAINID_TO_RPC.get(chain_id) - root = repo_root() - book = load_address_book(root) - - version = args.version or "1.3.0" - onchain: dict = {} - if args.onchain: - if not rpc: - die("no RPC for --onchain; pass --rpc") - try: - onchain["nonce"] = int(rpc_call(rpc, safe, "0xaffed0e0") or "0x0", 16) - onchain["threshold"] = int(rpc_call(rpc, safe, "0xe75235b8") or "0x0", 16) - ver_raw = rpc_call(rpc, safe, "0xffa1ad74") # VERSION() - if ver_raw and len(ver_raw) > 130: - strlen = int(ver_raw[66:130], 16) - version = bytes.fromhex(ver_raw[130:130 + strlen * 2]).decode() or version - if args.version: - version = args.version - except Exception as e: # noqa: BLE001 - die(f"--onchain RPC call failed: {e}") - - if args.nonce is None: - die("pass --nonce N (the queued tx nonce to verify)") - - if not short: - die(f"no Safe tx-service short-name known for chain {chain_id}; use a known chain prefix") - api = f"{TX_SERVICE.format(short=short)}/safes/{safe}/multisig-transactions/?nonce={args.nonce}" - results = http_get_json(api).get("results", []) - if not results: - die(f"no queued tx at nonce {args.nonce} for {safe} on chain {chain_id}") - if len(results) > 1 and args.index is None: - # Competing proposals at one nonce: never silently pick one — the signer must choose explicitly, - # otherwise --json / redirected stdout could verify a different tx than the one being signed. - lines = [f"{len(results)} proposals share nonce {args.nonce}; re-run with --index N to pick one:"] - for i, r in enumerate(results): - lines.append(f" --index {i} safeTxHash={r.get('safeTxHash')} to={r.get('to')} op={r.get('operation')}") - die("\n".join(lines)) - idx = args.index or 0 - if idx < 0 or idx >= len(results): - die(f"--index {idx} out of range (found {len(results)} proposal(s) at nonce {args.nonce})") - tx = results[idx] - tx.setdefault("baseGas", 0) - - computed = compute_hashes(tx, safe, chain_id, version) - backend_hash = (tx.get("safeTxHash") or "").lower() - match = backend_hash == computed["safeTxHash"].lower() - - if args.as_json: - print(json.dumps({ - "safe": safe, "chainId": chain_id, "version": version, "nonce": args.nonce, - "params": {k: tx.get(k) for k in - ("to", "value", "data", "operation", "safeTxGas", "baseGas", "gasPrice", - "gasToken", "refundReceiver", "nonce")}, - "computed": computed, "backendSafeTxHash": backend_hash, - "backendMatches": match, "onchain": onchain, - }, indent=2)) - sys.exit(0 if match else 2) - - print(c("=== Safe transaction verification ===", "1")) - print(f"Safe: {_fmt_addr(safe, book)}") - print(f"Chain: {chain_id} ({short}) Safe version: {version}") - print(f"Nonce: {args.nonce}") - if onchain: - print(f"On-chain: nonce={onchain.get('nonce')} threshold={onchain.get('threshold')}") - if onchain.get("nonce", args.nonce) > args.nonce: - print(c(" ! on-chain nonce is already past this tx's nonce", "31")) - print() - print(c("Ledger will show:", "1")) - print(f" domain hash: {computed['domainHash']}") - print(f" message hash: {c(computed['messageHash'], '1;36')} <- verify THIS on the device") - print(f" safeTxHash: {computed['safeTxHash']}") - print() - print(f"Backend safeTxHash: {backend_hash}") - print(" " + (c("MATCH — backend params are consistent with the hash", "1;32") if match - else c("MISMATCH — do NOT sign; params/hash disagree", "1;31"))) - print() - print(c("Decoded action:", "1")) - print(f" to: {_fmt_addr(to_checksum_address(tx['to']), book)}") - print(f" value: {tx.get('value')} operation: {tx.get('operation')} ({'delegatecall' if int(tx.get('operation', 0)) == 1 else 'call'})") - sels = load_selectors(root) - for line in decode_calldata(tx.get("data") or "0x", sels, book): - print(" " + line) - sys.exit(0 if match else 2) - - -if __name__ == "__main__": - main()