Implement IPNS validation (#16)

* chore: minor clean-up
* Better exception errors: add print statement for error message and type
* remove duplicate codecs
* correct errors in testing_data
* Implement IPNS validation
* Add support for IPNS subdomain and path gateway
* Add support for IPNS paths
* Add testing data
* Add tests for new implementation
* Bump package version

Author: Barabazs

README.MD

@@ -21,14 +21,16 @@ print(Validator("QmYjtig7VJQ6XsnUjqqJvj7QaMcCAwtrgNdahSiFofrE7o").is_ipfs())
 * [x] CID
   * [x] v0
   * [x] v1
-* [ ] URL
-  * [x] IPFS
-  * [ ] IPNS
-* [ ] Subdomain
-  * [x] IPFS
-  * [ ] IPNS
-* [ ] Path
+* [x] URL
+  * [x] Path Gateway
+    * [x] IPFS
+    * [x] IPNS
+  * [x] Subdomain Gateway
+    * [x] IPFS
+    * [x] IPNS
+* [x] Path
   * [x] IPFS
+  * [x] IPNS
 * [ ] ...
 
 ## License

is_ipfs/is_ipfs.py

@@ -1,5 +1,6 @@
 import re
 import typing
+from urllib.parse import urlparse
 
 import cid
 from multibase import decode
@@ -13,21 +14,27 @@ class Validator:
 
     def __init__(self, input: typing.Any):
         self.input = input
-        self.pathGatewayPattern = re.compile(
+        self.path_gateway_pattern = re.compile(
             r"^https?://[^/]+/(?P<protocol>ip[fn]s)/(?P<hash>[^/?#]+)"
         )
-        self.subdomainGatewayPattern = re.compile(
+        self.subdomain_gateway_pattern = re.compile(
             r"^https?://(?P<hash>[^/]+)\.(?P<protocol>ip[fn]s)\.[^/?]+"
         )
-        self.pathPattern = re.compile(r"^/(?P<protocol>ip[fn]s)/(?P<hash>[^/?#]+)")
+        self.path_pattern = re.compile(r"^/(?P<protocol>ip[fn]s)/(?P<hash>[^/?#]+)")
 
     def is_ipfs(self) -> bool:
         """
         Returns True if the provided input is a valid IPFS resource/object or False otherwise.
         """
-        return self._is_CID() or self._is_ipfs_url() or self._is_ipfs_path()
+        return (
+            self._is_cid()
+            or self._is_ipfs_url()
+            or self._is_ipns_url()
+            or self._is_ipfs_path()
+            or self._is_ipns_path()
+        )
 
-    def _is_CID(self) -> bool:
+    def _is_cid(self) -> bool:
         """
         Returns True if the provided string or CID object represents a valid CID or False otherwise.
         """
@@ -36,27 +43,25 @@ def _is_CID(self) -> bool:
         elif type(self.input) == bytes:
             try:
                 return cid.is_cid(decode(self.input))
-            except:
+            except Exception as error:
+                print(f"Unexpected {type(error)}, {error}")
                 return False
         else:
             try:
                 if isinstance(self.input, (cid.CIDv0, cid.CIDv1)):
                     return cid.is_cid(str(self.input))
-            except:
+            except Exception as error:
+                print(f"Unexpected {type(error)}, {error}")
                 return False
         return False
 
-    def _is_ipfs_url(self) -> bool:
-        """
-        Returns True if the provided string is a valid IPFS url or False otherwise.
-        """
-        return self._ipfs_path_url() or self._ipfs_subdomain_url()
-
     def _is_integral_ipfs_url(
         self,
         pattern: re.Pattern,
     ) -> bool:
-
+        """
+        Main logic for IPFS URL validation.
+        """
         formatted = str(self.input)
         if not formatted:
             return False
@@ -70,14 +75,15 @@ def _is_integral_ipfs_url(
 
         _hash = match["hash"]
 
-        if pattern == self.subdomainGatewayPattern:
+        if pattern == self.subdomain_gateway_pattern:
             _hash = _hash.lower()
             try:
                 if get_codec(_hash).encoding not in ["base32", "base36"]:
                     return False
-            except:
+            except Exception as error:
+                print(f"Unexpected {type(error)}, {error}")
                 return False
-        elif pattern == self.pathGatewayPattern:
+        elif pattern == self.path_gateway_pattern:
             if not str(_hash).startswith("Qm"):
                 try:
                     if get_codec(_hash).encoding not in [
@@ -90,34 +96,141 @@ def _is_integral_ipfs_url(
                         "base58flickr",
                         "base58btc",
                         "base64url",
-                        "base32",
-                        "base36",
                     ]:
                         return False
-                except:
-                    pass
+                except Exception as error:
+                    print(f"Unexpected {type(error)}, {error}")
+                    return False
 
-        return Validator(_hash)._is_CID()
+        return Validator(_hash)._is_cid()
 
     def _ipfs_subdomain_url(self) -> bool:
         """
         Returns True if the provided url string includes a valid IPFS subdomain (case-insensitive CIDv1) or False otherwise.
         """
         return self._is_integral_ipfs_url(
-            self.subdomainGatewayPattern,
+            self.subdomain_gateway_pattern,
         )
 
     def _ipfs_path_url(self) -> bool:
         """
         Returns True if the provided url string is a valid IPFS URL or False otherwise.
         """
-
         return self._is_integral_ipfs_url(
-            self.pathGatewayPattern,
+            self.path_gateway_pattern,
         )
 
+    def _is_ipfs_url(self) -> bool:
+        """
+        Returns True if the provided string is a valid IPFS url or False otherwise.
+        """
+        return self._ipfs_path_url() or self._ipfs_subdomain_url()
+
     def _is_ipfs_path(self) -> bool:
         """
-        Returns true if the provided string is a valid IPFS path or false otherwise.
+        Returns True if the provided string is a valid IPFS path or False otherwise.
         """
-        return self._is_integral_ipfs_url(self.pathPattern)
+        return self._is_integral_ipfs_url(self.path_pattern)
+
+    def _is_integral_ipns_url(
+        self,
+        pattern: re.Pattern,
+    ) -> bool:
+        """
+        Main logic for IPNS URL validation.
+        """
+        formatted = str(self.input)
+        if not formatted:
+            return False
+
+        match = re.match(pattern, formatted)
+        if not match:
+            return False
+
+        if match["protocol"] != "ipns":
+            return False
+
+        ipns_id = match["hash"]
+
+        if ipns_id and pattern == self.subdomain_gateway_pattern:
+            ipns_id = ipns_id.lower()
+
+            if Validator(ipns_id)._is_cid():
+                try:
+                    if get_codec(ipns_id).encoding == "base36":
+                        return True
+                except Exception as error:
+                    print(f"Unexpected {type(error)}, {error}")
+                    return False
+            try:
+                if "." not in ipns_id and "-" in ipns_id:
+                    ipns_id = (
+                        ipns_id.replace("--", "@").replace("-", ".").replace("@", "-")
+                    )
+
+                return self._id_is_explicit_tld(ipns_id)
+            except Exception as error:
+                print(f"Unexpected {type(error)}, {error}")
+                return False
+
+        elif pattern == self.path_gateway_pattern or pattern == self.path_pattern:
+            if not str(ipns_id).startswith("Qm"):
+                if self._id_is_explicit_tld(ipns_id):
+                    return True
+
+                try:
+                    if get_codec(ipns_id).encoding not in [
+                        "base2",
+                        "base16",
+                        "base32",
+                        "base32hex",
+                        "base36",
+                        "base36upper",
+                        "base58flickr",
+                        "base58btc",
+                        "base64url",
+                    ]:
+                        return False
+                except Exception as error:
+                    print(f"Unexpected {type(error)}, {error}")
+                    return False
+
+        return Validator(ipns_id)._is_cid()
+
+    def _ipns_subdomain_url(self) -> bool:
+        """
+        Returns True if the provided url string includes a valid IPFS subdomain (case-insensitive CIDv1) or False otherwise.
+        """
+        return self._is_integral_ipns_url(
+            self.subdomain_gateway_pattern,
+        )
+
+    def _ipns_path_url(self) -> bool:
+        """
+        Returns True if the provided url string is a valid IPFS URL or False otherwise.
+        """
+        return self._is_integral_ipns_url(
+            self.path_gateway_pattern,
+        )
+
+    def _is_ipns_url(self) -> bool:
+        """
+        Returns True if the provided string is a valid IPFS url or False otherwise.
+        """
+        return self._ipns_path_url() or self._ipns_subdomain_url()
+
+    def _is_ipns_path(self) -> bool:
+        """
+        Returns True if the provided string is a valid IPNS path or False otherwise.
+        """
+        return self._is_integral_ipns_url(self.path_pattern)
+
+    def _id_is_explicit_tld(self, input_string: str) -> bool:
+        """
+        Returns True if the provided url string has an explicit TLD, False otherwise.
+        """
+        fqdn_with_tld = re.compile(
+            r"(?=^.{4,253}\.?$)(^((?!-)[a-zA-Z0-9-]{1,63}(?<!-)\.)+[a-zA-Z]{2,63}$)"
+        )
+        hostname = urlparse(f"http://{input_string}").hostname
+        return bool(re.search(fqdn_with_tld, hostname))

setup.py

@@ -6,7 +6,7 @@
 
 setuptools.setup(
     name="py-is_ipfs",
-    version="0.0.6",
+    version="0.1.0",
     description="Python library to identify valid IPFS resources",
     long_description=long_description,
     long_description_content_type="text/markdown",
@@ -15,7 +15,7 @@
     packages=setuptools.find_packages(exclude=["tests", "tests.*"]),
     license="MIT",
     classifiers=[
-        "Development Status :: 3 - Alpha",
+        "Development Status :: 4 - Beta",
         "Intended Audience :: Developers",
         "License :: OSI Approved :: MIT License",
         "Natural Language :: English",

tests/integration/test_is_ipfs.py

@@ -41,6 +41,30 @@ def test_all(self):
             for entry in testing_data.invalid_entries["path"]["ipfs"]:
                 self.assertFalse(Validator(entry).is_ipfs())
 
+        with self.subTest("Test valid IPNS URL entries from fixtures"):
+            for entry in testing_data.valid_entries["url"]["ipns"]:
+                self.assertTrue(Validator(entry).is_ipfs())
+
+        with self.subTest("Test invalid IPNS URL entries from fixtures"):
+            for entry in testing_data.invalid_entries["url"]["ipns"]:
+                self.assertFalse(Validator(entry).is_ipfs())
+
+        with self.subTest("Test valid IPNS subdomain entries from fixtures"):
+            for entry in testing_data.valid_entries["subdomain"]["ipns"]:
+                self.assertTrue(Validator(entry).is_ipfs())
+
+        with self.subTest("Test invalid IPNS subdomain entries from fixtures"):
+            for entry in testing_data.invalid_entries["subdomain"]["ipns"]:
+                self.assertFalse(Validator(entry).is_ipfs())
+
+        with self.subTest("Test valid IPNS path entries from fixtures"):
+            for entry in testing_data.valid_entries["path"]["ipns"]:
+                self.assertTrue(Validator(entry).is_ipfs())
+
+        with self.subTest("Test invalid IPNS path entries from fixtures"):
+            for entry in testing_data.invalid_entries["path"]["ipns"]:
+                self.assertFalse(Validator(entry).is_ipfs())
+
 
 if __name__ == "__main__":
     unittest.main()