OCT-1895: basic holonym integration

backend/app/infrastructure/database/models.py

@@ -52,6 +52,14 @@ class GPStamps(BaseModel):
     stamps = Column(db.String, nullable=False)
 
 
+class HolonymSBT(BaseModel):
+    __tablename__ = "holonym_sbts"
+    id = Column(db.Integer, primary_key=True)
+    user_id = Column(db.Integer, db.ForeignKey("users.id"), nullable=False)
+    has_sbt = Column(db.Boolean, nullable=False, default=False)
+    sbt_details = Column(db.String, nullable=False, default="[]")
+
+
 class PatronModeEvent(BaseModel):
     __tablename__ = "patron_events"
 

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/external_api/holonym/antisybil.py

@@ -0,0 +1,56 @@
+import requests
+from typing import Tuple, List, Union
+from enum import StrEnum
+from eth_utils.address import to_checksum_address
+
+ACTION_ID = "123456789"  # Default, holonym-wide ActionID.
+# Use different ActionID to prevent tracking between protocols consuming Holonym.
+# https://docs.holonym.id/for-developers/custom-sybil-resistance#how-to-set-an-actionid
+
+
+class CredentialType(StrEnum):
+    """
+    CredentialType specifies different methods used by Holonym.
+
+    PHONE_NUMBER_VERIFICATION - phone number verification, code is sent also through e2e messengers
+    GOV_ID_KYC_WITH_ZK - KYC with zk barrier between provider and address
+    E_PASSPORT_ON_DEVICE_WITH_ZK - the most private, on-device only, currently android only
+    """
+
+    PHONE_NUMBER_VERIFICATION = "phone"
+    GOV_ID_KYC_WITH_ZK = "gov-id"
+    E_PASSPORT_ON_DEVICE_WITH_ZK = "epassport"
+
+
+def check(
+    address: str, cred: Union[CredentialType, None] = None
+) -> Tuple[bool, List[str]]:
+    """
+    Check if address has a holonym's SBT.
+
+    Specify credential type. If unspecified, function will query for
+    all credential types.
+
+    Returns a tuple. First element is True if address has at least one SBT.
+    Second element contains a list credential types for which active SBT was found.
+    """
+    address = to_checksum_address(address)
+    if cred is None:
+        creds = list(CredentialType)
+    else:
+        creds = [cred]
+
+    results = {}
+    for cred in creds:
+        results[cred] = _call(address, cred)
+
+    has_sbt = any(results.values())
+    sbt_type = [k for k, v in results.items() if v]
+    return has_sbt, sbt_type
+
+
+def _call(address: str, cred: str) -> bool:
+    dict = requests.get(
+        f"https://api.holonym.io/sybil-resistance/{cred}/optimism?user={address}&action-id={ACTION_ID}"
+    ).json()
+    return dict["result"]

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",
@@ -186,21 +190,24 @@ def patch(self, user_address: str):
 )
 class AntisybilStatus(OctantResource):
     @ns.doc(
-        description="Returns user's antisybil status.",
+        description="""Returns user's antisybil status.""",
     )
     @ns.marshal_with(user_antisybil_status_model)
     @ns.response(200, "User's cached antisybil status retrieved")
+    @ns.response(404, {"status": "Unknown"})
     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:
+        if antisybil_status == (None, None):
             return {"status": "Unknown"}, 404
-        score, expires_at = antisybil_status
+        gitcoin, passport = antisybil_status
         return {
+            "holonym": passport.has_sbt if passport else None,
             "status": "Known",
-            "score": score,
-            "expires_at": int(expires_at.timestamp()),
+            "score": gitcoin.score if gitcoin else None,
+            "expires_at": int(gitcoin.expires_at.timestamp()) if gitcoin else None,
         }, 200
 
     @ns.doc(
@@ -211,9 +218,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}")
+        passport, sbt = status
         app.logger.info(
-            f"User {user_address} antisybil status refreshed {[score, expires_at]}"
+            f"User {user_address} antisybil status refreshed {[passport.score, sbt.has_sbt, passport.expires_at]}"
         )
 
         return {}, 204

backend/app/modules/modules_factory/current.py

@@ -40,7 +40,8 @@
 )
 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.initial import GitcoinPassportAntisybil
+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
@@ -55,7 +56,8 @@ class CurrentServices(Model):
     user_allocations_nonce_service: UserAllocationNonceProtocol
     user_deposits_service: CurrentUserDeposits
     user_tos_service: UserTos
-    user_antisybil_service: GitcoinPassportAntisybil
+    user_antisybil_passport_service: GitcoinPassportAntisybil
+    user_antisybil_holonym_service: HolonymAntisybil
     octant_rewards_service: OctantRewards
     history_service: HistoryService
     simulated_pending_snapshot_service: SimulatePendingSnapshots
@@ -97,7 +99,8 @@ def create(chain_id: int) -> "CurrentServices":
         user_allocations = SavedUserAllocations()
         user_allocations_nonce = SavedUserAllocationsNonce()
         user_withdrawals = FinalizedWithdrawals()
-        user_antisybil_service = GitcoinPassportAntisybil()
+        user_antisybil_passport_service = GitcoinPassportAntisybil()
+        user_antisybil_holonym_service = HolonymAntisybil()
         tos_verifier = InitialUserTosVerifier()
         user_tos = InitialUserTos(verifier=tos_verifier)
         patron_donations = EventsBasedUserPatronMode()
@@ -111,7 +114,7 @@ def create(chain_id: int) -> "CurrentServices":
         score_delegation_verifier = SimpleObfuscationDelegationVerifier()
         score_delegation = SimpleObfuscationDelegation(
             verifier=score_delegation_verifier,
-            antisybil=user_antisybil_service,
+            antisybil=user_antisybil_passport_service,
             user_deposits_service=user_deposits,
         )
 
@@ -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,
         )
@@ -142,7 +146,8 @@ def create(chain_id: int) -> "CurrentServices":
             simulated_pending_snapshot_service=simulated_pending_snapshot_service,
             multisig_signatures_service=multisig_signatures,
             user_tos_service=user_tos,
-            user_antisybil_service=user_antisybil_service,
+            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

@@ -38,9 +38,10 @@
     PendingUserAllocations,
     PendingUserAllocationsVerifier,
 )
-from app.modules.user.antisybil.service.initial import (
+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/score_delegation/service/simple_obfuscation.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Protocol, runtime_checkable, Tuple, Container, List
+from typing import Protocol, runtime_checkable, Tuple, Container, List, Optional
 
 from app.context.manager import Context
 from app.extensions import db
@@ -9,6 +9,7 @@
 from app.modules.dto import ScoreDelegationPayload
 from app.modules.score_delegation import core
 from app.modules.score_delegation.core import ActionType
+from app.modules.user.antisybil.service.passport import GitcoinAntisybilDTO
 from app.pydantic import Model
 
 from flask import current_app as app
@@ -20,7 +21,7 @@
 class Antisybil(Protocol):
     def fetch_antisybil_status(
         self, _: Context, user_address: str
-    ) -> Tuple[float, datetime, any]:
+    ) -> Optional[GitcoinAntisybilDTO]:
         ...
 
     def update_antisybil_status(
@@ -85,7 +86,7 @@ def _delegation(
         hashed_addresses = get_hashed_addresses(
             payload.primary_addr, payload.secondary_addr
         )
-        score, expires_at, stamps = self.antisybil.fetch_antisybil_status(
+        score, stamps = self.antisybil.fetch_antisybil_status(
             context, payload.secondary_addr
         )
         secondary_budget = self.user_deposits_service.get_user_effective_deposit(
@@ -95,10 +96,10 @@ def _delegation(
             context,
             hashed_addresses=hashed_addresses,
             payload=payload,
-            score=score,
+            score=score.score,
             secondary_budget=secondary_budget,
             action_type=action,
         )
         self.antisybil.update_antisybil_status(
-            context, payload.primary_addr, score, expires_at, stamps
+            context, payload.primary_addr, score.score, score.expires_at, stamps
         )

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

@@ -1,6 +1,5 @@
-from datetime import datetime
 from decimal import Decimal
-from typing import Protocol, Optional, Tuple, runtime_checkable
+from typing import Protocol, Optional, runtime_checkable
 
 from app.context.manager import Context
 from app.infrastructure.database.uniqueness_quotient import (
@@ -9,13 +8,23 @@
 )
 from app.modules.uq.core import calculate_uq
 from app.pydantic import Model
+from app.modules.user.antisybil.service.holonym import HolonymAntisybilDTO
+from app.modules.user.antisybil.service.passport import GitcoinAntisybilDTO
 
 
 @runtime_checkable
-class Antisybil(Protocol):
+class Passport(Protocol):
     def get_antisybil_status(
         self, _: Context, user_address: str
-    ) -> Optional[Tuple[float, datetime]]:
+    ) -> Optional[GitcoinAntisybilDTO]:
+        ...
+
+
+@runtime_checkable
+class Holonym(Protocol):
+    def get_sbt_status(
+        self, _: Context, user_address: str
+    ) -> Optional[HolonymAntisybilDTO]:
         ...
 
 
@@ -26,7 +35,8 @@ 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 +66,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]