#337 сreate base airdropper service (#343)

* cherrypick part of changes

* create indexer.py

* remove solana_receipts_update.py

* Cherry pick files from old branch

* add requirement

* fix refactoring issues

* Fix inspection issues

* fix last issue

* simplify tests

* add test

Co-authored-by: ivanl <[email protected]>

proxy/indexer/airdropper.py

@@ -0,0 +1,175 @@
+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
+
+        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}')
+            return
+
+        self.airdrop_ready[eth_address] = create_acc
+
+
+    def process_trx_airdropper_mode(self, trx):
+        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_functions(self):
+        IndexerBase.process_functions(self)
+        logger.debug("Process receipts")
+        self.process_receipts()
+
+
+    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)
+
+
+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,176 @@
+import unittest
+from proxy.testing.mock_server import MockServer
+from proxy.indexer.airdropper import Airdropper
+from proxy.indexer.sql_dict import SQLDict
+import time
+from flask import request, Response
+from unittest.mock import MagicMock, patch, call, ANY
+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)
+
+        cls.evm_loader_id = evm_loader_addr
+        cls.wrapper_whitelist = wrapper_whitelist
+        cls.airdropper = Airdropper(f'http://{cls.address}:8899',
+                                    cls.evm_loader_id,
+                                    f'http://{cls.address}:{cls.faucet_port}',
+                                    cls.wrapper_whitelist,
+                                    cls.airdrop_amount,
+                                    'INFO')
+
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        cls.faucet.shutdown_server()
+        cls.faucet.join()
+
+
+    @patch.object(SQLDict, '__setitem__')
+    @patch.object(SQLDict, '__contains__')
+    def test_success_process_trx_with_one_airdrop(self,
+                                                  mock_sql_dict_contains,
+                                                  mock_sql_dict_setitem):
+        print("\n\nShould airdrop to new address - one target in transaction")
+        mock_sql_dict_contains.side_effect = [False] # new eth address
+        self.faucet.request_eth_token_mock.side_effect = [Response("{}", status=200, mimetype='application/json')]
+
+        self.airdropper.process_trx_airdropper_mode(pre_token_airdrop_trx1)
+
+        mock_sql_dict_contains.assert_called_once_with(token_airdrop_address1)
+        mock_sql_dict_setitem.assert_has_calls([call(token_airdrop_address1, ANY)])
+        json_req = {'wallet': token_airdrop_address1, 'amount': self.airdrop_amount}
+        self.faucet.request_eth_token_mock.assert_called_once_with(json_req)
+        self.faucet.request_eth_token_mock.reset_mock()
+
+
+    @patch.object(Airdropper, '_is_allowed_wrapper_contract')
+    @patch.object(SQLDict, '__setitem__')
+    @patch.object(SQLDict, '__contains__')
+    def test_failed_airdrop_contract_not_in_whitelist(self,
+                                                      mock_sql_dict_contains,
+                                                      mock_sql_dict_setitem,
+                                                      mock_is_allowed_contract):
+        print("\n\nShould not airdrop for contract that is not in whitelist")
+        mock_is_allowed_contract.side_effect = [False]
+        self.airdropper.process_trx_airdropper_mode(pre_token_airdrop_trx1)
+
+        mock_is_allowed_contract.assert_called_once()
+        mock_sql_dict_contains.assert_not_called()
+        mock_sql_dict_setitem.assert_not_called()
+        self.faucet.request_eth_token_mock.assert_not_called()
+        self.faucet.request_eth_token_mock.reset_mock()
+
+
+    @patch.object(SQLDict, '__setitem__')
+    @patch.object(SQLDict, '__contains__')
+    def test_faucet_failure(self,
+                            mock_sql_dict_contains,
+                            mock_sql_dict_setitem):
+        print("\n\nShould not add address to processed list due to faucet error")
+        mock_sql_dict_contains.side_effect = [False]  # new eth address
+        self.faucet.request_eth_token_mock.side_effect = [Response("{}", status=400, mimetype='application/json')]
+
+        self.airdropper.process_trx_airdropper_mode(pre_token_airdrop_trx1)
+
+        mock_sql_dict_contains.assert_called_once_with(token_airdrop_address1)
+        mock_sql_dict_setitem.assert_not_called()
+        json_req = {'wallet': token_airdrop_address1, 'amount': self.airdrop_amount}
+        self.faucet.request_eth_token_mock.assert_called_once_with(json_req)
+        self.faucet.request_eth_token_mock.reset_mock()
+
+
+    @patch.object(SQLDict, '__setitem__')
+    @patch.object(SQLDict, '__contains__')
+    def test_process_trx_with_one_airdrop_for_already_processed_address(self,
+                                                                        mock_sql_dict_contains,
+                                                                        mock_sql_dict_setitem):
+        print("\n\nShould not airdrop to repeated address")
+        mock_sql_dict_contains.side_effect = [True]  # eth address processed later
+
+        self.airdropper.process_trx_airdropper_mode(pre_token_airdrop_trx1)
+
+        mock_sql_dict_contains.assert_called_once_with(token_airdrop_address1)
+        mock_sql_dict_setitem.assert_not_called()
+        self.faucet.request_eth_token_mock.assert_not_called()
+        self.faucet.request_eth_token_mock.reset_mock()
+
+
+    @patch.object(SQLDict, '__setitem__')
+    @patch.object(SQLDict, '__contains__')
+    def test_complex_transation(self,
+                                mock_sql_dict_contains,
+                                mock_sql_dict_setitem):
+        print("\n\nShould airdrop to several targets in one transaction")
+        mock_sql_dict_contains.side_effect = [False, False] # both targets are new
+        self.faucet.request_eth_token_mock.side_effect = [Response("{}", status=200, mimetype='application/json'),
+                                                          Response("{}", status=200, mimetype='application/json')]
+
+        self.airdropper.process_trx_airdropper_mode(pre_token_airdrop_trx2)
+
+        mock_sql_dict_contains.assert_has_calls([call(token_airdrop_address3),
+                                                 call(token_airdrop_address2)])
+        mock_sql_dict_setitem.assert_has_calls([call(token_airdrop_address3, ANY),
+                                                call(token_airdrop_address2, ANY)])
+        json_req1 = {'wallet': token_airdrop_address2, 'amount': self.airdrop_amount}
+        json_req2 = {'wallet': token_airdrop_address3, 'amount': self.airdrop_amount}
+        self.faucet.request_eth_token_mock.assert_has_calls([call(json_req2), call(json_req1)])
+        self.faucet.request_eth_token_mock.reset_mock()
+
+
+    @patch.object(SQLDict, '__setitem__')
+    @patch.object(SQLDict, '__contains__')
+    def test_no_airdrop_instructions(self,
+                                     mock_sql_dict_contains,
+                                     mock_sql_dict_setitem):
+        print("\n\nShould not airdrop when instructions are inconsistent")
+        self.airdropper.process_trx_airdropper_mode(create_sol_acc_and_airdrop_trx)
+
+        mock_sql_dict_contains.assert_not_called()
+        mock_sql_dict_setitem.assert_not_called()
+        self.faucet.request_eth_token_mock.assert_not_called()
+        self.faucet.request_eth_token_mock.reset_mock()
+