feat(dev): disable blueprints, config validators;

- JSON string loading config validator.
- Safe url came_from.
- Disable JWT login windows by default.

Author: JVickery-TBS

ckanext/language_domains/assets/webassets.yml

@@ -2,7 +2,6 @@ css:
   contents:
     - styles/language_domains.css
   output: language_domains/%(version)s_language_domains.css
-  filters: cache_bust
 
 js_login_sender:
   contents:
@@ -11,7 +10,6 @@ js_login_sender:
     preload:
       - base/main  # need ckan JS and jQuery
   output: language_domains/%(version)s_language_login_sender.js
-  filters: cache_bust
 
 js_login_receiver:
   contents:
@@ -20,7 +18,6 @@ js_login_receiver:
     preload:
       - base/main  # need ckan JS and jQuery
   output: language_domains/%(version)s_language_login_receiver.js
-  filters: cache_bust
 
 js_logout_sender:
   contents:
@@ -29,7 +26,6 @@ js_logout_sender:
     preload:
       - base/main  # need ckan JS and jQuery
   output: language_domains/%(version)s_language_logout_sender.js
-  filters: cache_bust
 
 js_logout_receiver:
   contents:
@@ -38,4 +34,3 @@ js_logout_receiver:
     preload:
       - base/main  # need ckan JS and jQuery
   output: language_domains/%(version)s_language_logout_receiver.js
-  filters: cache_bust

ckanext/language_domains/blueprint.py

@@ -3,7 +3,6 @@
 from werkzeug.datastructures import ImmutableMultiDict
 from urllib.parse import urlsplit
 from logging import getLogger
-import json
 import jwt
 import datetime
 
@@ -56,17 +55,14 @@ def login_master():
         return h.redirect_to('user.login')
 
     came_from = request.args.get('_came_from', '')
+    if not h.url_is_local(came_from):
+        came_from = ''
 
     default_domain = config.get('ckan.site_url', '')
     uri_parts = urlsplit(default_domain)
     domain_scheme = uri_parts.scheme
 
-    language_domains = config.get('ckanext.language_domains.domain_map', '')
-    if not language_domains:
-        language_domains = {}
-    else:
-        language_domains = json.loads(language_domains)
-
+    language_domains = config.get('ckanext.language_domains.domain_map')
     current_domain = request.url
     uri_parts = urlsplit(current_domain)
     current_domain = uri_parts.netloc
@@ -109,11 +105,9 @@ def login():
         uri_parts = urlsplit(default_domain)
         domain_scheme = uri_parts.scheme
 
-        language_domains = config.get('ckanext.language_domains.domain_map', '')
+        language_domains = config.get('ckanext.language_domains.domain_map')
         trusted_domains = []
         if language_domains:
-            language_domains = json.loads(language_domains)
-
             current_domain = request.url
             uri_parts = urlsplit(current_domain)
             current_domain = uri_parts.netloc
@@ -210,17 +204,14 @@ def logout_master():
             return h.redirect_to('user.login')
 
         came_from = request.args.get('_came_from', '')
+        if not h.url_is_local(came_from):
+            came_from = ''
 
         default_domain = config.get('ckan.site_url', '')
         uri_parts = urlsplit(default_domain)
         domain_scheme = uri_parts.scheme
 
-        language_domains = config.get('ckanext.language_domains.domain_map', '')
-        if not language_domains:
-            language_domains = {}
-        else:
-            language_domains = json.loads(language_domains)
-
+        language_domains = config.get('ckanext.language_domains.domain_map')
         current_domain = request.url
         uri_parts = urlsplit(current_domain)
         current_domain = uri_parts.netloc
@@ -314,11 +305,9 @@ def logout():
         uri_parts = urlsplit(default_domain)
         domain_scheme = uri_parts.scheme
 
-        language_domains = config.get('ckanext.language_domains.domain_map', '')
+        language_domains = config.get('ckanext.language_domains.domain_map')
         trusted_domains = []
         if language_domains:
-            language_domains = json.loads(language_domains)
-
             current_domain = request.url
             uri_parts = urlsplit(current_domain)
             current_domain = uri_parts.netloc

ckanext/language_domains/config_declaration.yaml

@@ -7,13 +7,24 @@ groups:
         description: |
           Dictionary of language locales to domain list
         required: true
+        validators: load_json_string
       - key: ckanext.language_domains.root_paths
         example: {"example.com": "/data", "exemple.com": "/data"}
         description: |
           Dictionary of domains and their CKAN root_paths
         required: false
+        validators: load_json_string
       - key: ckanext.language_domains.secret
         example: 'thisisalegitsecret'
         description: |
           Secret key for JWT encoding.
         required: true
+      - key: ckanext.language_domains.enable_jwt_login
+        example: true
+        default: false
+        description: >
+          Enables multi-domain login and logout via JS
+          window messaging with JWT tokens. Requires
+          ckanext.language_domains.secret to be set.
+        required: false
+        validators: ignore_missing boolean_validator

ckanext/language_domains/helpers.py

@@ -4,7 +4,6 @@
 )
 from urllib.parse import urlparse, urlunparse
 from urllib.parse import urlsplit
-import json
 import re
 
 from typing import Any, cast, Union, Tuple, Dict
@@ -38,11 +37,7 @@ def _get_correct_language_domain() -> Tuple[str, str]:
     default_domain = config.get('ckan.site_url', '')
     uri_parts = urlsplit(default_domain)
     default_scheme = uri_parts.scheme
-    language_domains = config.get('ckanext.language_domains.domain_map', '')
-    if not language_domains:
-        language_domains = {}
-    else:
-        language_domains = json.loads(language_domains)
+    language_domains = config.get('ckanext.language_domains.domain_map')
     current_lang = h.lang()
     correct_lang_domain = current_domain
     domain_index_match = _get_domain_index(current_domain, language_domains)
@@ -79,11 +74,7 @@ def redirect_to(*args: Any, **kw: Any) -> Response:
             status_code = 301
             _url = _url[len(f'/{current_lang}'):]
         _scheme, _host = _get_correct_language_domain()
-        root_paths = config.get('ckanext.language_domains.root_paths', '')
-        if not root_paths:
-            root_paths = {}
-        else:
-            root_paths = json.loads(root_paths)
+        root_paths = config.get('ckanext.language_domains.root_paths')
         root_path = root_paths.get(_host, '').rstrip('/')
         root_path = root_path.replace('/{{LANG}}', '')
         if not _url.startswith(root_path):
@@ -146,11 +137,7 @@ def local_url(url_to_amend: str, **kw: Any):
 
     # ckan.root_path is defined when we have none standard language
     # position in the url
-    root_paths = config.get('ckanext.language_domains.root_paths', '')
-    if not root_paths:
-        root_paths = {}
-    else:
-        root_paths = json.loads(root_paths)
+    root_paths = config.get('ckanext.language_domains.root_paths')
     root_path = root_paths.get(host, '').rstrip('/')
     if root_path:
         # FIXME this can be written better once

ckanext/language_domains/plugin.py

@@ -1,16 +1,15 @@
 from logging import getLogger
-import json
 from urllib.parse import urlsplit
 
 from typing import Any, Optional, List, Tuple, Callable, Dict
 from flask import Blueprint
-from ckan.types import CKANApp
+from ckan.types import CKANApp, Validator
 from ckan.common import CKANConfig
 
 import ckan.plugins as plugins
 import ckan.lib.helpers as core_helpers
 
-from ckanext.language_domains import helpers
+from ckanext.language_domains import helpers, validators
 from ckanext.language_domains.blueprint import language_domain_views
 
 
@@ -21,6 +20,7 @@
 class LanguageDomainsPlugin(plugins.SingletonPlugin):
     plugins.implements(plugins.IMiddleware, inherit=True)
     plugins.implements(plugins.IConfigurer)
+    plugins.implements(plugins.IValidators)
     plugins.implements(plugins.ITemplateHelpers)
     plugins.implements(plugins.IBlueprint)
 
@@ -31,14 +31,23 @@ def make_middleware(self, app: CKANApp, config: 'CKANConfig') -> CKANApp:
     # IConfigurer
     def update_config(self, config: 'CKANConfig'):
         # NOTE: monkey patch these core helpers as the other helpers call them directly
+        # TODO: fix upstream helpers to always use h object helpers!!!
         core_helpers.redirect_to = helpers.redirect_to
         core_helpers.get_site_protocol_and_host = helpers.get_site_protocol_and_host
         core_helpers._local_url = helpers.local_url
 
-        plugins.toolkit.add_template_directory(config, 'templates')
-        plugins.toolkit.add_resource('assets', 'language_domain_assets')
+        if plugins.toolkit.config.get('ckanext.language_domains.'
+                                      'enable_jwt_login', False):
+            plugins.toolkit.add_template_directory(config, 'templates')
+            plugins.toolkit.add_resource('assets', 'language_domain_assets')
 
-        config['ckan.auth.route_after_login'] = 'language_domains.login_master'
+            config['ckan.auth.route_after_login'] = 'language_domains.login_master'
+
+    # IValidators
+    def get_validators(self) -> Dict[str, Validator]:
+        return {
+            'load_json_string': validators.load_json_string,
+        }
 
     # ITemplateHelpers
     def get_helpers(self) -> Dict[str, Callable[..., Any]]:
@@ -47,26 +56,22 @@ def get_helpers(self) -> Dict[str, Callable[..., Any]]:
 
     # IBlueprint
     def get_blueprint(self) -> List[Blueprint]:
-        return [language_domain_views]
+        if plugins.toolkit.config.get('ckanext.language_domains.'
+                                      'enable_jwt_login', False):
+            return [language_domain_views]
+        return []
+
 
 
 class LanguageDomainMiddleware(object):
     def __init__(self, app: Any, config: 'CKANConfig'):
         self.app = app
-        language_domains = config.get('ckanext.language_domains.domain_map', '')
-        if not language_domains:
-            self.language_domains = {}
-        else:
-            self.language_domains = json.loads(language_domains)
+        self.language_domains = config.get('ckanext.language_domains.domain_map')
         default_domain = config.get('ckan.site_url', '')
         uri_parts = urlsplit(default_domain)
         self.default_domain = uri_parts.netloc
         self.domain_scheme = uri_parts.scheme
-        root_paths = config.get('ckanext.language_domains.root_paths', '')
-        if not root_paths:
-            self.root_paths = {}
-        else:
-            self.root_paths = json.loads(root_paths)
+        self.root_paths = config.get('ckanext.language_domains.root_paths')
 
     def __call__(self, environ: Any, start_response: Any) -> Any:
         extra_response_headers = []

ckanext/language_domains/validators.py

@@ -0,0 +1,11 @@
+import json
+
+from typing import Any
+
+
+def load_json_string(value: Any):
+    if isinstance(value, dict):
+        return value
+    if not value:
+        return {}
+    return json.loads(value)