robojerk/tumbi-assembler
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add support for git-credential-helper

Browse source 
This patch adds an additional field `options` to scm_dict, which can be
used to provide additional information to the backends.

It implements a single new option for GitWrapper. This option allows
setting a custom git credentials wrapper. This can be useful if Pungi
needs to get files from a git repository that requires authentication.

The helper can be as simple as this (assuming the username is already
provided in the url):

    #!/bin/sh
    echo password=i-am-secret

The helper would need to be referenced by an absolute path from the
pungi configuration, or prefixed with ! to have git interpret it as a
shell script and look it up in PATH.

See https://git-scm.com/docs/gitcredentials for more details.

Signed-off-by: Lubomír Sedlář <lsedlar@redhat.com>
JIRA: RHELCMP-11808
This commit is contained in:
Lubomír Sedlář 2023-06-14 13:08:31 +02:00 • committed by lsedlar
parent e4c525ecbf
commit ada8f4e346
8 changed files with 139 additions and 51 deletions
Download patch file Download diff file Expand all files Collapse all files

8
pungi/checks.py
View file

@ -387,6 +387,7 @@ def _extend_with_default_and_alias(validator_class, offline=False):
instance[property]["branch"] = resolver(
instance[property]["repo"],
instance[property].get("branch") or "HEAD",
instance[property].get("options"),
)
for error in _hook_errors(properties, instance, schema):
@ -520,6 +521,13 @@ def make_schema():
"file": {"type": "string"},
"dir": {"type": "string"},
"command": {"type": "string"},
"options": {
"type": "object",
"properties": {
"credential_helper": {"type": "string"},
},
"additionalProperties": False,
},
},
"additionalProperties": False,
},

25
pungi/util.py
View file

@ -279,7 +279,7 @@ class GitUrlResolveError(RuntimeError):
pass
def resolve_git_ref(repourl, ref):
def resolve_git_ref(repourl, ref, credential_helper=None):
"""Resolve a reference in a Git repo to a commit.
Raises RuntimeError if there was an error. Most likely cause is failure to
@ -289,7 +289,7 @@ def resolve_git_ref(repourl, ref):
# This looks like a commit ID already.
return ref
try:
_, output = git_ls_remote(repourl, ref)
_, output = git_ls_remote(repourl, ref, credential_helper)
except RuntimeError as e:
raise GitUrlResolveError(
"ref does not exist in remote repo %s with the error %s %s"
@ -316,7 +316,7 @@ def resolve_git_ref(repourl, ref):
return lines[0].split()[0]
def resolve_git_url(url):
def resolve_git_url(url, credential_helper=None):
"""Given a url to a Git repo specifying HEAD or origin/<branch> as a ref,
replace that specifier with actual SHA1 of the commit.
@ -335,7 +335,7 @@ def resolve_git_url(url):
scheme = r.scheme.replace("git+", "")
baseurl = urllib.parse.urlunsplit((scheme, r.netloc, r.path, "", ""))
fragment = resolve_git_ref(baseurl, ref)
fragment = resolve_git_ref(baseurl, ref, credential_helper)
result = urllib.parse.urlunsplit((r.scheme, r.netloc, r.path, r.query, fragment))
if "?#" in url:
@ -354,13 +354,18 @@ class GitUrlResolver(object):
self.offline = offline
self.cache = {}
def __call__(self, url, branch=None):
def __call__(self, url, branch=None, options=None):
credential_helper = options.get("credential_helper") if options else None
if self.offline:
return branch or url
key = (url, branch)
if key not in self.cache:
try:
res = resolve_git_ref(url, branch) if branch else resolve_git_url(url)
res = (
resolve_git_ref(url, branch, credential_helper)
if branch
else resolve_git_url(url, credential_helper)
)
self.cache[key] = res
except GitUrlResolveError as exc:
self.cache[key] = exc
@ -991,8 +996,12 @@ def retry(timeout=120, interval=30, wait_on=Exception):
@retry(wait_on=RuntimeError)
def git_ls_remote(baseurl, ref):
return run(["git", "ls-remote", baseurl, ref], universal_newlines=True)
def git_ls_remote(baseurl, ref, credential_helper=None):
cmd = ["git"]
if credential_helper:
cmd.extend(["-c", "credential.useHttpPath=true"])
cmd.extend(["-c", "credential.helper=%s" % credential_helper])
return run(cmd + ["ls-remote", baseurl, ref], universal_newlines=True)
def get_tz_offset():

30
pungi/wrappers/scm.py
View file

@ -31,10 +31,11 @@ from .kojiwrapper import KojiWrapper
class ScmBase(kobo.log.LoggingBase):
def __init__(self, logger=None, command=None, compose=None):
def __init__(self, logger=None, command=None, compose=None, options=None):
kobo.log.LoggingBase.__init__(self, logger=logger)
self.command = command
self.compose = compose
self.options = options or {}
@retry(interval=60, timeout=300, wait_on=RuntimeError)
def retry_run(self, cmd, **kwargs):
@ -156,22 +157,31 @@ class GitWrapper(ScmBase):
if "://" not in repo:
repo = "file://%s" % repo
git_cmd = ["git"]
if "credential_helper" in self.options:
git_cmd.extend(["-c", "credential.useHttpPath=true"])
git_cmd.extend(
["-c", "credential.helper=%s" % self.options["credential_helper"]]
)
run(["git", "init"], workdir=destdir)
try:
run(["git", "fetch", "--depth=1", repo, branch], workdir=destdir)
run(git_cmd + ["fetch", "--depth=1", repo, branch], workdir=destdir)
run(["git", "checkout", "FETCH_HEAD"], workdir=destdir)
except RuntimeError as e:
# Fetch failed, to do a full clone we add a remote to our empty
# repo, get its content and check out the reference we want.
self.log_debug(
"Trying to do a full clone because shallow clone failed: %s %s"
% (e, e.output)
% (e, getattr(e, "output", ""))
)
try:
# Re-run git init in case of previous failure breaking .git dir
run(["git", "init"], workdir=destdir)
run(["git", "remote", "add", "origin", repo], workdir=destdir)
self.retry_run(["git", "remote", "update", "origin"], workdir=destdir)
self.retry_run(
git_cmd + ["remote", "update", "origin"], workdir=destdir
)
run(["git", "checkout", branch], workdir=destdir)
except RuntimeError:
if self.compose:
@ -361,15 +371,19 @@ def get_file_from_scm(scm_dict, target_path, compose=None):
scm_file = os.path.abspath(scm_dict)
scm_branch = None
command = None
options = {}
else:
scm_type = scm_dict["scm"]
scm_repo = scm_dict["repo"]
scm_file = scm_dict["file"]
scm_branch = scm_dict.get("branch", None)
command = scm_dict.get("command")
options = scm_dict.get("options", {})
logger = compose._logger if compose else None
scm = _get_wrapper(scm_type, logger=logger, command=command, compose=compose)
scm = _get_wrapper(
scm_type, logger=logger, command=command, compose=compose, options=options
)
files_copied = []
for i in force_list(scm_file):
@ -450,15 +464,19 @@ def get_dir_from_scm(scm_dict, target_path, compose=None):
scm_dir = os.path.abspath(scm_dict)
scm_branch = None
command = None
options = {}
else:
scm_type = scm_dict["scm"]
scm_repo = scm_dict.get("repo", None)
scm_dir = scm_dict["dir"]
scm_branch = scm_dict.get("branch", None)
command = scm_dict.get("command")
options = scm_dict.get("options", {})
logger = compose._logger if compose else None
scm = _get_wrapper(scm_type, logger=logger, command=command, compose=compose)
scm = _get_wrapper(
scm_type, logger=logger, command=command, compose=compose, options=options
)
with temp_dir(prefix="scm_checkout_") as tmp_dir:
scm.export_dir(scm_repo, scm_dir, scm_branch=scm_branch, target_dir=tmp_dir)