validators/schemas: python3 fixes str vs unicode and few test breaking fixes

rhodecode/model/validation_schema/preparers.py

@@ -1,88 +1,99 b''
 
 
 # Copyright (C) 2016-2020 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
-import unicodedata
+from rhodecode.lib.str_utils import convert_special_chars, safe_int, safe_bytes
+
+
+def to_bytes_preparer(value):
+
+    if value:
+        value = safe_bytes(value)
+    return value
 
 
 def strip_preparer(value):
     """
     strips given values using .strip() function
     """
 
     if value:
         value = value.strip()
     return value
 
 
 def slugify_preparer(value, keep_case=True):
     """
     Slugify given value to a safe representation for url/id
     """
     from rhodecode.lib.utils import repo_name_slug
     if value:
         value = repo_name_slug(value if keep_case else value.lower())
     return value
 
 
 def non_ascii_strip_preparer(value):
     """
     trie to replace non-ascii letters to their ascii representation
     eg::
 
         `żołw` converts into `zolw`
     """
     if value:
-        value = unicodedata.normalize('NFKD', value).encode('ascii', 'ignore')
+        value = convert_special_chars(value)
     return value
 
 
 def unique_list_preparer(value):
     """
     Converts an list to a list with only unique values
     """
 
     def make_unique(value):
         seen = []
         return [c for c in value if
                 not (c in seen or seen.append(c))]
 
     if isinstance(value, list):
         ret_val = make_unique(value)
     elif isinstance(value, set):
         ret_val = list(value)
     elif isinstance(value, tuple):
         ret_val = make_unique(value)
     elif value is None:
         ret_val = []
     else:
         ret_val = [value]
 
     return ret_val
 
 
 def unique_list_from_str_preparer(value):
     """
     Converts an list to a list with only unique values
     """
     from rhodecode.lib.utils2 import aslist
 
     if isinstance(value, str):
         value = aslist(value, ',')
-    return unique_list_preparer(value)
  No newline at end of file
+    return unique_list_preparer(value)
+
+
+def ensure_value_is_int(value):
+    return safe_int(value)

rhodecode/model/validation_schema/schemas/gist_schema.py

@@ -1,183 +1,194 b''
 
 
 # Copyright (C) 2016-2020 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import os
 
 import colander
 
+from rhodecode.lib.str_utils import safe_str
 from rhodecode.translation import _
 from rhodecode.model.validation_schema import preparers
 
 
 def nodes_to_sequence(nodes, colander_node=None):
     """
     Converts old style dict nodes to new list of dicts
 
     :param nodes: dict with key beeing name of the file
 
     """
     if not isinstance(nodes, dict):
-        msg = 'Nodes needs to be a dict, got {}'.format(type(nodes))
+        msg = f'Nodes needs to be a dict, got {type(nodes)}'
         raise colander.Invalid(colander_node, msg)
     out = []
 
     for key, val in nodes.items():
         val = (isinstance(val, dict) and val) or {}
         out.append(dict(
             filename=key,
             content=val.get('content'),
             mimetype=val.get('mimetype')
         ))
 
     out = Nodes().deserialize(out)
     return out
 
 
 def sequence_to_nodes(nodes, colander_node=None):
+
     if not isinstance(nodes, list):
         msg = 'Nodes needs to be a list, got {}'.format(type(nodes))
         raise colander.Invalid(colander_node, msg)
     nodes = Nodes().deserialize(nodes)
 
     out = {}
     try:
         for file_data in nodes:
             file_data_skip = file_data.copy()
             # if we got filename_org we use it as a key so we keep old
             # name as input and rename is-reflected inside the values as
             # filename and filename_org differences.
             filename_org = file_data.get('filename_org')
             filename = filename_org or file_data['filename']
             out[filename] = {}
             out[filename].update(file_data_skip)
 
     except Exception as e:
         msg = 'Invalid data format org_exc:`{}`'.format(repr(e))
         raise colander.Invalid(colander_node, msg)
     return out
 
 
 @colander.deferred
 def deferred_lifetime_validator(node, kw):
     options = kw.get('lifetime_options', [])
     return colander.All(
         colander.Range(min=-1, max=60 * 24 * 30 * 12),
         colander.OneOf([x for x in options]))
 
 
 def unique_gist_validator(node, value):
     from rhodecode.model.db import Gist
     existing = Gist.get_by_access_id(value)
     if existing:
-        msg = _(u'Gist with name {} already exists').format(value)
+        msg = _('Gist with name {} already exists').format(value)
         raise colander.Invalid(node, msg)
 
 
 def filename_validator(node, value):
     if value != os.path.basename(value):
-        msg = _(u'Filename {} cannot be inside a directory').format(value)
+        msg = _('Filename {} cannot be inside a directory').format(safe_str(value))
         raise colander.Invalid(node, msg)
 
 
 class NodeSchema(colander.MappingSchema):
     # if we perform rename this will be org filename
     filename_org = colander.SchemaNode(
-        colander.String(),
-        preparer=[preparers.strip_preparer,
-                  preparers.non_ascii_strip_preparer],
+        colander.String(encoding='utf-8'),
+        preparer=[
+            preparers.strip_preparer,
+            preparers.non_ascii_strip_preparer,
+            preparers.to_bytes_preparer,
+        ],
         validator=filename_validator,
         missing=None)
 
     filename = colander.SchemaNode(
-        colander.String(),
-        preparer=[preparers.strip_preparer,
-                  preparers.non_ascii_strip_preparer],
+        colander.String(encoding='utf-8'),
+        preparer=[
+            preparers.strip_preparer,
+            preparers.non_ascii_strip_preparer,
+            preparers.to_bytes_preparer,
+        ],
         validator=filename_validator)
 
     content = colander.SchemaNode(
-        colander.String())
+        colander.String(encoding='utf-8'),
+        preparer=[preparers.to_bytes_preparer])
+
     mimetype = colander.SchemaNode(
         colander.String(),
         missing=None)
 
 
 class Nodes(colander.SequenceSchema):
     filenames = NodeSchema()
 
     def validator(self, node, cstruct):
         if not isinstance(cstruct, list):
             return
 
         found_filenames = []
         for data in cstruct:
             filename = data['filename']
             if filename in found_filenames:
                 msg = _('Duplicated value for filename found: `{}`').format(
                     filename)
                 raise colander.Invalid(node, msg)
             found_filenames.append(filename)
 
 
 class GistSchema(colander.MappingSchema):
     """
     schema = GistSchema()
     schema.bind(
         lifetime_options = [1,2,3]
     )
     out = schema.deserialize(dict(
         nodes=[
             {'filename': 'x', 'content': 'xxx', },
             {'filename': 'docs/Z', 'content': 'xxx', 'mimetype': 'x'},
         ]
     ))
     """
 
     from rhodecode.model.db import Gist
 
     gistid = colander.SchemaNode(
         colander.String(),
         missing=None,
         preparer=[preparers.strip_preparer,
                   preparers.non_ascii_strip_preparer,
                   preparers.slugify_preparer],
         validator=colander.All(
             colander.Length(min=3),
             unique_gist_validator
         ))
 
     description = colander.SchemaNode(
         colander.String(),
-        missing=u'')
+        missing='')
 
     lifetime = colander.SchemaNode(
         colander.Integer(),
         validator=deferred_lifetime_validator)
 
     gist_acl_level = colander.SchemaNode(
         colander.String(),
+        missing=Gist.ACL_LEVEL_PRIVATE,
         validator=colander.OneOf([Gist.ACL_LEVEL_PUBLIC,
                                   Gist.ACL_LEVEL_PRIVATE]))
 
     gist_type = colander.SchemaNode(
         colander.String(),
-        missing=Gist.GIST_PUBLIC,
+        missing=Gist.GIST_PRIVATE,
         validator=colander.OneOf([Gist.GIST_PRIVATE, Gist.GIST_PUBLIC]))
 
     nodes = Nodes()

rhodecode/model/validation_schema/schemas/repo_schema.py

@@ -1,450 +1,460 b''
 
 
 # Copyright (C) 2016-2020 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import colander
 import deform.widget
 
 from rhodecode.translation import _
 from rhodecode.model.validation_schema.utils import convert_to_optgroup, username_converter
 from rhodecode.model.validation_schema import validators, preparers, types
 
 DEFAULT_LANDING_REF = 'rev:tip'
 
 
 def get_group_and_repo(repo_name):
     from rhodecode.model.repo_group import RepoGroupModel
     return RepoGroupModel()._get_group_name_and_parent(
         repo_name, get_object=True)
 
 
 def get_repo_group(repo_group_id):
     from rhodecode.model.repo_group import RepoGroup
     return RepoGroup.get(repo_group_id), RepoGroup.CHOICES_SEPARATOR
 
 
 @colander.deferred
 def deferred_repo_type_validator(node, kw):
     options = kw.get('repo_type_options', [])
     return colander.OneOf([x for x in options])
 
 
 @colander.deferred
 def deferred_repo_owner_validator(node, kw):
 
     def repo_owner_validator(node, value):
         from rhodecode.model.db import User
         value = username_converter(value)
         existing = User.get_by_username(value)
         if not existing:
             msg = _(u'Repo owner with id `{}` does not exists').format(value)
             raise colander.Invalid(node, msg)
 
     return repo_owner_validator
 
 
 @colander.deferred
 def deferred_landing_ref_validator(node, kw):
     options = kw.get(
         'repo_ref_options', [DEFAULT_LANDING_REF])
     return colander.OneOf([x for x in options])
 
 
 @colander.deferred
 def deferred_sync_uri_validator(node, kw):
     repo_type = kw.get('repo_type')
     validator = validators.CloneUriValidator(repo_type)
     return validator
 
 
 @colander.deferred
 def deferred_landing_ref_widget(node, kw):
     from rhodecode.model.scm import ScmModel
 
     repo_type = kw.get('repo_type')
     default_opts = []
     if repo_type:
         default_landing_ref, _lbl = ScmModel.backend_landing_ref(repo_type)
         default_opts.append((default_landing_ref, default_landing_ref))
 
     items = kw.get('repo_ref_items', default_opts)
     items = convert_to_optgroup(items)
     return deform.widget.Select2Widget(values=items)
 
 
 @colander.deferred
 def deferred_fork_of_validator(node, kw):
     old_values = kw.get('old_values') or {}
 
     def fork_of_validator(node, value):
         from rhodecode.model.db import Repository, RepoGroup
         existing = Repository.get_by_repo_name(value)
         if not existing:
             msg = _(u'Fork with id `{}` does not exists').format(value)
             raise colander.Invalid(node, msg)
         elif old_values['repo_name'] == existing.repo_name:
             msg = _(u'Cannot set fork of '
                     u'parameter of this repository to itself').format(value)
             raise colander.Invalid(node, msg)
 
     return fork_of_validator
 
 
 @colander.deferred
 def deferred_can_write_to_group_validator(node, kw):
     request_user = kw.get('user')
     old_values = kw.get('old_values') or {}
 
     def can_write_to_group_validator(node, value):
         """
         Checks if given repo path is writable by user. This includes checks if
         user is allowed to create repositories under root path or under
         repo group paths
         """
 
         from rhodecode.lib.auth import (
             HasPermissionAny, HasRepoGroupPermissionAny)
         from rhodecode.model.repo_group import RepoGroupModel
 
         messages = {
             'invalid_repo_group':
                 _(u"Repository group `{}` does not exist"),
             # permissions denied we expose as not existing, to prevent
             # resource discovery
             'permission_denied':
                 _(u"Repository group `{}` does not exist"),
             'permission_denied_root':
                 _(u"You do not have the permission to store "
                   u"repositories in the root location.")
         }
 
         value = value['repo_group_name']
 
         is_root_location = value is types.RootLocation
         # NOT initialized validators, we must call them
         can_create_repos_at_root = HasPermissionAny('hg.admin', 'hg.create.repository')
 
         # if values is root location, we simply need to check if we can write
         # to root location !
         if is_root_location:
 
             if can_create_repos_at_root(user=request_user):
                 # we can create repo group inside tool-level. No more checks
                 # are required
                 return
             else:
                 old_name = old_values.get('repo_name')
                 if old_name and old_name == old_values.get('submitted_repo_name'):
                     # since we didn't change the name, we can skip validation and
                     # allow current users without store-in-root permissions to update
                     return
 
                 # "fake" node name as repo_name, otherwise we oddly report
                 # the error as if it was coming form repo_group
                 # however repo_group is empty when using root location.
                 node.name = 'repo_name'
                 raise colander.Invalid(node, messages['permission_denied_root'])
 
         # parent group not exists ? throw an error
         repo_group = RepoGroupModel().get_by_group_name(value)
         if value and not repo_group:
             raise colander.Invalid(
                 node, messages['invalid_repo_group'].format(value))
 
         gr_name = repo_group.group_name
 
         # create repositories with write permission on group is set to true
         create_on_write = HasPermissionAny(
             'hg.create.write_on_repogroup.true')(user=request_user)
 
         group_admin = HasRepoGroupPermissionAny('group.admin')(
             gr_name, 'can write into group validator', user=request_user)
         group_write = HasRepoGroupPermissionAny('group.write')(
             gr_name, 'can write into group validator', user=request_user)
 
         forbidden = not (group_admin or (group_write and create_on_write))
 
         # TODO: handling of old values, and detecting no-change in path
         # to skip permission checks in such cases. This only needs to be
         # implemented if we use this schema in forms as well
 
         # gid = (old_data['repo_group'].get('group_id')
         #        if (old_data and 'repo_group' in old_data) else None)
         # value_changed = gid != safe_int(value)
         # new = not old_data
 
         # do check if we changed the value, there's a case that someone got
         # revoked write permissions to a repository, he still created, we
         # don't need to check permission if he didn't change the value of
         # groups in form box
         # if value_changed or new:
         #     # parent group need to be existing
         # TODO: ENDS HERE
 
         if repo_group and forbidden:
             msg = messages['permission_denied'].format(value)
             raise colander.Invalid(node, msg)
 
     return can_write_to_group_validator
 
 
 @colander.deferred
 def deferred_unique_name_validator(node, kw):
     request_user = kw.get('user')
     old_values = kw.get('old_values') or {}
 
     def unique_name_validator(node, value):
         from rhodecode.model.db import Repository, RepoGroup
         name_changed = value != old_values.get('repo_name')
 
         existing = Repository.get_by_repo_name(value)
         if name_changed and existing:
             msg = _(u'Repository with name `{}` already exists').format(value)
             raise colander.Invalid(node, msg)
 
         existing_group = RepoGroup.get_by_group_name(value)
         if name_changed and existing_group:
             msg = _(u'Repository group with name `{}` already exists').format(
                 value)
             raise colander.Invalid(node, msg)
     return unique_name_validator
 
 
 @colander.deferred
 def deferred_repo_name_validator(node, kw):
     def no_git_suffix_validator(node, value):
         if value.endswith('.git'):
             msg = _('Repository name cannot end with .git')
             raise colander.Invalid(node, msg)
     return colander.All(
         no_git_suffix_validator, validators.valid_name_validator)
 
 
 @colander.deferred
 def deferred_repo_group_validator(node, kw):
-    options = kw.get(
-        'repo_repo_group_options')
-    return colander.OneOf([x for x in options])
+    options = kw.get('repo_repo_group_options')
+
+    def repo_group_validator(node, value):
+        choices = [x for x in options]
+        err = _('Group ID: `${val}` is not one of allowed ${choices}')
+
+        if preparers.ensure_value_is_int(value) not in choices:
+            choices = ', '.join(['%s' % x for x in sorted(choices)])
+            err = _(err, mapping={'val': value, 'choices': choices})
+            raise colander.Invalid(node, err)
+
+    return repo_group_validator
 
 
 @colander.deferred
 def deferred_repo_group_widget(node, kw):
     items = kw.get('repo_repo_group_items')
     return deform.widget.Select2Widget(values=items)
 
 
 class GroupType(colander.Mapping):
     def _validate(self, node, value):
         try:
             return dict(repo_group_name=value)
         except Exception as e:
             raise colander.Invalid(
                 node, '"${val}" is not a mapping type: ${err}'.format(
                     val=value, err=e))
 
     def deserialize(self, node, cstruct):
         if cstruct is colander.null:
             return cstruct
 
         appstruct = super(GroupType, self).deserialize(node, cstruct)
         validated_name = appstruct['repo_group_name']
 
         # inject group based on once deserialized data
         (repo_name_without_group,
          parent_group_name,
          parent_group) = get_group_and_repo(validated_name)
 
         appstruct['repo_name_with_group'] = validated_name
         appstruct['repo_name_without_group'] = repo_name_without_group
         appstruct['repo_group_name'] = parent_group_name or types.RootLocation
 
         if parent_group:
             appstruct['repo_group_id'] = parent_group.group_id
 
         return appstruct
 
 
 class GroupSchema(colander.SchemaNode):
     schema_type = GroupType
     validator = deferred_can_write_to_group_validator
     missing = colander.null
 
 
 class RepoGroup(GroupSchema):
     repo_group_name = colander.SchemaNode(
         types.GroupNameType())
     repo_group_id = colander.SchemaNode(
         colander.String(), missing=None)
     repo_name_without_group = colander.SchemaNode(
         colander.String(), missing=None)
 
 
 class RepoGroupAccessSchema(colander.MappingSchema):
     repo_group = RepoGroup()
 
 
 class RepoNameUniqueSchema(colander.MappingSchema):
     unique_repo_name = colander.SchemaNode(
         colander.String(),
         validator=deferred_unique_name_validator)
 
 
 class RepoSchema(colander.MappingSchema):
 
     repo_name = colander.SchemaNode(
         types.RepoNameType(),
         validator=deferred_repo_name_validator)
 
     repo_type = colander.SchemaNode(
         colander.String(),
         validator=deferred_repo_type_validator)
 
     repo_owner = colander.SchemaNode(
         colander.String(),
         validator=deferred_repo_owner_validator,
         widget=deform.widget.TextInputWidget())
 
     repo_description = colander.SchemaNode(
         colander.String(), missing='',
         widget=deform.widget.TextAreaWidget())
 
     repo_landing_commit_ref = colander.SchemaNode(
         colander.String(),
         validator=deferred_landing_ref_validator,
         preparers=[preparers.strip_preparer],
         missing=DEFAULT_LANDING_REF,
         widget=deferred_landing_ref_widget)
 
     repo_clone_uri = colander.SchemaNode(
         colander.String(),
         validator=deferred_sync_uri_validator,
         preparers=[preparers.strip_preparer],
         missing='')
 
     repo_push_uri = colander.SchemaNode(
         colander.String(),
         validator=deferred_sync_uri_validator,
         preparers=[preparers.strip_preparer],
         missing='')
 
     repo_fork_of = colander.SchemaNode(
         colander.String(),
         validator=deferred_fork_of_validator,
         missing=None)
 
     repo_private = colander.SchemaNode(
         types.StringBooleanType(),
         missing=False, widget=deform.widget.CheckboxWidget())
     repo_copy_permissions = colander.SchemaNode(
         types.StringBooleanType(),
         missing=False, widget=deform.widget.CheckboxWidget())
     repo_enable_statistics = colander.SchemaNode(
         types.StringBooleanType(),
         missing=False, widget=deform.widget.CheckboxWidget())
     repo_enable_downloads = colander.SchemaNode(
         types.StringBooleanType(),
         missing=False, widget=deform.widget.CheckboxWidget())
     repo_enable_locking = colander.SchemaNode(
         types.StringBooleanType(),
         missing=False, widget=deform.widget.CheckboxWidget())
 
     def deserialize(self, cstruct):
         """
         Custom deserialize that allows to chain validation, and verify
         permissions, and as last step uniqueness
         """
 
         # first pass, to validate given data
         appstruct = super(RepoSchema, self).deserialize(cstruct)
         validated_name = appstruct['repo_name']
 
         # second pass to validate permissions to repo_group
         if 'old_values' in self.bindings:
             # save current repo name for name change checks
             self.bindings['old_values']['submitted_repo_name'] = validated_name
         second = RepoGroupAccessSchema().bind(**self.bindings)
         appstruct_second = second.deserialize({'repo_group': validated_name})
         # save result
         appstruct['repo_group'] = appstruct_second['repo_group']
 
         # thirds to validate uniqueness
         third = RepoNameUniqueSchema().bind(**self.bindings)
         third.deserialize({'unique_repo_name': validated_name})
 
         return appstruct
 
 
 class RepoSettingsSchema(RepoSchema):
     repo_group = colander.SchemaNode(
         colander.Integer(),
         validator=deferred_repo_group_validator,
         widget=deferred_repo_group_widget,
+        preparers=[preparers.ensure_value_is_int],
         missing='')
 
     repo_clone_uri_change = colander.SchemaNode(
         colander.String(),
         missing='NEW')
 
     repo_clone_uri = colander.SchemaNode(
         colander.String(),
         preparers=[preparers.strip_preparer],
         validator=deferred_sync_uri_validator,
         missing='')
 
     repo_push_uri_change = colander.SchemaNode(
         colander.String(),
         missing='NEW')
 
     repo_push_uri = colander.SchemaNode(
         colander.String(),
         preparers=[preparers.strip_preparer],
         validator=deferred_sync_uri_validator,
         missing='')
 
     def deserialize(self, cstruct):
         """
         Custom deserialize that allows to chain validation, and verify
         permissions, and as last step uniqueness
         """
 
         # first pass, to validate given data
         appstruct = super(RepoSchema, self).deserialize(cstruct)
         validated_name = appstruct['repo_name']
         # because of repoSchema adds repo-group as an ID, we inject it as
         # full name here because validators require it, it's unwrapped later
         # so it's safe to use and final name is going to be without group anyway
 
         group, separator = get_repo_group(appstruct['repo_group'])
         if group:
             validated_name = separator.join([group.group_name, validated_name])
 
         # second pass to validate permissions to repo_group
         if 'old_values' in self.bindings:
             # save current repo name for name change checks
             self.bindings['old_values']['submitted_repo_name'] = validated_name
         second = RepoGroupAccessSchema().bind(**self.bindings)
         appstruct_second = second.deserialize({'repo_group': validated_name})
         # save result
         appstruct['repo_group'] = appstruct_second['repo_group']
 
         # thirds to validate uniqueness
         third = RepoNameUniqueSchema().bind(**self.bindings)
         third.deserialize({'unique_repo_name': validated_name})
 
         return appstruct

rhodecode/model/validation_schema/validators.py

@@ -1,160 +1,161 b''
 
 
 # Copyright (C) 2011-2020 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
-import os
 import re
 import logging
 
-
 import ipaddress
 import colander
 
 from rhodecode.translation import _
-from rhodecode.lib.utils2 import glob2re, safe_unicode
+from rhodecode.lib.utils2 import glob2re
+from rhodecode.lib.str_utils import safe_str
 from rhodecode.lib.ext_json import json
 
 log = logging.getLogger(__name__)
 
 
 def ip_addr_validator(node, value):
     try:
         # this raises an ValueError if address is not IpV4 or IpV6
-        ipaddress.ip_network(safe_unicode(value), strict=False)
+        ipaddress.ip_network(safe_str(value), strict=False)
     except ValueError:
-        msg = _(u'Please enter a valid IPv4 or IpV6 address')
+        msg = _('Please enter a valid IPv4 or IpV6 address')
         raise colander.Invalid(node, msg)
 
 
 class IpAddrValidator(object):
     def __init__(self, strict=True):
         self.strict = strict
 
     def __call__(self, node, value):
         try:
             # this raises an ValueError if address is not IpV4 or IpV6
-            ipaddress.ip_network(safe_unicode(value), strict=self.strict)
+            ipaddress.ip_network(safe_str(value), strict=self.strict)
         except ValueError:
-            msg = _(u'Please enter a valid IPv4 or IpV6 address')
+            msg = _('Please enter a valid IPv4 or IpV6 address')
             raise colander.Invalid(node, msg)
 
 
 def glob_validator(node, value):
     try:
         re.compile('^' + glob2re(value) + '$')
     except Exception:
-        msg = _(u'Invalid glob pattern')
+        msg = _('Invalid glob pattern')
         raise colander.Invalid(node, msg)
 
 
 def valid_name_validator(node, value):
     from rhodecode.model.validation_schema import types
     if value is types.RootLocation:
         return
 
     msg = _('Name must start with a letter or number. Got `{}`').format(value)
     if not re.match(r'^[a-zA-z0-9]{1,}', value):
         raise colander.Invalid(node, msg)
 
 
 class InvalidCloneUrl(Exception):
     allowed_prefixes = ()
 
 
 def url_validator(url, repo_type, config):
     from rhodecode.lib.vcs.backends.hg import MercurialRepository
     from rhodecode.lib.vcs.backends.git import GitRepository
     from rhodecode.lib.vcs.backends.svn import SubversionRepository
 
     if repo_type == 'hg':
         allowed_prefixes = ('http', 'svn+http', 'git+http')
 
         if 'http' in url[:4]:
             # initially check if it's at least the proper URL
             # or does it pass basic auth
 
             return MercurialRepository.check_url(url, config)
         elif 'svn+http' in url[:8]:  # svn->hg import
             SubversionRepository.check_url(url, config)
         elif 'git+http' in url[:8]:  # git->hg import
             raise NotImplementedError()
         else:
             exc = InvalidCloneUrl('Clone from URI %s not allowed. '
                                   'Allowed url must start with one of %s'
                                   % (url, ','.join(allowed_prefixes)))
             exc.allowed_prefixes = allowed_prefixes
             raise exc
 
     elif repo_type == 'git':
         allowed_prefixes = ('http', 'svn+http', 'hg+http')
         if 'http' in url[:4]:
             # initially check if it's at least the proper URL
             # or does it pass basic auth
             return GitRepository.check_url(url, config)
         elif 'svn+http' in url[:8]:  # svn->git import
             raise NotImplementedError()
         elif 'hg+http' in url[:8]:  # hg->git import
             raise NotImplementedError()
         else:
             exc = InvalidCloneUrl('Clone from URI %s not allowed. '
                                   'Allowed url must start with one of %s'
                                   % (url, ','.join(allowed_prefixes)))
             exc.allowed_prefixes = allowed_prefixes
             raise exc
     elif repo_type == 'svn':
         # no validation for SVN yet
         return
 
     raise InvalidCloneUrl('Invalid repo type specified: `{}`'.format(repo_type))
 
 
 class CloneUriValidator(object):
     def __init__(self, repo_type):
         self.repo_type = repo_type
 
     def __call__(self, node, value):
 
         from rhodecode.lib.utils import make_db_config
         try:
             config = make_db_config(clear_session=False)
             url_validator(value, self.repo_type, config)
         except InvalidCloneUrl as e:
             log.warning(e)
-            raise colander.Invalid(node, e.message)
-        except Exception:
+            raise colander.Invalid(node, str(e))
+        except Exception as e:
             log.exception('Url validation failed')
-            msg = _(u'invalid clone url or credentials for {repo_type} repository').format(
-                repo_type=self.repo_type)
+            reason = repr(e)
+            reason = reason.replace('<', '&lt;').replace('>', '&gt;')
+            msg = _('invalid clone url or credentials for {repo_type} repository. Reason: {reason}')\
+                .format(reason=reason, repo_type=self.repo_type)
             raise colander.Invalid(node, msg)
 
 
 def json_validator(node, value):
     try:
         json.loads(value)
-    except (Exception,) as e:
-        msg = _(u'Please enter a valid json object')
+    except (Exception,):
+        msg = _('Please enter a valid json object')
         raise colander.Invalid(node, msg)
 
 
 def json_validator_with_exc(node, value):
     try:
         json.loads(value)
     except (Exception,) as e:
-        msg = _(u'Please enter a valid json object: `{}`'.format(e))
+        msg = _('Please enter a valid json object: `{}`'.format(e))
         raise colander.Invalid(node, msg)