feat: DNSADDR protocol completed

Author: acul71

multiaddr/codecs/cid.py

@@ -1,19 +1,20 @@
-from typing import Dict, List
+import logging
 
 import base58
 import cid
-import varint
 
 from ..codecs import CodecBase
-from . import LENGTH_PREFIXED_VAR_SIZE
 from ..exceptions import BinaryParseError
+from . import LENGTH_PREFIXED_VAR_SIZE
+
+logger = logging.getLogger(__name__)
 
 SIZE = LENGTH_PREFIXED_VAR_SIZE
 IS_PATH = False
 
 
 # Spec: https://github.com/libp2p/specs/blob/master/peer-ids/peer-ids.md#string-representation
-CIDv0_PREFIX_TO_LENGTH: Dict[str, List[int]] = {
+CIDv0_PREFIX_TO_LENGTH: dict[str, list[int]] = {
     # base58btc prefixes for valid lengths 1 - 42 with the identity "hash" function
     "12": [5, 12, 19, 23, 30, 41, 52, 56],
     "13": [9, 16, 34, 45],
@@ -75,8 +76,15 @@ def _is_binary_cidv0_multihash(buf: bytes) -> bool:
     """Check if the given bytes represent a CIDv0 multihash."""
     try:
         # CIDv0 is just a base58btc encoded multihash
-        decoded = base58.b58decode(base58.b58encode(buf).decode("ascii"))
-        return len(decoded) == len(buf) and decoded == buf
+        # The first byte is the hash function code, second byte is the length
+        if len(buf) < 2:
+            return False
+        hash_code = buf[0]
+        length = buf[1]
+        if len(buf) != length + 2:  # +2 for the hash code and length bytes
+            return False
+        # For CIDv0, we only support sha2-256 (0x12) and identity (0x00)
+        return hash_code in (0x12, 0x00)
     except Exception:
         return False
 
@@ -85,62 +93,84 @@ class Codec(CodecBase):
     SIZE = SIZE
     IS_PATH = IS_PATH
 
-    def to_bytes(self, proto, value: str) -> bytes:
+    def to_bytes(self, proto, string: str) -> bytes:
         """Convert a CID string to its binary representation."""
-        if not value:
+        if not string:
             raise ValueError("CID string cannot be empty")
 
+        logger.debug(f"[DEBUG CID to_bytes] Input value: {string}")
+
         # First try to parse as CIDv0 (base58btc encoded multihash)
         try:
-            decoded = base58.b58decode(value)
+            decoded = base58.b58decode(string)
             if _is_binary_cidv0_multihash(decoded):
-                # Add length prefix for CIDv0
-                return varint.encode(len(decoded)) + decoded
-        except Exception:
-            pass
+                logger.debug(f"[DEBUG CID to_bytes] Parsed as CIDv0: {decoded.hex()}")
+                # Do not add length prefix here; the framework handles it
+                return decoded
+        except Exception as e:
+            logger.debug(f"[DEBUG CID to_bytes] Failed to parse as CIDv0: {e}")
 
         # If not CIDv0, try to parse as CIDv1
         try:
-            parsed = cid.make_cid(value)
-            # Add length prefix for CIDv1
-            return varint.encode(len(parsed.buffer)) + parsed.buffer
-        except ValueError:
-            raise ValueError(f"Invalid CID: {value}")
+            parsed = cid.make_cid(string)
+
+            # Do not add length prefix here; the framework handles it
+            if not isinstance(parsed.buffer, bytes):
+                raise ValueError("CID buffer must be bytes")
+            return parsed.buffer
+        except ValueError as e:
+            logger.debug(f"[DEBUG CID to_bytes] Failed to parse as CIDv1: {e}")
+            raise ValueError(f"Invalid CID: {string}")
 
     def to_string(self, proto, buf: bytes) -> str:
         """Convert a binary CID to its string representation."""
         if not buf:
             raise ValueError("CID buffer cannot be empty")
 
+        logger.debug(f"[DEBUG CID to_string] Input buffer: {buf.hex()}")
+        logger.debug(f"[DEBUG CID to_string] Protocol: {proto.name}")
+
         expected_codec = PROTO_NAME_TO_CIDv1_CODEC.get(proto.name)
+        logger.debug(f"[DEBUG CID to_string] Expected codec: {expected_codec}")
 
         try:
-            if _is_binary_cidv0_multihash(buf):  # CIDv0
-                if not expected_codec:
-                    # Simply encode as base58btc as there is nothing better to do
-                    return base58.b58encode(buf).decode("ascii")
-
-                # "Implementations SHOULD display peer IDs using the first (raw
-                #  base58btc encoded multihash) format until the second format is
-                #  widely supported."
-                return base58.b58encode(buf).decode("ascii")
-            else:  # CIDv1+
-                parsed = cid.from_bytes(buf)
-
-                # Ensure CID has correct codec for protocol
-                if expected_codec and parsed.codec != expected_codec:
-                    raise ValueError(
-                        '"{0}" multiaddr CIDs must use the "{1}" multicodec'.format(
-                            proto.name, expected_codec
-                        )
+            # First try to parse as CIDv0
+            if _is_binary_cidv0_multihash(buf):
+                result = base58.b58encode(buf).decode("ascii")
+                logger.debug(f"[DEBUG CID to_string] Parsed as CIDv0: {result}")
+                return result
+
+            # If not CIDv0, try to parse as CIDv1
+            parsed = cid.from_bytes(buf)
+            logger.debug(f"[DEBUG CID to_string] Parsed as CIDv1: {parsed}")
+
+            # Ensure CID has correct codec for protocol
+            if expected_codec and parsed.codec != expected_codec:
+                raise ValueError(
+                    '"{}" multiaddr CIDs must use the "{}" multicodec'.format(
+                        proto.name, expected_codec
                     )
-
-                # "Implementations SHOULD display peer IDs using the first (raw
-                #  base58btc encoded multihash) format until the second format is
-                #  widely supported."
-                if expected_codec and _is_binary_cidv0_multihash(parsed.multihash):
-                    return base58.b58encode(parsed.multihash).decode("ascii")
-
-                return parsed.encode("base32").decode("ascii")
+                )
+
+            # For peer IDs (p2p/ipfs), always try to use CIDv0 format if possible
+            if expected_codec:
+                # Try to convert to CIDv0 format
+                try:
+                    # Extract the multihash bytes
+                    multihash = parsed.multihash
+                    logger.debug(f"[DEBUG CID to_string] Extracted multihash: {multihash.hex()}")
+                    # Check if it's a valid CIDv0 multihash
+                    if _is_binary_cidv0_multihash(multihash):
+                        result = base58.b58encode(multihash).decode("ascii")
+                        logger.debug(f"[DEBUG CID to_string] Converted to CIDv0: {result}")
+                        return result
+                except Exception as e:
+                    logger.debug(f"[DEBUG CID to_string] Failed to convert to CIDv0: {e}")
+
+            # If we can't convert to CIDv0, use base32 CIDv1 format
+            result = parsed.encode("base32").decode("ascii")
+            logger.debug(f"[DEBUG CID to_string] Using CIDv1 format: {result}")
+            return result
         except Exception as e:
+            logger.debug(f"[DEBUG CID to_string] Error: {e}")
             raise BinaryParseError(str(e), buf, proto.name, e) from e

multiaddr/codecs/domain.py

@@ -1,6 +1,7 @@
 import idna
-from . import CodecBase, LENGTH_PREFIXED_VAR_SIZE
+
 from ..exceptions import BinaryParseError
+from . import LENGTH_PREFIXED_VAR_SIZE, CodecBase
 
 SIZE = LENGTH_PREFIXED_VAR_SIZE  # Variable size for length-prefixed values
 IS_PATH = False
@@ -10,19 +11,21 @@ class Codec(CodecBase):
     SIZE = SIZE
     IS_PATH = IS_PATH
 
-    def to_bytes(self, proto, value: str) -> bytes:
-        """Convert a domain name string to its binary representation (UTF-8), validating with IDNA."""
-        if not value:
+    def to_bytes(self, proto, string: str) -> bytes:
+        """Convert a domain name string to its binary representation (UTF-8),
+        validating with IDNA."""
+        if not string:
             raise ValueError("Domain name cannot be empty")
         try:
             # Validate using IDNA, but store as UTF-8
-            idna.encode(value, uts46=True)
-            return value.encode("utf-8")
+            idna.encode(string, uts46=True)
+            return string.encode("utf-8")
         except idna.IDNAError as e:
-            raise ValueError(f"Invalid domain name: {str(e)}")
+            raise ValueError(f"Invalid domain name: {e!s}")
 
     def to_string(self, proto, buf: bytes) -> str:
-        """Convert a binary domain name to its string representation (UTF-8), validating with IDNA."""
+        """Convert a binary domain name to its string representation (UTF-8),
+        validating with IDNA."""
         if not buf:
             raise ValueError("Domain name buffer cannot be empty")
         try:
@@ -31,7 +34,7 @@ def to_string(self, proto, buf: bytes) -> str:
             idna.encode(value, uts46=True)
             return value
         except (UnicodeDecodeError, idna.IDNAError) as e:
-            raise BinaryParseError(f"Invalid domain name encoding: {str(e)}", buf, proto.name, e)
+            raise BinaryParseError(f"Invalid domain name encoding: {e!s}", buf, proto.name, e)
 
 
 def to_bytes(proto, string):

multiaddr/codecs/fspath.py

@@ -1,7 +1,10 @@
-import os
+import logging
 import urllib.parse
-from . import CodecBase, LENGTH_PREFIXED_VAR_SIZE
+
 from ..exceptions import BinaryParseError
+from . import LENGTH_PREFIXED_VAR_SIZE, CodecBase
+
+logger = logging.getLogger(__name__)
 
 SIZE = LENGTH_PREFIXED_VAR_SIZE
 IS_PATH = True
@@ -11,47 +14,59 @@ class Codec(CodecBase):
     SIZE = SIZE
     IS_PATH = IS_PATH
 
-    def to_bytes(self, proto, value: str) -> bytes:
+    def to_bytes(self, proto, string: str) -> bytes:
         """Convert a filesystem path to its binary representation."""
-        if not value:
+        logger.debug(f"[DEBUG fspath.to_bytes] input value: {string}")
+        if not string:
             raise ValueError("Path cannot be empty")
 
         # Normalize path separators
-        value = value.replace("\\", "/")
-        
-        # Remove leading/trailing slashes
-        value = value.strip("/")
-        
+        string = string.replace("\\", "/")
+
+        # Remove leading/trailing slashes but preserve path components
+        string = string.strip("/")
+
         # Handle empty path after normalization
-        if not value:
+        if not string:
             raise ValueError("Path cannot be empty after normalization")
-            
+
         # URL decode to handle special characters
-        value = urllib.parse.unquote(value)
-        
+        string = urllib.parse.unquote(string)
+
         # Encode as UTF-8
-        return value.encode("utf-8")
+        encoded = string.encode("utf-8")
+        logger.debug(f"[DEBUG fspath.to_bytes] encoded bytes: {encoded}")
+        return encoded
 
     def to_string(self, proto, buf: bytes) -> str:
         """Convert a binary filesystem path to its string representation."""
+        logger.debug(f"[DEBUG fspath.to_string] input bytes: {buf}")
         if not buf:
             raise ValueError("Path buffer cannot be empty")
 
         try:
             # Decode from UTF-8
             value = buf.decode("utf-8")
-            
+            logger.debug(f"[DEBUG fspath.to_string] decoded value: {value}")
+
             # Normalize path separators
             value = value.replace("\\", "/")
-            
-            # Remove leading/trailing slashes
+
+            # Remove leading/trailing slashes but preserve path components
             value = value.strip("/")
-            
+
             # Handle empty path after normalization
             if not value:
                 raise ValueError("Path cannot be empty after normalization")
-                
+
             # URL encode special characters
-            return urllib.parse.quote(value)
+            result = urllib.parse.quote(value)
+            logger.debug(f"[DEBUG fspath.to_string] output string: {result}")
+
+            # Add leading slash for Unix socket paths
+            if proto.name == "unix":
+                result = "/" + result
+
+            return result
         except UnicodeDecodeError as e:
-            raise BinaryParseError(f"Invalid UTF-8 encoding: {str(e)}", buf, proto.name, e)
+            raise BinaryParseError(f"Invalid UTF-8 encoding: {e!s}", buf, proto.name, e)

multiaddr/codecs/ip4.py

@@ -1,6 +1,7 @@
 import netaddr
 
 from ..codecs import CodecBase
+from ..exceptions import BinaryParseError
 
 SIZE = 32
 IS_PATH = False
@@ -14,4 +15,7 @@ def to_bytes(self, proto, string):
         return netaddr.IPAddress(string, version=4).packed
 
     def to_string(self, proto, buf):
-        return str(netaddr.IPAddress(int.from_bytes(buf, byteorder='big'), version=4))
+        try:
+            return str(netaddr.IPAddress(int.from_bytes(buf, byteorder='big'), version=4))
+        except (ValueError, netaddr.AddrFormatError):
+            raise BinaryParseError("Invalid IPv4 address bytes", buf, "ip4")