[WIP] holonym integration

backend/app/infrastructure/database/user_antisybil.py

@@ -2,7 +2,9 @@
 from datetime import datetime
 import json
 
-from app.infrastructure.database.models import GPStamps
+from typing import List
+
+from app.infrastructure.database.models import GPStamps, HolonymSBT
 from app.infrastructure.database.user import get_by_address
 from app.exceptions import UserNotFound
 from app.extensions import db
@@ -37,3 +39,29 @@ def get_score_by_address(user_address: str) -> Optional[GPStamps]:
         .filter_by(user_id=user.id)
         .first()
     )
+
+
+def get_sbt_by_address(user_address: str) -> Optional[HolonymSBT]:
+    user = get_by_address(user_address)
+    if user is None:
+        raise UserNotFound(user_address)
+
+    return (
+        HolonymSBT.query.order_by(HolonymSBT.created_at.desc())
+        .filter_by(user_id=user.id)
+        .first()
+    )
+
+
+def add_sbt(user_address: str, has_sbt: bool, sbt_details: List[str]) -> HolonymSBT:
+    user = get_by_address(user_address)
+
+    if user is None:
+        raise UserNotFound(user_address)
+
+    verification = HolonymSBT(
+        user_id=user.id, has_sbt=has_sbt, sbt_details=json.dumps(sbt_details)
+    )
+    db.session.add(verification)
+
+    return verification

backend/app/infrastructure/routes/user.py

@@ -27,6 +27,10 @@
         "expires_at": fields.String(
             required=False, description="Expiry date, unix timestamp"
         ),
+        "holonym": fields.String(
+            required=False,
+            description="True, if user has Holonym SBT",
+        ),
         "score": fields.String(
             required=False,
             description="Score, parses as a float",
@@ -192,12 +196,14 @@ class AntisybilStatus(OctantResource):
     @ns.response(200, "User's cached antisybil status retrieved")
     def get(self, user_address: str):
         app.logger.debug(f"Getting user {user_address} cached antisybil status")
+
         antisybil_status = get_user_antisybil_status(user_address)
         app.logger.debug(f"User {user_address} antisybil status: {antisybil_status}")
         if antisybil_status is None:
             return {"status": "Unknown"}, 404
-        score, expires_at = antisybil_status
+        (score, expires_at), (has_sbt, _) = antisybil_status
         return {
+            "holonym": has_sbt,
             "status": "Known",
             "score": score,
             "expires_at": int(expires_at.timestamp()),
@@ -211,9 +217,11 @@ def get(self, user_address: str):
     @ns.response(504, "Could not refresh antisybil status. Upstream is unavailable.")
     def put(self, user_address: str):
         app.logger.info(f"Updating user {user_address} antisybil status")
-        score, expires_at = update_user_antisybil_status(user_address)
+        status = update_user_antisybil_status(user_address)
+        app.logger.info(f"Got status for user {user_address} = {status}")
+        (score, expires_at), (has_sbt, _) = status
         app.logger.info(
-            f"User {user_address} antisybil status refreshed {[score, expires_at]}"
+            f"User {user_address} antisybil status refreshed {[score, has_sbt, expires_at]}"
         )
 
         return {}, 204

backend/app/modules/modules_factory/current.py

@@ -41,6 +41,7 @@
 from app.modules.user.patron_mode.service.events_based import EventsBasedUserPatronMode
 from app.modules.user.tos.service.initial import InitialUserTos, InitialUserTosVerifier
 from app.modules.user.antisybil.service.passport import GitcoinPassportAntisybil
+from app.modules.user.antisybil.service.holonym import HolonymAntisybil
 from app.modules.withdrawals.service.finalized import FinalizedWithdrawals
 from app.pydantic import Model
 from app.shared.blockchain_types import compare_blockchain_types, ChainTypes
@@ -56,6 +57,7 @@ class CurrentServices(Model):
     user_deposits_service: CurrentUserDeposits
     user_tos_service: UserTos
     user_antisybil_passport_service: GitcoinPassportAntisybil
+    user_antisybil_holonym_service: HolonymAntisybil
     octant_rewards_service: OctantRewards
     history_service: HistoryService
     simulated_pending_snapshot_service: SimulatePendingSnapshots
@@ -98,6 +100,7 @@ def create(chain_id: int) -> "CurrentServices":
         user_allocations_nonce = SavedUserAllocationsNonce()
         user_withdrawals = FinalizedWithdrawals()
         user_antisybil_passport_service = GitcoinPassportAntisybil()
+        user_antisybil_holonym_service = HolonymAntisybil()
         tos_verifier = InitialUserTosVerifier()
         user_tos = InitialUserTos(verifier=tos_verifier)
         patron_donations = EventsBasedUserPatronMode()
@@ -129,7 +132,8 @@ def create(chain_id: int) -> "CurrentServices":
         )
         uq_threshold = UQ_THRESHOLD_MAINNET if is_mainnet else UQ_THRESHOLD_NOT_MAINNET
         uniqueness_quotients = PreliminaryUQ(
-            antisybil=GitcoinPassportAntisybil(),
+            passport=user_antisybil_passport_service,
+            holonym=user_antisybil_holonym_service,
             budgets=user_budgets,
             uq_threshold=uq_threshold,
         )
@@ -143,6 +147,7 @@ def create(chain_id: int) -> "CurrentServices":
             multisig_signatures_service=multisig_signatures,
             user_tos_service=user_tos,
             user_antisybil_passport_service=user_antisybil_passport_service,
+            user_antisybil_holonym_service=user_antisybil_holonym_service,
             projects_metadata_service=StaticProjectsMetadataService(),
             user_budgets_service=user_budgets,
             score_delegation_service=score_delegation,

backend/app/modules/modules_factory/pending.py

@@ -41,6 +41,7 @@
 from app.modules.user.antisybil.service.passport import (
     GitcoinPassportAntisybil,
 )
+from app.modules.user.antisybil.service.holonym import HolonymAntisybil
 from app.modules.user.budgets.service.saved import SavedUserBudgets
 from app.modules.user.deposits.service.saved import SavedUserDeposits
 from app.modules.user.patron_mode.service.events_based import EventsBasedUserPatronMode
@@ -99,7 +100,8 @@ def create(chain_id: int) -> "PendingServices":
         is_mainnet = compare_blockchain_types(chain_id, ChainTypes.MAINNET)
         uq_threshold = UQ_THRESHOLD_MAINNET if is_mainnet else UQ_THRESHOLD_NOT_MAINNET
         uniqueness_quotients = PreliminaryUQ(
-            antisybil=GitcoinPassportAntisybil(),
+            passport=GitcoinPassportAntisybil(),
+            holonym=HolonymAntisybil(),
             budgets=saved_user_budgets,
             uq_threshold=uq_threshold,
         )

backend/app/modules/modules_factory/protocols.py

@@ -1,5 +1,5 @@
 from decimal import Decimal
-from typing import Protocol, List, Dict, Tuple, Optional, runtime_checkable, Set
+from typing import Protocol, List, Dict, Tuple, Optional, runtime_checkable, Container
 
 from app.context.manager import Context
 from app.engine.projects.rewards import ProjectRewardDTO, ProjectRewardsResult
@@ -240,7 +240,9 @@ def delegate(self, context: Context, payload: ScoreDelegationPayload):
     def recalculate(self, context: Context, payload: ScoreDelegationPayload):
         ...
 
-    def check(self, context: Context, addresses: List[str]) -> Set[Tuple[str, str]]:
+    def check(
+        self, context: Context, addresses: List[str]
+    ) -> Container[Tuple[str, str]]:
         ...
 
 

backend/app/modules/uq/service/preliminary.py

@@ -1,6 +1,6 @@
 from datetime import datetime
 from decimal import Decimal
-from typing import Protocol, Optional, Tuple, runtime_checkable
+from typing import Protocol, List, Optional, Tuple, runtime_checkable
 
 from app.context.manager import Context
 from app.infrastructure.database.uniqueness_quotient import (
@@ -12,21 +12,30 @@
 
 
 @runtime_checkable
-class Antisybil(Protocol):
+class Passport(Protocol):
     def get_antisybil_status(
         self, _: Context, user_address: str
     ) -> Optional[Tuple[float, datetime]]:
         ...
 
 
+@runtime_checkable
+class Holonym(Protocol):
+    def get_sbt_status(
+        self, _: Context, user_address: str
+    ) -> Optional[Tuple[bool, List[str]]]:
+        ...
+
+
 @runtime_checkable
 class UserBudgets(Protocol):
     def get_budget(self, context: Context, user_address: str) -> int:
         ...
 
 
 class PreliminaryUQ(Model):
-    antisybil: Antisybil
+    passport: Passport
+    holonym: Holonym
     budgets: UserBudgets
     uq_threshold: int
 
@@ -56,7 +65,7 @@ def calculate(self, context: Context, user_address: str) -> Decimal:
         return calculate_uq(gp_score, self.uq_threshold)
 
     def _get_gp_score(self, context: Context, address: str) -> float:
-        antisybil_status = self.antisybil.get_antisybil_status(context, address)
-        if antisybil_status is None:
+        passport_status = self.passport.get_antisybil_status(context, address)
+        if passport_status is None:
             return 0.0
-        return antisybil_status[0]
+        return passport_status[0]

backend/app/modules/user/antisybil/controller.py

@@ -1,25 +1,41 @@
 from datetime import datetime
-from typing import Tuple
+from typing import List, Optional, Tuple
 
 from app.context.epoch_state import EpochState
 from app.context.manager import state_context
 from app.modules.registry import get_services
 
 
-def get_user_antisybil_status(user_address: str) -> Tuple[int, datetime]:
+def get_user_antisybil_status(
+    user_address: str,
+) -> Optional[Tuple[Tuple[int, datetime], Tuple[bool, List[str]]]]:
     context = state_context(EpochState.CURRENT)
-    service = get_services(context.epoch_state).user_antisybil_passport_service
-    return service.get_antisybil_status(context, user_address)
+    passport = get_services(context.epoch_state).user_antisybil_passport_service
+    passport_status = passport.get_antisybil_status(context, user_address)
+    holonym = get_services(context.epoch_state).user_antisybil_holonym_service
+    holonym_status = holonym.get_sbt_status(context, user_address)
+    if passport_status is None or holonym_status is None:
+        return None
 
+    return (passport_status, holonym_status)
 
-def update_user_antisybil_status(user_address: str) -> Tuple[int, datetime]:
+
+def update_user_antisybil_status(
+    user_address: str,
+) -> Tuple[Tuple[int, datetime], Tuple[bool, List[str]]]:
     context = state_context(EpochState.CURRENT)
-    service = get_services(context.epoch_state).user_antisybil_passport_service
+    passport = get_services(context.epoch_state).user_antisybil_passport_service
 
-    score, expires_at, all_stamps = service.fetch_antisybil_status(
+    score, expires_at, all_stamps = passport.fetch_antisybil_status(
         context, user_address
     )
-    service.update_antisybil_status(
+    passport.update_antisybil_status(
         context, user_address, score, expires_at, all_stamps
     )
-    return service.get_antisybil_status(context, user_address)
+
+    holonym = get_services(context.epoch_state).user_antisybil_holonym_service
+
+    has_sbt, cred_type = holonym.fetch_sbt_status(context, user_address)
+    holonym.update_sbt_status(context, user_address, has_sbt, cred_type)
+
+    return ((score, expires_at), (has_sbt, cred_type))

backend/app/modules/user/antisybil/service/holonym.py

@@ -0,0 +1,44 @@
+from flask import current_app as app
+
+from eth_utils.address import to_checksum_address
+
+import json
+from typing import List, Optional, Tuple
+
+from app.extensions import db
+from app.infrastructure import database
+from app.context.manager import Context
+from app.pydantic import Model
+from app.exceptions import UserNotFound
+
+
+class HolonymAntisybil(Model):
+    def get_sbt_status(
+        self, _: Context, user_address: str
+    ) -> Optional[Tuple[bool, List[str]]]:
+        user_address = to_checksum_address(user_address)
+        try:
+            entry = database.user_antisybil.get_sbt_by_address(user_address)
+        except UserNotFound as ex:
+            app.logger.debug(
+                f"User {user_address} antisybil status: except UserNotFound"
+            )
+            raise ex
+
+        if entry is not None:
+            return entry.has_sbt, json.loads(entry.sbt_details)
+        return None
+
+    def fetch_sbt_status(
+        self, _: Context, user_address: str
+    ) -> Optional[Tuple[bool, List[str]]]:
+        from app.infrastructure.external_api.holonym.antisybil import check
+
+        user_address = to_checksum_address(user_address)
+        return check(user_address)
+
+    def update_sbt_status(
+        self, _: Context, user_address: str, has_sbt: bool, cred_type: List[str]
+    ):
+        database.user_antisybil.add_sbt(user_address, has_sbt, cred_type)
+        db.session.commit()

backend/migrations/versions/8b425b454a86_fix_created_at_field.py

@@ -19,14 +19,16 @@
 
 
 def upgrade():
-    if op.get_bind().engine.name == "sqlite":
-        return
-    query = f"UPDATE score_delegation SET created_at = make_date(2024, 7, 17) WHERE hashed_addr IN ('{HASH1}', '{HASH2}', '{HASH3}');"
-    op.execute(query)
+    return
+    # if op.get_bind().engine.name == "sqlite":
+    #     return
+    # query = f"UPDATE score_delegation SET created_at = '2024-07-17T00:00:01.000000' WHERE hashed_addr IN ('{HASH1}', '{HASH2}', '{HASH3}');"
+    # op.execute(query)
 
 
 def downgrade():
-    if op.get_bind().engine.name == "sqlite":
-        return
-    query = f"UPDATE score_delegation SET created_at = NULL WHERE hashed_addr IN ('{HASH1}', '{HASH2}', '{HASH3}');"
-    op.execute(query)
+    return
+    # if op.get_bind().engine.name == "sqlite":
+    #     return
+    # query = f"UPDATE score_delegation SET created_at = NULL WHERE hashed_addr IN ('{HASH1}', '{HASH2}', '{HASH3}');"
+    # op.execute(query)

backend/tests/api-e2e/test_api_antisybil.py

@@ -6,7 +6,25 @@
 
 
 @pytest.mark.api
-def test_antisybil(client: Client, ua_alice: UserAccount):
+def test_holonym(client: Client):
+    address_with_sbt = "0x76273DCC41356e5f0c49bB68e525175DC7e83417"
+    database.user.add_user(address_with_sbt)
+    # check status for a known address with SBT before caching
+    _, code = client.get_antisybil_score(address_with_sbt)
+    assert code == 404  # score for this user is not cached
+
+    _, code = client.refresh_antisybil_score(address_with_sbt)
+    assert code == 204
+
+    score, code = client.get_antisybil_score(address_with_sbt)
+    assert code == 200  # score available
+    assert score["holonym"] == "True"
+    # check after caching
+    # check for address that can't have an SBT (known non-wallet smart-contract)
+
+
+@pytest.mark.api
+def test_passport(client: Client, ua_alice: UserAccount):
     database.user.add_user(ua_alice.address)
     # flow for an address known to GP
     _, code = client.get_antisybil_score(ua_alice.address)

backend/tests/modules/modules_factory/test_modules_factory.py

@@ -34,6 +34,7 @@
 )
 from app.modules.user.allocations.service.saved import SavedUserAllocations
 from app.modules.user.antisybil.service.passport import GitcoinPassportAntisybil
+from app.modules.user.antisybil.service.holonym import HolonymAntisybil
 from app.modules.user.budgets.service.saved import SavedUserBudgets
 from app.modules.user.deposits.service.calculated import CalculatedUserDeposits
 from app.modules.user.deposits.service.contract_balance import (
@@ -141,7 +142,8 @@ def test_pending_services_factory():
     saved_user_budgets = SavedUserBudgets()
     user_nonce = SavedUserAllocationsNonce()
     uniqueness_quotients = PreliminaryUQ(
-        antisybil=GitcoinPassportAntisybil(),
+        passport=GitcoinPassportAntisybil(),
+        holonym=HolonymAntisybil(),
         budgets=saved_user_budgets,
         uq_threshold=UQ_THRESHOLD_MAINNET,
     )

backend/tests/modules/uq/conftest.py

@@ -15,9 +15,17 @@ def mock_antisybil():
 
 
 @pytest.fixture
-def service(mock_antisybil, mock_user_budgets):
+def mock_holonym():
+    mock = Mock()
+    mock.get_sbt_status.return_value = (False, ["phone"])
+    return mock
+
+
+@pytest.fixture
+def service(mock_antisybil, mock_holonym, mock_user_budgets):
     return PreliminaryUQ(
-        antisybil=mock_antisybil,
+        passport=mock_antisybil,
+        holonym=mock_holonym,
         budgets=mock_user_budgets,
         uq_threshold=UQ_THRESHOLD_MAINNET,
     )