fix(workspace): honor explicit provider host when injecting git clone tokens

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

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."""