Feat/ens avatar

ens/abis.py

@@ -1209,3 +1209,23 @@
         "type": "constructor",
     },
 ]
+
+ERC_721_TOKEN_URI = [
+    {
+        "name": "tokenURI",
+        "type": "function",
+        "stateMutability": "view",
+        "inputs": [{"name": "tokenId", "type": "uint256"}],
+        "outputs": [{"name": "", "type": "string"}],
+    }
+]
+
+ERC_1155_TOKEN_URI = [
+    {
+        "name": "uri",
+        "type": "function",
+        "stateMutability": "view",
+        "inputs": [{"name": "_id", "type": "uint256"}],
+        "outputs": [{"name": "", "type": "string"}],
+    }
+]

ens/async_ens.py

@@ -1,6 +1,7 @@
 from copy import (
     deepcopy,
 )
+import re
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -47,13 +48,16 @@
 )
 from ens.exceptions import (
     AddressMismatch,
+    EnsAvatarError,
     ENSValueError,
     ResolverNotFound,
     UnauthorizedError,
     UnownedName,
     UnsupportedFunction,
 )
 from ens.utils import (
+    UriItem,
+    _parse_avatar_uri,
     address_in,
     address_to_reverse_domain,
     default,
@@ -64,6 +68,7 @@
     label_to_hash,
     normal_name_to_hash,
     normalize_name,
+    parse_nft_uri,
     raw_name_to_hash,
 )
 
@@ -431,6 +436,65 @@ async def set_text(
             name, r.functions.setText, (node, key, value), transact
         )
 
+    async def get_avatar(
+        self, name: str, gateway_urls: Optional[dict[str, str]] = None
+    ) -> Optional[str]:
+        """
+        Resolve the 'avatar' text record for `name` into a usable URI when possible.
+        For more details checkout out [ENSIP 15](https://docs.ens.domains/ensip/15/)
+
+        :param str name: ENS name to look up
+        :param dict gateway_urls: optional mapping
+        of gateway overrides,
+        e.g. {"ipfs": "...", "arweave": "..."}
+        :return: resolved avatar URI or None
+        :raises EnsAvatarError: when avatar resolution fails in specific ways
+        """
+        record = await self.get_text(name, "avatar")
+        if not record:
+            return None
+        gateway_urls = gateway_urls or {}
+        # If the avatar record contains an eip155 NFT URI, resolve token URI first
+        if re.search(r"eip155:", record, re.IGNORECASE):
+            return await self._parse_nft_avatar_uri(record, gateway_urls)
+        item = _parse_avatar_uri(record, gateway_urls)
+        if isinstance(item, UriItem):
+            return item.uri
+        raise EnsAvatarError(f"Unexpected avatar parse result for {name!r}")
+
+    async def _parse_nft_avatar_uri(
+        self, record: str, gateway_urls: Optional[dict[str, str]] = None
+    ) -> Optional[str]:
+        parsed_nft = parse_nft_uri(record)
+        nft_uri: Optional[str] = None
+        if parsed_nft.namespace == "erc721":
+            nft_token_contract = self.w3.eth.contract(
+                abi=abis.ERC_721_TOKEN_URI,
+                address=self.w3.to_checksum_address(parsed_nft.contract_address),
+            )
+            nft_uri = await nft_token_contract.functions.tokenURI(
+                int(parsed_nft.token_id)
+            ).call()
+        if parsed_nft.namespace == "erc1155":
+            nft_token_contract = self.w3.eth.contract(
+                abi=abis.ERC_1155_TOKEN_URI,
+                address=self.w3.to_checksum_address(parsed_nft.contract_address),
+            )
+            nft_uri = await nft_token_contract.functions.uri(
+                int(parsed_nft.token_id)
+            ).call()
+        if not nft_uri:
+            raise EnsAvatarError(
+                f"""
+                Could not resolve to token uri
+                given contract address of {parsed_nft.contract_address}
+                """
+            )
+        avatar = _parse_avatar_uri(nft_uri, gateway_urls)
+        if isinstance(avatar, UriItem):
+            return avatar.uri
+        raise EnsAvatarError(f"Unexpected avatar parse result for {record}")
+
     # -- private methods -- #
 
     async def _get_resolver(

ens/constants.py

@@ -1,3 +1,5 @@
+import re
+
 from eth_typing import (
     ChecksumAddress,
     HexAddress,
@@ -32,3 +34,30 @@
 ENS_CONTENT_HASH_INTERFACE_ID = HexStr("0xbc1c58d1")
 ENS_MULTICHAIN_ADDRESS_INTERFACE_ID = HexStr("0xf1cb7e06")  # ENSIP-9
 ENS_EXTENDED_RESOLVER_INTERFACE_ID = HexStr("0x9061b923")  # ENSIP-10
+
+# --- avatar resolution regex --- #
+NETWORK_REGEX = re.compile(
+    r"""
+    (?P<protocol>https?:\/\/[^/]*|ipfs:\/|ipns:\/|ar:\/)?
+    (?P<root>\/)?
+    (?P<subpath>ipfs\/|ipns\/)?
+    (?P<target>[\w\-.]+)
+    (?P<subtarget>\/.*)?
+    """,
+    re.VERBOSE,
+)
+
+IPFS_HASH_REGEX = re.compile(
+    r"""
+    ^(Qm[1-9A-HJ-NP-Za-km-z]{44,}
+    |b[A-Za-z2-7]{58,}
+    |B[A-Z2-7]{58,}
+    |z[1-9A-HJ-NP-Za-km-z]{48,}
+    |F[0-9A-F]{50,})
+    (\/(?P<target>[\w\-.]+))?
+    (?P<subtarget>\/.*)?$
+    """,
+    re.VERBOSE,
+)
+BASE64_REGEX = re.compile(r"^data:([a-zA-Z\-/+]*);base64,([^\"].*)")
+DATA_URI_REGEX = re.compile(r"^data:([a-zA-Z\-/+]*)?(;[a-zA-Z0-9].*?)?(,)")

ens/ens.py

@@ -1,6 +1,7 @@
 from copy import (
     deepcopy,
 )
+import re
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -46,13 +47,16 @@
 )
 from .exceptions import (
     AddressMismatch,
+    EnsAvatarError,
     ENSValueError,
     ResolverNotFound,
     UnauthorizedError,
     UnownedName,
     UnsupportedFunction,
 )
 from .utils import (
+    UriItem,
+    _parse_avatar_uri,
     address_in,
     address_to_reverse_domain,
     default,
@@ -63,6 +67,7 @@
     label_to_hash,
     normal_name_to_hash,
     normalize_name,
+    parse_nft_uri,
     raw_name_to_hash,
 )
 
@@ -422,7 +427,61 @@ def set_text(
             name, r.functions.setText, (node, key, value), transact
         )
 
+    def get_avatar(
+        self, name: str, gateway_urls: Optional[dict[str, str]] = None
+    ) -> Optional[str]:
+        """
+        Resolve the 'avatar' text record for `name` into a usable URI when possible.
+        For more details checkout out [ENSIP 15](https://docs.ens.domains/ensip/15/)
+
+        :param str name: ENS name to look up
+        :param dict gateway_urls: optional mapping of gateway overrides,
+        e.g. {"ipfs": "...", "arweave": "..."}
+        :return: resolved avatar URI or None
+        :raises EnsAvatarError: when avatar resolution fails in specific ways
+        """
+        record = self.get_text(name, "avatar")
+        if not record:
+            return None
+        gateway_urls = gateway_urls or {}
+        if re.search(r"eip155:", record, re.IGNORECASE):
+            return self._parse_nft_avatar_uri(record, gateway_urls)
+        item = _parse_avatar_uri(record, gateway_urls)
+        if isinstance(item, UriItem):
+            return item.uri
+        raise EnsAvatarError(f"Unexpected avatar parse result for {name!r}")
+
     # -- private methods -- #
+    def _parse_nft_avatar_uri(
+        self, record: str, gateway_urls: Optional[dict[str, str]] = None
+    ) -> Optional[str]:
+        parsed_nft = parse_nft_uri(record)
+        nft_uri: Optional[str] = None
+        if parsed_nft.namespace == "erc721":
+            nft_token_contrat = self.w3.eth.contract(
+                abi=abis.ERC_721_TOKEN_URI,
+                address=self.w3.to_checksum_address(parsed_nft.contract_address),
+            )
+            nft_uri = nft_token_contrat.functions.tokenURI(
+                int(parsed_nft.token_id)
+            ).call()
+        if parsed_nft.namespace == "erc1155":
+            nft_token_contrat = self.w3.eth.contract(
+                abi=abis.ERC_1155_TOKEN_URI,
+                address=self.w3.to_checksum_address(parsed_nft.contract_address),
+            )
+            nft_uri = nft_token_contrat.functions.uri(int(parsed_nft.token_id)).call()
+        if not nft_uri:
+            raise EnsAvatarError(
+                f"""
+                Could not resolve to token uri
+                given contract address of {parsed_nft.contract_address}
+                """
+            )
+        avatar = _parse_avatar_uri(nft_uri, gateway_urls)
+        if isinstance(avatar, UriItem):
+            return avatar.uri
+        raise EnsAvatarError(f"Unexpected avatar parse result for {record}")
 
     def _get_resolver(
         self,

ens/exceptions.py

@@ -104,3 +104,21 @@ class ENSValidationError(ENSException):
     """
     Raised if there is a validation error
     """
+
+
+class EnsAvatarError(ENSException):
+    """
+    Raised when an avatar resolution fails in specific ways
+    """
+
+
+class EnsAvatarUriResolutionError(ENSException):
+    """
+    Raised when an avatar URI cannot be resolved to a usable form.
+    """
+
+
+class EnsAvatarInvalidNftUriError(ENSException):
+    """
+    Raised when an NFT avatar URI is malformed or unsupported.
+    """