diff --git a/openhands-sdk/openhands/sdk/workspace/repo.py b/openhands-sdk/openhands/sdk/workspace/repo.py index 6dee56fea0..ddc485fd3f 100644 --- a/openhands-sdk/openhands/sdk/workspace/repo.py +++ b/openhands-sdk/openhands/sdk/workspace/repo.py @@ -294,10 +294,17 @@ def _get_unique_dir_name(base_name: str, existing_dirs: set[str]) -> str: } -def _build_clone_url(url: str, provider: GitProvider, token: str | None) -> str: +def _build_clone_url( + url: str, + provider: GitProvider, + token: str | None, + *, + explicit_provider: bool = False, +) -> str: """Build authenticated clone URL based on the repository URL and provider. - Uses proper URL parsing to prevent token injection into malicious URLs. + The token is injected into the provider's public host, or into the URL's own + host when the caller set `provider` explicitly (self-hosted instances). """ config = _PROVIDER_CONFIG.get(provider) if not config: @@ -311,16 +318,21 @@ def _build_clone_url(url: str, provider: GitProvider, token: str | None) -> str: if is_short_format: return f"https://{auth_prefix}{base_url}/{url}.git" - # Handle full URLs - inject authentication only if hostname matches exactly - if token: - parsed = urllib.parse.urlparse(url) - if parsed.netloc.lower() == base_url: - # Replace only the first occurrence to prevent double injection - return url.replace( - f"https://{base_url}", f"https://{auth_prefix}{base_url}", 1 - ) + if not token: + return url - return url + parsed = urllib.parse.urlparse(url) + hostname = (parsed.hostname or "").lower() + if parsed.scheme != "https" or parsed.username or not hostname: + return url + if hostname != base_url: + # An auto-detected provider is derived from the URL, so it cannot + # authorize another host - and never a lookalike of the public one. + if not explicit_provider or hostname.startswith(f"{base_url}."): + return url + + netloc = hostname if parsed.port is None else f"{hostname}:{parsed.port}" + return urllib.parse.urlunparse(parsed._replace(netloc=f"{auth_prefix}{netloc}")) # Type for functions that fetch tokens by name (e.g., "github_token" -> token value) @@ -367,7 +379,9 @@ def _clone_single_repo(repo: RepoSource, dest: Path, token: str | None) -> bool: """Clone a single repository. Returns True on success.""" try: provider = repo.get_provider() - clone_url = _build_clone_url(repo.url, provider, token) + clone_url = _build_clone_url( + repo.url, provider, token, explicit_provider=repo.provider is not None + ) provider_str = provider.value except ValueError: # No provider detected (e.g., file:// URLs) - use URL as-is diff --git a/tests/workspace/test_cloud_workspace_repos.py b/tests/workspace/test_cloud_workspace_repos.py index a555419edc..0e3f4acf72 100644 --- a/tests/workspace/test_cloud_workspace_repos.py +++ b/tests/workspace/test_cloud_workspace_repos.py @@ -260,6 +260,57 @@ def test_build_clone_url_no_token_passthrough(self): ) assert url == "https://github.com/owner/repo" + def test_build_clone_url_self_hosted_gitlab_with_explicit_provider(self): + """A self-hosted GitLab host gets the token when provider is explicit.""" + url = _build_clone_url( + "https://gitlab.mycompany.com/owner/repo", + GitProvider.GITLAB, + "gltoken123", + explicit_provider=True, + ) + assert url == "https://oauth2:gltoken123@gitlab.mycompany.com/owner/repo" + + def test_build_clone_url_self_hosted_host_without_explicit_provider(self): + """An auto-detected provider must not inject a token into an unrelated host.""" + url = _build_clone_url( + "https://gitlab.mycompany.com/owner/repo", + GitProvider.GITLAB, + "gltoken123", + explicit_provider=False, + ) + assert url == "https://gitlab.mycompany.com/owner/repo" + + @pytest.mark.parametrize("explicit_provider", [False, True]) + def test_build_clone_url_lookalike_host_not_injected(self, explicit_provider): + """A lookalike of the public host never receives the token.""" + url = _build_clone_url( + "https://github.com.evil.com/owner/repo", + GitProvider.GITHUB, + "ghtoken123", + explicit_provider=explicit_provider, + ) + assert url == "https://github.com.evil.com/owner/repo" + + def test_build_clone_url_self_hosted_host_is_normalized(self): + """Host case and non-default port survive token injection.""" + url = _build_clone_url( + "https://GitLab.MyCompany.com:8443/owner/repo", + GitProvider.GITLAB, + "gltoken123", + explicit_provider=True, + ) + assert url == "https://oauth2:gltoken123@gitlab.mycompany.com:8443/owner/repo" + + def test_build_clone_url_existing_credentials_preserved(self): + """A URL that already carries credentials keeps its own.""" + url = _build_clone_url( + "https://oauth2:embedded@gitlab.mycompany.com/owner/repo", + GitProvider.GITLAB, + "gltoken123", + explicit_provider=True, + ) + assert url == "https://oauth2:embedded@gitlab.mycompany.com/owner/repo" + class TestGetReposContext: """Tests for get_repos_context function.""" @@ -448,6 +499,30 @@ def token_fetcher(name: str) -> str | None: assert "github_token" in fetched_tokens assert "gitlab_token" in fetched_tokens + @patch("subprocess.run") + def test_clone_self_hosted_gitlab_with_token(self, mock_run): + """A self-hosted GitLab instance clones with the token authenticated, + instead of silently sending an unauthenticated request.""" + mock_run.return_value = MagicMock(returncode=0, stderr="") + + def token_fetcher(name: str) -> str | None: + return "gltoken123" if name == "gitlab_token" else None + + with tempfile.TemporaryDirectory() as tmpdir: + repos = [ + RepoSource( + url="https://gitlab.mycompany.com/owner/repo", + provider="gitlab", + ) + ] + clone_repos(repos, Path(tmpdir), token_fetcher=token_fetcher) + + call_args = mock_run.call_args[0][0] + assert any( + "oauth2:gltoken123@gitlab.mycompany.com" in str(arg) + for arg in call_args + ) + @patch("subprocess.run") def test_directory_name_collision(self, mock_run): """Test handling of directory name collisions."""