basicswap/basicswap/basicswap_util.py

# -*- coding: utf-8 -*-

# Copyright (c) 2021-2024 tecnovert
# Copyright (c) 2024 The Basicswap developers
# Distributed under the MIT software license, see the accompanying
# file LICENSE or http://www.opensource.org/licenses/mit-license.php.


import struct
import hashlib
from enum import IntEnum, auto
from .util.address import (
    encodeAddress,
    decodeAddress,
)
from .chainparams import (
    chainparams,
)


class TxLockTypes(IntEnum):
    SEQUENCE_LOCK_BLOCKS = 1
    SEQUENCE_LOCK_TIME = 2
    ABS_LOCK_BLOCKS = 3
    ABS_LOCK_TIME = 4


class KeyTypes(IntEnum):
    KBVL = 1
    KBSL = 2
    KAL = 3
    KBVF = 4
    KBSF = 5
    KAF = 6


class MessageTypes(IntEnum):
    OFFER = auto()
    BID = auto()
    BID_ACCEPT = auto()

    XMR_OFFER = auto()
    XMR_BID_FL = auto()
    XMR_BID_SPLIT = auto()
    XMR_BID_ACCEPT_LF = auto()
    XMR_BID_TXN_SIGS_FL = auto()
    XMR_BID_LOCK_SPEND_TX_LF = auto()
    XMR_BID_LOCK_RELEASE_LF = auto()
    OFFER_REVOKE = auto()

    ADS_BID_LF = auto()
    ADS_BID_ACCEPT_FL = auto()


class AddressTypes(IntEnum):
    OFFER = auto()
    BID = auto()
    RECV_OFFER = auto()
    SEND_OFFER = auto()


class SwapTypes(IntEnum):
    SELLER_FIRST = auto()
    BUYER_FIRST = auto()
    SELLER_FIRST_2MSG = auto()
    BUYER_FIRST_2MSG = auto()
    XMR_SWAP = auto()
    XMR_BCH_SWAP = auto()


class OfferStates(IntEnum):
    OFFER_SENT = 1
    OFFER_RECEIVED = 2
    OFFER_ABANDONED = 3
    OFFER_EXPIRED = 4


class BidStates(IntEnum):
    BID_SENT = 1
    BID_RECEIVING = 2  # Partially received
    BID_RECEIVED = 3
    BID_RECEIVING_ACC = 4  # Partially received accept message
    BID_ACCEPTED = 5  # BidAcceptMessage received/sent
    SWAP_INITIATED = 6  # Initiate txn validated
    SWAP_PARTICIPATING = 7  # Participate txn validated
    SWAP_COMPLETED = 8  # All swap txns spent
    XMR_SWAP_SCRIPT_COIN_LOCKED = 9
    XMR_SWAP_HAVE_SCRIPT_COIN_SPEND_TX = 10
    XMR_SWAP_NOSCRIPT_COIN_LOCKED = 11
    XMR_SWAP_LOCK_RELEASED = 12
    XMR_SWAP_SCRIPT_TX_REDEEMED = 13
    XMR_SWAP_SCRIPT_TX_PREREFUND = 14  # script txo moved into pre-refund tx
    XMR_SWAP_NOSCRIPT_TX_REDEEMED = 15
    XMR_SWAP_NOSCRIPT_TX_RECOVERED = 16
    XMR_SWAP_FAILED_REFUNDED = 17
    XMR_SWAP_FAILED_SWIPED = 18
    XMR_SWAP_FAILED = 19
    SWAP_DELAYING = 20
    SWAP_TIMEDOUT = 21
    BID_ABANDONED = 22  # Bid will no longer be processed
    BID_ERROR = 23  # An error occurred
    BID_STALLED_FOR_TEST = 24
    BID_REJECTED = 25
    BID_STATE_UNKNOWN = 26
    XMR_SWAP_MSG_SCRIPT_LOCK_TX_SIGS = 27  # XmrBidLockTxSigsMessage
    XMR_SWAP_MSG_SCRIPT_LOCK_SPEND_TX = 28  # XmrBidLockSpendTxMessage
    BID_REQUEST_SENT = 29
    BID_REQUEST_ACCEPTED = 30
    BID_EXPIRED = 31
    BID_AACCEPT_DELAY = 32
    BID_AACCEPT_FAIL = 33


class TxStates(IntEnum):
    TX_NONE = auto()
    TX_SENT = auto()
    TX_CONFIRMED = auto()
    TX_REDEEMED = auto()
    TX_REFUNDED = auto()
    TX_IN_MEMPOOL = auto()
    TX_IN_CHAIN = auto()


class TxTypes(IntEnum):
    ITX = auto()
    PTX = auto()
    ITX_REDEEM = auto()
    ITX_REFUND = auto()
    PTX_REDEEM = auto()
    PTX_REFUND = auto()

    XMR_SWAP_A_LOCK = auto()
    XMR_SWAP_A_LOCK_SPEND = auto()
    XMR_SWAP_A_LOCK_REFUND = auto()
    XMR_SWAP_A_LOCK_REFUND_SPEND = auto()
    XMR_SWAP_A_LOCK_REFUND_SWIPE = auto()
    XMR_SWAP_B_LOCK = auto()
    XMR_SWAP_B_LOCK_SPEND = auto()
    XMR_SWAP_B_LOCK_REFUND = auto()

    ITX_PRE_FUNDED = auto()

    BCH_MERCY = auto()


class ActionTypes(IntEnum):
    ACCEPT_BID = auto()
    ACCEPT_XMR_BID = auto()
    SIGN_XMR_SWAP_LOCK_TX_A = auto()
    SEND_XMR_SWAP_LOCK_TX_A = auto()
    SEND_XMR_SWAP_LOCK_TX_B = auto()
    SEND_XMR_LOCK_RELEASE = auto()
    REDEEM_XMR_SWAP_LOCK_TX_A = auto()  # Follower
    REDEEM_XMR_SWAP_LOCK_TX_B = auto()  # Leader
    RECOVER_XMR_SWAP_LOCK_TX_B = auto()
    SEND_XMR_SWAP_LOCK_SPEND_MSG = auto()
    REDEEM_ITX = auto()
    ACCEPT_AS_REV_BID = auto()


class EventLogTypes(IntEnum):
    FAILED_TX_B_LOCK_PUBLISH = auto()
    LOCK_TX_A_PUBLISHED = auto()
    LOCK_TX_B_PUBLISHED = auto()
    FAILED_TX_B_SPEND = auto()
    LOCK_TX_A_SEEN = auto()
    LOCK_TX_A_CONFIRMED = auto()
    LOCK_TX_B_SEEN = auto()
    LOCK_TX_B_CONFIRMED = auto()
    DEBUG_TWEAK_APPLIED = auto()
    FAILED_TX_B_REFUND = auto()
    LOCK_TX_B_INVALID = auto()
    LOCK_TX_A_REFUND_TX_PUBLISHED = auto()
    LOCK_TX_A_REFUND_SPEND_TX_PUBLISHED = auto()
    LOCK_TX_A_REFUND_SWIPE_TX_PUBLISHED = auto()
    LOCK_TX_B_REFUND_TX_PUBLISHED = auto()
    SYSTEM_WARNING = auto()
    LOCK_TX_A_SPEND_TX_PUBLISHED = auto()
    LOCK_TX_B_SPEND_TX_PUBLISHED = auto()
    LOCK_TX_A_REFUND_TX_SEEN = auto()
    LOCK_TX_A_REFUND_SPEND_TX_SEEN = auto()
    ERROR = auto()
    AUTOMATION_CONSTRAINT = auto()
    AUTOMATION_ACCEPTING_BID = auto()
    ITX_PUBLISHED = auto()
    ITX_REDEEM_PUBLISHED = auto()
    ITX_REFUND_PUBLISHED = auto()
    PTX_PUBLISHED = auto()
    PTX_REDEEM_PUBLISHED = auto()
    PTX_REFUND_PUBLISHED = auto()
    LOCK_TX_B_IN_MEMPOOL = auto()
    BCH_MERCY_TX_PUBLISHED = auto()
    BCH_MERCY_TX_FOUND = auto()


class XmrSplitMsgTypes(IntEnum):
    BID = auto()
    BID_ACCEPT = auto()


class DebugTypes(IntEnum):
    NONE = 0
    BID_STOP_AFTER_COIN_A_LOCK = auto()
    BID_DONT_SPEND_COIN_A_LOCK_REFUND = auto()
    BID_DONT_SPEND_COIN_A_LOCK_REFUND2 = auto()  # continues
    CREATE_INVALID_COIN_B_LOCK = auto()
    BUYER_STOP_AFTER_ITX = auto()
    MAKE_INVALID_PTX = auto()
    DONT_SPEND_ITX = auto()
    SKIP_LOCK_TX_REFUND = auto()
    SEND_LOCKED_XMR = auto()
    B_LOCK_TX_MISSED_SEND = auto()
    DUPLICATE_ACTIONS = auto()
    DONT_CONFIRM_PTX = auto()
    OFFER_LOCK_2_VALUE_INC = auto()
    BID_STOP_AFTER_COIN_B_LOCK = auto()
    BID_DONT_SPEND_COIN_B_LOCK = auto()
    WAIT_FOR_COIN_B_LOCK_BEFORE_REFUND = auto()
    BID_DONT_SPEND_COIN_A_LOCK = auto()


class NotificationTypes(IntEnum):
    NONE = 0
    OFFER_RECEIVED = auto()
    BID_RECEIVED = auto()
    BID_ACCEPTED = auto()


class AutomationOverrideOptions(IntEnum):
    DEFAULT = 0
    ALWAYS_ACCEPT = 1
    NEVER_ACCEPT = auto()


def strAutomationOverrideOption(option):
    if option == AutomationOverrideOptions.DEFAULT:
        return "Default"
    if option == AutomationOverrideOptions.ALWAYS_ACCEPT:
        return "Always Accept"
    if option == AutomationOverrideOptions.NEVER_ACCEPT:
        return "Never Accept"
    return "Unknown"


class VisibilityOverrideOptions(IntEnum):
    DEFAULT = 0
    HIDE = 1
    BLOCK = auto()


def strVisibilityOverrideOption(option):
    if option == VisibilityOverrideOptions.DEFAULT:
        return "Default"
    if option == VisibilityOverrideOptions.HIDE:
        return "Hide"
    if option == VisibilityOverrideOptions.BLOCK:
        return "Block"
    return "Unknown"


def strOfferState(state):
    if state == OfferStates.OFFER_SENT:
        return "Sent"
    if state == OfferStates.OFFER_RECEIVED:
        return "Received"
    if state == OfferStates.OFFER_ABANDONED:
        return "Abandoned"
    if state == OfferStates.OFFER_EXPIRED:
        return "Expired"
    return "Unknown"


def strBidState(state):
    if state == BidStates.BID_SENT:
        return "Sent"
    if state == BidStates.BID_RECEIVING:
        return "Receiving"
    if state == BidStates.BID_RECEIVING_ACC:
        return "Receiving accept"
    if state == BidStates.BID_RECEIVED:
        return "Received"
    if state == BidStates.BID_ACCEPTED:
        return "Accepted"
    if state == BidStates.SWAP_INITIATED:
        return "Initiated"
    if state == BidStates.SWAP_PARTICIPATING:
        return "Participating"
    if state == BidStates.SWAP_COMPLETED:
        return "Completed"
    if state == BidStates.SWAP_TIMEDOUT:
        return "Timed-out"
    if state == BidStates.BID_ABANDONED:
        return "Abandoned"
    if state == BidStates.BID_STALLED_FOR_TEST:
        return "Stalled (debug)"
    if state == BidStates.BID_ERROR:
        return "Error"
    if state == BidStates.BID_REJECTED:
        return "Rejected"
    if state == BidStates.XMR_SWAP_SCRIPT_COIN_LOCKED:
        return "Script coin locked"
    if state == BidStates.XMR_SWAP_HAVE_SCRIPT_COIN_SPEND_TX:
        return "Script coin spend tx valid"
    if state == BidStates.XMR_SWAP_NOSCRIPT_COIN_LOCKED:
        return "Scriptless coin locked"
    if state == BidStates.XMR_SWAP_LOCK_RELEASED:
        return "Script coin lock released"
    if state == BidStates.XMR_SWAP_SCRIPT_TX_REDEEMED:
        return "Script tx redeemed"
    if state == BidStates.XMR_SWAP_SCRIPT_TX_PREREFUND:
        return "Script pre-refund tx in chain"
    if state == BidStates.XMR_SWAP_NOSCRIPT_TX_REDEEMED:
        return "Scriptless tx redeemed"
    if state == BidStates.XMR_SWAP_NOSCRIPT_TX_RECOVERED:
        return "Scriptless tx recovered"
    if state == BidStates.XMR_SWAP_FAILED_REFUNDED:
        return "Failed, refunded"
    if state == BidStates.XMR_SWAP_FAILED_SWIPED:
        return "Failed, swiped"
    if state == BidStates.XMR_SWAP_FAILED:
        return "Failed"
    if state == BidStates.SWAP_DELAYING:
        return "Delaying"
    if state == BidStates.XMR_SWAP_MSG_SCRIPT_LOCK_TX_SIGS:
        return "Exchanged script lock tx sigs msg"
    if state == BidStates.XMR_SWAP_MSG_SCRIPT_LOCK_SPEND_TX:
        return "Exchanged script lock spend tx msg"
    if state == BidStates.BID_REQUEST_SENT:
        return "Request sent"
    if state == BidStates.BID_REQUEST_ACCEPTED:
        return "Request accepted"
    if state == BidStates.BID_STATE_UNKNOWN:
        return "Unknown bid state"
    if state == BidStates.BID_EXPIRED:
        return "Expired"
    if state == BidStates.BID_AACCEPT_DELAY:
        return "Auto accept delay"
    if state == BidStates.BID_AACCEPT_FAIL:
        return "Auto accept failed"
    return "Unknown" + " " + str(state)


def strTxState(state):
    if state == TxStates.TX_NONE:
        return "None"
    if state == TxStates.TX_SENT:
        return "Sent"
    if state == TxStates.TX_CONFIRMED:
        return "Confirmed"
    if state == TxStates.TX_REDEEMED:
        return "Redeemed"
    if state == TxStates.TX_REFUNDED:
        return "Refunded"
    if state == TxStates.TX_IN_MEMPOOL:
        return "In Mempool"
    if state == TxStates.TX_IN_CHAIN:
        return "In Chain"
    return "Unknown"


def strTxType(tx_type):
    if tx_type == TxTypes.XMR_SWAP_A_LOCK:
        return "Chain A Lock Tx"
    if tx_type == TxTypes.XMR_SWAP_A_LOCK_SPEND:
        return "Chain A Lock Spend Tx"
    if tx_type == TxTypes.XMR_SWAP_A_LOCK_REFUND:
        return "Chain A Lock Refund Tx"
    if tx_type == TxTypes.XMR_SWAP_A_LOCK_REFUND_SPEND:
        return "Chain A Lock Refund Spend Tx"
    if tx_type == TxTypes.XMR_SWAP_A_LOCK_REFUND_SWIPE:
        return "Chain A Lock Refund Swipe Tx"
    if tx_type == TxTypes.XMR_SWAP_B_LOCK:
        return "Chain B Lock Tx"
    if tx_type == TxTypes.ITX_PRE_FUNDED:
        return "Funded mock initiate Tx"
    if tx_type == TxTypes.BCH_MERCY:
        return "BCH Mercy Tx"
    return "Unknown"


def strAddressType(addr_type):
    if addr_type == AddressTypes.OFFER:
        return "Offer"
    if addr_type == AddressTypes.BID:
        return "Bid"
    if addr_type == AddressTypes.RECV_OFFER:
        return "Offer recv"
    if addr_type == AddressTypes.SEND_OFFER:
        return "Offer send"
    return "Unknown"


def getLockName(lock_type):
    if lock_type == TxLockTypes.SEQUENCE_LOCK_BLOCKS:
        return "Sequence lock, blocks"
    if lock_type == TxLockTypes.SEQUENCE_LOCK_TIME:
        return "Sequence lock, time"
    if lock_type == TxLockTypes.ABS_LOCK_BLOCKS:
        return "blocks"
    if lock_type == TxLockTypes.ABS_LOCK_TIME:
        return "time"


def describeEventEntry(event_type, event_msg):
    if event_type == EventLogTypes.FAILED_TX_B_LOCK_PUBLISH:
        return "Failed to publish lock tx B"
    if event_type == EventLogTypes.LOCK_TX_A_PUBLISHED:
        return "Lock tx A published"
    if event_type == EventLogTypes.LOCK_TX_B_PUBLISHED:
        return "Lock tx B published"
    if event_type == EventLogTypes.FAILED_TX_B_SPEND:
        return "Failed to publish lock tx B spend: " + event_msg
    if event_type == EventLogTypes.LOCK_TX_A_SEEN:
        return "Lock tx A seen in chain"
    if event_type == EventLogTypes.LOCK_TX_A_CONFIRMED:
        return "Lock tx A confirmed in chain"
    if event_type == EventLogTypes.LOCK_TX_B_SEEN:
        return "Lock tx B seen in chain"
    if event_type == EventLogTypes.LOCK_TX_B_CONFIRMED:
        return "Lock tx B confirmed in chain"
    if event_type == EventLogTypes.LOCK_TX_B_IN_MEMPOOL:
        return "Lock tx B seen in mempool"
    if event_type == EventLogTypes.DEBUG_TWEAK_APPLIED:
        return "Debug tweak applied " + event_msg
    if event_type == EventLogTypes.FAILED_TX_B_REFUND:
        return "Failed to publish lock tx B refund"
    if event_type == EventLogTypes.LOCK_TX_B_INVALID:
        return "Detected invalid lock Tx B"
    if event_type == EventLogTypes.LOCK_TX_A_REFUND_TX_PUBLISHED:
        return "Lock tx A refund tx published"
    if event_type == EventLogTypes.LOCK_TX_A_REFUND_SPEND_TX_PUBLISHED:
        return "Lock tx A refund spend tx published"
    if event_type == EventLogTypes.LOCK_TX_A_REFUND_SWIPE_TX_PUBLISHED:
        return "Lock tx A refund swipe tx published"
    if event_type == EventLogTypes.LOCK_TX_B_REFUND_TX_PUBLISHED:
        return "Lock tx B refund tx published"
    if event_type == EventLogTypes.LOCK_TX_A_SPEND_TX_PUBLISHED:
        return "Lock tx A spend tx published"
    if event_type == EventLogTypes.LOCK_TX_B_SPEND_TX_PUBLISHED:
        return "Lock tx B spend tx published"
    if event_type == EventLogTypes.LOCK_TX_A_REFUND_TX_SEEN:
        return "Lock tx A refund tx seen in chain"
    if event_type == EventLogTypes.LOCK_TX_A_REFUND_SPEND_TX_SEEN:
        return "Lock tx A refund spend tx seen in chain"
    if event_type == EventLogTypes.SYSTEM_WARNING:
        return "Warning: " + event_msg
    if event_type == EventLogTypes.ERROR:
        return "Error: " + event_msg
    if event_type == EventLogTypes.AUTOMATION_CONSTRAINT:
        return "Failed auto accepting"
    if event_type == EventLogTypes.AUTOMATION_ACCEPTING_BID:
        return "Auto accepting"
    if event_type == EventLogTypes.ITX_PUBLISHED:
        return "Initiate tx published"
    if event_type == EventLogTypes.ITX_REDEEM_PUBLISHED:
        return "Initiate tx redeem tx published"
    if event_type == EventLogTypes.ITX_REFUND_PUBLISHED:
        return "Initiate tx refund tx published"
    if event_type == EventLogTypes.PTX_PUBLISHED:
        return "Participate tx published"
    if event_type == EventLogTypes.PTX_REDEEM_PUBLISHED:
        return "Participate tx redeem tx published"
    if event_type == EventLogTypes.PTX_REFUND_PUBLISHED:
        return "Participate tx refund tx published"
    if event_type == EventLogTypes.BCH_MERCY_TX_FOUND:
        return "BCH mercy tx found"
    if event_type == EventLogTypes.BCH_MERCY_TX_PUBLISHED:
        return "Lock tx B mercy tx published"


def getVoutByAddress(txjs, p2sh):
    for o in txjs["vout"]:
        try:
            if "address" in o["scriptPubKey"] and o["scriptPubKey"]["address"] == p2sh:
                return o["n"]
            if p2sh in o["scriptPubKey"]["addresses"]:
                return o["n"]
        except Exception:
            pass
    raise ValueError("Address output not found in txn")


def getVoutByScriptPubKey(txjs, scriptPubKey_hex: str) -> int:
    for o in txjs["vout"]:
        try:
            if scriptPubKey_hex == o["scriptPubKey"]["hex"]:
                return o["n"]
        except Exception:
            pass
    raise ValueError("scriptPubKey output not found in txn")


def replaceAddrPrefix(addr, coin_type, chain_name, addr_type="pubkey_address"):
    return encodeAddress(
        bytes((chainparams[coin_type][chain_name][addr_type],))
        + decodeAddress(addr)[1:]
    )


def getOfferProofOfFundsHash(offer_msg, offer_addr):
    # TODO: Hash must not include proof_of_funds sig if it exists in offer_msg
    h = hashlib.sha256()
    h.update(offer_addr.encode("utf-8"))
    offer_bytes = offer_msg.to_bytes()
    h.update(offer_bytes)
    return h.digest()


def getLastBidState(packed_states):
    num_states = len(packed_states) // 12
    if num_states < 2:
        return BidStates.BID_STATE_UNKNOWN
    return struct.unpack_from("<i", packed_states[(num_states - 2) * 12 :])[0]
    try:
        num_states = len(packed_states) // 12
        if num_states < 2:
            return BidStates.BID_STATE_UNKNOWN
        return struct.unpack_from("<i", packed_states[(num_states - 2) * 12 :])[0]
    except Exception:
        return BidStates.BID_STATE_UNKNOWN


def strSwapType(swap_type):
    if swap_type == SwapTypes.SELLER_FIRST:
        return "seller_first"
    if swap_type == SwapTypes.XMR_SWAP:
        return "xmr_swap"
    return None


def strSwapDesc(swap_type):
    if swap_type == SwapTypes.SELLER_FIRST:
        return "Secret Hash"
    if swap_type == SwapTypes.XMR_SWAP:
        return "Adaptor Sig"
    return None


inactive_states = [
    BidStates.SWAP_COMPLETED,
    BidStates.BID_ERROR,
    BidStates.BID_REJECTED,
    BidStates.SWAP_TIMEDOUT,
    BidStates.BID_ABANDONED,
    BidStates.BID_EXPIRED,
]


def canAcceptBidState(state):
    return state in (
        BidStates.BID_RECEIVED,
        BidStates.BID_AACCEPT_DELAY,
        BidStates.BID_AACCEPT_FAIL,
    )


def isActiveBidState(state):
    if state >= BidStates.BID_ACCEPTED and state < BidStates.SWAP_COMPLETED:
        return True
    return state in (
        BidStates.SWAP_DELAYING,
        BidStates.XMR_SWAP_HAVE_SCRIPT_COIN_SPEND_TX,
        BidStates.XMR_SWAP_SCRIPT_COIN_LOCKED,
        BidStates.XMR_SWAP_NOSCRIPT_COIN_LOCKED,
        BidStates.XMR_SWAP_LOCK_RELEASED,
        BidStates.XMR_SWAP_NOSCRIPT_TX_REDEEMED,
        BidStates.XMR_SWAP_SCRIPT_TX_REDEEMED,
        BidStates.XMR_SWAP_SCRIPT_TX_PREREFUND,
        BidStates.XMR_SWAP_MSG_SCRIPT_LOCK_TX_SIGS,
        BidStates.XMR_SWAP_MSG_SCRIPT_LOCK_SPEND_TX,
        BidStates.XMR_SWAP_FAILED,
        BidStates.BID_REQUEST_ACCEPTED,
    )


def isErrorBidState(state):
    return state in (
        BidStates.BID_STALLED_FOR_TEST,
        BidStates.BID_ERROR,
    )


def isFailingBidState(state):
    return state in (
        BidStates.BID_STALLED_FOR_TEST,
        BidStates.BID_ERROR,
        BidStates.XMR_SWAP_SCRIPT_TX_PREREFUND,
        BidStates.XMR_SWAP_NOSCRIPT_TX_RECOVERED,
        BidStates.XMR_SWAP_FAILED_REFUNDED,
        BidStates.XMR_SWAP_FAILED_SWIPED,
        BidStates.XMR_SWAP_FAILED,
    )


def isFinalBidState(state):
    return state in inactive_states