fix: fix many typecheck errors

Author: acul71

multiaddr/codecs/__init__.py

@@ -1,22 +1,44 @@
 import importlib
+from typing import Dict, Union, Any
 
 
 # These are special sizes
 LENGTH_PREFIXED_VAR_SIZE = -1
 
 
-class NoneCodec:
-    SIZE = 0
-    IS_PATH = False
+class CodecBase:
+    SIZE: int
+    IS_PATH: bool
 
+    def to_string(self, proto: Any, buf: bytes) -> str:
+        raise NotImplementedError
 
-CODEC_CACHE = {}
+    def to_bytes(self, proto: Any, string: str) -> bytes:
+        raise NotImplementedError
 
 
-def codec_by_name(name):
-    if name is None:  # Special “do nothing – expect nothing” pseudo-codec
-        return NoneCodec
+class NoneCodec(CodecBase):
+    SIZE: int = 0
+    IS_PATH: bool = False
+
+    def to_string(self, proto: Any, buf: bytes) -> str:
+        return ""
+
+    def to_bytes(self, proto: Any, string: str) -> bytes:
+        return b""
+
+
+CODEC_CACHE: Dict[str, CodecBase] = {}
+
+
+def codec_by_name(name: Union[str, None]) -> CodecBase:
+    if name is None:  # Special "do nothing – expect nothing" pseudo-codec
+        return NoneCodec()
     codec = CODEC_CACHE.get(name)
-    if not codec:
-        codec = CODEC_CACHE[name] = importlib.import_module(".{0}".format(name), __name__)
+    if codec is None:
+        module = importlib.import_module(f".{name}", __name__)
+        codec_class = getattr(module, "Codec")
+        assert codec_class is not None, f"Codec {name} not found"
+        codec = codec_class()
+        CODEC_CACHE[name] = codec
     return codec

multiaddr/codecs/cid.py

@@ -1,16 +1,17 @@
 import base58
 import cid
+from typing import Dict, List
+from ..codecs import CodecBase
 
 from . import LENGTH_PREFIXED_VAR_SIZE
 
-
 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 = {
-    # base58btc prefixes for valid lengths 1 – 42 with the identity “hash” function
+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],
     '14': [27, 38, 49, 60],
@@ -62,71 +63,82 @@
     'Qm': [46],
 }
 
-PROTO_NAME_TO_CIDv1_CODEC = {
-    # The “p2p” multiaddr protocol requires all keys to use the “libp2p-key” multicodec
+PROTO_NAME_TO_CIDv1_CODEC: Dict[str, str] = {
+    # The "p2p" multiaddr protocol requires all keys to use the "libp2p-key" multicodec
     "p2p": "libp2p-key",
 }
 
 
-def to_bytes(proto, string):
-    expected_codec = PROTO_NAME_TO_CIDv1_CODEC.get(proto.name)
-
-    if len(string) in CIDv0_PREFIX_TO_LENGTH.get(string[0:2], ()):  # CIDv0
-        # Upgrade the wire (binary) representation of any received CIDv0 string
-        # to CIDv1 if we can determine which multicodec value to use
-        if expected_codec:
-            return cid.make_cid(1, expected_codec, base58.b58decode(string)).buffer
-
-        return base58.b58decode(string)
-    else:  # CIDv1+
-        parsed = cid.from_string(string)
-
-        # 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))
-
-        return parsed.buffer
-
-
-def _is_binary_cidv0_multihash(buf):
+class Codec(CodecBase):
+    SIZE = SIZE
+    IS_PATH = IS_PATH
+
+    def to_bytes(self, proto, string):
+        expected_codec = PROTO_NAME_TO_CIDv1_CODEC.get(proto.name)
+
+        if len(string) in CIDv0_PREFIX_TO_LENGTH.get(string[0:2], ()):  # CIDv0
+            # Upgrade the wire (binary) representation of any received CIDv0 string
+            # to CIDv1 if we can determine which multicodec value to use
+            if expected_codec:
+                cid_obj = cid.make_cid(1, expected_codec, base58.b58decode(string))
+                assert isinstance(cid_obj.buffer, bytes)
+                return cid_obj.buffer
+
+            return base58.b58decode(string)
+        else:  # CIDv1+
+            parsed = cid.from_string(string)
+
+            # 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
+                    )
+                )
+
+            return parsed.buffer
+
+    def to_string(self, proto, buf):
+        expected_codec = PROTO_NAME_TO_CIDv1_CODEC.get(proto.name)
+
+        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."
+            #
+            # In the future the following line should instead convert the multihash
+            # to CIDv1 and with the `expected_codec` and wrap it in base32:
+            #   return cid.make_cid(1, expected_codec, buf).encode("base32").decode("ascii")
+            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
+                    )
+                )
+
+            # "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")
+
+
+def _is_binary_cidv0_multihash(buf: bytes) -> bool:
     if buf.startswith(b"\x12\x20") and len(buf) == 34:  # SHA2-256
         return True
 
     if (buf[0] == 0x00 and buf[1] in range(43)) and len(buf) == (buf[1] + 2):  # Identity hash
         return True
 
     return False
-
-
-def to_string(proto, buf):
-    expected_codec = PROTO_NAME_TO_CIDv1_CODEC.get(proto.name)
-
-    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.”
-        #
-        # In the future the following line should instead convert the multihash
-        # to CIDv1 and with the `expected_codec` and wrap it in base32:
-        #   return cid.make_cid(1, expected_codec, buf).encode("base32").decode("ascii")
-        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))
-
-        # “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")

multiaddr/codecs/domain.py

@@ -1,12 +1,30 @@
 import idna
 
-from . import LENGTH_PREFIXED_VAR_SIZE
+from ..codecs import CodecBase
+from ..exceptions import BinaryParseError
 
 
-SIZE = LENGTH_PREFIXED_VAR_SIZE
+SIZE = -1
 IS_PATH = False
 
 
+class Codec(CodecBase):
+    SIZE = SIZE
+    IS_PATH = IS_PATH
+
+    def to_bytes(self, proto, string):
+        return string.encode('utf-8')
+
+    def to_string(self, proto, buf):
+        try:
+            string = buf.decode("utf-8")
+            for label in string.split("."):
+                idna.check_label(label)
+            return string
+        except (ValueError, UnicodeDecodeError) as e:
+            raise BinaryParseError(str(e), buf, proto)
+
+
 def to_bytes(proto, string):
     return idna.uts46_remap(string).encode("utf-8")
 

multiaddr/codecs/fspath.py

@@ -1,15 +1,27 @@
-import os
+from ..codecs import CodecBase
 
-from . import LENGTH_PREFIXED_VAR_SIZE
 
-
-SIZE = LENGTH_PREFIXED_VAR_SIZE
+SIZE = -1
 IS_PATH = True
 
 
-def to_bytes(proto, string):
-    return os.fsencode(string)
+class Codec(CodecBase):
+    SIZE = SIZE
+    IS_PATH = IS_PATH
 
+    def to_bytes(self, proto, string):
+        if len(string) == 0:
+            raise ValueError("{0} value must not be empty".format(proto.name))
+        # Remove leading slash unless the path is just '/'
+        if string != '/' and string.startswith('/'):
+            string = string[1:]
+        return string.encode('utf-8')
 
-def to_string(proto, buf):
-    return os.fsdecode(buf)
+    def to_string(self, proto, buf):
+        if len(buf) == 0:
+            raise ValueError("invalid length (should be > 0)")
+        string = buf.decode('utf-8')
+        # Always add a single leading slash
+        if not string.startswith('/'):
+            string = '/' + string
+        return string

multiaddr/codecs/ip4.py

@@ -1,13 +1,17 @@
 import netaddr
+from ..codecs import CodecBase
 
 
 SIZE = 32
 IS_PATH = False
 
 
-def to_bytes(proto, string):
-    return netaddr.IPAddress(string, version=4).packed
+class Codec(CodecBase):
+    SIZE = SIZE
+    IS_PATH = IS_PATH
 
+    def to_bytes(self, proto, string):
+        return netaddr.IPAddress(string, version=4).packed
 
-def to_string(proto, buf):
-    return str(netaddr.IPAddress(int.from_bytes(buf, byteorder='big'), version=4))
+    def to_string(self, proto, buf):
+        return str(netaddr.IPAddress(int.from_bytes(buf, byteorder='big'), version=4))

multiaddr/codecs/ip6.py

@@ -1,13 +1,17 @@
 import netaddr
+from ..codecs import CodecBase
 
 
 SIZE = 128
 IS_PATH = False
 
 
-def to_bytes(proto, string):
-    return netaddr.IPAddress(string, version=6).packed
+class Codec(CodecBase):
+    SIZE = SIZE
+    IS_PATH = IS_PATH
 
+    def to_bytes(self, proto, string):
+        return netaddr.IPAddress(string, version=6).packed
 
-def to_string(proto, buf):
-    return str(netaddr.IPAddress(int.from_bytes(buf, byteorder='big'), version=6))
+    def to_string(self, proto, buf):
+        return str(netaddr.IPAddress(int.from_bytes(buf, byteorder='big'), version=6))