Add instruction validation to anchor program indexer (#2867)

discovery-provider/src/app.py

@@ -550,6 +550,7 @@ def configure_celery(celery, test_config=None):
     # Initialize Anchor Indexer
     anchor_program_indexer = AnchorProgramIndexer(
         shared_config["solana"]["anchor_data_program_id"],
+        shared_config["solana"]["anchor_admin_storage_public_key"],
         "index_solana_user_data",
         redis_inst,
         db,

discovery-provider/src/solana/anchor_program_indexer.py

@@ -1,7 +1,7 @@
 import asyncio
 import base64
 import logging
-from typing import Any
+from typing import Any, Dict
 
 from solana.transaction import Transaction
 from sqlalchemy import desc
@@ -24,13 +24,15 @@ class AnchorProgramIndexer(SolanaProgramIndexer):
     def __init__(
         self,
         program_id: str,
+        admin_storage_public_key: str,
         label: str,
         redis: Any,
         db: Any,
         solana_client_manager: Any,
     ):
         super().__init__(program_id, label, redis, db, solana_client_manager)
         self.anchor_parser = AnchorParser(AUDIUS_DATA_IDL_PATH, program_id)
+        self.admin_storage_public_key = admin_storage_public_key
 
     def is_tx_in_db(self, session: Any, tx_sig: str):
         exists = False
@@ -88,7 +90,8 @@ async def parse_tx(self, tx_sig):
         tx_instructions = []
         for instruction in tx.instructions:
             parsed_instr = self.anchor_parser.parse_instruction(instruction)
-            tx_instructions.append(parsed_instr)
+            if self.is_valid_instruction(parsed_instr):
+                tx_instructions.append(parsed_instr)
 
         tx_metadata["instructions"] = tx_instructions
 
@@ -98,6 +101,21 @@ async def parse_tx(self, tx_sig):
         """
         return {"tx_sig": tx_sig, "tx_metadata": tx_metadata, "result": None}
 
+    def is_valid_instruction(self, parsed_instr: Dict):
+        if parsed_instr["instruction_name"] == "init_user":
+            if (
+                parsed_instr["account_names_map"]["admin"]
+                != self.admin_storage_public_key
+            ):
+                return False
+        # TODO implement remaining instruction validation
+        # consider creating classes for each instruction type
+        # then implementing instruction validation / updating user records for each.
+        # consider renaming admin accounts in program for consistency
+        # then dynamically validating.
+
+        return True
+
     def process_index_task(self):
         self.msg("Processing indexing task")
         # Retrieve transactions to process

discovery-provider/src/solana/solana_program_indexer.py

@@ -117,6 +117,14 @@ async def parse_tx(self, tx_sig):
         result: TransactionInfoResult = tx_info["result"]
         return {"tx_sig": tx_sig, "tx_metadata": {}, "result": result}
 
+    @abstractmethod
+    def is_valid_instruction(self, tx):
+        """
+        Returns a boolean value indicating whether an instruction is valid.
+        @param tx: transaction to be validated
+        """
+        raise Exception("Must be implemented in subclass")
+
     @abstractmethod
     async def fetch_ipfs_metadata(self, parsed_transactions):
         """