feat(logic): validators;

- Moved monkey patching excessive schemas to `IValidator` implementation.
- Moved monkey patching schema old user name to chained action.
- Added more config options for password requirements.

ckanext/security/logic/action.py

@@ -9,6 +9,7 @@
     reset_address_throttle,
     reset_totp
 )
+from ckanext.security import validators
 
 
 def security_throttle_user_reset(context, data_dict):
@@ -67,6 +68,9 @@ def user_update(up_func, context, data_dict):
     ckanext-security: reset throttling information for updated users
     to allow new login attempts after password reset
     """
+    # (canada fork only): update the user update form schema for username field
+    # TODO: upstream contrib??
+    context['schema']['name'].append(validators.old_username_validator)
     rval = up_func(context, data_dict)
     get_action('security_throttle_user_reset')(
         dict(context, ignore_auth=True), {'user': rval['name']})

ckanext/security/plugin/__init__.py

@@ -1,13 +1,12 @@
 import logging
 import ckan.plugins as p
 
-from ckanext.security import schema as ext_schema
 from ckan.plugins import toolkit as tk
-from ckan.logic import schema as core_schema
 from ckanext.security.model import define_security_tables
 from ckanext.security.resource_upload_validator import (
     validate_upload_type, validate_upload_presence
 )
+from ckanext.security import validators
 from ckanext.security.logic import auth, action
 from ckanext.security.helpers import security_disable_totp
 
@@ -27,30 +26,32 @@ class CkanSecurityPlugin(MixinPlugin, p.SingletonPlugin):
     p.implements(p.IActions)
     p.implements(p.IAuthFunctions)
     p.implements(p.ITemplateHelpers)
+    p.implements(p.IValidators, inherit=True)
 
     # BEGIN Hooks for IConfigurer
 
     def update_config(self, config):
         define_security_tables()  # map security models to db schema
 
-        # Monkeypatching all user schemas in order to enforce a stronger
-        # password policy. I tried monkeypatching
-        # `ckan.logic.validators.user_password_validator` instead
-        # without success.
-        core_schema.default_user_schema = \
-            ext_schema.default_user_schema
-        core_schema.user_new_form_schema = \
-            ext_schema.user_new_form_schema
-        core_schema.user_edit_form_schema = \
-            ext_schema.user_edit_form_schema
-        core_schema.default_update_user_schema = \
-            ext_schema.default_update_user_schema
+        # (canada fork only): remove monkey patching
+        # TODO: upstream contrib??
 
         tk.add_template_directory(config, '../templates')
         tk.add_resource('../fanstatic', 'security')
 
     # END Hooks for IConfigurer
 
+    # BEGIN Hooks for IValidators
+
+    def get_validators(self):
+        # (canada fork only): implement IValidators instead of monkey patching
+        # TODO: upstream contrib??
+        return {
+            'user_password_validator': validators.user_password_validator,
+        }
+
+    # END Hooks for IValidators
+
     # BEGIN Hooks for IResourceController
 
     def before_create(self, context, resource):

ckanext/security/validators.py

@@ -5,13 +5,16 @@
 from ckan import authz
 from ckan.common import _
 from ckan.lib.navl.dictization_functions import Missing, Invalid
+# (canada fork only): more configs
+from ckan.plugins.toolkit import config, asbool
 
-MIN_PASSWORD_LENGTH = 8
-MIN_LEN_ERROR = (
-    'Your password must be {} characters or longer, and consist of at least '
-    'three of the following character sets: uppercase characters, lowercase '
-    'characters, digits, punctuation & special characters.'
+MIN_PASSWORD_LENGTH = int(config.get('ckanext.security.min_password_length', 8))
+MIN_LEN_ERROR = 'Your password must be {} characters or longer.'
+COMPLEXITY_ERROR = (
+    'Your password must consist of at least three of the following character sets: '
+    'uppercase characters, lowercase characters, digits, punctuation & special characters.'
 )
+NZISM_COMPLIANT = asbool(config.get('ckanext.security.nzism_compliant_passwords', True))
 
 
 def user_password_validator(key, data, errors, context):
@@ -24,15 +27,20 @@ def user_password_validator(key, data, errors, context):
     elif value == '':
         pass  # Already handled in core
     else:
-        # NZISM compliant password rules
-        rules = [
-            any(x.isupper() for x in value),
-            any(x.islower() for x in value),
-            any(x.isdigit() for x in value),
-            any(x in string.punctuation for x in value)
-        ]
-        if len(value) < MIN_PASSWORD_LENGTH or sum(rules) < 3:
-            raise Invalid(_(MIN_LEN_ERROR.format(MIN_PASSWORD_LENGTH)))
+        # (canad fork only): better error messages
+        # TODO: upstream contrib??
+        if len(value) < MIN_PASSWORD_LENGTH:
+            errors[key].append(_(MIN_LEN_ERROR.format(MIN_PASSWORD_LENGTH)))
+        if NZISM_COMPLIANT:
+            # NZISM compliant password rules
+            rules = [
+                any(x.isupper() for x in value),
+                any(x.islower() for x in value),
+                any(x.isdigit() for x in value),
+                any(x in string.punctuation for x in value)
+            ]
+            if sum(rules) < 3:
+                errors[key].append(_(COMPLEXITY_ERROR))
 
 
 def old_username_validator(key, data, errors, context):