a script for detecting unrecoverable unlock locksroots

find_ghost_locksroots.py

#!/usr/bin/env python
import json
import sys
from collections import defaultdict

from raiden.storage.sqlite import SQLiteStorage


def read_args():
    if len(sys.argv) < 2 or "-h" in sys.argv or "--help" in sys.argv:
        print(f"Usage:\n\t{sys.argv[0]} <db_file>")
        sys.exit(0)

    storage = SQLiteStorage(sys.argv[1])
    return storage


def collect_known_locksroots(storage):
    known_locksroots = defaultdict(list)

    for x in [item for sublist in storage.batch_query_state_changes(500) for item in sublist]:
        if "balance_proof" in x.data:
            record = json.loads(x.data)
            if record["balance_proof"] is not None and record["balance_proof"]["locksroot"]:
                known_locksroots[record["balance_proof"]["locksroot"]].append(x)

    for x in [item for sublist in storage.batch_query_event_records(500) for item in sublist]:
        if "balance_proof" in x.data:
            record = json.loads(x.data)
            if record["balance_proof"] is not None and record["balance_proof"]["locksroot"]:
                known_locksroots[record["balance_proof"]["locksroot"]].append(x)

    return known_locksroots


def find_balance_proofs_for_unlock_locksroot(storage, our_address, known_locksroots):
    for x in [item for sublist in storage.batch_query_state_changes(500) for item in sublist]:
        if "ContractReceiveChannelBatchUnlock" in x.data:
            record = json.loads(x.data)
            if our_address.lower() in (record["participant"].lower(), record["partner"].lower()):
                if record["locksroot"] not in known_locksroots:
                    yield ("missing", record["locksroot"])
                else:
                    yield ("ok", record["locksroot"])


def get_address(storage):
    for entry in storage.get_state_changes():
        if "ActionInitChain" in entry:
            return json.loads(entry)["our_address"].lower()


if __name__ == "__main__":
    storage = read_args()
    our_address = get_address(storage)
    print(f"our address is {our_address}")
    known_locksroots = collect_known_locksroots(storage)
    print(f"found {len(known_locksroots)} locksroots")
    our_locksroots = list(
        find_balance_proofs_for_unlock_locksroot(storage, our_address, known_locksroots)
    )
    if our_locksroots:
        for result in our_locksroots:
            print(result)
    else:
        print('no ContractReceiveChannelBatchUnlock')