OCT-1655 create an endpoint to get linked pair of addresses (#238)

## Description

## Definition of Done

1. [ ] Acceptance criteria are met.
2. [ ] PR is manually tested before the merge by developer(s).
    - [ ] Happy path is manually checked.
3. [ ] PR is manually tested by QA when their assistance is required
(1).
- [ ] Octant Areas & Test Cases are checked for impact and updated if
required (2).
4. [ ] Unit tests are added unless there is a reason to omit them.
5. [ ] Automated tests are added when required.
6. [ ] The code is merged.
7. [ ] Tech documentation is added / updated, reviewed and approved
(including mandatory approval by a code owner, should such exist for
changed files).
    - [ ] BE: Swagger documentation is updated.
8. [ ] When required by QA:
    - [ ] Deployed to the relevant environment.
    - [ ] Passed system tests.

---

(1) Developer(s) in coordination with QA decide whether it's required.
For small tickets introducing small changes QA assistance is most
probably not required.

(2) [Octant Areas & Test
Cases](https://docs.google.com/spreadsheets/d/1cRe6dxuKJV3a4ZskAwWEPvrFkQm6rEfyUCYwLTYw_Cc).

backend/app/exceptions.py

@@ -300,6 +300,14 @@ def __init__(self):
         super().__init__(self.description, self.code)
 
 
+class DelegationCheckWrongParams(OctantException):
+    code = 400
+    description = "Please specify at least 2 and not more than 10 addresses"
+
+    def __init__(self):
+        super().__init__(self.description, self.code)
+
+
 class AntisybilScoreTooLow(OctantException):
     code = 400
     description = "Antisybil score {} is lower then {}"

backend/app/infrastructure/routes/delegation.py

@@ -34,6 +34,21 @@
 )
 
 
+score_delegation_check_model = api.model(
+    "ScoreDelegationCheckResult",
+    {
+        "primary": fields.String(
+            required=False,
+            description="Address that receives delegated score",
+        ),
+        "secondary": fields.String(
+            required=False,
+            description="Address that donates delegated score",
+        ),
+    },
+)
+
+
 @ns.route("/delegate")
 @ns.doc(
     description="Delegates UQ score from secondary address to primary address",
@@ -64,3 +79,18 @@ def put(self):
         controller.recalculate_uq_score(ns.payload)
 
         return {}, 204
+
+
+@ns.route("/check/<string:addresses>")
+@ns.doc(
+    description="Allows wallet to check if its accounts are delegating to each other. Implementation of this feature relies on a fact that Ethereum has > 250mil addresses, so blind enumeration is hard. We intend to replace it with proper zk-based delegation as soon as possible",
+    params={
+        "addresses": "Ethereum addresses in hexadecimal format (case-insensitive, prefixed with 0x), separated by comma. At most 10 addresses at the same time"
+    },
+)
+class UQScoreDelegationCheck(OctantResource):
+    @ns.marshal_with(score_delegation_check_model)
+    @ns.response(200, "User's delegations reconstructed")
+    def get(self, addresses: str):
+        secondary, primary = controller.delegation_check(addresses)
+        return {"primary": primary, "secondary": secondary}, 200

backend/app/modules/modules_factory/protocols.py

@@ -238,3 +238,6 @@ 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]]:
+        ...

backend/app/modules/score_delegation/controller.py

@@ -1,7 +1,10 @@
-from eth_utils import to_checksum_address
+from typing import Tuple
+
+from eth_utils.address import to_checksum_address
 
 from app.context.epoch_state import EpochState
 from app.context.manager import state_context
+from app.exceptions import DelegationCheckWrongParams, DelegationDoesNotExist
 from app.modules.dto import ScoreDelegationPayload
 from app.modules.modules_factory.current import CurrentServices
 from app.modules.registry import get_services
@@ -21,6 +24,22 @@ def recalculate_uq_score(payload: dict):
     services.score_delegation_service.recalculate(context, score_delegation_payload)
 
 
+def delegation_check(addresses: str) -> Tuple[str, str]:
+    tokens = addresses.split(",")
+    if len(tokens) < 2:
+        raise DelegationCheckWrongParams()
+    if len(tokens) > 10:
+        raise DelegationCheckWrongParams()
+    context = state_context(EpochState.CURRENT)
+    services: CurrentServices = get_services(EpochState.CURRENT)
+    pairs = list(services.score_delegation_service.check(context, tokens))
+    if not pairs:
+        raise DelegationDoesNotExist()
+    if len(pairs) > 1:
+        raise DelegationDoesNotExist()
+    return pairs[0]
+
+
 def _deserialize_payload(payload: dict) -> ScoreDelegationPayload:
     return ScoreDelegationPayload(
         primary_addr=to_checksum_address(payload["primaryAddr"]),

backend/app/modules/score_delegation/core.py

@@ -1,6 +1,9 @@
-import hashlib
-from collections import namedtuple
+from itertools import permutations
+from typing import NamedTuple
 from enum import Enum
+from eth_utils.address import to_checksum_address
+import hashlib
+from typing import Tuple
 
 from app.exceptions import (
     DelegationAlreadyExists,
@@ -18,9 +21,11 @@
 
 MIN_SCORE = 15
 
-HashedAddresses = namedtuple(
-    "HashedAddresses", ["primary_addr_hash", "secondary_addr_hash", "both_hash"]
-)
+
+class HashedAddresses(NamedTuple):
+    primary_addr_hash: str
+    secondary_addr_hash: str
+    both_hash: str
 
 
 class ActionType(Enum):
@@ -33,10 +38,18 @@ def build_score_delegation_message(primary_addr: str, secondary_addr: str) -> st
 
 
 def get_hashed_addresses(
-    payload: ScoreDelegationPayload, salt: str, salt_primary: str
+    payload: ScoreDelegationPayload,
+    salt: str,
+    salt_primary: str,
+    normalize: bool = True,
 ) -> HashedAddresses:
-    primary_addr_data = salt_primary + payload.primary_addr
-    secondary_addr_data = salt + payload.secondary_addr
+    primary = payload.primary_addr
+    secondary = payload.secondary_addr
+    if normalize:
+        primary = to_checksum_address(primary)
+        secondary = to_checksum_address(secondary)
+    primary_addr_data = salt_primary + primary
+    secondary_addr_data = salt + secondary
 
     hashed_primary = hashlib.sha256(primary_addr_data.encode()).hexdigest()
     hashed_secondary = hashlib.sha256(secondary_addr_data.encode()).hexdigest()
@@ -47,6 +60,29 @@ def get_hashed_addresses(
     return HashedAddresses(hashed_primary, hashed_secondary, hashed_both)
 
 
+def delegation_check(
+    addresses: list[str],
+    all_hashes: set[str],
+    salt: str,
+    salt_primary: str,
+    normalize=True,
+) -> set[Tuple[str, str]]:
+    result = []
+    for secondary, primary in permutations(addresses, 2):
+        payload = ScoreDelegationPayload(
+            primary_addr=primary,
+            secondary_addr=secondary,
+            primary_addr_signature=None,
+            secondary_addr_signature=None,
+        )
+        _, _, both = get_hashed_addresses(
+            payload, salt, salt_primary, normalize=normalize
+        )
+        if both in all_hashes:
+            result.append((secondary, primary))
+    return set(result)
+
+
 def verify_score_delegation(
     hashed_addresses: HashedAddresses,
     all_hashes: set[str],

backend/app/modules/score_delegation/service/simple_obfuscation.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Protocol, runtime_checkable
+from typing import Protocol, runtime_checkable, Tuple, Container
 
 from app.context.manager import Context
 from app.extensions import db
@@ -17,7 +17,7 @@
 class Antisybil(Protocol):
     def fetch_antisybil_status(
         self, _: Context, user_address: str
-    ) -> (float, datetime, any):
+    ) -> Tuple[float, datetime, any]:
         ...
 
     def update_antisybil_status(
@@ -66,6 +66,15 @@ def recalculate(self, context: Context, payload: ScoreDelegationPayload):
         self._delegation(context, payload, ActionType.RECALCULATION)
         db.session.commit()
 
+    def check(self, _: Context, addresses: list[str]) -> Container[Tuple[str, str]]:
+        all_hashes = database.score_delegation.get_all_delegations()
+        return core.delegation_check(
+            addresses,
+            all_hashes,
+            app.config["DELEGATION_SALT"],
+            app.config["DELEGATION_SALT_PRIMARY"],
+        )
+
     def _delegation(
         self, context: Context, payload: ScoreDelegationPayload, action: ActionType
     ):