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
Original file line number Diff line number Diff line change
Expand Up @@ -404,9 +404,10 @@ async def populate_runtime_credentials(context: RunnerContext) -> None:
if github_creds.get("email"):
git_user_email = github_creds["email"]

# Configure git identity and credential helper
# Configure git identity, credential helper, and gh CLI wrapper
await configure_git_identity(git_user_name, git_user_email)
install_git_credential_helper()
install_gh_wrapper()

if auth_failures:
raise PermissionError(
Expand Down Expand Up @@ -533,6 +534,43 @@ async def populate_mcp_server_credentials(context: RunnerContext) -> None:
logger.warning(f"Failed to fetch MCP credentials for {server_name}: {e}")


_GH_WRAPPER_DIR = "/tmp/bin"
_GH_WRAPPER_PATH = "/tmp/bin/gh"

# Wrapper script for the gh CLI. The `gh` CLI reads GITHUB_TOKEN from the
# process environment, but the CLI subprocess's env is fixed at spawn time.
# This wrapper reads the latest token from the token file (updated on every
# credential refresh) and exports GH_TOKEN before calling the real `gh`,
# ensuring mid-run refreshes are picked up.
_GH_WRAPPER_SCRIPT = """\
#!/bin/sh
# Ambient gh CLI wrapper — reads fresh GitHub token from file.
token=""
if [ -f "/tmp/.ambient_github_token" ]; then
token=$(cat /tmp/.ambient_github_token 2>/dev/null)
fi
if [ -n "$token" ]; then
export GH_TOKEN="$token"
fi
# Find the real gh binary, skipping this wrapper directory.
real_gh=""
IFS=:
for p in $PATH; do
if [ "$p" != "{wrapper_dir}" ] && [ -x "$p/gh" ]; then
real_gh="$p/gh"
break
fi
done
unset IFS
if [ -z "$real_gh" ]; then
echo "Error: gh CLI not found" >&2
exit 1
fi
exec "$real_gh" "$@"
""".format(wrapper_dir=_GH_WRAPPER_DIR)

_gh_wrapper_installed = False # reset on every new process / deployment

_GIT_CREDENTIAL_HELPER_PATH = "/tmp/git-credential-ambient"

# Injected into git's credential system so clean remote URLs (without embedded
Expand Down Expand Up @@ -627,6 +665,42 @@ def install_git_credential_helper() -> None:
logger.warning(f"Failed to install git credential helper: {e}")


def install_gh_wrapper() -> None:
"""Install a gh CLI wrapper that reads the fresh GitHub token from file.

The ``gh`` CLI prioritises the ``GITHUB_TOKEN`` env var over all other
credential sources. Since the CLI subprocess's environment is fixed at
spawn time, a stale ``GITHUB_TOKEN`` causes 401 errors after a mid-run
credential refresh. This wrapper reads from the token file (updated on
every refresh) and exports ``GH_TOKEN`` before exec-ing the real ``gh``.
"""
global _gh_wrapper_installed
if _gh_wrapper_installed:
return

import stat

try:
wrapper_dir = Path(_GH_WRAPPER_DIR)
wrapper_dir.mkdir(parents=True, exist_ok=True)

wrapper_path = Path(_GH_WRAPPER_PATH)
wrapper_path.write_text(_GH_WRAPPER_SCRIPT)
wrapper_path.chmod(
stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP | stat.S_IROTH | stat.S_IXOTH
) # 755

# Prepend wrapper dir to PATH so it is found before the real gh.
current_path = os.environ.get("PATH", "")
if _GH_WRAPPER_DIR not in current_path.split(":"):
os.environ["PATH"] = f"{_GH_WRAPPER_DIR}:{current_path}"

_gh_wrapper_installed = True
logger.info("Installed gh CLI wrapper at %s", _GH_WRAPPER_PATH)
except Exception as e:
logger.warning(f"Failed to install gh CLI wrapper: {e}")


def ensure_git_auth(
github_token: str | None = None,
gitlab_token: str | None = None,
Expand Down
110 changes: 110 additions & 0 deletions components/runners/ambient-runner/tests/test_shared_session_credentials.py
100644 → 100755
Original file line number Diff line number Diff line change
Expand Up @@ -4,17 +4,21 @@
import os
from http.server import BaseHTTPRequestHandler, HTTPServer
from io import BytesIO
from pathlib import Path
from threading import Thread
from unittest.mock import AsyncMock, MagicMock, patch
from urllib.error import HTTPError

import pytest

from ambient_runner.platform.auth import (
_GH_WRAPPER_DIR,
_GH_WRAPPER_PATH,
_GITHUB_TOKEN_FILE,
_GITLAB_TOKEN_FILE,
_fetch_credential,
clear_runtime_credentials,
install_gh_wrapper,
populate_runtime_credentials,
sanitize_user_context,
)
Expand Down Expand Up @@ -710,3 +714,109 @@ async def test_returns_success_on_successful_refresh(self):

assert result.get("isError") is None or result.get("isError") is False
assert "successfully" in result["content"][0]["text"].lower()


# ---------------------------------------------------------------------------
# gh CLI wrapper — ensures gh picks up refreshed tokens (issue #1135)
# ---------------------------------------------------------------------------


class TestGhWrapper:
"""The gh CLI wrapper reads the latest GitHub token from the token file.

This mirrors the git credential helper pattern: the CLI subprocess's
environment is fixed at spawn time so env var updates don't propagate.
The wrapper reads the token file on every invocation, ensuring `gh`
always uses the freshest token.
"""

def _cleanup(self):
"""Remove wrapper artifacts created during tests."""
import ambient_runner.platform.auth as _auth_mod

_auth_mod._gh_wrapper_installed = False
wrapper = Path(_GH_WRAPPER_PATH)
wrapper.unlink(missing_ok=True)
wrapper_dir = Path(_GH_WRAPPER_DIR)
if wrapper_dir.exists() and not any(wrapper_dir.iterdir()):
wrapper_dir.rmdir()

def test_install_creates_executable_wrapper(self):
"""install_gh_wrapper creates an executable script at _GH_WRAPPER_PATH."""
self._cleanup()
try:
install_gh_wrapper()
wrapper = Path(_GH_WRAPPER_PATH)
assert wrapper.exists(), "Wrapper script should be created"
assert os.access(str(wrapper), os.X_OK), "Wrapper should be executable"
content = wrapper.read_text()
assert "/tmp/.ambient_github_token" in content
assert "GH_TOKEN" in content
finally:
self._cleanup()

def test_install_prepends_to_path(self):
"""install_gh_wrapper prepends the wrapper dir to PATH."""
self._cleanup()
original_path = os.environ.get("PATH", "")
try:
# Remove wrapper dir from PATH if present
parts = [p for p in original_path.split(":") if p != _GH_WRAPPER_DIR]
os.environ["PATH"] = ":".join(parts)

install_gh_wrapper()

current_path = os.environ.get("PATH", "")
assert current_path.startswith(_GH_WRAPPER_DIR + ":"), (
"Wrapper dir should be first in PATH"
)
finally:
os.environ["PATH"] = original_path
self._cleanup()

def test_install_is_idempotent(self):
"""Calling install_gh_wrapper twice does not duplicate PATH entries."""
self._cleanup()
original_path = os.environ.get("PATH", "")
try:
parts = [p for p in original_path.split(":") if p != _GH_WRAPPER_DIR]
os.environ["PATH"] = ":".join(parts)

install_gh_wrapper()
install_gh_wrapper() # second call should be a no-op

current_path = os.environ.get("PATH", "")
count = current_path.split(":").count(_GH_WRAPPER_DIR)
assert count == 1, f"Wrapper dir should appear once in PATH, got {count}"
finally:
os.environ["PATH"] = original_path
self._cleanup()

@pytest.mark.asyncio
async def test_populate_installs_gh_wrapper(self):
"""populate_runtime_credentials installs the gh wrapper."""
self._cleanup()
try:
with patch("ambient_runner.platform.auth._fetch_credential") as mock_fetch:

async def _creds(ctx, ctype):
if ctype == "github":
return {
"token": "gh-test-token",
"userName": "user",
"email": "u@example.com",
}
return {}

mock_fetch.side_effect = _creds
ctx = _make_context()
await populate_runtime_credentials(ctx)

wrapper = Path(_GH_WRAPPER_PATH)
assert wrapper.exists(), (
"populate_runtime_credentials should install gh wrapper"
)
finally:
self._cleanup()
for key in ["GITHUB_TOKEN", "GIT_USER_NAME", "GIT_USER_EMAIL"]:
os.environ.pop(key, None)
Loading