Implement IPFS URL & subdomain validation (#12)

Author: Barabazs

.pre-commit-config.yaml

@@ -10,6 +10,6 @@ repos:
       - id: end-of-file-fixer
 
   - repo: https://github.com/psf/black
-    rev: 22.1.0
+    rev: 22.3.0
     hooks:
       - id: black

README.MD

@@ -1,6 +1,5 @@
 # py-is_ipfs
-[![Latest release](https://img.shields.io/pypi/v/py-is_ipfs?color=blue&label=release)](https://github.com/Barabazs/py-is_ipfs/releases/latest)
-![GitHub Workflow Status](https://img.shields.io/github/workflow/status/Barabazs/py-is_ipfs/Test?label=Test&logo=github)
+[![Latest release](https://img.shields.io/pypi/v/py-is_ipfs?color=blue&label=release)](https://github.com/Barabazs/py-is_ipfs/releases/latest) ![GitHub Workflow Status](https://img.shields.io/github/workflow/status/Barabazs/py-is_ipfs/Test?label=Test&logo=github)
 
 py-is_ipfs is a Python library to identify valid IPFS resources.  
 This project is a work in progress.
@@ -9,7 +8,7 @@ This project is a work in progress.
 ### Installation
 py-is_ipfs is available on PyPI:
 
-`python -m pip install py_is-ipfs`
+`python -m pip install py-is_ipfs`
 
 ### Usage
 
@@ -23,13 +22,17 @@ print(Validator("QmYjtig7VJQ6XsnUjqqJvj7QaMcCAwtrgNdahSiFofrE7o").is_ipfs())
   * [x] v0
   * [x] v1
 * [ ] URL
-  * [ ] IPFS
+  * [x] IPFS
   * [ ] IPNS
 * [ ] Subdomain
-  * [ ] IPFS
+  * [x] IPFS
   * [ ] IPNS
 * [ ] Path
 * [ ] ...
 
 ## License
 [MIT](https://github.com/Barabazs/py-is_ipfs/blob/main/LICENSE)
+
+## Acknowledgments
+* [Protocol Labs](https://protocol.ai/) for funding this project with a micro grant.
+* [ipfs-shipyard/is-ipfs](https://github.com/ipfs-shipyard/is-ipfs) for the heuristics implemented in JavaScript that this project used as a starting point.

is_ipfs/is_ipfs.py

@@ -1,3 +1,4 @@
+import re
 import typing
 
 import cid
@@ -11,16 +12,22 @@ class Validator:
 
     def __init__(self, input: typing.Any):
         self.input = input
+        self.pathGatewayPattern = re.compile(
+            r"^https?://[^/]+/(?P<protocol>ip[fn]s)/(?P<hash>[^/?#]+)"
+        )
+        self.subdomainGatewayPattern = re.compile(
+            r"^https?://(?P<hash>[^/]+)\.(?P<protocol>ip[fn]s)\.[^/?]+"
+        )
 
     def is_ipfs(self) -> bool:
         """
-        Returns True if the provided input is a valid IPFS CID or False otherwise.
+        Returns True if the provided input is a valid IPFS resource/object or False otherwise.
         """
-        return self._is_CID()
+        return self._is_CID() or self._is_ipfs_url()
 
     def _is_CID(self) -> bool:
         """
-        Returns true if the provided string or CID object represents a valid CID or false otherwise.
+        Returns True if the provided string or CID object represents a valid CID or False otherwise.
         """
         if type(self.input) == str:
             return cid.is_cid(self.input)
@@ -36,3 +43,49 @@ def _is_CID(self) -> bool:
             except:
                 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:
+
+        formatted = str(self.input)
+        if not formatted:
+            return False
+
+        match = re.match(pattern, formatted)
+        if not match:
+            return False
+
+        if match["protocol"] != "ipfs":
+            return False
+
+        _hash = match["hash"]
+
+        if pattern == self.subdomainGatewayPattern:
+            _hash = _hash.lower()
+
+        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,
+        )
+
+    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,
+        )

requirements-dev.txt

@@ -1,2 +1,2 @@
 pre-commit==2.17.0
-black==22.1.0
+black==22.3.0

setup.py

@@ -6,7 +6,7 @@
 
 setuptools.setup(
     name="py-is_ipfs",
-    version="0.0.3",
+    version="0.0.4",
     description="Python library to identify valid IPFS resources",
     long_description=long_description,
     long_description_content_type="text/markdown",

tests/integration/test_is_ipfs.py

@@ -1,49 +1,38 @@
 from is_ipfs import Validator
-import cid
 import unittest
+import tests.testing_data as testing_data
 
 
 class TestCase(unittest.TestCase):
-    def test_is_ipfs(self):
-        # Test valid CIDv0 strings
-        self.assertTrue(
-            Validator("QmYjtig7VJQ6XsnUjqqJvj7QaMcCAwtrgNdahSiFofrE7o").is_ipfs()
-        )
-        # Test invalid CIDv0 strings
-        self.assertFalse(
-            Validator("QmYjtig7VJQ6XsnUjqqJvj7QaMcCAwtrgNdahSiFofrE70").is_ipfs()
-        )
-
-        # Test valid CIDv1 strings
-        self.assertTrue(
-            Validator(
-                "bafybeie2reiz2q6rbcuwpy2etyztjnceolu4rdi7rp3th2lsky4r5ckeey"
-            ).is_ipfs()
-        )
-
-        # Test invalid CIDv1 strings
-        self.assertFalse(
-            Validator(
-                "bafybeie2reiz2q6rbcuwpy2etyztjnceolu4rdi7rp3th2lsky4r5ckee"
-            ).is_ipfs()
-        )
-
-        # Test invalid misc input
-        self.assertFalse(Validator("").is_ipfs())
-        self.assertFalse(Validator(1345).is_ipfs())
-
-        self.assertTrue(
-            Validator(
-                cid.from_string("QmYjtig7VJQ6XsnUjqqJvj7QaMcCAwtrgNdahSiFofrE7o")
-            ).is_ipfs()
-        )
-        self.assertFalse(
-            Validator(
-                cid.CIDv0("QmYjtig7VQ6XsnUjqqJvj7QaMcCAwtrgNdahSiFofrE7o")
-            ).is_ipfs()
-        )
-        self.assertFalse(Validator(cid.CIDv0("dfmqjdmfkjqdm")).is_ipfs())
-
-
-if __name__ == "__main__":  # pragma: no cover
+    def test_all(self):
+
+        with self.subTest("Test valid CID entries from fixtures"):
+            for key, value in testing_data.valid_entries["cid"].items():
+                for entries in value:
+                    self.assertTrue(Validator(entries).is_ipfs())
+
+        with self.subTest("Test invalid CID entries from fixtures"):
+            for key, value in testing_data.invalid_entries["cid"].items():
+                for entries in value:
+                    self.assertFalse(Validator(entries).is_ipfs())
+
+        with self.subTest("Test valid IPFS URL entries from fixtures"):
+            for key, value in testing_data.valid_entries["url"].items():
+                for entries in value:
+                    self.assertTrue(Validator(entries).is_ipfs())
+
+        with self.subTest("Test invalid IPFS URL entries from fixtures"):
+            for entry in testing_data.invalid_entries["url"]["ipfs"]:
+                self.assertFalse(Validator(entry).is_ipfs())
+
+        with self.subTest("Test valid IPFS subdomain entries from fixtures"):
+            for entry in testing_data.valid_entries["subdomain"]["ipfs"]:
+                self.assertTrue(Validator(entry).is_ipfs())
+
+        with self.subTest("Test invalid IPFS subdomain entries from fixtures"):
+            for entry in testing_data.invalid_entries["subdomain"]["ipfs"]:
+                self.assertFalse(Validator(entry).is_ipfs())
+
+
+if __name__ == "__main__":
     unittest.main()

tests/testing_data.py

@@ -47,6 +47,23 @@
             "f01701220c3c4733ec8affd06cf9e9ff50ffc6bcd2ec85a6170004bb709669c31de94391a",
         ],
     },
+    "url": {
+        "ipfs": [
+            "http://ipfs.io/ipfs/QmYHNYAaYK5hm3ZhZFx5W9H6xydKDGimjdgJMrMSdnctEm?arg=val#hash",
+            "http://ipfs.alexandria.media/ipfs/QmeWz9YZEeNFXQhHg4PnR5ZiNr5isttgi5n1tc1eD5EfGU/content/index.html?arg=val#hash",
+            "http://ipfs.io/ipfs/QmYHNYAaYK5hm3ZhZFx5W9H6xydKDGimjdgJMrMSdnctEm",
+            "https://gateway.pinata.cloud/ipfs/Qmb4sw3sqA7AZsaRZ7vtMwxCduk1ExJL5gVDPcpnP8kxFK/",
+            "https://gateway.pinata.cloud/ipfs/bafybeif5dwlk2sdx5yge4azff2ovsnar63cu37ncw4n24vnqixwamwhxui/",
+            "https://bafybeif5dwlk2sdx5yge4azff2ovsnar63cu37ncw4n24vnqixwamwhxui.ipfs.dweb.link/",
+        ],
+    },
+    "subdomain": {
+        "ipfs": [
+            "http://bafybeie5gq4jxvzmsym6hjlwxej4rwdoxt7wadqvmmwbqi7r27fclha2va.ipfs.dweb.link",
+            "http://bafybeidvtwx54qr44kidymvhfzefzxhgkieigwth6oswk75zhlzjdmunoy.ipfs.dweb.link/linkify-demo.html",
+            "http://bafybeie5gq4jxvzmsym6hjlwxej4rwdoxt7wadqvmmwbqi7r27fclha2va.ipfs.localhost:8080",
+        ],
+    },
 }
 invalid_entries = {
     "cid": {
@@ -67,4 +84,30 @@
             encode("base16", "QmNQuBJ8g4QN6mSLXHekxBbcToPwKxamWNrDdEugxMTDd"),
         ]
     },
+    "url": {
+        "ipfs": [
+            "http://ipfs.io/ipns/github.com/",
+            "https://Qmb4sw3sqA7AZsaRZ7vtMwxCduk1ExJL5gVDPcpnP8kxFK.ipfs.dweb.link/",
+            "http://ipfs.io/ipns/QmYHNYAaYK5hm3ZhZFx5W9H6xydKDGimjdgJMrMSdnctEm",
+            "https://github.com/ipfs/js-ipfs/blob/master/README.md",
+            "https://google.com",
+            "http://ipfs.io/ipns/github.com/",
+            "https://github.com/ipfs/js-ipfs/blob/master/README.md",
+            "http://ipfs.io/ipns/github.com/",
+            "https://github.com/ipfs/js-ipfs/blob/master/README.md",
+        ],
+    },
+    "subdomain": {
+        "ipfs": [
+            "http://bafybeiabc2xofh6tdi6vutusorpumwcikw3hf3st4ecjugo6j52f6xwc6q.ipns.dweb.link",
+            "http://not-a-cid.ipfs.dweb.link",
+            "http://QmbWqxBEKC3P8tqsKc98xmWNzrzDtRLMiMPL8wBuTGsMnR.ipfs.dweb.link",
+            "http://bafybeie5gq4jxvzmsym6hjlwxej4rwdoxt7wadqvmmwbqi7r27fclha2va.dweb.link",
+            "http://QmcNioXSC1bfJj1dcFErhUfyjFzoX2HodkRccsFFVJJvg8.ipns.dweb.link",
+            "http://bafybeiabc2xofh6tdi6vutusorpumwcikw3hf3st4ecjugo6j52f6xwc6q.dweb.link",
+            "http://invalid-hostname-.ipns.dweb.link",
+            "http://www.bafybeie5gq4jxvzmsym6hjlwxej4rwdoxt7wadqvmmwbqi7r27fclha2va.ipfs.dweb.link",
+            "http://not-a-cid-or-valid-hostname-.ipns.dweb.link",
+        ],
+    },
 }

tests/unit/test_subdomain.py

@@ -0,0 +1,18 @@
+from is_ipfs import Validator
+import unittest
+import tests.testing_data as testing_data
+
+
+class TestCase(unittest.TestCase):
+    def test_ipfs_subdomain(self):
+        with self.subTest("Test valid IPFS subdomain entries from fixtures"):
+            for entry in testing_data.valid_entries["subdomain"]["ipfs"]:
+                self.assertTrue(Validator(entry)._ipfs_subdomain_url())
+
+        with self.subTest("Test invalid IPFS subdomain entries from fixtures"):
+            for entry in testing_data.invalid_entries["subdomain"]["ipfs"]:
+                self.assertFalse(Validator(entry)._ipfs_subdomain_url())
+
+
+if __name__ == "__main__":  # pragma: no cover
+    unittest.main()

tests/unit/test_url.py

@@ -0,0 +1,19 @@
+from is_ipfs import Validator
+import unittest
+import tests.testing_data as testing_data
+
+
+class TestCase(unittest.TestCase):
+    def test_ipfs_url(self):
+        with self.subTest("Test valid IPFS URL entries from fixtures"):
+            for entry in testing_data.valid_entries["url"]["ipfs"]:
+                print(entry)
+                self.assertTrue(Validator(entry)._is_ipfs_url())
+
+        with self.subTest("Test invalid IPFS URL entries from fixtures"):
+            for entry in testing_data.invalid_entries["url"]["ipfs"]:
+                self.assertFalse(Validator(entry)._is_ipfs_url())
+
+
+if __name__ == "__main__":  # pragma: no cover
+    unittest.main()