kallithea Changeset - a6dfd14d4b20

Changeset - a6dfd14d4b20

Parent rev.

Child rev.

[Not reviewed]

default

0 2 0

Daniel Anderson - 11 years ago 2014-07-18 23:10:58
daniel@dattrix.com

Escape shell command parts for Git backend

Python 2.x uses pipes.quote and Python 3.3+ uses shlex.quote in
GitRepository._run_git_command()

2 files changed with 37 insertions and 4 deletions:

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

kallithea/tests/vcs/test_git.py

0 comments (0 inline, 0 general)

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

➞

Show inline comments

@@ @@ -8,24 +8,30 @@ @@
     :created_on: Apr 8, 2010
     :copyright: (c) 2010-2011 by Marcin Kuzminski, Lukasz Balcerzak.
 """
 import os
 import re
 import time
 import urllib
 import urllib2
 import logging
 import posixpath
 import string
 try:
     # Python <=2.7
     from pipes import quote
 except ImportError:
     # Python 3.3+
     from shlex import quote
 from dulwich.objects import Tag
 from dulwich.repo import Repo, NotGitRepository
 from dulwich.config import ConfigFile
 from kallithea.lib.vcs import subprocessio
 from kallithea.lib.vcs.backends.base import BaseRepository, CollectionGenerator
 from kallithea.lib.vcs.conf import settings
 from kallithea.lib.vcs.exceptions import (
     BranchDoesNotExistError, ChangesetDoesNotExistError, EmptyRepositoryError,
     RepositoryError, TagAlreadyExistError, TagDoesNotExistError
@@ @@ -645,51 +651,51 @@ class GitRepository(BaseRepository): @@
         :param update_after_clone: If set to ``False``, git won't checkout
           working directory
         :param bare: If set to ``True``, repository would be cloned into
           *bare* git repository (no working directory at all).
         """
         url = self._get_url(url)
         cmd = ['clone', '-q']
         if bare:
             cmd.append('--bare')
         elif not update_after_clone:
             cmd.append('--no-checkout')
-        cmd += ['--', '"%s"' % url, '"%s"' % self.path]
+        cmd += ['--', quote(url), self.path]
         cmd = ' '.join(cmd)
         # If error occurs run_git_command raises RepositoryError already
         self.run_git_command(cmd)
     def pull(self, url):
         """
         Tries to pull changes from external location.
         """
         url = self._get_url(url)
         cmd = ['pull', "--ff-only", url]
+        cmd = ['pull', "--ff-only", quote(url)]
         cmd = ' '.join(cmd)
         # If error occurs run_git_command raises RepositoryError already
         self.run_git_command(cmd)
     def fetch(self, url):
         """
         Tries to pull changes from external location.
         """
         url = self._get_url(url)
         so, se = self.run_git_command('ls-remote -h %s' % url)
+        so, se = self.run_git_command('ls-remote -h %s' % quote(url))
         refs = []
         for line in (x for x in so.splitlines()):
             sha, ref = line.split('\t')
             refs.append(ref)
         refs = ' '.join(('+%s:%s' % (r, r) for r in refs))
         cmd = '''fetch %s -- %s''' % (url, refs)
+        cmd = '''fetch %s -- %s''' % (quote(url), refs)
         self.run_git_command(cmd)
     def _update_server_info(self):
         """
         runs gits update-server-info command in this repo instance
         """
         from dulwich.server import update_server_info
         update_server_info(self._repo)
     @LazyProperty
     def workdir(self):
         """

kallithea/tests/vcs/test_git.py

➞

Show inline comments

 from __future__ import with_statement
 import os
 import mock
 import datetime
 import urllib2
 from kallithea.lib.vcs.backends.git import GitRepository, GitChangeset
 from kallithea.lib.vcs.exceptions import RepositoryError, VCSError, NodeDoesNotExistError
 from kallithea.lib.vcs.nodes import NodeKind, FileNode, DirNode, NodeState
 from kallithea.lib.vcs.utils.compat import unittest
 from kallithea.tests.vcs.base import BackendTestMixin
 from kallithea.tests.vcs.conf import TEST_GIT_REPO, TEST_GIT_REPO_CLONE, get_new_dir
 class GitRepositoryTest(unittest.TestCase):
     def __check_for_existing_repo(self):
         if os.path.exists(TEST_GIT_REPO_CLONE):
             self.fail('Cannot test git clone repo as location %s already '
                       'exists. You should manually remove it first.'
                       % TEST_GIT_REPO_CLONE)
     def setUp(self):
         self.repo = GitRepository(TEST_GIT_REPO)
     def test_wrong_repo_path(self):
         wrong_repo_path = '/tmp/errorrepo'
         self.assertRaises(RepositoryError, GitRepository, wrong_repo_path)
     def test_git_cmd_injection(self):
         remote_repo_url = 'https://github.com/codeinn/vcs.git'
         inject_remote = '%s;%s' % (remote_repo_url, '; echo "Cake";')
         with self.assertRaises(urllib2.URLError):
             # Should fail because URL will be: https://github.com/codeinn/vcs.git%3B%3B%20echo%20%22Cake%22%3B
             urlerror_fail_repo = GitRepository(get_new_dir('injection-repo'), src_url=inject_remote, update_after_clone=True, create=True)
         with self.assertRaises(RepositoryError):
             # Should fail on direct clone call, which as of this writing does not happen outside of class
             clone_fail_repo = GitRepository(get_new_dir('injection-repo'), create=True)
             clone_fail_repo.clone(inject_remote, update_after_clone=True,)
         successfully_cloned = GitRepository(get_new_dir('injection-repo'), src_url=remote_repo_url, update_after_clone=True, create=True)
         # Repo should have been created
         self.assertFalse(successfully_cloned._repo.bare)
         with self.assertRaises(RepositoryError):
             # Should fail because URL will be invalid repo
             inject_remote_var = '%s;%s' % (remote_repo_url, '; echo $PATH;')
             successfully_cloned.fetch(inject_remote_var)
         with self.assertRaises(RepositoryError):
             # Should fail because URL will be invalid repo
             inject_remote_ls = '%s;%s' % (remote_repo_url, '; ls -1 ~;')
             successfully_cloned.pull(inject_remote_ls)
     def test_repo_clone(self):
         self.__check_for_existing_repo()
         repo = GitRepository(TEST_GIT_REPO)
         repo_clone = GitRepository(TEST_GIT_REPO_CLONE,
             src_url=TEST_GIT_REPO, create=True, update_after_clone=True)
         self.assertEqual(len(repo.revisions), len(repo_clone.revisions))
         # Checking hashes of changesets should be enough
         for changeset in repo.get_changesets():
             raw_id = changeset.raw_id
             self.assertEqual(raw_id, repo_clone.get_changeset(raw_id).raw_id)
     def test_repo_clone_without_create(self):

0 comments (0 inline, 0 general)