Merge pull request #736 from NVIDIA/yashraj/isolate-http-mcp-git-credentials

fix(mcp): isolate Git credentials for HTTP scans

Author: rng1995

README.md

@@ -476,6 +476,12 @@ claude mcp add skillspector -- skillspector mcp
 > - Local paths and `file://` URLs are **automatically rejected** over HTTP to
 >   prevent unauthenticated callers from reading arbitrary host files. Only
 >   remote Git and `.zip` URLs are accepted.
+>
+> HTTP scans accept only credential-free `https://` targets. Git scans never use
+> the server's Git login or Git config. If a proxy or custom CA is needed, set
+> `HTTPS_PROXY`, `NO_PROXY` or `GIT_SSL_CAINFO` in the server environment.
+> Use stdio or the CLI for private repositories. Git redirects are rejected;
+> use the current repository URL after a rename or transfer.
 
 ## Vulnerability Patterns
 

src/skillspector/input_handler.py

@@ -767,7 +767,10 @@ class InputHandler:
     Normalizes all inputs to a local directory path for scanning.
     """
 
-    def __init__(self, transitive_budget: object | None = None) -> None:
+    def __init__(
+        self, transitive_budget: object | None = None, *, allow_git_credentials: bool = True
+    ) -> None:
+        self._allow_git_credentials = allow_git_credentials
         self._temp_dir: Path | None = None
         self._transitive_budget = transitive_budget
         self.primary_file_path: str | None = None
@@ -1129,23 +1132,99 @@ def _resolve_tree_ref(self, repository_url: str, segments: list[str]) -> tuple[s
             f"{repository_url} ({'/'.join(segments)})"
         )
 
+    def _git_invocation(self, url: str) -> tuple[list[str], dict[str, str] | None, str]:
+        """Build a Git policy for both ref discovery and cloning.
+
+        Remote MCP callers must not borrow the server's SSH keys, helpers,
+        authorization headers, URL rewrites, client certificates, or netrc.
+        Trusted local callers retain the operator's configured Git identity.
+        """
+        if self._allow_git_credentials:
+            return ["git"], None, url
+        parsed = urlparse(url)
+        if parsed.scheme != "https" or parsed.username is not None or parsed.password is not None:
+            raise ValueError("Untrusted Git scans require an unauthenticated HTTPS URL")
+        if parsed.hostname in {"gitlab.com", "bitbucket.org"}:
+            # Browser URLs need the Git endpoint without following a redirect
+            # whose destination has not passed the host/IP checks.
+            path = parsed.path.rstrip("/")
+            if path and not path.endswith(".git"):
+                path += ".git"
+            url = parsed._replace(path=path).geturl()
+        isolated_home = self._get_temp_dir() / "git-home"
+        isolated_home.mkdir(mode=0o700, exist_ok=True)
+        env = {
+            key: value
+            for key, value in os.environ.items()
+            if key.upper()
+            in {
+                "PATH",
+                "SYSTEMROOT",
+                "WINDIR",
+                "TEMP",
+                "TMP",
+                "TMPDIR",
+                "HTTP_PROXY",
+                "HTTPS_PROXY",
+                "ALL_PROXY",
+                "NO_PROXY",
+                "GIT_SSL_CAINFO",
+                "GIT_SSL_CAPATH",
+                "SSL_CERT_FILE",
+                "SSL_CERT_DIR",
+                "CURL_CA_BUNDLE",
+            }
+        }
+        env.update(
+            HOME=str(isolated_home),
+            USERPROFILE=str(isolated_home),
+            XDG_CONFIG_HOME=str(isolated_home),
+            GIT_CONFIG_GLOBAL=os.devnull,
+            GIT_CONFIG_SYSTEM=os.devnull,
+            GIT_CONFIG_NOSYSTEM="1",
+            GIT_TERMINAL_PROMPT="0",
+            GIT_ALLOW_PROTOCOL="https",
+            GIT_CEILING_DIRECTORIES=str(isolated_home.parent),
+        )
+        return (
+            [
+                "git",
+                "-C",
+                str(isolated_home),
+                "-c",
+                "credential.helper=",
+                "-c",
+                "core.askPass=",
+                "-c",
+                f"core.hooksPath={os.devnull}",
+                "-c",
+                "http.extraHeader=",
+                "-c",
+                "http.followRedirects=false",
+            ],
+            env,
+            url,
+        )
+
     def _list_remote_refs(self, repository_url: str) -> set[str]:
         """Return the branch/tag names advertised by the remote repository.
 
         Bounded by the ingest deadline; the host allowlist and private-IP
         checks from URL validation apply.
         """
+        git_argv, git_env, repository_url = self._git_invocation(repository_url)
         self._validate_url_host(repository_url, ALLOWED_GIT_HOSTS)
         deadline = self._deadline()
         self._check_deadline(deadline, "git")
         timeout = max(1.0, deadline - monotonic())
         try:
             process = subprocess.run(
-                ["git", "ls-remote", repository_url],
+                [*git_argv, "ls-remote", repository_url],
                 stdout=subprocess.PIPE,
                 stderr=subprocess.DEVNULL,
                 timeout=timeout,
                 check=False,
+                **({"env": git_env} if git_env is not None else {}),
             )
         except subprocess.TimeoutExpired as exc:
             raise IngestLimitExceededError("Git ref listing exceeded its time limit") from exc
@@ -1206,6 +1285,7 @@ def _clone_git(self, url: str, *, branch: str | None = None) -> Path:
             self._truncate("byte_budget_exhausted", "git")
         if remaining_artifacts is not None and remaining_artifacts <= 0:
             self._truncate("artifact_budget_exhausted", "git")
+        git_argv, git_env, url = self._git_invocation(url)
         self._validate_url_host(url, ALLOWED_GIT_HOSTS)
         deadline = self._deadline()
         self._check_deadline(deadline, "git")
@@ -1225,6 +1305,7 @@ def _clone_git(self, url: str, *, branch: str | None = None) -> Path:
             clone_command[6:6] = ["--branch", branch]
         if remaining_bytes is not None:
             clone_command.insert(6, f"--filter=blob:limit={remaining_bytes}")
+        clone_command[:1] = git_argv
         process: subprocess.Popen[bytes] | None = None
         final_measurement: _TreeMeasurement | None = None
         try:
@@ -1233,6 +1314,7 @@ def _clone_git(self, url: str, *, branch: str | None = None) -> Path:
                 stdout=subprocess.DEVNULL,
                 stderr=subprocess.DEVNULL,
                 shell=False,
+                **({"env": git_env} if git_env is not None else {}),
             )
             while True:
                 self._check_deadline(deadline, "git")

src/skillspector/mcp_server.py

@@ -31,6 +31,7 @@
 from ipaddress import ip_address
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
+from urllib.parse import urlparse
 
 from skillspector import __version__
 from skillspector.cleanup import TempDirTracker, cleanup_result
@@ -116,7 +117,7 @@ async def run_scan(
             :data:`VALID_FORMATS`.
         allow_local_targets: Whether local filesystem targets are allowed.
             HTTP MCP calls set this to ``False`` so routable servers do not
-            accept caller-controlled local paths.
+            accept caller-controlled local paths or use ambient Git credentials.
         yara_rules_dir: Optional directory of additional YARA rules.
 
     Returns:
@@ -133,12 +134,20 @@ async def run_scan(
         local_yara_rules = yara_rules_dir is not None and _is_local_target(yara_rules_dir)
         if local_target or local_yara_rules:
             raise ValueError("local targets are disabled for this MCP transport")
+        parsed_target = urlparse(target.strip())
+        if (
+            parsed_target.scheme != "https"
+            or parsed_target.username is not None
+            or parsed_target.password is not None
+        ):
+            raise ValueError("this MCP transport requires an unauthenticated HTTPS target")
 
     llm_preflight_available, _ = is_llm_available()
     llm_enabled = use_llm and llm_preflight_available
 
     state: dict[str, Any] = {
         "input_path": target,
+        "allow_git_credentials": allow_local_targets,
         "output_format": output_format,
         "use_llm": llm_enabled,
         "llm_requested": use_llm,

src/skillspector/nodes/resolve_input.py

@@ -56,7 +56,10 @@ def resolve_input(state: SkillspectorState) -> dict[str, object]:
     workflow_budget = ensure_workflow_resource_budget(state)
 
     if input_path and isinstance(input_path, str) and input_path.strip():
-        handler = InputHandler(transitive_budget=workflow_budget)
+        handler = InputHandler(
+            transitive_budget=workflow_budget,
+            allow_git_credentials=state.get("allow_git_credentials", True),
+        )
         try:
             resolved, source_type = handler.resolve(input_path.strip())
             temp_dir = handler.temp_dir_for_cleanup()

src/skillspector/state.py

@@ -235,6 +235,8 @@ class SkillspectorState(TypedDict, total=False):
 
     # Input: resolve_input node consumes input_path or skill_path, sets skill_path
     input_path: str | None
+    # Only trusted local transports may use the operator's Git credentials.
+    allow_git_credentials: bool
     skill_path: str | None
     # Relative identity of a wrapped/downloaded explicit file, distinct from passive assets.
     primary_file_path: str | None

tests/nodes/test_resolve_input.py

@@ -91,7 +91,10 @@ def test_workflow_budget_starts_before_input_materialization(
     class CapturingHandler:
         primary_file_path = None
 
-        def __init__(self, transitive_budget: object | None = None) -> None:
+        def __init__(
+            self, transitive_budget: object | None = None, *, allow_git_credentials: bool = True
+        ) -> None:
+            assert allow_git_credentials is True
             assert transitive_budget is not None
             assert getattr(transitive_budget, "started_at", None) is not None
             captured.append(transitive_budget)
@@ -116,7 +119,10 @@ def test_transitive_truncation_is_typed_sanitized_and_cleaned(
     cleaned: list[bool] = []
 
     class TruncatedHandler:
-        def __init__(self, transitive_budget: object | None = None) -> None:
+        def __init__(
+            self, transitive_budget: object | None = None, *, allow_git_credentials: bool = True
+        ) -> None:
+            assert allow_git_credentials is True
             assert transitive_budget is not None
 
         def resolve(self, _input_path: str) -> tuple[Path, str]:
@@ -323,3 +329,13 @@ def wait(self, timeout: float | None = None) -> int:
     assert [clone.terminated for clone in InterruptedClone.instances] == [True]
     assert created
     assert [path for path in created if path.exists()] == []
+
+
+def test_resolve_input_preserves_anonymous_git_policy() -> None:
+    with pytest.raises(ValueError, match="unauthenticated HTTPS"):
+        resolve_input(
+            {
+                "input_path": "git@github.com:org/private.git",
+                "allow_git_credentials": False,
+            }
+        )