htlc_decoder.py

#!/usr/bin/env python3
"""
htlc-decoder — decode COMIT / Liquality atomic-swap HTLC contracts from EVM bytecode.

The Liquality / COMIT cross-chain atomic-swap HTLC is a ~200-byte hand-written EVM
contract. Every instance shares one opcode skeleton; only four fields vary:

  secretHash  sha256 image the claimer must reveal          (PUSH32)
  expiration  unix timestamp after which refund is allowed   (PUSH5)
  buyer       receives funds on CLAIM (secret revealed)      (PUSH20)
  seller      receives funds on REFUND (after timeout)       (PUSH20)

Two sub-variants differ only in the gas pushed to the SHA-256 precompile
(0x48 vs 0xffff); this decoder handles both.

Usage:
  python htlc_decoder.py <hex_bytecode>   # decode raw runtime bytecode
  python htlc_decoder.py <address>        # fetch code from RPC, decode, show live state
                                          #   (set RPC_URL; defaults to a public node)

Read-only. No keys, no transactions. Decoding is not recovering.
"""

import json
import os
import re
import sys
import urllib.request
from datetime import datetime, timezone

RPC = os.environ.get("RPC_URL", "https://ethereum-rpc.publicnode.com")

# Shared opcode skeleton; () groups capture the four variable fields.
HTLC_RE = re.compile(
    r"^0x"
    r"602080600080376021816000806002(?:60[0-9a-f]{2}|61[0-9a-f]{4})f1"   # setup + sha256 precompile call
    r"36602014167f([0-9a-f]{64})"                                        # CALLDATASIZE==32, PUSH32 secretHash
    r"602151141660[0-9a-f]{2}573615"                                     # hash compare + claim jumpdest
    r"64([0-9a-f]{10})"                                                  # PUSH5 expiration
    r"42111660[0-9a-f]{2}57fe"                                           # block.timestamp > exp + refund jumpdest
    r"5b7f8c1d64e3bd87387709175b9ef4e7a1d7a8364559fc0e2ad9d77953909a0d1eb3"  # CLAIM event topic
    r"60206000a173([0-9a-f]{40})ff"                                      # LOG1(secret); PUSH20 buyer; SELFDESTRUCT
    r"5b7f5d26862916391bf49478b2f5103b0720a842b45ef145a268f2cd1fb2aed55178"  # REFUND event topic
    r"600080a173([0-9a-f]{40})ff$",                                      # LOG1(); PUSH20 seller; SELFDESTRUCT
    re.IGNORECASE,
)


def decode(bytecode):
    """Return the four HTLC fields, or None if the bytecode isn't a COMIT HTLC."""
    if not bytecode:
        return None
    m = HTLC_RE.match(bytecode.strip())
    if not m:
        return None
    secret_hash, exp_hex, buyer, seller = m.groups()
    return {
        "secret_hash": "0x" + secret_hash,
        "expiration": int(exp_hex, 16),
        "buyer": "0x" + buyer,
        "seller": "0x" + seller,
    }


def rpc(method, params):
    req = urllib.request.Request(
        RPC,
        data=json.dumps({"jsonrpc": "2.0", "method": method, "params": params, "id": 1}).encode(),
        headers={"Content-Type": "application/json", "User-Agent": "htlc-decoder"},
    )
    with urllib.request.urlopen(req, timeout=30) as r:
        return json.load(r)["result"]


def main(argv):
    if not argv:
        print(__doc__)
        return
    arg = argv[0].strip()

    if arg.lower().startswith("0x") and len(arg) == 42:        # an address
        code = rpc("eth_getCode", [arg, "latest"])
        info = decode(code)
        if not info:
            print(f"{arg}: not a COMIT/Liquality HTLC (or already self-destructed)")
            return
        bal = int(rpc("eth_getBalance", [arg, "latest"]), 16)
        now = int(rpc("eth_getBlockByNumber", ["latest", False])["timestamp"], 16)
        info["address"] = arg
        info["balance_eth"] = bal / 1e18
        info["expired"] = now > info["expiration"]
        info["refundable"] = bal > 0 and now > info["expiration"]
    else:                                                      # raw bytecode
        info = decode(arg if arg.lower().startswith("0x") else "0x" + arg)
        if not info:
            print("not a COMIT/Liquality HTLC bytecode")
            return

    info["expiration_utc"] = datetime.fromtimestamp(info["expiration"], timezone.utc).isoformat()
    print(json.dumps(info, indent=2))


if __name__ == "__main__":
    main(sys.argv[1:])