user-api: use simple schema validator to be consistent how we validate between API and web views.

rhodecode/api/tests/test_create_user.py

@@ -59,6 +59,21 @@ class TestCreateUser(object):
         expected = "email `%s` already exist" % (TEST_USER_REGULAR_EMAIL,)
         assert_error(id_, expected, given=response.body)
 
+    def test_api_create_user_with_wrong_username(self):
+        bad_username = '<> HELLO WORLD <>'
+        id_, params = build_data(
+            self.apikey, 'create_user',
+            username=bad_username,
+            email='new@email.com',
+            password='trololo')
+        response = api_call(self.app, params)
+
+        expected = {'username':
+                        "Username may only contain alphanumeric characters "
+                        "underscores, periods or dashes and must begin with "
+                        "alphanumeric character or underscore"}
+        assert_error(id_, expected, given=response.body)
+
     def test_api_create_user(self):
         username = 'test_new_api_user'
         email = username + "@foo.com"
@@ -175,7 +190,6 @@ class TestCreateUser(object):
             fixture.destroy_repo_group(username)
             fixture.destroy_user(usr.user_id)
 
-
     @mock.patch.object(UserModel, 'create_or_update', crash)
     def test_api_create_user_when_exception_happened(self):
 

rhodecode/api/views/user_api.py

@@ -20,7 +20,8 @@
 
 import logging
 
-from rhodecode.api import jsonrpc_method, JSONRPCError, JSONRPCForbidden
+from rhodecode.api import (
+    jsonrpc_method, JSONRPCError, JSONRPCForbidden, JSONRPCValidationError)
 from rhodecode.api.utils import (
     Optional, OAttr, has_superadmin_permission, get_user_or_error, store_update)
 from rhodecode.lib import audit_logger
@@ -29,6 +30,8 @@ from rhodecode.lib.exceptions import DefaultUserException
 from rhodecode.lib.utils2 import safe_int, str2bool
 from rhodecode.model.db import Session, User, Repository
 from rhodecode.model.user import UserModel
+from rhodecode.model import validation_schema
+from rhodecode.model.validation_schema.schemas import user_schema
 
 log = logging.getLogger(__name__)
 
@@ -238,17 +241,45 @@ def create_user(request, apiuser, username, email, password=Optional(''),
     if isinstance(create_repo_group, basestring):
         create_repo_group = str2bool(create_repo_group)
 
+    username = Optional.extract(username)
+    password = Optional.extract(password)
+    email = Optional.extract(email)
+    first_name = Optional.extract(firstname)
+    last_name = Optional.extract(lastname)
+    active = Optional.extract(active)
+    admin = Optional.extract(admin)
+    extern_type = Optional.extract(extern_type)
+    extern_name = Optional.extract(extern_name)
+
+    schema = user_schema.UserSchema().bind(
+        # user caller
+        user=apiuser)
+    try:
+        schema_data = schema.deserialize(dict(
+            username=username,
+            email=email,
+            password=password,
+            first_name=first_name,
+            last_name=last_name,
+            active=active,
+            admin=admin,
+            extern_type=extern_type,
+            extern_name=extern_name,
+            ))
+    except validation_schema.Invalid as err:
+        raise JSONRPCValidationError(colander_exc=err)
+
     try:
         user = UserModel().create_or_update(
-            username=Optional.extract(username),
-            password=Optional.extract(password),
-            email=Optional.extract(email),
-            firstname=Optional.extract(firstname),
-            lastname=Optional.extract(lastname),
-            active=Optional.extract(active),
-            admin=Optional.extract(admin),
-            extern_type=Optional.extract(extern_type),
-            extern_name=Optional.extract(extern_name),
+            username=schema_data['username'],
+            password=schema_data['password'],
+            email=schema_data['email'],
+            firstname=schema_data['first_name'],
+            lastname=schema_data['last_name'],
+            active=schema_data['active'],
+            admin=schema_data['admin'],
+            extern_type=schema_data['extern_type'],
+            extern_name=schema_data['extern_name'],
             force_password_change=Optional.extract(force_password_change),
             create_repo_group=create_repo_group
         )

rhodecode/api/views/user_group_api.py

@@ -20,8 +20,8 @@
 
 import logging
 
-from rhodecode.api import jsonrpc_method, JSONRPCError, JSONRPCForbidden, \
-    JSONRPCValidationError
+from rhodecode.api import (
+    jsonrpc_method, JSONRPCError, JSONRPCForbidden, JSONRPCValidationError)
 from rhodecode.api.utils import (
     Optional, OAttr, store_update, has_superadmin_permission, get_origin,
     get_user_or_error, get_user_group_or_error, get_perm_or_error)

rhodecode/model/validation_schema/schemas/user_schema.py

@@ -18,10 +18,12 @@
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
+import re
 import colander
 
 from rhodecode import forms
 from rhodecode.model.db import User
+from rhodecode.model.validation_schema import types, validators
 from rhodecode.translation import _
 from rhodecode.lib.auth import check_password
 
@@ -52,10 +54,72 @@ class ChangePasswordSchema(colander.Schema):
         widget=forms.widget.CheckedPasswordWidget(redisplay=True),
         validator=colander.Length(min=6))
 
-
     def validator(self, form, values):
         if values['current_password'] == values['new_password']:
             exc = colander.Invalid(form)
             exc['new_password'] = _('New password must be different '
                                     'to old password')
             raise exc
+
+
+@colander.deferred
+def deferred_username_validator(node, kw):
+
+    def name_validator(node, value):
+        msg = _(
+            u'Username may only contain alphanumeric characters '
+            u'underscores, periods or dashes and must begin with '
+            u'alphanumeric character or underscore')
+
+        if not re.match(r'^[\w]{1}[\w\-\.]{0,254}$', value):
+            raise colander.Invalid(node, msg)
+
+    return name_validator
+
+
+@colander.deferred
+def deferred_email_validator(node, kw):
+    # NOTE(marcink): we might provide uniqueness validation later here...
+    return colander.Email()
+
+
+class UserSchema(colander.Schema):
+    username = colander.SchemaNode(
+        colander.String(),
+        validator=deferred_username_validator)
+
+    email = colander.SchemaNode(
+        colander.String(),
+        validator=deferred_email_validator)
+
+    password = colander.SchemaNode(
+        colander.String(), missing='')
+
+    first_name = colander.SchemaNode(
+        colander.String(), missing='')
+
+    last_name = colander.SchemaNode(
+        colander.String(), missing='')
+
+    active = colander.SchemaNode(
+        types.StringBooleanType(),
+        missing=False)
+
+    admin = colander.SchemaNode(
+        types.StringBooleanType(),
+        missing=False)
+
+    extern_name = colander.SchemaNode(
+        colander.String(), missing='')
+
+    extern_type = colander.SchemaNode(
+        colander.String(), missing='')
+
+    def deserialize(self, cstruct):
+        """
+        Custom deserialize that allows to chain validation, and verify
+        permissions, and as last step uniqueness
+        """
+
+        appstruct = super(UserSchema, self).deserialize(cstruct)
+        return appstruct