Changeset - a6dfd14d4b20
[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:
0 comments (0 inline, 0 general)
kallithea/lib/vcs/backends/git/repository.py
Show inline comments
 
@@ -14,12 +14,18 @@ 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
 
@@ -651,39 +657,39 @@ class GitRepository(BaseRepository):
 
        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
 
        """
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
 
@@ -23,12 +24,38 @@ class GitRepositoryTest(unittest.TestCas
 
        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))
0 comments (0 inline, 0 general)