Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,12 @@ and this project adheres to [Semantic Versioning][semver].
### Fixed

- Detect signing scheme from key material instead of assuming RSA ([757])
- Clone no longer fails when the repository path contains a space (e.g. a Windows home directory with a space in the user name) ([762])
- Surface the underlying git error when a clone fails, instead of hiding it behind a generic access message ([762])
- Correct the clone access error that rendered as "Cannot None ..." and stop misattributing a local failure to an access/authentication problem ([762])


[762]: https://github.com/openlawlibrary/taf/pull/762
[759]: https://github.com/openlawlibrary/taf/pull/759
[757]: https://github.com/openlawlibrary/taf/pull/757

Expand Down
143 changes: 103 additions & 40 deletions taf/git.py
Original file line number Diff line number Diff line change
Expand Up @@ -290,16 +290,33 @@ def _git(self, cmd, *args, **kwargs):
log_success_msg = kwargs.pop("log_success_msg", "")
error_if_not_exists = kwargs.pop("error_if_not_exists", True)

if len(args):
cmd = cmd.format(*args)
# Build an argv list rather than one string: run() whitespace-splits a
# single-string command, which shatters any argument containing a space
# (e.g. a Windows path under a home dir with a space in the user name) -
# both `git -C <path>` and any `{}` argument such as a local clone
# source. The template is tokenized first and each `{}` placeholder is
# filled positionally in place, so a substituted value stays a single
# argv token no matter what it contains, while the command's own
# tokenization is unchanged. Callers must therefore pass paths/URLs as
# `{}` arguments, not interpolate them into the command string.
command = ["git", "-C", str(self.path)]
if self.allow_unsafe:
command = f"git -C {self.path} -c safe.directory={self.path} {cmd}"
else:
command = f"git -C {self.path} {cmd}"
command += ["-c", f"safe.directory={self.path}"]
args_iter = iter(args)
for token in cmd.split():
placeholders = token.count("{}")
if placeholders:
token = token.format(*[next(args_iter) for _ in range(placeholders)])
# an empty token means an optional flag arg was "" (e.g. push's
# force/no-verify flags); drop it, matching the old .split() which
# collapsed empties, so it does not reach git as an empty argument
if token != "":
command.append(token)
command_str = " ".join(command)
result = None
if log_error or log_error_msg:
try:
result = run(command, **kwargs)
result = run(*command, **kwargs)
if log_success_msg:
self._log_debug(log_success_msg)
except subprocess.CalledProcessError as e:
Expand All @@ -314,7 +331,7 @@ def _git(self, cmd, *args, **kwargs):
error = GitError(self, message=log_error_msg, error=e)
self._log_error(error.message)
else:
error = GitError(self, command=command, error=e)
error = GitError(self, command=command_str, error=e)
# not every git error indicates a problem
# if it does, we expect that either custom error message will be provided
# or that the error will be reraised
Expand All @@ -323,9 +340,9 @@ def _git(self, cmd, *args, **kwargs):
raise error
else:
try:
result = run(command, **kwargs)
result = run(*command, **kwargs)
except subprocess.CalledProcessError as e:
raise GitError(self, command=command, error=e)
raise GitError(self, command=command_str, error=e)
if log_success_msg:
self._log_debug(log_success_msg)
return result
Expand Down Expand Up @@ -834,27 +851,33 @@ def clone(
joined_params = " ".join(params)

cloned = False
clone_errors = []
for url in self.urls:
self._log_info(f"trying to clone from {url}")
try:
# url is a `{}` arg (kept as a single token); joined_params holds
# controlled flags meant to expand into multiple tokens, so it
# stays in the template string
self._git(
"clone {} . {}",
f"clone {{}} . {joined_params}",
url,
joined_params,
log_success_msg=f"successfully cloned from {url}",
reraise_error=True,
timeout=60,
error_if_not_exists=False,
)
except GitError as e:
self._log_debug(f"could not clone from {url} due to {e}")
clone_errors.append(f"{url}: {e}")
else:
self._log_info(f"successfully cloned from {url}")
cloned = True
break

if not cloned:
self.raise_git_access_error(CloneRepoException)
self.raise_git_access_error(
CloneRepoException, operation="clone", underlying_errors=clone_errors
)

# the path is now a repository; drop any cached negative result from
# before the clone
Expand Down Expand Up @@ -889,8 +912,11 @@ def clone_from_disk(
# file-by-file object copy pygit2 performs. git silently falls back to
# copying when hardlinks are not possible, so this is never slower.
bare_flag = "--bare " if is_bare else ""
# pass the source path as a `{}` arg (the `{{}}` survives the f-string)
# so a path containing a space is not split into separate tokens
self._git(
f"clone --local {bare_flag}{local_path} .",
f"clone --local {bare_flag}{{}} .",
str(local_path),
error_if_not_exists=False,
reraise_error=True,
)
Expand Down Expand Up @@ -1502,7 +1528,9 @@ def get_last_remote_commit(
raise FetchException(
"Could not fetch the last remote commit. URL not found"
)
last_commit = self._git(f"--no-pager ls-remote {url} {branch}", log_error=True)
last_commit = self._git(
"--no-pager ls-remote {} {}", url, branch, log_error=True
)
if last_commit:
last_commit = last_commit.split("\t", 1)[0]
# in some cases (e.g. upstream is defined the result might contain a warning line)
Expand Down Expand Up @@ -1802,37 +1830,56 @@ def push(
raise PushFailedError(self, message=f"Push operation failed: {e}")

def raise_git_access_error(
self, error_cls=GitAccessDeniedException, operation=None, error_msg=""
self,
error_cls=GitAccessDeniedException,
operation="access",
error_msg="",
underlying_errors=None,
):
"""Raise ``error_cls`` explaining why a git network operation failed.

``operation`` names the attempted action (e.g. ``"clone"``) in the
message. ``error_msg``, when given, replaces the auto-selected guidance.
``underlying_errors`` is the list of concrete per-attempt git failures
(typically one per URL); they are surfaced on the exception so the real
cause is never hidden behind the guidance.
"""
# Pick the guidance that best fits what we can observe about the remote,
# then surface the concrete git failure(s) alongside it. The guidance is
# a best guess (an unauthenticated probe cannot tell "private" from
# "missing"); the underlying errors are the ground truth and must not be
# swallowed - historically only the guidance was shown, hiding the real
# cause (e.g. a command-construction failure, not an access problem).
hosts = {
h for h in (extract_hostname(url) for url in self.urls) if h is not None
}
unknown_hosts = [host for host in hosts if not is_host_known(host)]
if len(unknown_hosts):
message = _no_hosts_error_format.format(hostname=",".join(unknown_hosts))
raise error_cls(self, operation=operation, message=message)
repo_exists = any(repository_exists(url) for url in self.urls)
if repo_exists:
if error_msg:
guidance = error_msg
elif len(unknown_hosts):
guidance = _no_hosts_error_format.format(hostname=",".join(unknown_hosts))
elif any(repository_exists(url) for url in self.urls):
uses_ssh = any(url.startswith("git@") for url in self.urls)
if uses_ssh:
raise error_cls(
self,
operation=operation,
message=(
_clone_or_pull_error_message if error_msg == "" else error_msg
),
)
else:
raise error_cls(
self,
operation=operation,
message=(
_clone_or_pull_error_message_no_ssh
if error_msg == ""
else error_msg
),
)
raise error_cls(self, operation=operation)
guidance = (
_clone_or_pull_error_message
if uses_ssh
else _clone_or_pull_error_message_no_ssh
)
else:
guidance = _repo_not_found_error_message

parts = []
if underlying_errors:
parts.append(
"The git command failed with:\n" + "\n".join(underlying_errors)
)
if guidance:
parts.append(guidance)
raise error_cls(
self,
operation=operation,
message="\n\n".join(parts) if parts else None,
)

def remove_remote(self, remote_name: str) -> None:
try:
Expand Down Expand Up @@ -1931,7 +1978,9 @@ def safely_get_json(self, commit: Commitish, path: str) -> Optional[Dict]:
return None

def set_remote_url(self, new_url: str, remote: Optional[str] = "origin") -> None:
self._git(f"remote set-url {remote} {new_url}")
# new_url may be a local filesystem path containing spaces - pass it as
# a `{}` arg so it is not split into separate tokens
self._git("remote set-url {} {}", remote, new_url)

def set_head_to_branch(self, branch_name: str) -> None:
"""Point HEAD at the given local branch without checking it out."""
Expand Down Expand Up @@ -2226,6 +2275,20 @@ def _find_url(path, url):
)


_repo_not_found_error_message = (
"The remote repository could not be reached or found. If a git error is shown "
"above, start there - it is the actual failure. Otherwise, note that an "
"unauthenticated check cannot tell a private repository apart from a missing "
"one, so common causes are:\n\n"
"1. A wrong or mistyped repository URL.\n"
"2. A private repository your account has not been granted access to.\n"
"3. Missing or misconfigured credentials for this host:\n"
" - For SSH URLs, ensure your SSH key is added to your Git hosting account and "
"loaded (e.g. `ssh -T git@github.com`).\n"
" - For HTTPS URLs, ensure a valid token/credential helper is configured.\n"
)


_no_hosts_error_format = (
"Error: The host(s) '{hostname}' not recognized."
"To add a host to your known_hosts file, manually connect via SSH using the command:\n"
Expand Down
Loading