(feat) Implemented OFAC list address check

Author: shibaeff

pyinjective/composer.py

@@ -11,6 +11,7 @@
 from pyinjective.constant import ADDITIONAL_CHAIN_FORMAT_DECIMALS, INJ_DECIMALS, INJ_DENOM
 from pyinjective.core.market import BinaryOptionMarket, DerivativeMarket, SpotMarket
 from pyinjective.core.token import Token
+from pyinjective.ofac import OfacChecker
 from pyinjective.proto.cosmos.authz.v1beta1 import authz_pb2 as cosmos_authz_pb, tx_pb2 as cosmos_authz_tx_pb
 from pyinjective.proto.cosmos.bank.v1beta1 import bank_pb2 as bank_pb, tx_pb2 as cosmos_bank_tx_pb
 from pyinjective.proto.cosmos.base.v1beta1 import coin_pb2 as base_coin_pb
@@ -147,6 +148,7 @@ def __init__(
             self.derivative_markets = derivative_markets
             self.binary_option_markets = binary_option_markets
             self.tokens = tokens
+        self._ofac_checker = OfacChecker()
 
     def Coin(self, amount: int, denom: str):
         """
@@ -471,6 +473,8 @@ def MsgBid(self, sender: str, bid_amount: float, round: float):
 
     # region Authz module
     def MsgGrantGeneric(self, granter: str, grantee: str, msg_type: str, expire_in: int):
+        if self._ofac_checker.is_blacklisted(granter):
+            raise Exception(f"Address {granter} is in the OFAC list")
         auth = cosmos_authz_pb.GenericAuthorization(msg=msg_type)
         any_auth = any_pb2.Any()
         any_auth.Pack(auth, type_url_prefix="")
@@ -2125,6 +2129,9 @@ def MsgGrantTyped(
         subaccount_id: str,
         **kwargs,
     ):
+        if self._ofac_checker.is_blacklisted(granter):
+            raise Exception(f"Address {granter} is in the OFAC list")
+
         auth = None
         if msg_type == "CreateSpotLimitOrderAuthz":
             auth = injective_authz_pb.CreateSpotLimitOrderAuthz(

pyinjective/core/broadcaster.py

@@ -12,6 +12,7 @@
 from pyinjective.constant import GAS_PRICE
 from pyinjective.core.gas_limit_estimator import GasLimitEstimator
 from pyinjective.core.network import Network
+from pyinjective.ofac import OfacChecker
 
 
 class BroadcasterAccountConfig(ABC):
@@ -62,6 +63,10 @@ def __init__(
         self._client = client
         self._composer = composer
         self._fee_calculator = fee_calculator
+        self._ofac_checker = OfacChecker()
+
+        if self._ofac_checker.is_blacklisted(address=self._account_config.trading_injective_address):
+            raise Exception(f"Address {self._account_config.trading_injective_address} is in the OFAC list")
 
     @classmethod
     def new_using_simulation(

pyinjective/ofac.json

@@ -0,0 +1,48 @@
+[
+  "0x179f48c78f57a3a78f0608cc9197b8972921d1d2",
+  "0x1967d8af5bd86a497fb3dd7899a020e47560daaf",
+  "0x19aa5fe80d33a56d56c78e82ea5e50e5d80b4dff",
+  "0x19aa5fe80d33a56d56c78e82ea5e50e5d80b4dff",
+  "0x1da5821544e25c636c1417ba96ade4cf6d2f9b5a",
+  "0x2f389ce8bd8ff92de3402ffce4691d17fc4f6535",
+  "0x2f389ce8bd8ff92de3402ffce4691d17fc4f6535",
+  "0x2f50508a8a3d323b91336fa3ea6ae50e55f32185",
+  "0x308ed4b7b49797e1a98d3818bff6fe5385410370",
+  "0x3cbded43efdaf0fc77b9c55f6fc9988fcc9b757d",
+  "0x3efa30704d2b8bbac821307230376556cf8cc39e",
+  "0x48549a34ae37b12f6a30566245176994e17c6b4a",
+  "0x4f47bc496083c727c5fbe3ce9cdf2b0f6496270c",
+  "0x4f47bc496083c727c5fbe3ce9cdf2b0f6496270c",
+  "0x4f47bc496083c727c5fbe3ce9cdf2b0f6496270c",
+  "0x530a64c0ce595026a4a556b703644228179e2d57",
+  "0x5512d943ed1f7c8a43f3435c85f7ab68b30121b0",
+  "0x5a7a51bfb49f190e5a6060a5bc6052ac14a3b59f",
+  "0x5f48c2a71b2cc96e3f0ccae4e39318ff0dc375b2",
+  "0x6be0ae71e6c41f2f9d0d1a3b8d0f75e6f6a0b46e",
+  "0x6f1ca141a28907f78ebaa64fb83a9088b02a8352",
+  "0x746aebc06d2ae31b71ac51429a19d54e797878e9",
+  "0x77777feddddffc19ff86db637967013e6c6a116c",
+  "0x797d7ae72ebddcdea2a346c1834e04d1f8df102b",
+  "0x8576acc5c05d6ce88f4e49bf65bdf0c62f91353c",
+  "0x901bb9583b24d97e995513c6778dc6888ab6870e",
+  "0x961c5be54a2ffc17cf4cb021d863c42dacd47fc1",
+  "0x97b1043abd9e6fc31681635166d430a458d14f9c",
+  "0x9c2bc757b66f24d60f016b6237f8cdd414a879fa",
+  "0x9f4cda013e354b8fc285bf4b9a60460cee7f7ea9",
+  "0xa7e5d5a720f06526557c513402f2e6b5fa20b008",
+  "0xb6f5ec1a0a9cd1526536d3f0426c429529471f40",
+  "0xb6f5ec1a0a9cd1526536d3f0426c429529471f40",
+  "0xb6f5ec1a0a9cd1526536d3f0426c429529471f40",
+  "0xc455f7fd3e0e12afd51fba5c106909934d8a0e4a",
+  "0xca0840578f57fe71599d29375e16783424023357",
+  "0xd0975b32cea532eadddfc9c60481976e39db3472",
+  "0xd882cfc20f52f2599d84b8e8d58c7fb62cfe344b",
+  "0xd882cfc20f52f2599d84b8e8d58c7fb62cfe344b",
+  "0xe1d865c3d669dcc8c57c8d023140cb204e672ee4",
+  "0xe7aa314c77f4233c18c6cc84384a9247c0cf367b",
+  "0xed6e0a7e4ac94d976eebfb82ccf777a3c6bad921",
+  "0xf3701f445b6bdafedbca97d1e477357839e4120d",
+  "0xfac583c0cf07ea434052c49115a4682172ab6b4f",
+  "0xfec8a60023265364d066a1212fde3930f6ae8da7",
+  "0xffbac21a641dcfe4552920138d90f3638b3c9fba"
+]

pyinjective/ofac.py

@@ -0,0 +1,55 @@
+import asyncio
+import json
+import os
+
+import aiohttp
+
+OFAC_LIST_URL = "https://raw.githubusercontent.com/InjectiveLabs/injective-lists/master/wallets/ofac.json"
+OFAC_LIST_FILENAME_DEFAULT = "ofac.json"
+OFAC_LIST_FILENAME = OFAC_LIST_FILENAME_DEFAULT
+
+
+class OfacChecker:
+    def __init__(self):
+        self._ofac_list_path = self.get_ofac_list_path()
+        if not os.path.exists(self._ofac_list_path):
+            raise Exception(
+                "OFAC list is missing on the disk. Please, download it by running python3 pyinjective/ofac_list.py"
+            )
+
+        with open(self._ofac_list_path, "r") as f:
+            self._ofac_list = set(json.load(f))
+
+    @classmethod
+    def get_ofac_list_path(cls):
+        current_directory = os.getcwd()
+        while os.path.basename(current_directory) != "sdk-python":
+            current_directory = os.path.dirname(current_directory)
+        return os.path.join(os.path.join(current_directory, "pyinjective"), OFAC_LIST_FILENAME)
+
+    @classmethod
+    async def download_ofac_list(cls):
+        async with aiohttp.ClientSession() as session:
+            try:
+                async with session.get(OFAC_LIST_URL) as response:
+                    response.raise_for_status()
+                    text_content = await response.text()
+                    ofac_list = json.loads(text_content)
+                    ofac_file_path = cls.get_ofac_list_path()
+                    with open(ofac_file_path, "w") as f:
+                        json.dump(ofac_list, f, indent=2)
+                        f.write("\n")
+                    return
+            except Exception as e:
+                raise Exception(f"Error fetching OFAC list: {e}")
+
+    def is_blacklisted(self, address: str) -> bool:
+        return address in self._ofac_list
+
+
+async def main():
+    await OfacChecker.download_ofac_list()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

tests/core/test_broadcaster.py

@@ -0,0 +1,48 @@
+import json
+import os
+import unittest
+from unittest.mock import Mock
+
+import pytest
+
+import pyinjective.ofac as ofac
+from pyinjective import PrivateKey
+from pyinjective.async_client import AsyncClient
+from pyinjective.composer import Composer
+from pyinjective.core.broadcaster import MsgBroadcasterWithPk, StandardAccountBroadcasterConfig
+from pyinjective.core.network import Network
+
+
+class TestBroadcastAddressInOfacList(unittest.TestCase):
+    def setUp(self):
+        ofac.OFAC_LIST_FILENAME = "ofac_test.json"
+        self.private_key_banned = PrivateKey.from_mnemonic("test mnemonic never use other places")
+        public_key_banned = self.private_key_banned.to_public_key()
+        address_banned = public_key_banned.to_address()
+
+        ofac_list = [address_banned.to_acc_bech32()]
+
+        with open(ofac.OfacChecker.get_ofac_list_path(), "w") as ofac_file:
+            json.dump(ofac_list, ofac_file)
+
+    def tearDown(self):
+        os.remove(ofac.OfacChecker.get_ofac_list_path())
+        ofac.OFAC_LIST_FILENAME = ofac.OFAC_LIST_FILENAME_DEFAULT
+
+    def test_broadcast_address_in_ofac_list(self):
+        network = Network.local()
+        client = AsyncClient(
+            network=Network.local(),
+        )
+        composer = Mock(spec=Composer)
+
+        account_config = StandardAccountBroadcasterConfig(private_key=self.private_key_banned.to_hex())
+
+        with pytest.raises(Exception):
+            _ = MsgBroadcasterWithPk(
+                network=network,
+                account_config=account_config,
+                client=client,
+                composer=composer,
+                fee_calculator=Mock(),
+            )