Skip to content

Use GITHUB_TOKEN (PAT) for branch detection on url_pull repos #3061

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
ashwindasr wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Use GITHUB_TOKEN (PAT) for branch detection on url_pull repos #3061

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
32 changes: 25 additions & 7 deletions doozer/doozerlib/source_resolver.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
from artcommonlib import assertion, constants, exectools
from artcommonlib import util as art_util
from artcommonlib.git_helper import git_clone
from artcommonlib.github_auth import get_github_client_for_org, get_github_git_auth_env
from artcommonlib.github_auth import get_github_client_for_org, get_github_git_auth_env, get_github_git_pat_env
from artcommonlib.lock import get_named_semaphore
from artcommonlib.model import ListModel, Missing, Model
from github import GithubException, UnknownObjectException
Expand Down Expand Up @@ -306,13 +306,22 @@ def detect_remote_source_branch(
if use_source_fallback_branch not in ["yes", "always", "never"]:
raise ValueError(f"Invalid value for use_source_fallback_branch: {use_source_fallback_branch}")

git_url = source_details.get("url_pull", source_details["url"]) # Use pull URL if available
pull_url = source_details.get("url_pull")
push_url = source_details["url"]
git_url = pull_url or push_url
# When url_pull points to a personal fork, the GitHub App won't have
# access. Use GITHUB_TOKEN (PAT) for branch detection against url_pull
# and fall back to the push url (where the App is installed) if needed.
use_pat = bool(pull_url)
Comment on lines +309 to +315

@coderabbitai coderabbitai Bot Jun 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔒 Security & Privacy | 🟠 Major | ⚡ Quick win

Gate PAT auth to GitHub pull remotes only.

use_pat = bool(pull_url) is too broad here. If url_pull points at GitLab or another non-GitHub HTTPS host, _get_remote_branch_ref(..., use_pat=True) will inject GITHUB_TOKEN via get_github_git_pat_env(). Unlike get_github_git_auth_env(url=...), that helper does not suppress credentials for non-GitHub remotes, so this can leak the GitHub PAT to the foreign host during git ls-remote.

Suggested direction 
-        use_pat = bool(pull_url)
+        normalized_pull_url = art_util.convert_remote_git_to_https(pull_url) if pull_url else ""
+        use_pat = normalized_pull_url.startswith("https://github.com/")

Also applies to: 392-392

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@doozer/doozerlib/source_resolver.py` around lines 309 - 315, The current
logic sets use_pat = bool(pull_url) which enables injecting the GitHub PAT for
any pull URL (including non-GitHub hosts) and can leak credentials; change the
gating so use_pat is true only when url_pull is a GitHub HTTPS remote. Update
the code around source_details["url_pull"], source_details["url"], git_url and
use_pat to detect GitHub hosts (e.g. by validating the URL's hostname or using
the same GitHub-detection used by get_github_git_auth_env) before setting
use_pat, and leave use_pat False for non-GitHub remotes so
_get_remote_branch_ref(...) will not call get_github_git_pat_env for foreign
hosts; apply the same fix to the other occurrence at the later block (around
line 392) that uses use_pat.

branches = source_details["branch"]

stage_branch = branches.get("stage", None) if stage else None
if stage_branch:
LOGGER.info('Normal branch overridden by --stage option, using "{}"'.format(stage_branch))
result = SourceResolver._get_remote_branch_ref(git_url, stage_branch)
result = SourceResolver._get_remote_branch_ref(git_url, stage_branch, use_pat=use_pat)
if not result and pull_url:
LOGGER.info("Branch check against url_pull %s failed, retrying with push url %s", pull_url, push_url)
result = SourceResolver._get_remote_branch_ref(push_url, stage_branch)
Comment on lines +321 to +324

@coderabbitai coderabbitai Bot Jun 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

Don't treat "branch missing on url_pull" the same as "couldn't query url_pull".

Right now _get_remote_branch_ref() returns None both when git ls-remote errors and when the branch genuinely does not exist. These sites retry push_url on any falsy result, so a branch that exists only on push_url can be selected even though resolve_source() will still clone from pull_url. That makes branch resolution succeed and the later clone fail. Only retry push_url when the url_pull probe itself failed; a clean "no such ref" result should stay authoritative for the fork.

Also applies to: 343-346, 354-357

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@doozer/doozerlib/source_resolver.py` around lines 321 - 324,
_source_resolver.py currently treats "no such ref" and "probe failed" the same
because SourceResolver._get_remote_branch_ref returns falsy for both; change its
contract to return a two-tuple (ref, error) where ref is the ref string or None
when not present, and error is None on a successful probe (even if ref is
missing) or an Exception/str when the git ls-remote call failed; then update
callers (the blocks that call SourceResolver._get_remote_branch_ref, e.g. the
snippet where result = SourceResolver._get_remote_branch_ref(git_url,
stage_branch, use_pat=use_pat) and the similar calls around the other mentions)
to only attempt the retry against push_url when error is non-None, and treat
ref==None with error==None as authoritative "branch missing on url_pull" (no
retry). Ensure any call sites that expected a single return value are adjusted
to unpack the tuple and propagate/handle errors appropriately.

if result:
return stage_branch, result
raise IOError(
Expand All @@ -331,15 +340,21 @@ def detect_remote_source_branch(
if SourceResolver.is_branch_commit_hash(branch):
return branch, branch

result = SourceResolver._get_remote_branch_ref(git_url, branch)
result = SourceResolver._get_remote_branch_ref(git_url, branch, use_pat=use_pat)
if not result and pull_url:
LOGGER.info("Branch check against url_pull %s failed, retrying with push url %s", pull_url, push_url)
result = SourceResolver._get_remote_branch_ref(push_url, branch)
if result:
return branch, result

if not fallback_branch:
raise IOError('Requested target branch {} does not exist and no fallback provided'.format(branch))

LOGGER.info('Target branch does not exist in {}, checking fallback branch {}'.format(git_url, fallback_branch))
result = SourceResolver._get_remote_branch_ref(git_url, fallback_branch)
result = SourceResolver._get_remote_branch_ref(git_url, fallback_branch, use_pat=use_pat)
if not result and pull_url:
LOGGER.info("Branch check against url_pull %s failed, retrying with push url %s", pull_url, push_url)
result = SourceResolver._get_remote_branch_ref(push_url, fallback_branch)
if result:
return fallback_branch, result
raise IOError('Requested fallback branch {} does not exist'.format(branch))
Expand All @@ -363,15 +378,18 @@ def is_branch_commit_hash(branch):
return False

@staticmethod
def _get_remote_branch_ref(git_url, branch):
def _get_remote_branch_ref(git_url, branch, use_pat=False):
"""
Detect whether a single branch exists on a remote repo; returns git hash if found
:param git_url: The URL to the git repo to check.
:param branch: The name of the branch. If the name is not a branch and appears to be a commit
hash, the hash will be returned without modification.
:param use_pat: If True, use GITHUB_TOKEN (PAT) instead of GitHub App auth.
This is needed for url_pull repos (personal forks) where the
GitHub App is not installed.
"""
git_url = art_util.ensure_github_https_url(git_url)
auth_env = get_github_git_auth_env(url=git_url)
auth_env = get_github_git_pat_env() if use_pat else get_github_git_auth_env(url=git_url)
LOGGER.info('Checking if target branch {} exists in {}'.format(branch, git_url))

try:
Expand Down
53 changes: 47 additions & 6 deletions doozer/tests/test_source_resolver.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,18 @@ def test_get_remote_branch_ref(self, mock_ensure, mock_auth):
flexmock(exectools).should_receive("cmd_assert").once().and_raise(Exception("whatever"))
self.assertIsNone(sr._get_remote_branch_ref("giturl", "branch"))

@patch("doozerlib.source_resolver.get_github_git_pat_env", return_value={"GIT_ASKPASS": "/tmp/askpass.sh", "GIT_PASSWORD": "pat-token", "GIT_TERMINAL_PROMPT": "0"})
@patch("doozerlib.source_resolver.get_github_git_auth_env", return_value={"GIT_ASKPASS": "/tmp/askpass.sh"})
@patch("doozerlib.source_resolver.art_util.ensure_github_https_url", side_effect=lambda u: u)
def test_get_remote_branch_ref_use_pat(self, mock_ensure, mock_auth, mock_pat):
"""When use_pat=True, get_github_git_pat_env should be used instead of get_github_git_auth_env."""
sr = self.create_source_resolver()
flexmock(exectools).should_receive("cmd_assert").once().and_return("abc123", "")
res = sr._get_remote_branch_ref("giturl", "branch", use_pat=True)
self.assertEqual(res, "abc123")
mock_pat.assert_called_once()
mock_auth.assert_not_called()

def test_detect_remote_source_branch(self):
sr = self.create_source_resolver()
source_details = dict(
Expand Down Expand Up @@ -155,8 +167,8 @@ def test_https_pull_url_property_without_pull_url(self):
result = resolution.https_pull_url
self.assertEqual(result, "https://push.example.com/repo.git")

def test_detect_remote_source_branch_uses_pull_url(self):
"""Test that detect_remote_source_branch uses url_pull when available."""
def test_detect_remote_source_branch_uses_pull_url_with_pat(self):
"""Test that detect_remote_source_branch uses url_pull with PAT auth when available."""
sr = self.create_source_resolver()
source_details = dict(
url='https://push.example.com/repo.git',
Expand All @@ -167,14 +179,44 @@ def test_detect_remote_source_branch_uses_pull_url(self):
),
)

# Mock the _get_remote_branch_ref to expect pull URL
flexmock(SourceResolver).should_receive("_get_remote_branch_ref").with_args(
'https://pull.example.com/repo.git', 'main_branch'
'https://pull.example.com/repo.git', 'main_branch', use_pat=True
).once().and_return("commit_hash")

result = sr.detect_remote_source_branch(source_details, stage=False)
self.assertEqual(result, ("main_branch", "commit_hash"))

def test_detect_remote_source_branch_pull_url_fallback_to_push_url(self):
"""When url_pull branch check fails (e.g. GitHub App has no access to
a personal fork), fall back to checking the push url."""
sr = self.create_source_resolver()
source_details = dict(
url='https://push.example.com/repo.git',
url_pull='https://pull.example.com/repo.git',
branch=dict(
target='main_branch',
fallback='fallback_branch',
),
)

(
flexmock(SourceResolver)
.should_receive("_get_remote_branch_ref")
.with_args('https://pull.example.com/repo.git', 'main_branch', use_pat=True)
.once()
.and_return(None)
)
(
flexmock(SourceResolver)
.should_receive("_get_remote_branch_ref")
.with_args('https://push.example.com/repo.git', 'main_branch')
.once()
.and_return("commit_hash_from_push")
)

result = sr.detect_remote_source_branch(source_details, stage=False)
self.assertEqual(result, ("main_branch", "commit_hash_from_push"))

def test_detect_remote_source_branch_falls_back_to_url(self):
"""Test that detect_remote_source_branch falls back to url when url_pull is not available."""
sr = self.create_source_resolver()
Expand All @@ -186,9 +228,8 @@ def test_detect_remote_source_branch_falls_back_to_url(self):
),
)

# Mock the _get_remote_branch_ref to expect main URL
flexmock(SourceResolver).should_receive("_get_remote_branch_ref").with_args(
'https://push.example.com/repo.git', 'main_branch'
'https://push.example.com/repo.git', 'main_branch', use_pat=False
).once().and_return("commit_hash")

result = sr.detect_remote_source_branch(source_details, stage=False)
Expand Down
Loading