diff --git a/hermes_cli/git_credentials.py b/hermes_cli/git_credentials.py index fbef4f0ea5..f968fb6371 100644 --- a/hermes_cli/git_credentials.py +++ b/hermes_cli/git_credentials.py @@ -17,7 +17,9 @@ Resolution order for an ``https://`` URL: missing one fails in ~100 ms instead of asking. The credential is attached only after an anonymous attempt is refused -(:func:`run_git_with_credential_fallback`). Most catalog repos are public, and a stale or +(:func:`run_git_with_credential_fallback`), and a refused credential yields to the next one in +the list rather than ending the run — a stale ``GITHUB_TOKEN`` in ``.env`` otherwise shadows a +``gh auth login`` that still works. Most catalog repos are public, and a stale or revoked stored token sent pre-emptively turns a clone that works anonymously into a 401 that git can only answer with the prompt this module disables — "could not read Username for 'https://github.com': terminal prompts disabled". @@ -32,7 +34,7 @@ import re import shutil import subprocess import urllib.parse -from typing import Mapping, Optional +from typing import Iterator, Mapping, Optional from hermes_cli._subprocess_compat import noninteractive_git_env, windows_hide_flags @@ -51,18 +53,23 @@ def _https_origin(url: str) -> Optional[str]: return f"https://{host}" -def _github_token() -> Optional[str]: +def _env_github_token() -> Optional[str]: from agent.secret_scope import get_secret - token = get_secret("GITHUB_TOKEN") or get_secret("GH_TOKEN") - if token: - return token + return get_secret("GITHUB_TOKEN") or get_secret("GH_TOKEN") or None + + +def _gh_cli_token() -> Optional[str]: gh = shutil.which("gh") if not gh: return None try: env = noninteractive_git_env() env["GH_PROMPT_DISABLED"] = "1" + # gh echoes an exported GH_TOKEN/GITHUB_TOKEN back instead of its keyring login; that + # token is already candidate 1, and a dead one would hide the login that still works. + for var in ("GH_TOKEN", "GITHUB_TOKEN"): + env.pop(var, None) result = subprocess.run( [gh, "auth", "token"], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=10, stdin=subprocess.DEVNULL, env=env, creationflags=windows_hide_flags()) @@ -74,6 +81,10 @@ def _github_token() -> Optional[str]: return result.stdout.strip() or None +def _github_token() -> Optional[str]: + return _env_github_token() or _gh_cli_token() + + def _credential_fill(origin: str) -> Optional[tuple[str, str]]: """``(username, password)`` from the user's own git credential helpers, never prompting.""" git = shutil.which("git") @@ -103,27 +114,47 @@ def _credential_fill(origin: str) -> Optional[tuple[str, str]]: return None -def resolve_git_basic_auth(url: str) -> Optional[tuple[str, str]]: - """``(username, password)`` for *url*, or None for non-HTTPS URLs / no stored credential.""" +def iter_git_basic_auth(url: str) -> Iterator[tuple[str, tuple[str, str]]]: + """``(source, (username, password))`` for every credential the user owns for *url*'s host, + in precedence order and without duplicates. The remote decides which one works: an expired + ``GITHUB_TOKEN`` left in ``.env`` must not shadow a live ``gh auth login`` (#115257).""" origin = _https_origin(url) if origin is None: - return None + return + seen: set[tuple[str, str]] = set() + sources = [] if urllib.parse.urlsplit(origin).hostname in _GITHUB_HOSTS: - token = _github_token() - if token: - return "x-access-token", token - return _credential_fill(origin) + sources += [("GITHUB_TOKEN/GH_TOKEN", lambda: _token_pair(_env_github_token())), + ("gh auth token", lambda: _token_pair(_gh_cli_token()))] + sources.append(("git credential helper", lambda: _credential_fill(origin))) + for source, lookup in sources: + auth = lookup() + if auth is not None and auth not in seen: + seen.add(auth) + yield source, auth -def with_git_auth(env: Mapping[str, str], url: str) -> dict[str, str]: +def _token_pair(token: Optional[str]) -> Optional[tuple[str, str]]: + return ("x-access-token", token) if token else None + + +def resolve_git_basic_auth(url: str) -> Optional[tuple[str, str]]: + """``(username, password)`` for *url*, or None for non-HTTPS URLs / no stored credential.""" + return next((auth for _source, auth in iter_git_basic_auth(url)), None) + + +def with_git_auth(env: Mapping[str, str], url: str, + auth: Optional[tuple[str, str]] = None) -> dict[str, str]: """Copy of *env* (a :func:`noninteractive_git_env` result) that authenticates HTTPS requests to *url*'s origin via a ``GIT_CONFIG_*`` ``http./.extraheader`` entry when a credential is - available; unchanged otherwise. The header lives only in this process environment.""" + available (*auth*, else the first stored one); unchanged otherwise. The header lives only in + this process environment.""" env = dict(env) origin = _https_origin(url) if origin is None: return env - auth = resolve_git_basic_auth(url) + if auth is None: + auth = resolve_git_basic_auth(url) if auth is None: return env encoded = base64.b64encode(f"{auth[0]}:{auth[1]}".encode()).decode() @@ -134,6 +165,10 @@ def with_git_auth(env: Mapping[str, str], url: str) -> dict[str, str]: return env +def _auth_headers(env: Mapping[str, str]) -> set[str]: + return {v for k, v in env.items() if k.startswith("GIT_CONFIG_VALUE_") and v.startswith("Authorization:")} + + # What git prints when the remote wants a credential it could not obtain: the prompt that # GIT_TERMINAL_PROMPT=0 refused, a rejected credential, or a bare 401/403 from the server. _CREDENTIAL_REQUIRED_RE = re.compile( @@ -158,9 +193,10 @@ def run_git_with_credential_fallback( argv: list[str], url: str, *, env: Mapping[str, str], **run_kwargs, ) -> subprocess.CompletedProcess: """Run the git network verb *argv* against *url* anonymously; when the remote refuses with - the credential-required class and the user owns a credential for that host, rerun once with - it attached. *env* is a :func:`noninteractive_git_env`; *run_kwargs* must capture output so - the refusal can be classified. Empty *url* means no fallback (local verbs).""" + the credential-required class, rerun with each credential the user owns for that host, in + precedence order, until one is accepted or the failure stops being about credentials. *env* + is a :func:`noninteractive_git_env`; *run_kwargs* must capture output so the refusal can be + classified. Empty *url* means no fallback (local verbs).""" run_kwargs.setdefault("stdin", subprocess.DEVNULL) env = dict(env) # An inherited askpass (VS Code terminal, ksshaskpass) would swallow the remote's 401 into a @@ -174,4 +210,26 @@ def run_git_with_credential_fallback( auth_env = with_git_auth(env, url) if auth_env == env: return result - return subprocess.run(argv, env=auth_env, **run_kwargs) + result = subprocess.run(argv, env=auth_env, **run_kwargs) + sent = _auth_headers(auth_env) + rejected: list[str] = [] + # The first credential (typically GITHUB_TOKEN from .env) was refused too: it is stale or + # revoked, not missing. Try the remaining ones the user owns instead of failing on it. + for source, auth in iter_git_basic_auth(url): + if result.returncode == 0 or not is_credential_required_error(result): + break + candidate_env = with_git_auth(env, url, auth) + headers = _auth_headers(candidate_env) + if headers <= sent: + rejected.append(source) + continue + sent |= headers + logger.warning("%s rejected the credential from %s; retrying with %s", + _https_origin(url), ", ".join(rejected) or "the stored credential", source) + result = subprocess.run(argv, env=candidate_env, **run_kwargs) + if result.returncode != 0 and is_credential_required_error(result): + rejected.append(source) + if result.returncode != 0 and "GITHUB_TOKEN/GH_TOKEN" in rejected and isinstance(result.stderr, str): + result.stderr += ("\nhint: the GITHUB_TOKEN/GH_TOKEN in your .env was rejected by GitHub;" + " replace it or remove it (gh auth login is used when it is absent).") + return result diff --git a/tests/hermes_cli/test_plugin_private_repo_auth.py b/tests/hermes_cli/test_plugin_private_repo_auth.py index da497a6bce..7f4179fb91 100644 --- a/tests/hermes_cli/test_plugin_private_repo_auth.py +++ b/tests/hermes_cli/test_plugin_private_repo_auth.py @@ -34,7 +34,7 @@ def _seed_bare_upstream(tmp_path) -> None: subprocess.run(["git", "init", "-q", "--bare", str(upstream)], check=True) work = tmp_path / "work" subprocess.run(["git", "clone", "-q", str(upstream), str(work)], check=True) - (work / "plugin.yaml").write_text("name: probe\ndescription: d\nversion: '1'\n") + (work / "plugin.yaml").write_text("name: probe\ndescription: d\nversion: '1'\n", encoding="utf-8") subprocess.run(["git", "-C", str(work), "-c", "user.name=t", "-c", "user.email=t@t", "add", "."], check=True) subprocess.run(["git", "-C", str(work), "-c", "user.name=t", "-c", "user.email=t@t", "commit", "-qm", "i"], check=True) subprocess.run(["git", "-C", str(work), "push", "-q", "origin", "HEAD"], check=True) @@ -76,8 +76,8 @@ def test_private_clone_falls_back_to_auth_after_credential_required_error(tmp_pa assert len(clone_calls) == 2, f"expected anonymous + auth fallback, got {len(clone_calls)} clone attempts" assert clone_calls[0] == [], "first (anonymous) clone attempt must not carry an Authorization header" assert clone_calls[1] == [f"Authorization: basic {expected}"], "fallback clone must inject the stored credential" - assert "s3cret" not in (dest / ".git" / "config").read_text() - assert expected not in (dest / ".git" / "config").read_text() + assert "s3cret" not in (dest / ".git" / "config").read_text(encoding="utf-8") + assert expected not in (dest / ".git" / "config").read_text(encoding="utf-8") # Non-HTTPS URLs get no header; the hardened base env is otherwise untouched. base = noninteractive_git_env() assert git_credentials.with_git_auth(base, "git@github.com:acme/probe.git") == dict(base) @@ -223,13 +223,75 @@ def test_anonymous_attempt_fails_fast_under_inherited_askpass(tmp_path, monkeypa @pytest.mark.skipif(sys.platform == "win32", reason="POSIX shell stub credential helper") def test_credential_fill_uses_stored_helper_and_never_prompts(tmp_path, monkeypatch): helper = tmp_path / "helper.sh" - helper.write_text("#!/bin/sh\n[ \"$1\" = get ] && printf 'username=bob\\npassword=pw-from-helper\\n'\n") + helper.write_text("#!/bin/sh\n[ \"$1\" = get ] && printf 'username=bob\\npassword=pw-from-helper\\n'\n", encoding="utf-8") helper.chmod(0o755) gitconfig = tmp_path / "gitconfig" - gitconfig.write_text(f'[credential "https://git.example.test"]\n\thelper = !{helper}\n') + gitconfig.write_text(f'[credential "https://git.example.test"]\n\thelper = !{helper}\n', encoding="utf-8") monkeypatch.setenv("GIT_CONFIG_GLOBAL", str(gitconfig)) monkeypatch.setenv("GIT_ASKPASS", "/nonexistent/askpass-must-not-run") assert git_credentials.resolve_git_basic_auth("https://git.example.test/acme/x.git") == ("bob", "pw-from-helper") # Unknown host: no helper answers → None quickly, no prompt attempt escaped. assert git_credentials.resolve_git_basic_auth("https://nothing.example.test/x.git") is None + + +def _refused(argv): + return subprocess.CompletedProcess( + argv, 128, stdout="", + stderr="fatal: could not read Username for 'https://github.com': terminal prompts disabled\n") + + +def test_rejected_env_token_falls_back_to_the_next_owned_credential(monkeypatch): + """An expired GITHUB_TOKEN in .env must not shadow a live ``gh auth login`` (#115257): after + the remote refuses the first credential, the run retries with the next one and stops there.""" + monkeypatch.setattr(git_credentials, "iter_git_basic_auth", lambda url: iter([ + ("GITHUB_TOKEN/GH_TOKEN", ("x-access-token", "ghp_dead")), + ("gh auth token", ("x-access-token", "gho_live")), + ("git credential helper", ("bob", "never-needed")), + ])) + attempts: list[list[str]] = [] + + def fake_run(argv, **kw): + headers = _auth_headers_for(kw["env"], "https://github.com") + attempts.append(headers) + live = base64.b64encode(b"x-access-token:gho_live").decode() + if headers == [f"Authorization: basic {live}"]: + return subprocess.CompletedProcess(argv, 0, stdout="", stderr="") + return _refused(argv) + + monkeypatch.setattr(git_credentials.subprocess, "run", fake_run) + url = "https://github.com/acme/private.git" + result = git_credentials.run_git_with_credential_fallback( + ["git", "clone", url, "dest"], url, env=noninteractive_git_env(), capture_output=True, text=True) + + dead = base64.b64encode(b"x-access-token:ghp_dead").decode() + live = base64.b64encode(b"x-access-token:gho_live").decode() + assert result.returncode == 0 + assert attempts == [[], [f"Authorization: basic {dead}"], [f"Authorization: basic {live}"]] + + +def test_every_credential_rejected_names_the_dead_env_token(monkeypatch): + """When GitHub refuses every owned credential the git error gains a hint naming the .env token, + and a failure that stops being about credentials ends the retries without one.""" + monkeypatch.setattr(git_credentials, "iter_git_basic_auth", lambda url: iter([ + ("GITHUB_TOKEN/GH_TOKEN", ("x-access-token", "ghp_dead")), + ("gh auth token", ("x-access-token", "gho_dead_too")), + ])) + monkeypatch.setattr(git_credentials.subprocess, "run", lambda argv, **kw: _refused(argv)) + url = "https://github.com/acme/private.git" + result = git_credentials.run_git_with_credential_fallback( + ["git", "clone", url, "dest"], url, env=noninteractive_git_env(), capture_output=True, text=True) + assert result.returncode != 0 and "GITHUB_TOKEN/GH_TOKEN in your .env was rejected" in result.stderr + + calls = [] + + def not_found_once_authed(argv, **kw): + calls.append(kw["env"]) + if _auth_headers_for(kw["env"], "https://github.com"): + return subprocess.CompletedProcess(argv, 128, stdout="", stderr=f"fatal: repository '{url}/' not found\n") + return _refused(argv) + + monkeypatch.setattr(git_credentials.subprocess, "run", not_found_once_authed) + result = git_credentials.run_git_with_credential_fallback( + ["git", "clone", url, "dest"], url, env=noninteractive_git_env(), capture_output=True, text=True) + assert len(calls) == 2 and "not found" in result.stderr and "hint:" not in result.stderr