From 920bb59f1e7526372a22605e2a9383d3fd272c3c Mon Sep 17 00:00:00 2001 From: zzoubian <35345489+zzoubian@users.noreply.github.com> Date: Sun, 5 Apr 2026 21:30:01 +0000 Subject: [PATCH 1/2] fix: enhance git remote validation by trying SSH URLs for generic hosts --- src/apm_cli/commands/install.py | 31 +++++++++++++++++++++++-------- 1 file changed, 23 insertions(+), 8 deletions(-) diff --git a/src/apm_cli/commands/install.py b/src/apm_cli/commands/install.py index 10813f347..1dd6ab7f4 100644 --- a/src/apm_cli/commands/install.py +++ b/src/apm_cli/commands/install.py @@ -431,14 +431,29 @@ def _validate_package_exists(package, verbose=False, auth_resolver=None): if verbose_log: verbose_log(f"Trying git ls-remote for {dep_ref.host}") - cmd = ["git", "ls-remote", "--heads", "--exit-code", package_url] - result = subprocess.run( - cmd, - capture_output=True, - text=True, - timeout=30, - env=validate_env, - ) + # For generic hosts, try SSH first (no credentials needed when SSH + # keys are configured) before falling back to HTTPS. + urls_to_try = [] + if is_generic: + ssh_url = ado_downloader._build_repo_url( + dep_ref.repo_url, use_ssh=True, dep_ref=dep_ref + ) + urls_to_try = [ssh_url, package_url] + else: + urls_to_try = [package_url] + + result = None + for probe_url in urls_to_try: + cmd = ["git", "ls-remote", "--heads", "--exit-code", probe_url] + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=30, + env=validate_env, + ) + if result.returncode == 0: + break if verbose_log: if result.returncode == 0: From 01640358e0191d348360be6cceac63ebab7c5ecc Mon Sep 17 00:00:00 2001 From: zzoubian <35345489+zzoubian@users.noreply.github.com> Date: Mon, 6 Apr 2026 22:24:04 -0400 Subject: [PATCH 2/2] add unit tests and add encoding to subprocess call --- src/apm_cli/commands/install.py | 1 + tests/unit/test_install_command.py | 131 +++++++++++++++++++++++++++++ 2 files changed, 132 insertions(+) diff --git a/src/apm_cli/commands/install.py b/src/apm_cli/commands/install.py index 1dd6ab7f4..5f422b16c 100644 --- a/src/apm_cli/commands/install.py +++ b/src/apm_cli/commands/install.py @@ -449,6 +449,7 @@ def _validate_package_exists(package, verbose=False, auth_resolver=None): cmd, capture_output=True, text=True, + encoding="utf-8", timeout=30, env=validate_env, ) diff --git a/tests/unit/test_install_command.py b/tests/unit/test_install_command.py index 282b6e0b9..5dc8d3b4f 100644 --- a/tests/unit/test_install_command.py +++ b/tests/unit/test_install_command.py @@ -862,3 +862,134 @@ def test_global_rejects_tilde_local_path(self): assert "not supported at user scope" in result.output finally: os.chdir(self.original_dir) + +# --------------------------------------------------------------------------- +# Generic-host SSH-first validation tests +# --------------------------------------------------------------------------- + +class TestGenericHostSshFirstValidation: + """Tests for the SSH-first ls-remote logic added for generic (non-GitHub/ADO) hosts.""" + + def _make_completed_process(self, returncode, stderr=""): + """Return a minimal subprocess.CompletedProcess-like mock.""" + mock = MagicMock() + mock.returncode = returncode + mock.stderr = stderr + mock.stdout = "" + return mock + + @patch("subprocess.run") + def test_generic_host_tries_ssh_first_and_succeeds(self, mock_run): + """SSH URL is tried first for generic hosts and used when it succeeds.""" + from apm_cli.commands.install import _validate_package_exists + + # SSH probe succeeds on the first call + mock_run.return_value = self._make_completed_process(returncode=0) + + result = _validate_package_exists( + "git@git.example.org:org/group/repo.git", verbose=False + ) + + assert result is True + # subprocess.run must have been called at least once + assert mock_run.call_count >= 1 + # First call must use the SSH URL + first_call_cmd = mock_run.call_args_list[0][0][0] + assert any("git@git.example.org" in arg for arg in first_call_cmd), ( + f"Expected SSH URL in first ls-remote call, got: {first_call_cmd}" + ) + + @patch("subprocess.run") + def test_generic_host_falls_back_to_https_when_ssh_fails(self, mock_run): + """HTTPS fallback is used for generic hosts when SSH ls-remote fails.""" + from apm_cli.commands.install import _validate_package_exists + + # SSH probe fails, HTTPS succeeds + mock_run.side_effect = [ + self._make_completed_process(returncode=128, stderr="ssh: connect to host"), + self._make_completed_process(returncode=0), + ] + + result = _validate_package_exists( + "git@git.example.org:org/group/repo.git", verbose=False + ) + + assert result is True + assert mock_run.call_count == 2 + # First call: SSH + first_cmd = mock_run.call_args_list[0][0][0] + assert any("git@git.example.org" in arg for arg in first_cmd), ( + f"Expected SSH URL in first call, got: {first_cmd}" + ) + # Second call: HTTPS + second_cmd = mock_run.call_args_list[1][0][0] + assert any("https://git.example.org" in arg for arg in second_cmd), ( + f"Expected HTTPS URL in second call, got: {second_cmd}" + ) + + @patch("subprocess.run") + def test_generic_host_returns_false_when_both_transports_fail(self, mock_run): + """Validation returns False when both SSH and HTTPS fail for a generic host.""" + from apm_cli.commands.install import _validate_package_exists + + mock_run.return_value = self._make_completed_process( + returncode=128, stderr="fatal: could not read Username" + ) + + result = _validate_package_exists( + "git@git.example.org:org/group/repo.git", verbose=False + ) + + assert result is False + assert mock_run.call_count == 2 # tried SSH then HTTPS + + @patch("subprocess.run") + def test_github_host_skips_ssh_attempt(self, mock_run): + """GitHub.com repositories do NOT go through the SSH-first ls-remote path.""" + + import urllib.request + import urllib.error + + from apm_cli.commands.install import _validate_package_exists + + with patch("urllib.request.urlopen") as mock_urlopen: + mock_urlopen.side_effect = urllib.error.HTTPError( + url="https://api.github.com/repos/owner/repo", + code=404, msg="Not Found", hdrs={}, fp=None, + ) + result = _validate_package_exists("owner/repo", verbose=False) + + assert result is False + # No ls-remote call should have been made for a github.com host + ls_remote_calls = [ + call for call in mock_run.call_args_list + if "ls-remote" in (call[0][0] if call[0] else []) + ] + assert len(ls_remote_calls) == 0, ( + f"Expected no ls-remote calls for github.com, got: {ls_remote_calls}" + ) + + @patch("subprocess.run") + def test_ghes_host_skips_ssh_attempt(self, mock_run): + """A GHES host is treated as GitHub, not generic SSH probe is skipped.""" + from apm_cli.commands.install import _validate_package_exists + + mock_run.return_value = self._make_completed_process(returncode=0) + + result = _validate_package_exists( + "company.ghe.com/team/internal-repo", verbose=False + ) + + assert result is True + ls_remote_calls = [ + call for call in mock_run.call_args_list + if "ls-remote" in (call[0][0] if call[0] else []) + ] + assert len(ls_remote_calls) == 1, ( + f"Expected exactly 1 ls-remote call for GHES host, got: {ls_remote_calls}" + ) + only_cmd = ls_remote_calls[0][0][0] + # Must use HTTPS, not SSH + assert all("git@" not in arg for arg in only_cmd), ( + f"Expected HTTPS-only URL for GHES host, got: {only_cmd}" + )