Skip to content

Commit

Permalink
Add Ens Client (#814)
Browse files Browse the repository at this point in the history
  • Loading branch information
@falvaradorodriguez
falvaradorodriguez authored Mar 5, 2024
1 parent 4eee07e commit e4aeea4
Show file tree
Hide file tree
Showing 3 changed files with 293 additions and 0 deletions.
2 changes: 2 additions & 0 deletions gnosis/eth/clients/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
BlockScoutConfigurationProblem,
)
from .contract_metadata import ContractMetadata
from .ens_client import EnsClient
from .etherscan_client import (
EtherscanClient,
EtherscanClientConfigurationProblem,
Expand All @@ -22,6 +23,7 @@
"BlockscoutClient",
"BlockscoutClientException",
"ContractMetadata",
"EnsClient",
"EtherscanClient",
"EtherscanClientConfigurationProblem",
"EtherscanClientException",
Expand Down
149 changes: 149 additions & 0 deletions gnosis/eth/clients/ens_client.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,149 @@
from typing import Any, Dict, List, Optional, Union

import requests
from eth_typing import HexStr
from hexbytes import HexBytes

from gnosis.eth import EthereumNetwork
from gnosis.util import cache


class EnsClient:
"""
Resolves Ethereum Name Service domains using ``thegraph`` API
"""

def __init__(self, network_id: int):
self.ethereum_network = EthereumNetwork(network_id)
if network_id == self.ethereum_network.SEPOLIA:
url = (
"https://api.studio.thegraph.com/proxy/49574/enssepolia/version/latest/"
)
else: # Fallback to mainnet
url = "https://api.thegraph.com/subgraphs/name/ensdomains/ens/"
self.url = url
self.request_timeout = 5 # Seconds
self.request_session = requests.Session()

def is_available(self) -> bool:
"""
:return: True if service is available, False if it's down
"""
try:
return self.request_session.get(self.url, timeout=self.request_timeout).ok
except IOError:
return False

@staticmethod
def domain_hash_to_hex_str(domain_hash: Union[HexStr, bytes, int]) -> HexStr:
"""
:param domain_hash:
:return: Domain hash as an hex string of 66 chars (counting with 0x), padding with zeros if needed
"""
if not domain_hash:
domain_hash = b""
return HexStr("0x" + HexBytes(domain_hash).hex()[2:].rjust(64, "0"))

@cache
def _query_by_domain_hash(self, domain_hash_str: HexStr) -> Optional[str]:
query = """
{
domains(where: {labelhash: "domain_hash"}) {
labelName
}
}
""".replace(
"domain_hash", domain_hash_str
)
try:
response = self.request_session.post(
self.url, json={"query": query}, timeout=self.request_timeout
)
except IOError:
return None

"""
Example:
{
"data": {
"domains": [
{
"labelName": "safe-multisig"
}
]
}
}
"""
if response.ok:
data = response.json()
if data:
domains = data.get("data", {}).get("domains")
if domains:
return domains[0].get("labelName")
return None

def query_by_domain_hash(
self, domain_hash: Union[HexStr, bytes, int]
) -> Optional[str]:
"""
Get domain label from domain_hash (keccak of domain name without the TLD, don't confuse with namehash)
used for ENS ERC721 token_id. Use another method for caching purposes (use same parameter type)
:param domain_hash: keccak of domain name without the TLD, don't confuse with namehash. E.g. For
batman.eth it would be just keccak('batman')
:return: domain label if found
"""
domain_hash_str = self.domain_hash_to_hex_str(domain_hash)
return self._query_by_domain_hash(domain_hash_str)

def query_by_account(self, account: str) -> Optional[List[Dict[str, Any]]]:
"""
:param account: ethereum account to search for ENS registered addresses
:return: None if there's a problem or not found, otherwise example of dictionary returned:
{
"registrations": [
{
"domain": {
"isMigrated": true,
"labelName": "gilfoyle",
"labelhash": "0xadfd886b420023026d5c0b1be0ffb5f18bb2f37143dff545aeaea0d23a4ba910",
"name": "gilfoyle.eth",
"parent": {
"name": "eth"
}
},
"expiryDate": "1905460880"
}
]
}
"""
query = """query getRegistrations {
account(id: "account_id") {
registrations {
expiryDate
domain {
labelName
labelhash
name
isMigrated
parent {
name
}
}
}
}
}""".replace(
"account_id", account.lower()
)
try:
response = self.request_session.post(
self.url, json={"query": query}, timeout=self.request_timeout
)
except IOError:
return None

if response.ok:
data = response.json()
if data:
return data.get("data", {}).get("account")
return None
142 changes: 142 additions & 0 deletions gnosis/eth/tests/clients/test_ens_client.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,142 @@
from unittest import mock

from django.test import TestCase

from eth_utils import keccak
from requests import Session

from gnosis.eth.ethereum_client import EthereumNetwork

from ...clients import EnsClient


class TestEnsClient(TestCase):
def test_domain_hash_to_hex_str(self):
domain_hash_bytes = keccak(text="gnosis")
domain_hash_int = int.from_bytes(domain_hash_bytes, byteorder="big")

result = EnsClient.domain_hash_to_hex_str(domain_hash_bytes)
self.assertEqual(result, EnsClient.domain_hash_to_hex_str(domain_hash_int))
self.assertEqual(len(result), 66)

self.assertEqual(len(EnsClient.domain_hash_to_hex_str(b"")), 66)
self.assertEqual(len(EnsClient.domain_hash_to_hex_str(None)), 66)
self.assertEqual(len(EnsClient.domain_hash_to_hex_str(2)), 66)

def test_query_by_account(self):
ens_client = EnsClient(EthereumNetwork.MAINNET.value)
if not ens_client.is_available():
self.skipTest("ENS Mainnet Client is not available")

self.assertEqual(
ens_client.query_by_account("0x70608b1809c93Ec57160C266a38322144E9A9d28"),
{
"registrations": [
{
"expiryDate": "1763372829",
"domain": {
"labelName": "safe-treasury",
"labelhash": "0x136ff778d0f4bb244b1284dd5835c78a9fb425680d3a75aab24db723042494af",
"name": "safe-treasury.eth",
"isMigrated": True,
"parent": {"name": "eth"},
},
},
{
"expiryDate": "1775002833",
"domain": {
"labelName": "gnosis-safe",
"labelhash": "0x162be7f136f104c8cc5ce333cdb2ef94fa8270f4ca186ba6083634b8b93efa82",
"name": "gnosis-safe.eth",
"isMigrated": True,
"parent": {"name": "eth"},
},
},
{
"expiryDate": "1763371973",
"domain": {
"labelName": "safe-dao",
"labelhash": "0x3dcf430070cc5f52fbe66433a72fc6eed2860b28527f9016933599d41cbf6d9e",
"name": "safe-dao.eth",
"isMigrated": True,
"parent": {"name": "eth"},
},
},
{
"expiryDate": "1763373078",
"domain": {
"labelName": "safe-foundation",
"labelhash": "0x50270c4c4cf9837870f71a836cc4ab37d29e0a452eda3caa1b39cc8a29b96e90",
"name": "safe-foundation.eth",
"isMigrated": True,
"parent": {"name": "eth"},
},
},
{
"expiryDate": "1824427603",
"domain": {
"labelName": "safe",
"labelhash": "0xc318ae71df18dafd8fbd063284586ea242aa3d51bc2950f71d70d7fc205b875f",
"name": "safe.eth",
"isMigrated": True,
"parent": {"name": "eth"},
},
},
{
"expiryDate": "1763372390",
"domain": {
"labelName": "safe-token",
"labelhash": "0xc9ccb8a54110c76c01d4f63e9a9d760d8fd803aba14f4d2fa408200cc6b68cba",
"name": "safe-token.eth",
"isMigrated": True,
"parent": {"name": "eth"},
},
},
{
"expiryDate": "1763309068",
"domain": {
"labelName": "safe-multisig",
"labelhash": "0xce3f8bfd04bb347a13abbf6faca8dc5e4a281345a316019206742b60b6f1b053",
"name": "safe-multisig.eth",
"isMigrated": True,
"parent": {"name": "eth"},
},
},
{
"expiryDate": "1764337847",
"domain": {
"labelName": "takebackownership",
"labelhash": "0xedc916efb805eea66b4d5496f670c0166ccd9d2453ded805fe1d82738944e8df",
"name": "takebackownership.eth",
"isMigrated": True,
"parent": {"name": "eth"},
},
},
]
},
)

def test_query_by_domain_hash(self):
ens_client = EnsClient(EthereumNetwork.MAINNET.value) # Mainnet
if not ens_client.is_available():
self.skipTest("ENS Mainnet Client is not available")

# Query for gnosis domain
domain_hash = keccak(text="gnosis")
self.assertEqual("gnosis", ens_client.query_by_domain_hash(domain_hash))

domain_hash_2 = keccak(
text="notverycommon-domain-name-made-up-by-me-with-forbidden-word-ñ"
)
self.assertIsNone(ens_client.query_by_domain_hash(domain_hash_2))

def test_is_available(self):
for ethereum_network in (
EthereumNetwork.ROPSTEN,
EthereumNetwork.MAINNET,
):
with self.subTest(ethereum_network=ethereum_network):
ens_client = EnsClient(ethereum_network)
self.assertTrue(ens_client.is_available())
with mock.patch.object(Session, "get", side_effect=IOError()):
self.assertFalse(ens_client.is_available())

0 comments on commit e4aeea4

Please sign in to comment.