From b7a22e620c6f5c5fd74522ffa2f51931d62f52f2 Mon Sep 17 00:00:00 2001 From: Lukas Geiger Date: Thu, 29 Oct 2026 18:12:57 +0100 Subject: [PATCH 1/2] Add account-bound bot Git transport --- README.md | 2 +- README_de.md | 2 +- src/githubbot_core/transport.py | 42 +++- src/githubbot_core/typed_actions.py | 265 ++++++++++++++++++++-- tests/test_typed_actions.py | 336 ++++++++++++++++++++++++++-- 5 files changed, 601 insertions(+), 46 deletions(-) diff --git a/README.md b/README.md index 9a72683..c0a45bf 100644 --- a/README.md +++ b/README.md @@ -117,7 +117,7 @@ not a silent override. `permissions.enabled` must stay inside the ceiling. | `actions.dispatch` | disabled | `POST …/actions/workflows/{workflow}/dispatches` for an active workflow | | `actions.rerun` | disabled | `POST …/actions/runs/{id}/rerun` for a completed, non-fork run of the exact `sha` | | `pr.create` | disabled | `POST …/pulls` from a bot-prefixed branch onto the default branch | -| `git.push_branch` | disabled | `git push origin :refs/heads/` without force, bot prefix only (`typed_actions.push_branch_prefix`, default `bot/`), never the default branch, verified with `ls-remote` | +| `git.push_branch` | disabled | `git push origin :refs/heads/` without force, bot prefix only (`typed_actions.push_branch_prefix`, default `bot/`), never the default branch. Requires `multi_bot.merge_account` and the existing `GH_TOKEN` credential reader; the same token is checked against that account before Git runs. Only one exact HTTPS `github.com//` fetch and push URL is accepted. Per-call AskPass disables credential-helper fallback; readback must match the exact SHA and ref. | The eight typed actions sit in the default ceiling but are never enabled by default. Each one also needs its `multi_bot` switch (`merge_enabled`, `policy.draft_*`, `policy.trigger_missing_ci`, diff --git a/README_de.md b/README_de.md index 21d3a05..940a4b9 100644 --- a/README_de.md +++ b/README_de.md @@ -119,7 +119,7 @@ stille Überschreibung. `permissions.enabled` muss innerhalb der Obergrenze blei | `actions.dispatch` | aus | `POST …/actions/workflows/{workflow}/dispatches` für einen aktiven Workflow | | `actions.rerun` | aus | `POST …/actions/runs/{id}/rerun` für einen abgeschlossenen Lauf des exakten `sha`, nie aus Forks | | `pr.create` | aus | `POST …/pulls` von einem Branch mit Bot-Präfix auf den Default-Branch | -| `git.push_branch` | aus | `git push origin :refs/heads/` ohne Force, nur Bot-Präfix (`typed_actions.push_branch_prefix`, Standard `bot/`), nie der Default-Branch, Prüfung per `ls-remote` | +| `git.push_branch` | aus | `git push origin :refs/heads/` ohne Force, nur Bot-Präfix (`typed_actions.push_branch_prefix`, Standard `bot/`), nie der Default-Branch. Benötigt `multi_bot.merge_account` und den vorhandenen `GH_TOKEN`-Credential-Reader; derselbe Token wird vor Git gegen dieses Konto geprüft. Zulässig ist genau eine passende HTTPS-Fetch- und Push-URL auf `github.com//`. AskPass gilt nur für den Aufruf und schaltet Credential-Helper-Fallbacks ab; das Rücklesen muss exakt SHA und Ref bestätigen. | Die acht getypten Aktionen stehen in der Standard-Obergrenze, sind aber nie standardmäßig aktiv. Jede braucht zusätzlich ihren `multi_bot`-Schalter (`merge_enabled`, `policy.draft_*`, diff --git a/src/githubbot_core/transport.py b/src/githubbot_core/transport.py index 937d319..ec89108 100644 --- a/src/githubbot_core/transport.py +++ b/src/githubbot_core/transport.py @@ -13,6 +13,7 @@ import json import logging +import os import subprocess from typing import Any, Callable, Optional, Protocol @@ -39,7 +40,14 @@ def __init__(self, gh: str = "gh", timeout: int = 60, runner: Optional[Callable] self.timeout = timeout self._runner = runner - def _invoke(self, args: list[str], stdin: Optional[str] = None) -> tuple[str, Optional[str]]: + def _invoke( + self, + args: list[str], + stdin: Optional[str] = None, + *, + env: Optional[dict[str, str]] = None, + generic_errors: bool = False, + ) -> tuple[str, Optional[str]]: cmd = [self.gh, "api", "-H", f"X-GitHub-Api-Version: {API_VERSION}", *args] run = self._runner or subprocess.run try: @@ -51,16 +59,44 @@ def _invoke(self, args: list[str], stdin: Optional[str] = None) -> tuple[str, Op encoding="utf-8", errors="replace", timeout=self.timeout, + **({"env": env} if env is not None else {}), ) except subprocess.TimeoutExpired: - return "", "Timeout" + return "", "GitHub request failed" if generic_errors else "Timeout" except Exception as exc: # the transport reports, it never raises (callers collect partial errors) - return "", f"{type(exc).__name__}: {exc}" + return "", "GitHub request failed" if generic_errors else f"{type(exc).__name__}: {exc}" if result.returncode != 0: + if generic_errors: + return "", "GitHub request failed" detail = " ".join((result.stderr or result.stdout or "").split()) return "", detail[:500] or f"gh api exit {result.returncode}" return result.stdout or "", None + def get_for_token(self, token: str, endpoint: str) -> tuple[Any, Optional[str]]: + """Read GitHub.com with exactly this token; never use ambient auth fallback.""" + if not isinstance(token, str) or not token: + return None, "authorized credential unavailable" + env = os.environ.copy() + for name in ("GITHUB_TOKEN", "GH_ENTERPRISE_TOKEN", "GITHUB_ENTERPRISE_TOKEN"): + env.pop(name, None) + env["GH_TOKEN"] = token + env["GH_HOST"] = "github.com" + try: + text, error = self._invoke( + ["--hostname", "github.com", "-H", "Accept: application/vnd.github+json", endpoint], + env=env, + generic_errors=True, + ) + finally: + env.pop("GH_TOKEN", None) + env.pop("GH_HOST", None) + if error: + return None, error + try: + return json.loads(text), None + except (TypeError, json.JSONDecodeError): + return None, "GitHub response unreadable" + @staticmethod def _json(text: str) -> tuple[Any, Optional[str]]: if not text.strip(): diff --git a/src/githubbot_core/typed_actions.py b/src/githubbot_core/typed_actions.py index 7e16a40..f1763f7 100644 --- a/src/githubbot_core/typed_actions.py +++ b/src/githubbot_core/typed_actions.py @@ -11,9 +11,16 @@ from __future__ import annotations import re +import os import subprocess +import sys +import tempfile +from contextlib import contextmanager +from pathlib import Path from typing import Any, Optional -from urllib.parse import quote +from urllib.parse import quote, urlsplit + +from .credentials import CredentialUnavailable, EnvTokenProvider, resolve_token SHA = re.compile(r"[0-9a-f]{40}") REPO = re.compile(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+") @@ -113,6 +120,10 @@ def param_denial(kind: str, params: dict, config: dict) -> Optional[str]: return "body must be nonempty review text of at most 65536 characters" if not str((config.get("multi_bot") or {}).get("merge_account") or "").strip(): return "pr.review requires the configured multi_bot.merge_account" + if kind == "git.push_branch": + account = (config.get("multi_bot") or {}).get("merge_account") + if not isinstance(account, str) or not account.strip(): + return "git.push_branch requires the configured multi_bot.merge_account" if kind == "pr.merge" and params.get("method", "merge") not in MERGE_METHODS: return f"method must be one of {list(MERGE_METHODS)}" if kind == "actions.dispatch": @@ -458,38 +469,254 @@ def exec_create_pr(transport, params: dict, config: dict): return {"readback": {"number": data["number"], "url": data.get("html_url")}}, None -def _run_git(path: str, *args: str) -> tuple[str, Optional[str]]: - result = _git(["git", "-C", path, *args], capture_output=True, text=True, encoding="utf-8", errors="replace") +def _git_auth_config(repo: str) -> tuple[str, ...]: + """Clear every Git credential/header source that could bypass the askpass token.""" + owner = repo.split("/", 1)[0] + scopes = ( + "https://github.com", + "https://github.com/", + f"https://github.com/{owner}", + f"https://github.com/{owner}/", + f"https://github.com/{repo}", + f"https://github.com/{repo}/", + f"https://github.com/{repo}.git", + ) + args = [ + "-c", + "credential.helper=", + "-c", + "http.extraHeader=", + "-c", + "http.sslVerify=true", + "-c", + "http.followRedirects=false", + "-c", + "http.sslCert=", + "-c", + "http.sslKey=", + "-c", + "http.cookieFile=", + ] + for scope in scopes: + for setting in ( + f"credential.{scope}.helper=", + f"http.{scope}.extraheader=", + f"http.{scope}.sslverify=true", + f"http.{scope}.followredirects=false", + f"http.{scope}.sslcert=", + f"http.{scope}.sslkey=", + f"http.{scope}.cookiefile=", + ): + args.extend(("-c", setting)) + return tuple(args) + + +def _git_base_environment() -> dict[str, str]: + """Isolate Git path/auth overrides while preserving one config view for all checks.""" + env = os.environ.copy() + for name in tuple(env): + if ( + name + in ( + "GIT_DIR", + "GIT_WORK_TREE", + "GIT_COMMON_DIR", + "GIT_INDEX_FILE", + "GIT_PREFIX", + "GIT_ASKPASS", + "SSH_ASKPASS", + "GH_TOKEN", + "GITHUB_TOKEN", + "GH_ENTERPRISE_TOKEN", + "GITHUB_ENTERPRISE_TOKEN", + "GIT_CURL_VERBOSE", + "GCM_INTERACTIVE", + "GIT_SSL_NO_VERIFY", + "GIT_SSL_CAINFO", + "GIT_SSL_CAPATH", + "GIT_SSL_CERT", + "GIT_SSL_KEY", + ) + or name.startswith("GIT_CONFIG") + or name.startswith("GIT_TRACE") + ): + env.pop(name, None) + env["GIT_CONFIG_NOSYSTEM"] = "1" + env["GIT_CONFIG_GLOBAL"] = os.devnull + return env + + +@contextmanager +def _git_auth_environment(token: str, account: str): + """Prepare a short-lived, secret-free AskPass helper and a process-only env.""" + with tempfile.TemporaryDirectory(prefix="githubbot-git-askpass-") as temp_dir: + root = Path(temp_dir) + helper = root / ("askpass.cmd" if os.name == "nt" else "askpass") + script = root / "askpass.py" + script.write_text( + "import os, sys\n" + "prompt = ' '.join(sys.argv[1:]).casefold()\n" + "name = 'GITHUBBOT_GIT_USERNAME' if 'username' in prompt else 'GITHUBBOT_GIT_TOKEN'\n" + "value = os.environ.get(name, '')\n" + "if not value:\n" + " raise SystemExit(1)\n" + "sys.stdout.write(value)\n", + encoding="utf-8", + newline="\n", + ) + if os.name == "nt": + helper.write_text( + '@echo off\r\n"%GITHUBBOT_ASKPASS_PYTHON%" "%~dp0askpass.py" %*\r\nexit /b %ERRORLEVEL%\r\n', + encoding="utf-8", + newline="", + ) + else: + helper.write_text( + '#!/bin/sh\nexec "$GITHUBBOT_ASKPASS_PYTHON" "$(dirname "$0")/askpass.py" "$@"\n', + encoding="utf-8", + newline="\n", + ) + helper.chmod(0o700) + + env = _git_base_environment() + env.update( + { + "GIT_ASKPASS": str(helper), + "GIT_TERMINAL_PROMPT": "0", + "GIT_CONFIG_NOSYSTEM": "1", + "GIT_CONFIG_GLOBAL": os.devnull, + "GITHUBBOT_ASKPASS_PYTHON": sys.executable, + "GITHUBBOT_GIT_USERNAME": account, + "GITHUBBOT_GIT_TOKEN": token, + } + ) + try: + yield env + finally: + for name in ("GITHUBBOT_GIT_TOKEN", "GITHUBBOT_GIT_USERNAME"): + env.pop(name, None) + + +def _run_git( + path: str, + *args: str, + env: Optional[dict[str, str]] = None, + repo: Optional[str] = None, +) -> tuple[str, Optional[str]]: + command = ["git", "-C", path] + if env is not None: + command.extend(_git_auth_config(repo or "")) + command.extend(args) + run_env = env.copy() if env is not None else _git_base_environment() + try: + result = _git( + command, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=120, + env=run_env, + ) + except (OSError, subprocess.TimeoutExpired): + return "", "git command failed" + except Exception: + return "", "git command failed" if result.returncode: - return "", " ".join((result.stderr or result.stdout or "").split())[:500] or f"git exit {result.returncode}" - return result.stdout.strip(), None + return "", f"git command failed (exit {result.returncode})" + return (result.stdout or "").strip(), None def _same_repo(url: str, repo: str) -> bool: - url = url.strip().removesuffix(".git").removesuffix("/").casefold() - return url.endswith("github.com/" + repo.casefold()) or url.endswith("github.com:" + repo.casefold()) + if not REPO.fullmatch(repo): + return False + try: + parsed = urlsplit(url.strip()) + if ( + parsed.scheme.casefold() != "https" + or (parsed.hostname or "").casefold() != "github.com" + or parsed.netloc.casefold() != "github.com" + or parsed.port is not None + or parsed.username is not None + or parsed.password is not None + or parsed.query + or parsed.fragment + or "?" in url + or "#" in url + or "%" in parsed.path + ): + return False + except ValueError: + return False + path = parsed.path.removesuffix(".git") + return path == f"/{repo}" + + +def _exact_remote_ref(output: str, sha: str, ref: str) -> bool: + lines = [line for line in output.splitlines() if line] + if len(lines) != 1: + return False + fields = lines[0].split("\t") + return len(fields) == 2 and fields == [sha, ref] def exec_push_branch(transport, params: dict, config: dict): repo, branch, sha, path = params["repo"], params["branch"], params["sha"], params["path"] - default, error = _default_branch(transport, repo) - if error: - return None, error + account = str((config.get("multi_bot") or {}).get("merge_account") or "").strip() + if not account: + return None, "git.push_branch requires the configured multi_bot.merge_account" + try: + token, _provider = resolve_token([EnvTokenProvider(names=("GH_TOKEN",))]) + except CredentialUnavailable: + return None, "authorized bot credential unavailable" + except Exception: + return None, "authorized bot credential unavailable" + get_for_token = getattr(transport, "get_for_token", None) + if not callable(get_for_token): + return None, "account-bound GitHub transport unavailable" + try: + user, error = get_for_token(token, "user") + except Exception: + return None, "configured bot identity unreadable" + if error or not isinstance(user, dict) or user.get("login") != account: + return None, "credential identity does not match the configured bot account" + try: + repository, error = get_for_token(token, f"repos/{repo}") + except Exception: + return None, "target repository unreadable for the configured bot account" + if error or not isinstance(repository, dict) or not isinstance(repository.get("default_branch"), str): + return None, "target repository unreadable for the configured bot account" + default = repository["default_branch"] if branch == default: return None, "never pushes the default branch" - url, error = _run_git(path, "remote", "get-url", "origin") - if error or not _same_repo(url, repo): + fetch_urls, error = _run_git(path, "remote", "get-url", "--all", "origin") + if error: + return None, "origin of the checkout is not the target repository" + push_urls, error = _run_git(path, "remote", "get-url", "--push", "--all", "origin") + if error: + return None, "origin of the checkout is not the target repository" + if len(fetch_urls.splitlines()) != 1 or len(push_urls.splitlines()) != 1: + return None, "origin of the checkout is not the target repository" + if not _same_repo(fetch_urls, repo) or not _same_repo(push_urls, repo): return None, "origin of the checkout is not the target repository" resolved, error = _run_git(path, "rev-parse", "--verify", f"{sha}^{{commit}}") if error or resolved != sha: return None, "commit not present in the checkout" - # No '+' refspec and no --force: a non-fast-forward is rejected by git itself. - _, error = _run_git(path, "push", "origin", f"{sha}:refs/heads/{branch}") - if error: - return None, error - remote, error = _run_git(path, "ls-remote", "origin", f"refs/heads/{branch}") - if error or not remote.startswith(sha): - return None, "push not confirmed by ls-remote" + checked_branch, error = _run_git(path, "check-ref-format", "--branch", branch) + if error or checked_branch != branch: + return None, "branch ref is invalid" + ref = f"refs/heads/{branch}" + try: + with _git_auth_environment(token, account) as env: + # No '+' refspec and no --force: a non-fast-forward is rejected by git itself. + _, error = _run_git(path, "push", "origin", f"{sha}:{ref}", env=env, repo=repo) + if error: + return None, "git push failed or its outcome is unknown; read the target before retrying" + remote, error = _run_git(path, "ls-remote", "--exit-code", "origin", ref, env=env, repo=repo) + if error or not _exact_remote_ref(remote, sha, ref): + return None, "push readback is unconfirmed; read the target before retrying" + except Exception: + return None, "account-bound Git transport failed; read the target before retrying" return {"readback": {"branch": branch, "sha": sha}}, None diff --git a/tests/test_typed_actions.py b/tests/test_typed_actions.py index f6df93d..75bd077 100644 --- a/tests/test_typed_actions.py +++ b/tests/test_typed_actions.py @@ -3,6 +3,9 @@ from __future__ import annotations import json +import os +import subprocess +from pathlib import Path from types import SimpleNamespace import pytest @@ -10,7 +13,9 @@ from githubbot_core import actions, cli, typed_actions from githubbot_core.actions import Action from githubbot_core.config import default_config, merge_layers +from githubbot_core.credentials import EnvTokenProvider, resolve_token from githubbot_core.multibot import permission_contract +from githubbot_core.transport import GhCliTransport REPO = "example-org/example-repo" SHA = "a" * 40 @@ -60,6 +65,19 @@ def request(self, method, endpoint, body=None): return self.write_result, None +class TokenTransport(Transport): + def __init__(self, gets, *, login="example-bot"): + super().__init__(gets) + self.login = login + self.token_reads = [] + + def get_for_token(self, token, endpoint): + self.token_reads.append((token, endpoint)) + if endpoint == "user": + return {"login": self.login}, None + return self.get(endpoint) + + def base_gets(**pull_overrides): return { f"repos/{REPO}": {"default_branch": "main"}, @@ -97,6 +115,11 @@ def test_static_parameters_are_checked_in_preview(): push = {"repo": REPO, "branch": "feature/x", "sha": SHA, "path": "."} assert "prefix" in actions.denial_reason(Action("git.push_branch", push), cfg) assert "invalid" in actions.denial_reason(Action("git.push_branch", {**push, "branch": "bot/../main"}), cfg) + missing_account = config() + missing_account["multi_bot"]["merge_account"] = " " + assert "merge_account" in actions.denial_reason( + Action("git.push_branch", {**push, "branch": "bot/fix"}), missing_account + ) def test_merge_is_sha_bound_and_read_back(): @@ -250,44 +273,313 @@ def test_create_pr_targets_the_default_branch_from_a_bot_branch(): assert transport.writes[0][2]["base"] == "main" and transport.writes[0][2]["head"] == "bot/fix" -def test_push_branch_never_forces_and_checks_origin(monkeypatch): +def _mock_push_token(monkeypatch, token="synthetic-bot-token"): + seen = [] + + def reader(providers): + providers = list(providers) + seen.append(providers) + assert len(providers) == 1 + assert isinstance(providers[0], EnvTokenProvider) + assert providers[0].names == ("GH_TOKEN",) + return token, providers[0].name + + monkeypatch.setattr(typed_actions, "resolve_token", reader) + return token, seen + + +def _fake_git_output(command, *, origin=None, pushurl=None, readback=None, fail_push=False): + op_index = next( + i for i, part in enumerate(command) if part in {"remote", "rev-parse", "check-ref-format", "push", "ls-remote"} + ) + op = command[op_index] + if op == "remote": + if "--push" in command: + output = pushurl if pushurl is not None else origin + else: + output = origin + elif op == "rev-parse": + output = SHA + elif op == "check-ref-format": + output = "bot/fix" + elif op == "push": + if fail_push: + return SimpleNamespace( + returncode=128, stdout="remote rejected synthetic-bot-token", stderr="auth synthetic-bot-token" + ) + output = "" + else: + output = readback if readback is not None else f"{SHA}\trefs/heads/bot/fix\n" + return SimpleNamespace(returncode=0, stdout=f"{output}\n" if output else "", stderr="") + + +def _push_action(branch="bot/fix"): + return Action("git.push_branch", {"repo": REPO, "branch": branch, "sha": SHA, "path": "."}) + + +def _push_transport(login="example-bot"): + return TokenTransport({f"repos/{REPO}": {"default_branch": "main"}}, login=login) + + +def test_push_branch_uses_same_bot_token_exact_remote_and_exact_readback(monkeypatch): + token, readers = _mock_push_token(monkeypatch) calls = [] + temp_dirs = [] def git(cmd, **kwargs): - calls.append(cmd[3:]) - out = { - "remote": f"https://github.com/{REPO}.git\n", - "rev-parse": SHA + "\n", - "push": "", - "ls-remote": f"{SHA}\trefs/heads/bot/fix\n", - }[cmd[3]] - return SimpleNamespace(returncode=0, stdout=out, stderr="") + if "GITHUBBOT_GIT_TOKEN" in kwargs.get("env", {}): + env = kwargs["env"] + helper = env["GIT_ASKPASS"] + temp_dirs.append(os.path.dirname(helper)) + assert "GH_TOKEN" not in env and "GITHUB_TOKEN" not in env + assert env["GITHUBBOT_GIT_TOKEN"] == token + assert env["GIT_TERMINAL_PROMPT"] == "0" + assert all(token not in arg for arg in cmd) + assert token not in Path(helper).read_text(encoding="utf-8") + assert "credential.helper=" in cmd + assert "http.extraHeader=" in cmd + calls.append((cmd, kwargs)) + return _fake_git_output( + cmd, + origin=f"https://github.com/{REPO}.git", + pushurl=f"https://github.com/{REPO}.git", + ) monkeypatch.setattr(typed_actions, "_git", git) - action = Action("git.push_branch", {"repo": REPO, "branch": "bot/fix", "sha": SHA, "path": "."}) - transport = Transport({f"repos/{REPO}": {"default_branch": "main"}}) - receipt = actions.apply([action], transport, config())["receipts"][0] + transport = _push_transport() + receipt = actions.apply([_push_action()], transport, config())["receipts"][0] assert receipt["status"] == "applied" - push = next(c for c in calls if c[0] == "push") - assert push == ["push", "origin", f"{SHA}:refs/heads/bot/fix"] - assert not any(arg.startswith("+") or "force" in arg for c in calls for arg in c) + # Both API preflights use the token resolved by the one GH_TOKEN-only provider. + assert transport.token_reads == [(token, "user"), (token, f"repos/{REPO}")] + assert len(readers[0]) == 1 + push = next(cmd for cmd, _ in calls if "push" in cmd) + assert push[push.index("push") + 1 :] == ["origin", f"{SHA}:refs/heads/bot/fix"] + assert not any(arg.startswith("+") or "force" in arg for cmd, _ in calls for arg in cmd) + assert temp_dirs and all(not os.path.exists(directory) for directory in temp_dirs) + + +def test_push_branch_fails_closed_for_wrong_identity_or_missing_credential(monkeypatch): + _mock_push_token(monkeypatch) + calls = [] + monkeypatch.setattr(typed_actions, "_git", lambda cmd, **kwargs: calls.append(cmd)) + wrong_identity = actions.apply([_push_action()], _push_transport("lukisch"), config())["receipts"][0] + assert wrong_identity["status"] == "failed" + assert "identity" in wrong_identity["detail"] + assert calls == [] + + def missing_reader(providers): + providers = list(providers) + assert len(providers) == 1 and providers[0].names == ("GH_TOKEN",) + return resolve_token([EnvTokenProvider(names=("GH_TOKEN",), environ={})]) - calls.clear() + monkeypatch.setattr(typed_actions, "resolve_token", missing_reader) + missing = actions.apply([_push_action()], _push_transport(), config())["receipts"][0] + assert missing["status"] == "failed" + assert "credential unavailable" in missing["detail"] + assert calls == [] + + +def test_push_branch_rejects_bad_prefix_before_git_or_credential_lookup(monkeypatch): + calls = [] monkeypatch.setattr( - typed_actions, - "_git", - lambda cmd, **k: SimpleNamespace(returncode=0, stdout="https://github.com/other/repo\n", stderr=""), + typed_actions, "resolve_token", lambda _: (_ for _ in ()).throw(AssertionError("reader called")) ) - assert actions.apply([action], transport, config())["receipts"][0]["status"] == "failed" + monkeypatch.setattr(typed_actions, "_git", lambda cmd, **kwargs: calls.append(cmd)) + action = _push_action(branch="feature/x") + receipt = actions.apply([action], _push_transport(), config())["receipts"][0] + assert receipt["status"] == "denied" + assert "prefix" in receipt["detail"] + assert calls == [] + + +@pytest.mark.parametrize( + "origin,pushurl", + [ + ("https://github.com.attacker.invalid/example-org/example-repo.git", None), + ("https://github.com/other/repo.git", None), + ("https://token@github.com/example-org/example-repo.git", None), + (f"https://github.com:443/{REPO}.git", None), + (f"https://github.com/{REPO}.git?", None), + (f"git@github.com:{REPO}.git", None), + (f"https://github.com/{REPO}.git", "https://github.com/other/repo.git"), + ], +) +def test_push_branch_rejects_unbound_origin_and_pushurl(monkeypatch, origin, pushurl): + _mock_push_token(monkeypatch) + calls = [] + + def git(cmd, **kwargs): + calls.append(cmd) + return _fake_git_output(cmd, origin=origin, pushurl=pushurl) + + monkeypatch.setattr(typed_actions, "_git", git) + receipt = actions.apply([_push_action()], _push_transport(), config())["receipts"][0] + assert receipt["status"] == "failed" + assert not any("push" in cmd for cmd in calls) + + +def test_push_branch_rejects_empty_port_separator_remote(monkeypatch): + _mock_push_token(monkeypatch) + calls = [] + + def git(cmd, **kwargs): + calls.append(cmd) + return _fake_git_output(cmd, origin=f"https://github.com:/{REPO}.git") + + monkeypatch.setattr(typed_actions, "_git", git) + receipt = actions.apply([_push_action()], _push_transport(), config())["receipts"][0] + assert receipt["status"] == "failed" + assert "origin of the checkout" in receipt["detail"] + assert not any("push" in cmd for cmd in calls) + + +@pytest.mark.parametrize( + "readback", + [ + f"{SHA}\trefs/heads/bot/other\n", + f"{OTHER}\trefs/heads/bot/fix\n", + f"{SHA}\trefs/heads/bot/fix\n{OTHER}\trefs/heads/bot/other\n", + ], +) +def test_push_branch_rejects_wrong_readback_oid_or_ref(monkeypatch, readback): + _mock_push_token(monkeypatch) + calls = [] + + def git(cmd, **kwargs): + calls.append(cmd) + return _fake_git_output( + cmd, + origin=f"https://github.com/{REPO}.git", + pushurl=f"https://github.com/{REPO}.git", + readback=readback, + ) + + monkeypatch.setattr(typed_actions, "_git", git) + receipt = actions.apply([_push_action()], _push_transport(), config())["receipts"][0] + assert receipt["status"] == "failed" + assert "read the target before retrying" in receipt["detail"] + assert any("push" in cmd for cmd in calls) + + +def test_push_branch_redacts_secret_like_git_errors(monkeypatch, caplog): + token, _ = _mock_push_token(monkeypatch) + temp_dirs = [] + + def git(cmd, **kwargs): + if "GITHUBBOT_GIT_TOKEN" in kwargs.get("env", {}): + temp_dirs.append(os.path.dirname(kwargs["env"]["GIT_ASKPASS"])) + return _fake_git_output( + cmd, + origin=f"https://github.com/{REPO}.git", + pushurl=f"https://github.com/{REPO}.git", + fail_push="push" in cmd, + ) + + monkeypatch.setattr(typed_actions, "_git", git) + receipt = actions.apply([_push_action()], _push_transport(), config())["receipts"][0] + assert receipt["status"] == "failed" + assert token not in receipt["detail"] and token not in caplog.text + assert "read the target before retrying" in receipt["detail"] + assert temp_dirs and all(not os.path.exists(directory) for directory in temp_dirs) + + +def test_git_base_environment_removes_ambient_auth_and_repo_overrides(monkeypatch): + monkeypatch.setenv("GH_TOKEN", "synthetic-ambient-token") + monkeypatch.setenv("GITHUB_TOKEN", "synthetic-ambient-token") + monkeypatch.setenv("GIT_DIR", "wrong-repository") + monkeypatch.setenv("GIT_WORK_TREE", "wrong-worktree") + monkeypatch.setenv("GIT_CONFIG_COUNT", "1") + monkeypatch.setenv("GIT_TRACE", "2") + env = typed_actions._git_base_environment() + assert "GH_TOKEN" not in env and "GITHUB_TOKEN" not in env + assert "GIT_DIR" not in env and "GIT_WORK_TREE" not in env + assert "GIT_CONFIG_COUNT" not in env and "GIT_TRACE" not in env + assert env["GIT_CONFIG_NOSYSTEM"] == "1" + assert env["GIT_CONFIG_GLOBAL"] == os.devnull + + +def test_git_askpass_helper_uses_child_env_and_cleans_tempfiles(): + token = "synthetic-askpass-token" + directory = None + with typed_actions._git_auth_environment(token, "example-bot") as env: + helper = Path(env["GIT_ASKPASS"]) + directory = helper.parent + script = helper.with_name("askpass.py") + source = script.read_text(encoding="utf-8") + assert token not in source + commands = [] + + def run_askpass(prompt): + if os.name == "nt": + args = f'call "{helper}" "{prompt}"' + command = args + shell = True + else: + args = [str(helper), prompt] + command = args + shell = False + commands.append(args) + return subprocess.run( + command, + capture_output=True, + text=True, + encoding="utf-8", + env=env, + shell=shell, + check=True, + ) + + username = run_askpass("Username for https://github.com:") + password = run_askpass("Password for https://github.com:") + assert username.stdout == "example-bot" + assert password.stdout == token + assert all(token not in str(command) for command in commands) + assert directory is not None and not directory.exists() + + +def test_transport_reads_identity_with_exact_token_without_argv_or_raw_error(): + token = "synthetic-api-token" + seen = [] + + def runner(command, **kwargs): + seen.append((command, kwargs, kwargs["env"].get("GH_TOKEN"), "GITHUB_TOKEN" in kwargs["env"])) + return SimpleNamespace(returncode=1, stdout=f"{token}", stderr=f"failed {token}") + + transport = GhCliTransport(runner=runner) + user, error = transport.get_for_token(token, "user") + assert user is None and error == "GitHub request failed" + command, kwargs, child_token, has_ambiguous_token = seen[0] + assert "--hostname" in command and "github.com" in command + assert all(token not in arg for arg in command) + assert child_token == token + assert not has_ambiguous_token + assert "GH_TOKEN" not in kwargs["env"] + + +def test_transport_invoke_preserves_legacy_args_and_stdin_call_shapes(): + calls = [] + + def runner(command, **kwargs): + calls.append((command, kwargs)) + assert "env" not in kwargs + return SimpleNamespace(returncode=0, stdout='{"ok":true}', stderr="") + + transport = GhCliTransport(runner=runner) + assert transport._invoke(["user"]) == ('{"ok":true}', None) + assert transport._invoke(["user"], "request-body") == ('{"ok":true}', None) + assert calls[0][1]["input"] is None + assert calls[1][1]["input"] == "request-body" def test_push_branch_refuses_the_default_branch_even_with_bot_prefix(monkeypatch): + _mock_push_token(monkeypatch) monkeypatch.setattr(typed_actions, "_git", lambda *a, **k: (_ for _ in ()).throw(AssertionError("git called"))) cfg = config() cfg["typed_actions"] = {"push_branch_prefix": "release"} action = Action("git.push_branch", {"repo": REPO, "branch": "release/main", "sha": SHA, "path": "."}) - transport = Transport({f"repos/{REPO}": {"default_branch": "release/main"}}) - assert actions.apply([action], transport, cfg)["receipts"][0]["status"] == "failed" + transport = TokenTransport({f"repos/{REPO}": {"default_branch": "release/main"}}) + receipt = actions.apply([action], transport, cfg)["receipts"][0] + assert receipt["status"] == "failed" and "never pushes" in receipt["detail"] def test_run_contract_reports_effective_enabled(tmp_path): From d83171347045fcf56534e6e0bbd0b29e84d501f8 Mon Sep 17 00:00:00 2001 From: Lukas Geiger Date: Thu, 29 Oct 2026 19:16:21 +0100 Subject: [PATCH 2/2] Make Git AskPass isolated and report safe failures --- src/githubbot_core/typed_actions.py | 22 ++++++++------ tests/test_typed_actions.py | 46 ++++++++++++++++++++++++++++- 2 files changed, 58 insertions(+), 10 deletions(-) diff --git a/src/githubbot_core/typed_actions.py b/src/githubbot_core/typed_actions.py index f1763f7..fdaa865 100644 --- a/src/githubbot_core/typed_actions.py +++ b/src/githubbot_core/typed_actions.py @@ -566,13 +566,13 @@ def _git_auth_environment(token: str, account: str): ) if os.name == "nt": helper.write_text( - '@echo off\r\n"%GITHUBBOT_ASKPASS_PYTHON%" "%~dp0askpass.py" %*\r\nexit /b %ERRORLEVEL%\r\n', + '@echo off\r\n"%GITHUBBOT_ASKPASS_PYTHON%" -I "%~dp0askpass.py" %*\r\nexit /b %ERRORLEVEL%\r\n', encoding="utf-8", newline="", ) else: helper.write_text( - '#!/bin/sh\nexec "$GITHUBBOT_ASKPASS_PYTHON" "$(dirname "$0")/askpass.py" "$@"\n', + '#!/bin/sh\nexec "$GITHUBBOT_ASKPASS_PYTHON" -I "$(dirname "$0")/askpass.py" "$@"\n', encoding="utf-8", newline="\n", ) @@ -618,12 +618,14 @@ def _run_git( timeout=120, env=run_env, ) - except (OSError, subprocess.TimeoutExpired): - return "", "git command failed" - except Exception: - return "", "git command failed" + except subprocess.TimeoutExpired: + return "", "git process timed out" + except OSError as exc: + return "", f"git process could not start ({type(exc).__name__})" + except Exception as exc: + return "", f"git process failed ({type(exc).__name__})" if result.returncode: - return "", f"git command failed (exit {result.returncode})" + return "", f"git process exited with code {result.returncode}" return (result.stdout or "").strip(), None @@ -711,9 +713,11 @@ def exec_push_branch(transport, params: dict, config: dict): # No '+' refspec and no --force: a non-fast-forward is rejected by git itself. _, error = _run_git(path, "push", "origin", f"{sha}:{ref}", env=env, repo=repo) if error: - return None, "git push failed or its outcome is unknown; read the target before retrying" + return None, f"{error}; push outcome is unknown; read the target before retrying" remote, error = _run_git(path, "ls-remote", "--exit-code", "origin", ref, env=env, repo=repo) - if error or not _exact_remote_ref(remote, sha, ref): + if error: + return None, f"push readback could not be confirmed ({error}); read the target before retrying" + if not _exact_remote_ref(remote, sha, ref): return None, "push readback is unconfirmed; read the target before retrying" except Exception: return None, "account-bound Git transport failed; read the target before retrying" diff --git a/tests/test_typed_actions.py b/tests/test_typed_actions.py index 75bd077..a8035a1 100644 --- a/tests/test_typed_actions.py +++ b/tests/test_typed_actions.py @@ -479,10 +479,52 @@ def git(cmd, **kwargs): receipt = actions.apply([_push_action()], _push_transport(), config())["receipts"][0] assert receipt["status"] == "failed" assert token not in receipt["detail"] and token not in caplog.text + assert "git process exited with code 128" in receipt["detail"] assert "read the target before retrying" in receipt["detail"] assert temp_dirs and all(not os.path.exists(directory) for directory in temp_dirs) +@pytest.mark.parametrize( + ("failure", "expected"), + [ + ("start", "git process could not start (FileNotFoundError)"), + ("timeout", "git process timed out"), + ("exit", "git process exited with code 128"), + ], +) +def test_push_failure_receipt_reports_only_safe_process_category(monkeypatch, caplog, failure, expected): + token, _ = _mock_push_token(monkeypatch) + temp_dirs = [] + + def git(cmd, **kwargs): + if "push" in cmd: + temp_dirs.append(Path(kwargs["env"]["GIT_ASKPASS"]).parent) + if failure == "start": + raise FileNotFoundError(f"synthetic credential-shaped text {token}") + if failure == "timeout": + raise subprocess.TimeoutExpired( + cmd, timeout=120, output=f"synthetic {token}", stderr=f"rejected {token}" + ) + return SimpleNamespace( + returncode=128, + stdout=f"remote rejected {token}", + stderr=f"authentication failed {token}", + ) + return _fake_git_output( + cmd, + origin=f"https://github.com/{REPO}.git", + pushurl=f"https://github.com/{REPO}.git", + ) + + monkeypatch.setattr(typed_actions, "_git", git) + receipt = actions.apply([_push_action()], _push_transport(), config())["receipts"][0] + assert receipt["status"] == "failed" + assert expected in receipt["detail"] + assert "read the target before retrying" in receipt["detail"] + assert token not in receipt["detail"] and token not in caplog.text + assert temp_dirs and all(not directory.exists() for directory in temp_dirs) + + def test_git_base_environment_removes_ambient_auth_and_repo_overrides(monkeypatch): monkeypatch.setenv("GH_TOKEN", "synthetic-ambient-token") monkeypatch.setenv("GITHUB_TOKEN", "synthetic-ambient-token") @@ -498,9 +540,10 @@ def test_git_base_environment_removes_ambient_auth_and_repo_overrides(monkeypatc assert env["GIT_CONFIG_GLOBAL"] == os.devnull -def test_git_askpass_helper_uses_child_env_and_cleans_tempfiles(): +def test_git_askpass_helper_uses_child_env_and_cleans_tempfiles(monkeypatch, tmp_path): token = "synthetic-askpass-token" directory = None + monkeypatch.setenv("PYTHONHOME", str(tmp_path / "invalid-pythonhome")) with typed_actions._git_auth_environment(token, "example-bot") as env: helper = Path(env["GIT_ASKPASS"]) directory = helper.parent @@ -533,6 +576,7 @@ def run_askpass(prompt): password = run_askpass("Password for https://github.com:") assert username.stdout == "example-bot" assert password.stdout == token + assert env["PYTHONHOME"].endswith("invalid-pythonhome") assert all(token not in str(command) for command in commands) assert directory is not None and not directory.exists()