Cherry pick files from old branch

proxy/indexer/airdropper.py

@@ -0,0 +1,167 @@
+from proxy.indexer.indexer_base import IndexerBase, logger
+import os
+import requests
+import base58
+import json
+import logging
+
+try:
+    from utils import check_error
+    from sql_dict import SQLDict
+except ImportError:
+    from .utils import check_error
+    from .sql_dict import SQLDict
+
+class Airdropper(IndexerBase):
+    def __init__(self,
+                 solana_url,
+                 evm_loader_id,
+                 faucet_url = '',
+                 wrapper_whitelist = [],
+                 airdrop_amount = 10,
+                 log_level = 'INFO'):
+        IndexerBase.__init__(self, solana_url, evm_loader_id, log_level)
+
+        # collection of eth-address-to-create-accout-trx mappings
+        # for every addresses that was already funded with airdrop
+        self.airdrop_ready = SQLDict(tablename="airdrop_ready")
+        self.wrapper_whitelist = wrapper_whitelist
+        self.airdrop_amount = airdrop_amount
+        self.faucet_url = faucet_url
+
+
+    # helper function checking if given contract address is in whitelist
+    def _is_allowed_wrapper_contract(self, contract_addr):
+        return contract_addr in self.wrapper_whitelist
+
+
+    # helper function checking if given 'create account' corresponds to 'create erc20 token account' instruction
+    def _check_create_instr(self, account_keys, create_acc, create_token_acc):
+        # Must use the same Ethereum account
+        if account_keys[create_acc['accounts'][1]] != account_keys[create_token_acc['accounts'][2]]:
+            return False
+        # Must use the same token program
+        if account_keys[create_acc['accounts'][5]] != account_keys[create_token_acc['accounts'][6]]:
+            return False
+        # Token program must be system token program
+        if account_keys[create_acc['accounts'][5]] != 'TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA':
+            return False
+        # CreateERC20TokenAccount instruction must use ERC20-wrapper from whitelist
+        if not self._is_allowed_wrapper_contract(account_keys[create_token_acc['accounts'][3]]):
+            return False
+        return True
+
+
+    # helper function checking if given 'create erc20 token account' corresponds to 'token transfer' instruction
+    def _check_transfer(self, account_keys, create_token_acc, token_transfer) -> bool:
+        return account_keys[create_token_acc['accounts'][1]] == account_keys[token_transfer['accounts'][1]]
+
+
+    def _airdrop_to(self, create_acc):
+        eth_address = "0x" + bytearray(base58.b58decode(create_acc['data'])[20:][:20]).hex()
+
+        if eth_address in self.airdrop_ready:  # transaction already processed
+            return
+        self.airdrop_ready[eth_address] = create_acc
+
+        logger.info(f"Airdrop to address: {eth_address}")
+
+        json_data = { 'wallet': eth_address, 'amount': self.airdrop_amount }
+        resp = requests.post(self.faucet_url + '/request_eth_token', json = json_data)
+        if not resp.ok:
+            logger.warning(f'Failed to airdrop: {resp.status_code}')
+
+
+    def process_trx_airdropper_mode(self, trx, signature):
+        if check_error(trx):
+            return
+
+        # helper function finding all instructions that satisfies predicate
+        def find_instructions(trx, predicate):
+            return [instr for instr in trx['transaction']['message']['instructions'] if predicate(instr)]
+
+        account_keys = trx["transaction"]["message"]["accountKeys"]
+
+        # Finding instructions specific for airdrop.
+        # Airdrop triggers on sequence:
+        # neon.CreateAccount -> neon.CreateERC20TokenAccount -> spl.Transfer (maybe shuffled)
+        # First: select all instructions that can form such chains
+        predicate = lambda instr: account_keys[instr['programIdIndex']] == self.evm_loader_id \
+                                  and base58.b58decode(instr['data'])[0] == 0x02
+        create_acc_list = find_instructions(trx, predicate)
+
+        predicate = lambda  instr: account_keys[instr['programIdIndex']] == self.evm_loader_id \
+                                   and base58.b58decode(instr['data'])[0] == 0x0f
+        create_token_acc_list = find_instructions(trx, predicate)
+
+        predicate = lambda instr: account_keys[instr['programIdIndex']] == 'TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA' \
+                                  and base58.b58decode(instr['data'])[0] == 0x03
+        token_transfer_list = find_instructions(trx, predicate)
+
+        # Second: Find exact chains of instructions in sets created previously
+        for create_acc in create_acc_list:
+            for create_token_acc in create_token_acc_list:
+                if not self._check_create_instr(account_keys, create_acc, create_token_acc):
+                    continue
+                for token_transfer in token_transfer_list:
+                    if not self._check_transfer(account_keys, create_token_acc, token_transfer):
+                        continue
+                    self._airdrop_to(create_acc)
+
+
+    def process_receipts(self):
+        counter = 0
+        for signature in self.transaction_order:
+            counter += 1
+            if signature in self.transaction_receipts:
+                trx = self.transaction_receipts[signature]
+                if trx is None:
+                    logger.error("trx is None")
+                    del self.transaction_receipts[signature]
+                    continue
+                if 'slot' not in trx:
+                    logger.debug("\n{}".format(json.dumps(trx, indent=4, sort_keys=True)))
+                    exit()
+                if trx['transaction']['message']['instructions'] is not None:
+                    self.process_trx_airdropper_mode(trx, signature)
+
+
+def run_airdropper(solana_url,
+                   evm_loader_id,
+                   faucet_url = '',
+                   wrapper_whitelist = [],
+                   airdrop_amount = 10,
+                   log_level = 'INFO'):
+    logging.basicConfig(format='%(asctime)s - pid:%(process)d [%(levelname)-.1s] %(funcName)s:%(lineno)d - %(message)s')
+    logger.setLevel(logging.DEBUG)
+    logger.info(f"""Running indexer with params:
+        solana_url: {solana_url},
+        evm_loader_id: {evm_loader_id},
+        log_level: {log_level},
+        faucet_url: {faucet_url},
+        wrapper_whitelist: {wrapper_whitelist},
+        airdrop_amount: {airdrop_amount}""")
+
+    airdropper = Airdropper(solana_url,
+                            evm_loader_id,
+                            faucet_url,
+                            wrapper_whitelist,
+                            airdrop_amount,
+                            log_level)
+    airdropper.run()
+
+
+if __name__ == "__main__":
+    solana_url = os.environ.get('SOLANA_URL', 'http://localhost:8899')
+    evm_loader_id = os.environ.get('EVM_LOADER_ID', '53DfF883gyixYNXnM7s5xhdeyV8mVk9T4i2hGV9vG9io')
+    faucet_url = os.environ.get('FAUCET_URL', 'http://localhost:3333')
+    wrapper_whitelist = os.environ.get('INDEXER_ERC20_WRAPPER_WHITELIST', '').split(',')
+    airdrop_amount = os.environ.get('AIRDROP_AMOUNT', 0)
+    log_level = os.environ.get('LOG_LEVEL', 'INFO')
+
+    run_airdropper(solana_url,
+                   evm_loader_id,
+                   faucet_url,
+                   wrapper_whitelist,
+                   airdrop_amount,
+                   log_level)

proxy/testing/mock_server.py

@@ -0,0 +1,28 @@
+import requests
+
+from flask import Flask, request
+from threading import Thread
+
+class MockServer(Thread):
+    def __init__(self, port):
+        super().__init__()
+        self.port = port
+        self.app = Flask(__name__)
+        self.url = "http://localhost:%s" % self.port
+        self.app.add_url_rule("/shutdown", view_func=self._shutdown_server)
+
+    def add_url_rule(self, url, callback, methods):
+        self.app.add_url_rule(url, view_func=callback, methods=methods)
+
+    def _shutdown_server(self):
+        if not 'werkzeug.server.shutdown' in request.environ:
+            raise RuntimeError('Not running the development server')
+        request.environ['werkzeug.server.shutdown']()
+        return 'Server shutting down...'
+
+    def run(self):
+        self.app.run(port=self.port)
+
+    def shutdown_server(self):
+        requests.get("http://localhost:%s/shutdown" % self.port)
+        self.join()

proxy/testing/test_airdropper.py

@@ -0,0 +1,135 @@
+import unittest
+from proxy.testing.mock_server import MockServer
+from proxy.indexer.airdropper import run_airdropper
+from multiprocessing import Process
+import time
+from flask import request
+from unittest.mock import MagicMock, patch, call
+from solana.rpc.api import Client
+from solana.rpc.types import RPCResponse
+from solana.rpc.providers.http import HTTPProvider
+from typing import cast
+import itertools
+from proxy.testing.transactions import pre_token_airdrop_trx1, pre_token_airdrop_trx2,\
+    create_sol_acc_and_airdrop_trx, wrapper_whitelist, evm_loader_addr, token_airdrop_address1, \
+    token_airdrop_address2, token_airdrop_address3
+
+class MockFaucet(MockServer):
+    def __init__(self, port):
+        super().__init__(port)
+        self.request_eth_token_mock = MagicMock()
+        self.request_eth_token_mock.side_effect = itertools.repeat({})
+        self.add_url_rule("/request_eth_token", callback=self.request_eth_token, methods=['POST'])
+
+    def request_eth_token(self):
+        req = request.get_json()
+        return self.request_eth_token_mock(req)
+
+
+def create_signature_for_address(signature: str):
+    return {
+        'blockTime': 1638177745, # not make sense
+        'confirmationStatus': 'finalized',
+        'err': None,
+        'memo': None,
+        'signature': signature,
+        'slot': 9748200 # not make sense
+    }
+
+
+def create_get_signatures_for_address(signatures: list):
+    return {
+        'jsonrpc': '2.0',
+        'result': [ create_signature_for_address(sign) for sign in signatures ],
+        'id': 1
+    }
+
+
+class Test_Airdropper(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        print("testing indexer in airdropper mode")
+        cls.address = 'localhost'
+        cls.faucet_port = 3333
+        cls.airdrop_amount = 10
+
+        cls.faucet = MockFaucet(cls.faucet_port)
+        cls.faucet.start()
+        time.sleep(0.2)
+
+    def _run_test(self):
+        indexer = Process(target=run_airdropper,
+                          args=(f'http://localhost:8899',  # solana_url
+                                evm_loader_addr,
+                                f'http://{self.address}:{self.faucet_port}',  # faucet_url
+                                wrapper_whitelist,
+                                self.airdrop_amount,
+                                'INFO'))
+        indexer.start()
+        time.sleep(2)  # make sure airdropper processed event
+        indexer.terminate()
+        indexer.join()
+
+
+    @patch.object(Client, 'get_confirmed_transaction')
+    @patch.object(HTTPProvider, 'make_request')
+    @patch.object(Client, 'get_slot')
+    def test_simple_case_one_account_one_airdrop(self, get_slot, make_request, get_confirmed_transaction):
+        # Return the same slot on every call (not make sense)
+        get_slot.side_effect = itertools.repeat(cast(RPCResponse, { 'error': None, 'id': 1, 'result': 1 }))
+        # Will return 2 signatures on first call, empty list all other times
+        make_request.side_effect = itertools.chain([create_get_signatures_for_address([ 'signature1', 'signature2' ])],
+                                                   itertools.repeat(create_get_signatures_for_address([])))
+        # Will return same transaction (with same eth address) for every signature
+        get_confirmed_transaction.side_effect = [ pre_token_airdrop_trx1, pre_token_airdrop_trx1 ]
+
+        self._run_test()
+
+        # Should be only one call to faucet
+        req_json = {"wallet": token_airdrop_address1, "amount": self.airdrop_amount}
+        self.faucet.request_eth_token_mock.assert_called_once_with(req_json)
+        self.faucet.request_eth_token_mock.reset_mock()
+
+
+    @patch.object(Client, 'get_confirmed_transaction')
+    @patch.object(HTTPProvider, 'make_request')
+    @patch.object(Client, 'get_slot')
+    def test_complex_case_two_accounts_two_airdrops(self, get_slot, make_request, get_confirmed_transaction):
+        # Return the same slot on every call (not make sense)
+        get_slot.side_effect = itertools.repeat(cast(RPCResponse, {'error': None, 'id': 1, 'result': 1}))
+        # Will return signature on first call, empty list all other times
+        make_request.side_effect = itertools.chain([create_get_signatures_for_address(['signature3'])],
+                                                   itertools.repeat(create_get_signatures_for_address([])))
+        # Will return complex transaction containing 2 account creations and transfers
+        get_confirmed_transaction.side_effect = [pre_token_airdrop_trx2]
+
+        self._run_test()
+
+        # Should be 2 calls to faucet with different addresses
+        calls = [ call({"wallet": token_airdrop_address3, "amount": self.airdrop_amount}),
+                  call({"wallet": token_airdrop_address2, "amount": self.airdrop_amount}) ]
+        self.faucet.request_eth_token_mock.assert_has_calls(calls)
+        self.faucet.request_eth_token_mock.reset_mock()
+
+    @patch.object(Client, 'get_confirmed_transaction')
+    @patch.object(HTTPProvider, 'make_request')
+    @patch.object(Client, 'get_slot')
+    def test_should_not_call_faucet(self, get_slot, make_request, get_confirmed_transaction):
+        # Return the same slot on every call (not make sense)
+        get_slot.side_effect = itertools.repeat(cast(RPCResponse, {'error': None, 'id': 1, 'result': 1}))
+        # Will return 2 signatures on first call, empty list all other times
+        make_request.side_effect = itertools.chain([create_get_signatures_for_address(['signature4'])],
+                                                   itertools.repeat(create_get_signatures_for_address([])))
+        # Will return not interesting create_sol_acc_and_airdrop_trx
+        get_confirmed_transaction.side_effect = [create_sol_acc_and_airdrop_trx]
+
+        self._run_test()
+
+        self.faucet.request_eth_token_mock.assert_not_called()
+        self.faucet.request_eth_token_mock.reset_mock()
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        cls.faucet.shutdown_server()
+        cls.faucet.join()
+