feat: kick_user option; config declarations

DataShades · Nov 26, 2023 · 2d3cfe7 · 2d3cfe7
1 parent 7f26f7f
commit 2d3cfe7
Show file tree

Hide file tree

Showing 9 changed files with 132 additions and 89 deletions.
diff --git a/.gitignore b/.gitignore
@@ -1,4 +1,5 @@
 .ropeproject
+.mypy_cache/
 node_modules
 bower_components
 

diff --git a/ckanext/drupal_idp/drupal.py b/ckanext/drupal_idp/drupal.py
@@ -12,18 +12,16 @@
 import ckan.plugins.toolkit as tk
 from ckan.exceptions import CkanConfigurationException
 
-import ckanext.drupal_idp.utils as utils
+from ckanext.drupal_idp import utils, config
 
 log = logging.getLogger(__name__)
-CONFIG_PUBLIC_PATH = "ckanext.drupal_idp.public_path"
-DEFAULT_PUBLIC_PATH = "/sites/default/files/"
 
 
 def db_url() -> str:
-    url = tk.config.get(utils.CONFIG_DB_URL)
+    url = config.db_url()
     if not url:
         raise CkanConfigurationException(
-            f"drupal_idp plugin requires {utils.CONFIG_DB_URL} config option."
+            f"drupal_idp plugin requires {config.CONFIG_DB_URL} config option."
         )
     return url
 
@@ -127,9 +125,7 @@ def get_avatar(self, uid: utils.DrupalId):
         public_prefix = "public://"
         if path.startswith(public_prefix):
             path = os.path.join(
-                tk.config.get(CONFIG_PUBLIC_PATH, DEFAULT_PUBLIC_PATH).rstrip(
-                    "/"
-                ),
+                config.public_path().rstrip("/"),
                 path[len(public_prefix) :],
             )
         return path
@@ -151,7 +147,7 @@ def get_field(self, uid: utils.DrupalId, field: str) -> list[Any]:
         return [r[0] for r in query]
 
 
-_mapping = {"9": Drupal9}
+_mapping = {"9": Drupal9, "10": Drupal9, "11": Drupal9}
 
 
 def get_adapter(version: str) -> BaseDrupal:

diff --git a/ckanext/drupal_idp/logic/action.py b/ckanext/drupal_idp/logic/action.py
@@ -7,12 +7,10 @@
 
 from ckanext.toolbelt.decorators import Collector
 
-import ckanext.drupal_idp.utils as utils
+from ckanext.drupal_idp import utils, config
 
 log = logging.getLogger(__name__)
 action, get_actions = Collector("drupal_idp").split()
-CONFIG_FOCE_SYNC = "ckanext.drupal_idp.synchronization.force"
-DEFAULT_FORCE_SYNC = False
 
 
 @action
@@ -37,7 +35,7 @@ def user_synchronize(context, data_dict):
     if not details:
         raise tk.ObjectNotFound("drupal_user")
 
-    force = tk.asbool(tk.config.get(CONFIG_FOCE_SYNC, DEFAULT_FORCE_SYNC))
+    force = config.force_sync()
     return utils.synchronize(user, details, force)
 
 
@@ -83,7 +81,7 @@ def chained_user_show(next_, context, data_dict):
     if not url:
         return user
 
-    host = tk.config.get(utils.CONFIG_STATIC_HOST)
+    host = config.static_host()
     if host and not url.startswith("http"):
         url = "//" + host.rstrip("/") + url
 

diff --git a/ckanext/drupal_idp/plugin.py b/ckanext/drupal_idp/plugin.py
@@ -6,14 +6,19 @@
 
 from ckanext.drupal_idp.logic import action
 from ckanext.drupal_idp.logic import auth
-from ckanext.drupal_idp import helpers, utils, drupal, cli, views
-
-CONFIG_SKIP_STATIC = "ckanext.drupal_idp.skip_static"
-DEFAULT_SKIP_STATIC = False
+from ckanext.drupal_idp import helpers, utils, drupal, cli, views, config
 
 log = logging.getLogger(__name__)
 
+try:
+    config_declarations = tk.blanket.config_declarations
+except AttributeError:
+
+    def config_declarations(cls):
+        return cls
 
+
+@config_declarations
 class DrupalIdpPlugin(plugins.SingletonPlugin):
     plugins.implements(plugins.IAuthenticator, inherit=True)
     plugins.implements(plugins.IConfigurer)
@@ -57,20 +62,20 @@ def identify(self):
             ("static", "index"),
             ("webassets", "index"),
         }
-        if (
-            tk.asbool(tk.config.get(CONFIG_SKIP_STATIC, DEFAULT_SKIP_STATIC))
-            and tk.get_endpoint() in static
-        ):
+        if config.skip_static() and tk.get_endpoint() in static:
             log.debug("Skip static route")
             return
 
         cookie_sid = tk.request.cookies.get(utils.session_cookie_name())
         if not cookie_sid:
             log.debug("No session cookie found")
             return
+
         sid = utils.decode_sid(cookie_sid)
         uid = utils.sid_into_uid(sid)
         if not uid:
+            if tk.check_ckan_version("2.10") and config.kick_missing_session():
+                tk.logout_user()
             return
 
         try:

diff --git a/ckanext/drupal_idp/tests/test_drupal.py b/ckanext/drupal_idp/tests/test_drupal.py
@@ -2,14 +2,14 @@
 
 from ckan.exceptions import CkanConfigurationException
 
-import ckanext.drupal_idp.utils as utils
-import ckanext.drupal_idp.drupal as drupal
+from ckanext.drupal_idp import config, drupal
+
 
 class TestDbUrl:
     def test_default_value(self):
         assert drupal.db_url()
 
-    @pytest.mark.ckan_config(utils.CONFIG_DB_URL, "")
+    @pytest.mark.ckan_config(config.CONFIG_DB_URL, "")
     def test_exception_with_missing_config(self):
         with pytest.raises(CkanConfigurationException):
             drupal.db_url()
diff --git a/ckanext/drupal_idp/tests/test_utils.py b/ckanext/drupal_idp/tests/test_utils.py
@@ -4,7 +4,7 @@
 import ckan.lib.munge as munge
 
 
-import ckanext.drupal_idp.utils as utils
+from ckanext.drupal_idp import utils, config
 
 
 class TestDetails:
@@ -25,21 +25,21 @@ def test_into_user(self, details_data):
         }
 
     def test_sysadmin_ignored_by_default(self, details_data):
-        details_data["roles"].append(utils.DEFAULT_ADMIN_ROLE)
+        details_data["roles"].append(config.DEFAULT_ADMIN_ROLE)
         assert not utils.Details(**details_data).is_sysadmin()
 
-    @pytest.mark.ckan_config(utils.CONFIG_INHERIT_ADMIN_ROLE, "true")
+    @pytest.mark.ckan_config(config.CONFIG_INHERIT_ADMIN_ROLE, "true")
     def test_sysadmin(self, details_data):
         assert not utils.Details(**details_data).is_sysadmin()
         details_data["roles"].append("not-an-admin")
         assert not utils.Details(**details_data).is_sysadmin()
-        details_data["roles"].append(utils.DEFAULT_ADMIN_ROLE)
+        details_data["roles"].append(config.DEFAULT_ADMIN_ROLE)
         assert utils.Details(**details_data).is_sysadmin()
 
-    @pytest.mark.ckan_config(utils.CONFIG_INHERIT_ADMIN_ROLE, "true")
-    @pytest.mark.ckan_config(utils.CONFIG_ADMIN_ROLE_NAME, "custom-admin")
+    @pytest.mark.ckan_config(config.CONFIG_INHERIT_ADMIN_ROLE, "true")
+    @pytest.mark.ckan_config(config.CONFIG_ADMIN_ROLE_NAME, "custom-admin")
     def test_sysadmin_with_custom_role(self, details_data):
-        details_data["roles"].append(utils.DEFAULT_ADMIN_ROLE)
+        details_data["roles"].append(config.DEFAULT_ADMIN_ROLE)
         assert not utils.Details(**details_data).is_sysadmin()
         details_data["roles"].append("custom-admin")
         assert utils.Details(**details_data).is_sysadmin()
@@ -49,7 +49,7 @@ class TestSynchronizationEnabled:
     def test_default_value(self):
         assert not utils.is_synchronization_enabled()
 
-    @pytest.mark.ckan_config(utils.CONFIG_SYNCHRONIZATION_ENABLED, "true")
+    @pytest.mark.ckan_config(config.CONFIG_SYNCHRONIZATION_ENABLED, "true")
     def test_updated_value(self):
         assert utils.is_synchronization_enabled()
 
@@ -70,7 +70,7 @@ def test_https(self, ckan_config, monkeypatch):
             == "SSESS49960de5880e8c687434170f6476605b"
         )
 
-    @pytest.mark.ckan_config(utils.CONFIG_STATIC_HOST, "my.site.com")
+    @pytest.mark.ckan_config(config.CONFIG_STATIC_HOST, "my.site.com")
     def test_static_host(self):
         assert (
             utils.session_cookie_name()
@@ -111,7 +111,7 @@ def test_default_native_id(self, details_data):
         userdict = utils.get_or_create_from_details(details)
         assert userdict["id"] != details_data["id"]
 
-    @pytest.mark.ckan_config(utils.CONFIG_SAME_ID, "true")
+    @pytest.mark.ckan_config(config.CONFIG_SAME_ID, "true")
     def test_same_id(self, details_data):
         details = utils.Details(**details_data)
         userdict = utils.get_or_create_from_details(details)

diff --git a/ckanext/drupal_idp/utils.py b/ckanext/drupal_idp/utils.py
@@ -12,20 +12,7 @@
 import ckan.model as model
 import ckan.plugins.toolkit as tk
 import ckan.lib.munge as munge
-from . import signals
-
-CONFIG_DB_URL = "ckanext.drupal_idp.db_url"
-CONFIG_SYNCHRONIZATION_ENABLED = "ckanext.drupal_idp.synchronization.enabled"
-CONFIG_STATIC_HOST = "ckanext.drupal_idp.host"
-CONFIG_INHERIT_ADMIN_ROLE = "ckanext.drupal_idp.admin_role.inherit"
-CONFIG_ADMIN_ROLE_NAME = "ckanext.drupal_idp.admin_role.name"
-CONFIG_DRUPAL_VERSION = "ckanext.drupal_idp.drupal.version"
-CONFIG_SAME_ID = "ckanext.drupal_idp.same_id"
-CONFIG_EXTRA_FIELDS = "ckanext.drupal_idp.extra_fields"
-
-DEFAULT_EXTRA_FIELDS = []
-DEFAULT_ADMIN_ROLE = "administrator"
-DEFAULT_DRUPAL_VERSION = "9"
+from . import signals, config
 
 log = logging.getLogger(__name__)
 
@@ -53,9 +40,8 @@ class Details:
 
     def is_sysadmin(self):
         return (
-            tk.asbool(tk.config.get(CONFIG_INHERIT_ADMIN_ROLE))
-            and tk.config.get(CONFIG_ADMIN_ROLE_NAME, DEFAULT_ADMIN_ROLE)
-            in self.roles
+            config.inherit_admin_role()
+            and config.admin_role_name() in self.roles
         )
 
     def make_userdict(self):
@@ -68,15 +54,15 @@ def make_userdict(self):
 
 
 def is_synchronization_enabled() -> bool:
-    return tk.asbool(tk.config.get(CONFIG_SYNCHRONIZATION_ENABLED))
+    return config.synchronization_enabled()
 
 
 def _make_password():
     return secrets.token_urlsafe(60)
 
 
 def _get_host() -> str:
-    host = tk.config.get(CONFIG_STATIC_HOST)
+    host = config.static_host()
     if not host:
         host = tk.request.environ["HTTP_HOST"].split(":")[0]
     return host
@@ -136,19 +122,15 @@ def sid_into_uid(sid: str) -> DrupalId | None:
     """Fetch user data from Drupal's database."""
     import ckanext.drupal_idp.drupal as drupal
 
-    adapter = drupal.get_adapter(
-        tk.config.get(CONFIG_DRUPAL_VERSION, DEFAULT_DRUPAL_VERSION)
-    )
+    adapter = drupal.get_adapter(config.drupal_version())
     return adapter.get_uid_by_sid(sid)
 
 
 def get_user_details(uid: DrupalId) -> Optional[Details]:
     """Fetch user data from Drupal's database."""
     import ckanext.drupal_idp.drupal as drupal
 
-    adapter = drupal.get_adapter(
-        tk.config.get(CONFIG_DRUPAL_VERSION, DEFAULT_DRUPAL_VERSION)
-    )
+    adapter = drupal.get_adapter(config.drupal_version())
 
     user = adapter.get_user_by_uid(uid)
     if not user:
@@ -158,9 +140,7 @@ def get_user_details(uid: DrupalId) -> Optional[Details]:
     details_data["avatar"] = adapter.get_avatar(user.id)
     details_data["roles"] = roles
 
-    extra_fields = tk.aslist(
-        tk.config.get(CONFIG_EXTRA_FIELDS, DEFAULT_EXTRA_FIELDS)
-    )
+    extra_fields = config.extra_fields()
     details_data["fields"] = adapter.get_fields(user.id, extra_fields)
 
     return Details(**details_data)
@@ -185,7 +165,7 @@ def _create_from_details(details: Details) -> UserDict:
     """
     user = details.make_userdict()
     user["password"] = _make_password()
-    if tk.asbool(tk.config.get(CONFIG_SAME_ID)):
+    if config.same_id():
         user["id"] = str(details.id)
 
     admin = tk.get_action("get_site_user")({"ignore_auth": True}, {})