Bloxroute bsc rescue

bscrescue.py

#!/usr/bin/env python3
# Python 3.7 or higher required due to the use of asyncio.run()
import argparse
import asyncio, json, ssl
import math
import os
import json
import re

import websocket
import web3
from eth_account import Account
from eth_account.signers.local import LocalAccount
from eth_typing import ChecksumAddress
from eth_utils import to_hex, to_checksum_address, to_wei, from_wei
import dotenv
from web3.types import TxParams

from lib.style import PrettyText

dotenv.load_dotenv()
AUTH_HEADER = os.environ.get('BLOX_AUTH')
EIP20_ABI = json.loads('[{"constant":true,"inputs":[],"name":"name","outputs":[{"name":"","type":"string"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":false,"inputs":[{"name":"_spender","type":"address"},{"name":"_value","type":"uint256"}],"name":"approve","outputs":[{"name":"","type":"bool"}],"payable":false,"stateMutability":"nonpayable","type":"function"},{"constant":true,"inputs":[],"name":"totalSupply","outputs":[{"name":"","type":"uint256"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":false,"inputs":[{"name":"_from","type":"address"},{"name":"_to","type":"address"},{"name":"_value","type":"uint256"}],"name":"transferFrom","outputs":[{"name":"","type":"bool"}],"payable":false,"stateMutability":"nonpayable","type":"function"},{"constant":true,"inputs":[],"name":"decimals","outputs":[{"name":"","type":"uint8"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[{"name":"_owner","type":"address"}],"name":"balanceOf","outputs":[{"name":"","type":"uint256"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[],"name":"symbol","outputs":[{"name":"","type":"string"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":false,"inputs":[{"name":"_to","type":"address"},{"name":"_value","type":"uint256"}],"name":"transfer","outputs":[{"name":"","type":"bool"}],"payable":false,"stateMutability":"nonpayable","type":"function"},{"constant":true,"inputs":[{"name":"_owner","type":"address"},{"name":"_spender","type":"address"}],"name":"allowance","outputs":[{"name":"","type":"uint256"}],"payable":false,"stateMutability":"view","type":"function"},{"anonymous":false,"inputs":[{"indexed":true,"name":"_from","type":"address"},{"indexed":true,"name":"_to","type":"address"},{"indexed":false,"name":"_value","type":"uint256"}],"name":"Transfer","type":"event"},{"anonymous":false,"inputs":[{"indexed":true,"name":"_owner","type":"address"},{"indexed":true,"name":"_spender","type":"address"},{"indexed":false,"name":"_value","type":"uint256"}],"name":"Approval","type":"event"}]')
BEP20_ABI = json.loads("""[{"anonymous": false,"inputs": [{"indexed": true,"internalType": "address","name": "owner","type": "address"},{"indexed": true,"internalType": "address","name": "spender","type": "address"},{"indexed": false,"internalType": "uint256","name": "value","type": "uint256"}],"name": "Approval","type": "event"},{"anonymous": false,"inputs": [{"indexed": true,"internalType": "address","name": "from","type": "address"},{"indexed": true,"internalType": "address","name": "to","type": "address"},{"indexed": false,"internalType": "uint256","name": "value","type": "uint256"}],"name": "Transfer","type": "event"},{"constant": true,"inputs": [{"internalType": "address","name": "_owner","type": "address"},{"internalType": "address","name": "spender","type": "address"}],"name": "allowance","outputs": [{"internalType": "uint256","name": "","type": "uint256"}],"payable": false,"stateMutability": "view","type": "function"},{"constant": false,"inputs": [{"internalType": "address","name": "spender","type": "address"},{"internalType": "uint256","name": "amount","type": "uint256"}],"name": "approve","outputs": [{"internalType": "bool","name": "","type": "bool"}],"payable": false,"stateMutability": "nonpayable","type": "function"},{"constant": true,"inputs": [{"internalType": "address","name": "account","type": "address"}],"name": "balanceOf","outputs": [{"internalType": "uint256","name": "","type": "uint256"}],"payable": false,"stateMutability": "view","type": "function"},{"constant": true,"inputs": [],"name": "decimals","outputs": [{"internalType": "uint256","name": "","type": "uint256"}],"payable": false,"stateMutability": "view","type": "function"},{"constant": true,"inputs": [],"name": "getOwner","outputs": [{"internalType": "address","name": "","type": "address"}],"payable": false,"stateMutability": "view","type": "function"},{"constant": true,"inputs": [],"name": "name","outputs": [{"internalType": "string","name": "","type": "string"}],"payable": false,"stateMutability": "view","type": "function"},{"constant": true,"inputs": [],"name": "symbol","outputs": [{"internalType": "string","name": "","type": "string"}],"payable": false,"stateMutability": "view","type": "function"},{"constant": true,"inputs": [],"name": "totalSupply","outputs": [{"internalType": "uint256","name": "","type": "uint256"}],"payable": false,"stateMutability": "view","type": "function"},{"constant": false,"inputs": [{"internalType": "address","name": "recipient","type": "address"},{"internalType": "uint256","name": "amount","type": "uint256"}],"name": "transfer","outputs": [{"internalType": "bool","name": "","type": "bool"}],"payable": false,"stateMutability": "nonpayable","type": "function"},{"constant": false,"inputs": [{"internalType": "address","name": "sender","type": "address"},{"internalType": "address","name": "recipient","type": "address"},{"internalType": "uint256","name": "amount","type": "uint256"}],"name": "transferFrom","outputs": [{"internalType": "bool","name": "","type": "bool"}],"payable": false,"stateMutability": "nonpayable","type": "function"}]""")
assert AUTH_HEADER is not None


class Helpers:
    @staticmethod
    def load_json(file: str) -> dict:
        if os.path.exists(file):
            with open(file, 'r')as f:
                return json.load(f)
        return {}


class Web3ConnectionCheck:
    def __init__(self, w3: web3.Web3, name: str = None):
        self.name = name
        self._w3: web3.Web3 = w3
        self.cid: int = 0
        self.connected = False
        self.check()

    def check(self):
        if hasattr(self._w3, 'is_connected'):
            connected = self._w3.is_connected()
        else:
            connected = self._w3.isConnected()

        if connected:
            self.connected = True
            self.cid = self._w3.eth.chain_id
            printer.good(f'Web3 Connected: to chain with ID: {self.cid}')
            return True
        else:
            printer.warning('Web3 is not connected!')
            return False

    @property
    def w3(self):
        return self._w3


class Web3Connector:
    def __init__(self, chain: str, endpoint: str = None, name: str = None):
        self.chain = chain
        self.endpoint = endpoint
        self.name = name
        self._w3: (web3.Web3, None) = None

    def get_endpoint(self, chain: str):
        _endpoint = None
        if self.chain and not self.endpoint:
            endpoint = os.environ.get('%s_ws_endpoint' % chain)
            if endpoint:
                _endpoint = endpoint
            else:
                endpoint = os.environ.get('%s_http_endpoint' % chain)
                if endpoint:
                    _endpoint = endpoint
        return _endpoint

    def create_w3(self, endpoint_: str) -> (web3.Web3, None):
        _w3 = None
        if re.match('http', endpoint_):
            _w3 = web3.Web3(web3.HTTPProvider(endpoint_))
        if re.match('wss', endpoint_):
            _w3 = web3.Web3(web3.WebsocketProvider(endpoint_))
        if _w3 is not None:
            self._w3 = _w3
            return _w3
        return None

    def setup(self) -> (web3.Web3, None):
        """
        Attempts to configure a web3 connection.
        :return: Web3, None
        """
        dotenv.load_dotenv()
        if not self.endpoint:
            self.endpoint = self.get_endpoint(self.chain)
        assert self.endpoint is not None
        _w3_ = self.create_w3(self.endpoint)
        w3_check = Web3ConnectionCheck(_w3_, self.name)
        if w3_check.connected:
            return _w3_


class BloxRescue:
    def __init__(self, w3: web3.Web3, sender_account: LocalAccount, target_account: LocalAccount,
                 token_address: ChecksumAddress, gas_multiplier: float = 1.1):
        self.w3 = w3
        if w3.eth.chain_id == 56:
            self.token_object = w3.eth.contract(address=to_checksum_address(token_address), abi=BEP20_ABI)
        else:
            self.token_object = w3.eth.contract(address=to_checksum_address(token_address), abi=EIP20_ABI)
        self.sender_account = sender_account
        self.target_account = target_account
        self.gas_multiplier = gas_multiplier
        self.token_transfer_gas_est: int = 0

    def calculate_bribe_gwei(self, units: int = 21000):
        ret = 0
        gwei = 0
        while ret < 0.004:
            gwei += 10
            ret = float((from_wei(units * gwei, 'gwei')))

        return gwei

    def calculate_tx_fee(self, units: int = 21000, raw=False) -> (float, int):
        abs_low_fee = float(from_wei(units * int(from_wei(self.w3.eth.gas_price, 'gwei')), 'gwei'))
        if not raw:
            return abs_low_fee * self.gas_multiplier
        return int(math.ceil(to_wei(abs_low_fee * self.gas_multiplier, 'ether')))

    def create_eth_transfer_tx(self, gas_units_for_token_transfer: int) -> dict:
        fee_ether: float = self.calculate_tx_fee(gas_units_for_token_transfer, False)
        tx1: TxParams = {
            'to': self.target_account.address,
            "value": to_wei(fee_ether, "ether"),
            "gas": 21000,
            "gasPrice": to_wei(5, 'gwei'),
            # "maxFeePerGas": to_wei(200, "gwei"),
            # "maxPriorityFeePerGas": to_wei(50, "gwei"),
            'nonce': w3.eth.get_transaction_count(self.sender_account.address),
            'chainId': to_hex(self.w3.eth.chain_id),
            # 'type': 2
        }
        tx1_signed = w3.eth.account.sign_transaction(tx1, self.sender_account.key)
        raw_tx = tx1_signed.rawTransaction.hex()
        return raw_tx

    def create_token_transfer_tx(self):
        printer.normal(f'Checking token balance of target {self.target_account.address} ... ')
        token_balance = self.token_object.functions.balanceOf(self.token_object.address).call()
        printer.normal('Simulating a transfer ... ')
        try:
            self.token_object.functions.transfer(self.sender_account.address, token_balance).call(
                {'from': self.target_account.address})
        except web3.exceptions.ContractLogicError as err:
            printer.error(f'Transfer will fail because execution will revert because: {err}')
            exit(1)
        else:
            gas_est = self.token_object.functions.transfer(self.sender_account.address, token_balance).estimate_gas()
            self.token_transfer_gas_est = gas_est

        # Tx2: withdraw 1 KEEY token
        tx2 = self.token_object.functions.transfer(self.sender_account.address, token_balance).build_transaction({
            'chainId': to_hex(5),
            'gas': gas_est,  # High gas for miner reward
            "maxFeePerGas": to_wei(200, "gwei"),
            "maxPriorityFeePerGas": to_wei(50, "gwei"),
            'nonce': w3.eth.get_transaction_count(eth_receiver.address),
        })
        tx2_signed = w3.eth.account.sign_transaction(tx2, eth_receiver.key)
        raw_tx = tx2_signed.rawTransaction.hex()
        return raw_tx

    def create_transaction_list(self):
        raw_tx_2 = self.create_token_transfer_tx()
        raw_tx_1 = self.create_eth_transfer_tx(self.token_transfer_gas_est)
        return [raw_tx_1, raw_tx_2]

    async def main(self):
        txs = self.create_token_transfer_tx()
        try:
            ws = websocket.create_connection('wss://api.blxrbdn.com/ws',
                                             header=["Authorization:{}".format(AUTH_HEADER)],
                                             sslopt={"cert_reqs": ssl.CERT_NONE})
            # ETH Example
            # request = json.dumps({"id": 1, "method": "blxr_tx", "params": {"transaction": "f86b0184...e0b58219"}})

            # BSC Example
            request = json.dumps({"id": 1, "method": "blxr_tx",
                                  "params": {"transaction": txs, "blockchain_network": "BSC-Mainnet",
                                             "block_number": to_hex(w3.eth.block_number + 10)}})

            ws.send(str(request))
            while True:
                response = json.loads(ws.recv())
                print(response)  # or process it generally
        except Exception as e:
            print(f'Connection failed, Reason: {e}')








if __name__ == '__main__':

    args = argparse.ArgumentParser()
    args.add_argument('-w', '--wallet', type=str, default='keys/default_wallet.json',
                      help='Json wallet file for gas account.')
    args.add_argument('-k', '--key', type=str, help='Key of target account', required=True)
    args.add_argument('-t', '--token', type=str, help='Token to move.')
    """ example wallet.json file
    {
      "wallet": {
      "address":  "0xEFEFEFEEFFFFFFFFFFFFFFFFFFFFFFFFFFF",
      "private_key": "094039403943940390954394039409545905993094340903403494923493049"
      }
    }
    """
    args = args.parse_args()
    dotenv.load_dotenv()
    printer = PrettyText()
    w3: web3.Web3 = Web3Connector('bsc', None, 'normal rpc').setup()
    priv_key = Helpers().load_json(args.wallet).get('wallet').get('private_key')
    eth_sender: LocalAccount = Account.from_key(priv_key)
    eth_receiver: LocalAccount = Account.from_key(args.key)
    printer.normal(f"Gas Account Address: {eth_sender.address}")
    printer.normal(f"Target address: {eth_receiver.address}")
    blox = BloxRescue(w3, eth_sender, eth_receiver, args.token, 1.1)
    asyncio.run(blox.main())