diff --git a/kallithea/lib/vcs/backends/hg/repository.py b/kallithea/lib/vcs/backends/hg/repository.py --- a/kallithea/lib/vcs/backends/hg/repository.py +++ b/kallithea/lib/vcs/backends/hg/repository.py @@ -320,7 +320,7 @@ class MercurialRepository(BaseRepository req = urllib.request.Request( "%s?%s" % ( - test_uri, + safe_str(test_uri), urllib.parse.urlencode({ 'cmd': 'between', 'pairs': "%s-%s" % ('0' * 40, '0' * 40), diff --git a/kallithea/lib/vcs/utils/helpers.py b/kallithea/lib/vcs/utils/helpers.py --- a/kallithea/lib/vcs/utils/helpers.py +++ b/kallithea/lib/vcs/utils/helpers.py @@ -11,6 +11,7 @@ import urllib.request import mercurial.url from kallithea.lib.vcs.exceptions import RepositoryError, VCSError +from kallithea.lib.vcs.utils import safe_str from kallithea.lib.vcs.utils.paths import abspath @@ -226,9 +227,21 @@ def get_urllib_request_handlers(url_obj) test_uri, authinfo = url_obj.authinfo() if authinfo: + # authinfo is a tuple (realm, uris, user, password) where 'uris' itself + # is a tuple of URIs. + # If url_obj is obtained via mercurial.util.url, the obtained authinfo + # values will be bytes, e.g. + # (None, (b'http://127.0.0.1/repo', b'127.0.0.1'), b'user', b'pass') + # However, urllib expects strings, not bytes, so we must convert them. + # create a password manager passmgr = urllib.request.HTTPPasswordMgrWithDefaultRealm() - passmgr.add_password(*authinfo) + passmgr.add_password( + safe_str(authinfo[0]) if authinfo[0] else None, # realm + tuple(safe_str(x) for x in authinfo[1]), # uris + safe_str(authinfo[2]), # user + safe_str(authinfo[3]), # password + ) handlers.extend((mercurial.url.httpbasicauthhandler(passmgr), mercurial.url.httpdigestauthhandler(passmgr))) diff --git a/kallithea/tests/functional/test_admin_repos.py b/kallithea/tests/functional/test_admin_repos.py --- a/kallithea/tests/functional/test_admin_repos.py +++ b/kallithea/tests/functional/test_admin_repos.py @@ -344,6 +344,19 @@ class _BaseTestCase(base.TestController) _session_csrf_secret_token=self.session_csrf_secret_token())) response.mustcontain('Invalid repository URL') + def test_create_remote_repo_wrong_clone_uri_http_auth(self): + self.log_user() + repo_name = self.NEW_REPO + description = 'description for newly created repo' + response = self.app.post(base.url('repos'), + fixture._get_repo_create_params(repo_private=False, + repo_name=repo_name, + repo_type=self.REPO_TYPE, + repo_description=description, + clone_uri='http://user:pass@127.0.0.1/repo', + _session_csrf_secret_token=self.session_csrf_secret_token())) + response.mustcontain('Invalid repository URL') + def test_delete(self): self.log_user() repo_name = 'vcs_test_new_to_delete_%s' % self.REPO_TYPE