Feature/unt 701 arbitrum bridge requirement

core/constraints/abstract.py

@@ -11,6 +11,7 @@ class ConstraintApp(Enum):
     ENS = "ENS"
     EAS = "EAS"
     GITCOIN_PASSPORT = "gitcoin_passport"
+    ARBITRUM = "arbitrum"
 
     @classmethod
     def choices(cls):
@@ -34,6 +35,8 @@ class ConstraintParam(Enum):
     VALUE = "value"
     EAS_SCHEMA_ID = "eas_schema_id"
     FARCASTER_CHANNEL_ID = "FARCASTER_CHANNEL_ID"
+    SOURCE_CHAIN = "source_chain"
+    DESTINATION_CHAIN = "destination_chain"
 
     @classmethod
     def choices(cls):

core/constraints/arbitrum.py

@@ -0,0 +1,58 @@
+from core.constraints.abstract import (
+    ConstraintApp,
+    ConstraintParam,
+    ConstraintVerification,
+)
+from core.thirdpartyapp.arbitrum_bridge import ArbitrumBridgeUtils
+
+
+class HasBridgedToken(ConstraintVerification):
+    _param_keys = [
+        ConstraintParam.ADDRESS,
+        ConstraintParam.MINIMUM,
+        ConstraintParam.SOURCE_CHAIN,
+        ConstraintParam.DESTINATION_CHAIN,
+    ]
+    app_name = ConstraintApp.ARBITRUM.value
+
+    def __init__(self, user_profile) -> None:
+        super().__init__(user_profile)
+
+    def is_observed(self, *args, **kwargs) -> bool:
+        minimum_amount = int(self.param_values[ConstraintParam.MINIMUM.name])
+        sourec_chain = self.param_values[ConstraintParam.SOURCE_CHAIN.name]
+        # dest_chain = self.param_values[ConstraintParam.DESTINATION_CHAIN.name]
+        token_address = self.param_values[ConstraintParam.ADDRESS.name]
+
+        arb_utils = ArbitrumBridgeUtils()
+
+        if token_address.lower() in ["eth", ""]:
+            token_address = None
+
+        user_wallets = self.user_addresses
+        for wallet in user_wallets:
+            if sourec_chain == "any":
+                bridging_results = arb_utils.check_all_bridge_transactions(
+                    wallet, token_address, minimum_amount
+                )
+                if any(bridging_results.values()):
+                    return True
+            else:
+                source_chain, target_chain = self._parse_chain(sourec_chain)
+                if arb_utils.check_bridge_transactions(
+                    wallet, token_address, minimum_amount, source_chain, target_chain
+                ):
+                    return True
+
+        return False
+
+    def _parse_chain(self, chain):
+        chain_mapping = {
+            "ethereum_to_arbitrum": ("ethereum", "arbitrum"),
+            "arbitrum_to_ethereum": ("arbitrum", "ethereum"),
+            "arbitrum_to_nova": ("arbitrum", "nova"),
+            "nova_to_arbitrum": ("nova", "arbitrum"),
+        }
+        if chain not in chain_mapping:
+            raise ValueError(f"Invalid chain parameter: {chain}")
+        return chain_mapping[chain]

core/tests.py

@@ -4,7 +4,9 @@
 from rest_framework.test import APITestCase
 
 from authentication.models import GitcoinPassportConnection, UserProfile, Wallet
+from core.constraints.arbitrum import HasBridgedToken
 from core.models import Chain, NetworkTypes, WalletAccount
+from core.thirdpartyapp.arbitrum_bridge import ArbitrumUtils
 
 from .constraints import (
     Attest,
@@ -363,3 +365,108 @@ def test_gitcoin_passport_connection_constraint_fail(self):
         constraint = HasGitcoinPassportProfile(self.not_connected_user_profile)
 
         self.assertEqual(constraint.is_observed(), False)
+
+
+class TestArbitrumBridgeConstraint(BaseTestCase):
+    def setUp(self):
+        super().setUp()
+        self.user = User.objects.create_user(username="testuser", password="12345")
+        self.user_profile = UserProfile.objects.create(
+            user=self.user, initial_context_id="testuser", username="testuser"
+        )
+        self.address1 = "0x0cE49AF5d8c5A70Edacd7115084B2b3041fE4fF6"
+        self.address2 = "0x319B32d11e29dB4a6dB9E4E3da91Fc7FA2D2ff92"
+        Wallet.objects.create(
+            user_profile=self.user_profile, address=self.address1, wallet_type="EVM"
+        )
+        Wallet.objects.create(
+            user_profile=self.user_profile, address=self.address2, wallet_type="EVM"
+        )
+        self.token_address = "0x1234567890123456789012345678901234567890"
+        self.minimum_amount = 1.0
+
+    @patch.object(ArbitrumUtils, "check_all_bridge_transactions")
+    def test_has_bridged_token_any_chain_success(self, mock_check_all):
+        mock_check_all.return_value = {
+            "Ethereum to Arbitrum": True,
+            "Arbitrum to Ethereum": False,
+        }
+        constraint = HasBridgedToken(self.user_profile)
+        constraint.param_values = {
+            "MINIMUM": self.minimum_amount,
+            "CHAIN": "any",
+            "ADDRESS": self.token_address,
+        }
+        self.assertTrue(constraint.is_observed())
+
+    @patch.object(ArbitrumUtils, "check_all_bridge_transactions")
+    def test_has_bridged_token_any_chain_fail(self, mock_check_all):
+        mock_check_all.return_value = {
+            "Ethereum to Arbitrum": False,
+            "Arbitrum to Ethereum": False,
+        }
+        constraint = HasBridgedToken(self.user_profile)
+        constraint.param_values = {
+            "MINIMUM": self.minimum_amount,
+            "CHAIN": "any",
+            "ADDRESS": self.token_address,
+        }
+        self.assertFalse(constraint.is_observed())
+
+    @patch.object(ArbitrumUtils, "check_bridge_transactions")
+    def test_has_bridged_token_specific_chain_success(self, mock_check):
+        mock_check.return_value = True
+        constraint = HasBridgedToken(self.user_profile)
+        constraint.param_values = {
+            "MINIMUM": self.minimum_amount,
+            "CHAIN": "ethereum_to_arbitrum",
+            "ADDRESS": self.token_address,
+        }
+        self.assertTrue(constraint.is_observed())
+
+    @patch.object(ArbitrumUtils, "check_bridge_transactions")
+    def test_has_bridged_token_specific_chain_fail(self, mock_check):
+        mock_check.return_value = False
+        constraint = HasBridgedToken(self.user_profile)
+        constraint.param_values = {
+            "MINIMUM": self.minimum_amount,
+            "CHAIN": "ethereum_to_arbitrum",
+            "ADDRESS": self.token_address,
+        }
+        self.assertFalse(constraint.is_observed())
+
+    @patch.object(ArbitrumUtils, "check_bridge_transactions")
+    def test_has_bridged_eth_success(self, mock_check):
+        mock_check.return_value = True
+        constraint = HasBridgedToken(self.user_profile)
+        constraint.param_values = {
+            "MINIMUM": self.minimum_amount,
+            "CHAIN": "ethereum_to_arbitrum",
+            "ADDRESS": "eth",
+        }
+        self.assertTrue(constraint.is_observed())
+        mock_check.assert_called_with(
+            self.address1, None, self.minimum_amount, "ethereum", "arbitrum"
+        )
+
+    @patch.object(ArbitrumUtils, "check_bridge_transactions")
+    def test_has_bridged_token_multiple_wallets(self, mock_check):
+        mock_check.side_effect = [False, True]
+        constraint = HasBridgedToken(self.user_profile)
+        constraint.param_values = {
+            "MINIMUM": self.minimum_amount,
+            "CHAIN": "ethereum_to_arbitrum",
+            "ADDRESS": self.token_address,
+        }
+        self.assertTrue(constraint.is_observed())
+        self.assertEqual(mock_check.call_count, 2)
+
+    def test_has_bridged_token_invalid_chain(self):
+        constraint = HasBridgedToken(self.user_profile)
+        constraint.param_values = {
+            "MINIMUM": self.minimum_amount,
+            "CHAIN": "invalid_chain",
+            "ADDRESS": self.token_address,
+        }
+        with self.assertRaises(ValueError):
+            constraint.is_observed()

core/thirdpartyapp/__init__.py

@@ -1,3 +1,4 @@
+from .arbitrum_bridge import ArbitrumBridgeUtils
 from .EAS import EASUtils
 from .ens import ENSUtil  # noqa: F401
 from .farcaster import FarcasterUtil  # noqa: F401

core/thirdpartyapp/arbitrum_bridge.py

@@ -0,0 +1,120 @@
+from typing import Optional
+
+from core.thirdpartyapp.config import ARB_BRIDGE_ADDRESSES
+from core.utils import Web3Utils
+
+
+class ArbitrumBridgeUtils:
+    ARB_BRIDGE_ABI = [
+        {
+            "anonymous": False,
+            "inputs": [
+                {"indexed": True, "name": "l1Token", "type": "address"},
+                {"indexed": True, "name": "from", "type": "address"},
+                {"indexed": True, "name": "to", "type": "address"},
+                {"indexed": False, "name": "amount", "type": "uint256"},
+            ],
+            "name": "ERC20DepositInitiated",
+            "type": "event",
+        },
+        {
+            "anonymous": False,
+            "inputs": [
+                {"indexed": True, "name": "from", "type": "address"},
+                {"indexed": True, "name": "to", "type": "address"},
+                {"indexed": False, "name": "amount", "type": "uint256"},
+            ],
+            "name": "DepositInitiated",
+            "type": "event",
+        },
+        {
+            "anonymous": False,
+            "inputs": [
+                {"indexed": True, "name": "l1Token", "type": "address"},
+                {"indexed": True, "name": "from", "type": "address"},
+                {"indexed": True, "name": "to", "type": "address"},
+                {"indexed": False, "name": "amount", "type": "uint256"},
+            ],
+            "name": "WithdrawalInitiated",
+            "type": "event",
+        },
+        {
+            "anonymous": False,
+            "inputs": [
+                {"indexed": True, "name": "from", "type": "address"},
+                {"indexed": True, "name": "to", "type": "address"},
+                {"indexed": False, "name": "amount", "type": "uint256"},
+            ],
+            "name": "ETHWithdrawalInitiated",
+            "type": "event",
+        },
+    ]
+
+    def __init__(self, chain_name: str, rpc_url_private: str, poa: bool = False):
+        self.w3 = Web3Utils(rpc_url_private=rpc_url_private, poa=poa)
+        self.w3.set_contract(ARB_BRIDGE_ADDRESSES.get(chain_name), self.BRIDGE_ABI)
+        self.contracts = {
+            "ethereum": self.ETH_W3.set_contract(
+                self.BRIDGE_ADDRESSES["ethereum"], self.BRIDGE_ABI
+            ),
+            "arbitrum": self.ARB_W3.set_contract(
+                self.BRIDGE_ADDRESSES["arbitrum"], self.BRIDGE_ABI
+            ),
+            "nova": self.NOVA_W3.set_contract(
+                self.BRIDGE_ADDRESSES["nova"], self.BRIDGE_ABI
+            ),
+        }
+
+    def check_bridge_transactions(
+        self,
+        wallet_address: str,
+        token_address: Optional[str],
+        amount: float,
+        source_chain: str,
+        target_chain: str,
+    ) -> bool:
+        if source_chain not in self.contracts or target_chain not in self.contracts:
+            raise ValueError(f"Invalid chain: {source_chain} or {target_chain}")
+
+        contract = self.contracts[source_chain]
+
+        if token_address is None:  # ETH transaction
+            if source_chain == "ethereum":
+                event_name = "DepositInitiated"
+            else:
+                event_name = "ETHWithdrawalInitiated"
+
+            event = getattr(contract.events, event_name)
+            event_filter = event.createFilter(
+                fromBlock=0, argument_filters={"from": wallet_address}
+            )
+        else:  # ERC-20 transaction
+            if source_chain == "ethereum":
+                event_name = "ERC20DepositInitiated"
+            else:
+                event_name = "WithdrawalInitiated"
+
+            event = getattr(contract.events, event_name)
+            event_filter = event.createFilter(
+                fromBlock=0,
+                argument_filters={"from": wallet_address, "l1Token": token_address},
+            )
+
+        events = event_filter.get_all_entries()
+        return any(event["args"]["amount"] >= amount for event in events)
+
+    def check_all_bridge_transactions(
+        self, wallet_address: str, token_address: Optional[str], amount: float
+    ) -> dict[str, bool]:
+        results = {}
+        chains = ["ethereum", "arbitrum", "nova"]
+
+        for source in chains:
+            for target in chains:
+                if source != target:
+                    key = f"{source.capitalize()} to {target.capitalize()}"
+                    results[key] = self.check_bridge_transactions(
+                        wallet_address, token_address, amount, source, target
+                    )
+
+        return results

core/thirdpartyapp/config.py

@@ -22,3 +22,9 @@
     ),
 }
 SUBGRAPH_BASE_URL = os.getenv("SUBGRAPH_BASE_URL", "https://api.studio.thegraph.com")
+
+ARB_BRIDGE_ADDRESSES = {
+    "ethereum": "0x8315177aB297bA92A06054cE80a67Ed4DBd7ed3a",
+    "arbitrum": "0x0000000000000000000000000000000000000064",
+    "nova": "0x0000000000000000000000000000000000000064",
+}