kallithea Changeset - d2f59de17bef

Changeset - d2f59de17bef

Parent rev.

Child rev.

[Not reviewed]

default

0 2 0

Mads Kiilerich (mads) - 5 years ago 2020-12-03 11:16:44
mads@kiilerich.com

Grafted from: 4581409d9119

vcs: use urlparse in _check_url

This makes URL checking more standard and more explicit.

2 files changed with 32 insertions and 11 deletions:

kallithea/lib/vcs/backends/git/repository.py

kallithea/lib/vcs/backends/hg/repository.py

0 comments (0 inline, 0 general)

kallithea/lib/vcs/backends/git/repository.py

➞

Show inline comments

@@ @@ -112,149 +112,155 @@ class GitRepository(BaseRepository): @@
         gitenv['GIT_CONFIG_NOGLOBAL'] = '1'
         assert isinstance(cmd, list), cmd
         cmd = [settings.GIT_EXECUTABLE_PATH, '-c', 'core.quotepath=false'] + cmd
         try:
             p = subprocessio.SubprocessIOChunker(cmd, cwd=cwd, env=gitenv, shell=False)
         except (EnvironmentError, OSError) as err:
             # output from the failing process is in str(EnvironmentError)
             msg = ("Couldn't run git command %s.\n"
                    "Subprocess failed with '%s': %s\n" %
                    (cmd, type(err).__name__, err)
             ).strip()
             log.error(msg)
             raise RepositoryError(msg)
         try:
             stdout = b''.join(p.output)
             stderr = b''.join(p.error)
         finally:
             p.close()
         # TODO: introduce option to make commands fail if they have any stderr output?
         if stderr:
             log.debug('stderr from %s:\n%s', cmd, stderr)
         else:
             log.debug('stderr from %s: None', cmd)
         return stdout, stderr
     def run_git_command(self, cmd):
         """
         Runs given ``cmd`` as git command with cwd set to current repo.
         Returns stdout as unicode str ... or raise RepositoryError.
         """
         cwd = None
         if os.path.isdir(self.path):
             cwd = self.path
         stdout, _stderr = self._run_git_command(cmd, cwd=cwd)
         return safe_str(stdout)
     @staticmethod
     def _check_url(url):
         r"""
         Raise URLError if url doesn't seem like a valid safe Git URL. We
         only allow http, https, git, and ssh URLs.
         For http and https URLs, make a connection and probe to see if it is valid.
         >>> GitRepository._check_url('git://example.com/my%20fine repo')
         >>> GitRepository._check_url('http://example.com:65537/repo')
         Traceback (most recent call last):
         ...
         urllib.error.URLError: <urlopen error Error parsing URL: 'http://example.com:65537/repo'>
         >>> GitRepository._check_url('foo')
         Traceback (most recent call last):
         ...
         urllib.error.URLError: <urlopen error Unsupported protocol in URL 'foo'>
         >>> GitRepository._check_url('file:///repo')
         Traceback (most recent call last):
         ...
         urllib.error.URLError: <urlopen error Unsupported protocol in URL 'file:///repo'>
         >>> GitRepository._check_url('git+http://example.com/repo')
         Traceback (most recent call last):
         ...
         urllib.error.URLError: <urlopen error Unsupported protocol in URL 'git+http://example.com/repo'>
         >>> GitRepository._check_url('git://example.com/%09')
         Traceback (most recent call last):
         ...
         urllib.error.URLError: <urlopen error Invalid escape character in path: '%'>
         >>> GitRepository._check_url('git://example.com/%x00')
         Traceback (most recent call last):
         ...
         urllib.error.URLError: <urlopen error Invalid escape character in path: '%'>
         >>> GitRepository._check_url(r'git://example.com/\u0009')
         Traceback (most recent call last):
         ...
         urllib.error.URLError: <urlopen error Invalid escape character in path: '\'>
         >>> GitRepository._check_url(r'git://example.com/\t')
         Traceback (most recent call last):
         ...
         urllib.error.URLError: <urlopen error Invalid escape character in path: '\'>
         >>> GitRepository._check_url('git://example.com/\t')
         Traceback (most recent call last):
         ...
         urllib.error.URLError: <urlopen error Invalid whitespace character in path: '\t'>
         """
         try:
             parsed_url = urllib.parse.urlparse(url)
             parsed_url.port  # trigger netloc parsing which might raise ValueError
         except ValueError:
             raise urllib.error.URLError("Error parsing URL: %r" % url)
         # check first if it's not an local url
         if os.path.isabs(url) and os.path.isdir(url):
             return
         if url.startswith('git://'):
             try:
                 _git_colon, _empty, _host, path = url.split('/', 3)
             except ValueError:
                 raise urllib.error.URLError("Invalid URL: %r" % url)
         if parsed_url.scheme == 'git':
             # Mitigate problems elsewhere with incorrect handling of encoded paths.
             # Don't trust urllib.parse.unquote but be prepared for more flexible implementations elsewhere.
             # Space is the only allowed whitespace character - directly or % encoded. No other % or \ is allowed.
             for c in path.replace('%20', ' '):
+            for c in parsed_url.path.replace('%20', ' '):
                 if c in '%\\':
                     raise urllib.error.URLError("Invalid escape character in path: '%s'" % c)
                 if c.isspace() and c != ' ':
                     raise urllib.error.URLError("Invalid whitespace character in path: %r" % c)
             return
-        if not url.startswith('http://') and not url.startswith('https://'):
+        if parsed_url.scheme not in ['http', 'https']:
             raise urllib.error.URLError("Unsupported protocol in URL %r" % url)
         url_obj = mercurial.util.url(safe_bytes(url))
         test_uri, handlers = get_urllib_request_handlers(url_obj)
         if not test_uri.endswith(b'info/refs'):
             test_uri = test_uri.rstrip(b'/') + b'/info/refs'
         url_obj.passwd = b'*****'
         cleaned_uri = str(url_obj)
         o = urllib.request.build_opener(*handlers)
         o.addheaders = [('User-Agent', 'git/1.7.8.0')]  # fake some git
         req = urllib.request.Request(
             "%s?%s" % (
                 safe_str(test_uri),
                 urllib.parse.urlencode({"service": 'git-upload-pack'})
             ))
         try:
             resp = o.open(req)
             if resp.code != 200:
                 raise Exception('Return Code is not 200')
         except Exception as e:
             # means it cannot be cloned
             raise urllib.error.URLError("[%s] org_exc: %s" % (cleaned_uri, e))
         # now detect if it's proper git repo
         gitdata = resp.read()
         if b'service=git-upload-pack' not in gitdata:
             raise urllib.error.URLError(
                 "url [%s] does not look like an git" % cleaned_uri)
     def _get_repo(self, create, src_url=None, update_after_clone=False,
                   bare=False):
         if create and os.path.exists(self.path):
             raise RepositoryError("Location already exist")
         if src_url and not create:
             raise RepositoryError("Create should be set to True if src_url is "
                                   "given (clone operation creates repository)")
         try:
             if create and src_url:
                 GitRepository._check_url(src_url)
                 self.clone(src_url, update_after_clone, bare)
                 return Repo(self.path)
             elif create:
                 os.makedirs(self.path)
                 if bare:

kallithea/lib/vcs/backends/hg/repository.py

➞

Show inline comments

@@ @@ -246,110 +246,125 @@ class MercurialRepository(BaseRepository @@
         :param rev1: Entry point from which diff is shown. Can be
           ``self.EMPTY_CHANGESET`` - in this case, patch showing all
           the changes since empty state of the repository until ``rev2``
         :param rev2: Until which revision changes should be shown.
         :param ignore_whitespace: If set to ``True``, would not show whitespace
           changes. Defaults to ``False``.
         :param context: How many lines before/after changed lines should be
           shown. Defaults to ``3``. If negative value is passed-in, it will be
           set to ``0`` instead.
         """
         # Negative context values make no sense, and will result in
         # errors. Ensure this does not happen.
         if context < 0:
             context = 0
         if hasattr(rev1, 'raw_id'):
             rev1 = getattr(rev1, 'raw_id')
         if hasattr(rev2, 'raw_id'):
             rev2 = getattr(rev2, 'raw_id')
         # Check if given revisions are present at repository (may raise
         # ChangesetDoesNotExistError)
         if rev1 != self.EMPTY_CHANGESET:
             self.get_changeset(rev1)
         self.get_changeset(rev2)
         if path:
             file_filter = mercurial.match.exact([safe_bytes(path)])
         else:
             file_filter = None
         return b''.join(mercurial.patch.diff(self._repo, rev1, rev2, match=file_filter,
                           opts=mercurial.mdiff.diffopts(git=True,
                                         showfunc=True,
                                         ignorews=ignore_whitespace,
                                         context=context)))
     @staticmethod
     def _check_url(url, repoui=None):
         r"""
         Raise URLError if url doesn't seem like a valid safe Hg URL. We
         only allow http, https, ssh, and hg-git URLs.
         For http, https and git URLs, make a connection and probe to see if it is valid.
         On failures it'll raise urllib2.HTTPError, exception is also thrown
         when the return code is non 200
         >>> MercurialRepository._check_url('file:///repo')
         >>> MercurialRepository._check_url('http://example.com:65537/repo')
         Traceback (most recent call last):
         ...
         urllib.error.URLError: <urlopen error Error parsing URL: 'http://example.com:65537/repo'>
         """
         try:
             parsed_url = urllib.parse.urlparse(url)
             parsed_url.port  # trigger netloc parsing which might raise ValueError
         except ValueError:
             raise urllib.error.URLError("Error parsing URL: %r" % url)
         # check first if it's not an local url
         url = safe_bytes(url)
         if os.path.isdir(url) or url.startswith(b'file:'):
         if os.path.isdir(url) or parsed_url.scheme == 'file':
             # When creating repos, _get_url will use file protocol for local paths
             return
         if url.startswith(b'ssh:'):
         url = safe_bytes(url)
         if parsed_url.scheme == 'ssh':
             # in case of invalid uri or authentication issues, sshpeer will
             # throw an exception.
             mercurial.sshpeer.instance(repoui or mercurial.ui.ui(), url, False).lookup(b'tip')
             return
         url_prefix = None
-        if b'+' in url[:url.find(b'://')]:
+        if '+' in parsed_url.scheme:
             url_prefix, url = url.split(b'+', 1)
         url_obj = mercurial.util.url(url)
         test_uri, handlers = get_urllib_request_handlers(url_obj)
         url_obj.passwd = b'*****'
         cleaned_uri = str(url_obj)
         o = urllib.request.build_opener(*handlers)
         o.addheaders = [('Content-Type', 'application/mercurial-0.1'),
                         ('Accept', 'application/mercurial-0.1')]
         req = urllib.request.Request(
             "%s?%s" % (
                 safe_str(test_uri),
                 urllib.parse.urlencode({
                     'cmd': 'between',
                     'pairs': "%s-%s" % ('0' * 40, '0' * 40),
                 })
             ))
         try:
             resp = o.open(req)
             if resp.code != 200:
                 raise Exception('Return Code is not 200')
         except Exception as e:
             # means it cannot be cloned
             raise urllib.error.URLError("[%s] org_exc: %s" % (cleaned_uri, e))
         if not url_prefix: # skip git+http://... etc
             # now check if it's a proper hg repo
             try:
                 mercurial.httppeer.instance(repoui or mercurial.ui.ui(), url, False).lookup(b'tip')
             except Exception as e:
                 raise urllib.error.URLError(
                     "url [%s] does not look like an hg repo org_exc: %s"
                     % (cleaned_uri, e))
     def _get_repo(self, create, src_url=None, update_after_clone=False):
         """
         Function will check for mercurial repository in given path and return
         a localrepo object. If there is no repository in that path it will
         raise an exception unless ``create`` parameter is set to True - in
         that case repository would be created and returned.
         If ``src_url`` is given, would try to clone repository from the
         location at given clone_point. Additionally it'll make update to
         working copy accordingly to ``update_after_clone`` flag
         """

0 comments (0 inline, 0 general)