Revert to using datajoint.plugin

Author: ethho

datajoint.pub

@@ -0,0 +1,6 @@
+-----BEGIN PUBLIC KEY-----
+MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDUMOo2U7YQ1uOrKU/IreM3AQP2
+AXJC3au+S9W+dilxHcJ3e98bRVqrFeOofcGeRPoNc38fiLmLDUiBskJeVrpm29Wo
+AkH6yhZWk1o8NvGMhK4DLsJYlsH6tZuOx9NITKzJuOOH6X1I5Ucs7NOSKnmu7g5g
+WTT5kCgF5QAe5JN8WQIDAQAB
+-----END PUBLIC KEY-----

datajoint/errors.py

@@ -17,7 +17,19 @@ class DataJointError(Exception):
     """
 
     def __init__(self, *args):
-        super().__init__(*args)
+        from .plugin import connection_plugins, type_plugins
+
+        self.__cause__ = (
+            PluginWarning("Unverified DataJoint plugin detected.")
+            if any(
+                [
+                    any([not plugins[k]["verified"] for k in plugins])
+                    for plugins in [connection_plugins, type_plugins]
+                    if plugins
+                ]
+            )
+            else None
+        )
 
     def suggest(self, *args):
         """

datajoint/plugin.py

@@ -0,0 +1,44 @@
+from .settings import config
+import pkg_resources
+from pathlib import Path
+from cryptography.exceptions import InvalidSignature
+from otumat import hash_pkg, verify
+import logging
+
+logger = logging.getLogger(__name__.split(".")[0])
+
+
+def _update_error_stack(plugin_name):
+    try:
+        base_name = "datajoint"
+        base_meta = pkg_resources.get_distribution(base_name)
+        plugin_meta = pkg_resources.get_distribution(plugin_name)
+
+        data = hash_pkg(pkgpath=str(Path(plugin_meta.module_path, plugin_name)))
+        signature = plugin_meta.get_metadata(f"{plugin_name}.sig")
+        pubkey_path = str(Path(base_meta.egg_info, f"{base_name}.pub"))
+        verify(pubkey_path=pubkey_path, data=data, signature=signature)
+        logger.info(f"DataJoint verified plugin `{plugin_name}` detected.")
+        return True
+    except (FileNotFoundError, InvalidSignature):
+        logger.warning(f"Unverified plugin `{plugin_name}` detected.")
+        return False
+
+
+def _import_plugins(category):
+    return {
+        entry_point.name: dict(
+            object=entry_point,
+            verified=_update_error_stack(entry_point.module_name.split(".")[0]),
+        )
+        for entry_point in pkg_resources.iter_entry_points(
+            "datajoint_plugins.{}".format(category)
+        )
+        if "plugin" not in config
+        or category not in config["plugin"]
+        or entry_point.module_name.split(".")[0] in config["plugin"][category]
+    }
+
+
+connection_plugins = _import_plugins("connection")
+type_plugins = _import_plugins("datatype")

tests/test_plugin.py

@@ -0,0 +1,59 @@
+import pytest
+import datajoint.errors as djerr
+import datajoint.plugin as p
+import pkg_resources
+from os import path
+
+
+def test_check_pubkey():
+    base_name = "datajoint"
+    base_meta = pkg_resources.get_distribution(base_name)
+    pubkey_meta = base_meta.get_metadata("{}.pub".format(base_name))
+
+    with open(
+        path.join(path.abspath(path.dirname(__file__)), "..", "datajoint.pub"), "r"
+    ) as f:
+        assert f.read() == pubkey_meta
+
+
+def test_normal_djerror():
+    try:
+        raise djerr.DataJointError
+    except djerr.DataJointError as e:
+        assert e.__cause__ is None
+
+
+def test_verified_djerror(category="connection"):
+    try:
+        curr_plugins = getattr(p, "{}_plugins".format(category))
+        setattr(
+            p,
+            "{}_plugins".format(category),
+            dict(test_plugin_id=dict(verified=True, object="example")),
+        )
+        raise djerr.DataJointError
+    except djerr.DataJointError as e:
+        setattr(p, "{}_plugins".format(category), curr_plugins)
+        assert e.__cause__ is None
+
+
+def test_verified_djerror_type():
+    test_verified_djerror(category="type")
+
+
+def test_unverified_djerror(category="connection"):
+    try:
+        curr_plugins = getattr(p, "{}_plugins".format(category))
+        setattr(
+            p,
+            "{}_plugins".format(category),
+            dict(test_plugin_id=dict(verified=False, object="example")),
+        )
+        raise djerr.DataJointError("hello")
+    except djerr.DataJointError as e:
+        setattr(p, "{}_plugins".format(category), curr_plugins)
+        assert isinstance(e.__cause__, djerr.PluginWarning)
+
+
+def test_unverified_djerror_type():
+    test_unverified_djerror(category="type")