fix: improve subprocess exception in git_utils and add tests (#5812)

* fix: improve subprocess exception in git_utils and add tests

* fix: add generic exception handling for subprocess in git_utils

* fix: handle non-standard exception constructors in catch-all

Author: lucasjia-aws

src/sagemaker/git_utils.py

@@ -332,6 +332,22 @@ def _clone_command_for_codecommit_https(git_config, dest_dir):
     _run_clone_command(updated_url, dest_dir)
 
 
+def _redact_credentials_from_url(url):
+    """Redact credentials embedded in an HTTPS Git URL.
+
+    Replaces any username, password, or token embedded before the '@' in an
+    HTTPS URL with a placeholder so that credentials are never exposed in
+    logs or exception messages.
+
+    Args:
+        url (str): The Git repository URL that may contain embedded credentials.
+
+    Returns:
+        str: The URL with credentials replaced by '<credentials-redacted>'.
+    """
+    return re.sub(r"(https://)([^@]+)@", r"\1<credentials-redacted>@", url)
+
+
 def _run_clone_command(repo_url, dest_dir):
     """Run the 'git clone' command with the repo url and the directory to clone the repo into.
 
@@ -345,7 +361,24 @@ def _run_clone_command(repo_url, dest_dir):
     my_env = os.environ.copy()
     if repo_url.startswith("https://"):
         my_env["GIT_TERMINAL_PROMPT"] = "0"
-        subprocess.check_call(["git", "clone", repo_url, dest_dir], env=my_env)
+        try:
+            subprocess.check_call(["git", "clone", repo_url, dest_dir], env=my_env)
+        except subprocess.CalledProcessError as e:
+            # Re-raise with credentials redacted from the command to prevent
+            # plaintext tokens/passwords from leaking into logs or tracebacks.
+            safe_url = _redact_credentials_from_url(repo_url)
+            raise subprocess.CalledProcessError(
+                e.returncode,
+                ["git", "clone", safe_url, dest_dir],
+                output=e.output,
+                stderr=e.stderr,
+            ) from None
+        except Exception as e:
+            safe_url = _redact_credentials_from_url(repo_url)
+            try:
+                raise type(e)(str(e).replace(repo_url, safe_url)) from None
+            except TypeError:
+                raise RuntimeError(str(e).replace(repo_url, safe_url)) from None
     elif repo_url.startswith("git@") or repo_url.startswith("ssh://"):
         try:
             with tempfile.TemporaryDirectory() as tmp_dir:

tests/unit/test_git_utils.py

@@ -704,3 +704,108 @@ def test_sanitize_git_url_comprehensive_attack_scenarios(self):
                     "Invalid characters in hostname",
                 ]
             )
+
+
+class TestCredentialRedaction:
+    """Test cases for credential redaction in clone error handling."""
+
+    def test_redact_token_from_url(self):
+        """Test that a token embedded in an HTTPS URL is redacted."""
+        url = "https://ghp_SuperSecretToken123@github.com/user/repo.git"
+        result = git_utils._redact_credentials_from_url(url)
+        assert "ghp_SuperSecretToken123" not in result
+        assert result == "https://<credentials-redacted>@github.com/user/repo.git"
+
+    def test_redact_username_password_from_url(self):
+        """Test that username:password embedded in an HTTPS URL is redacted."""
+        url = "https://myuser:mypassword@github.com/user/repo.git"
+        result = git_utils._redact_credentials_from_url(url)
+        assert "myuser" not in result
+        assert "mypassword" not in result
+        assert result == "https://<credentials-redacted>@github.com/user/repo.git"
+
+    def test_redact_url_encoded_password(self):
+        """Test that URL-encoded credentials are redacted."""
+        url = "https://user:p%40ss%20word@git-codecommit.us-east-1.amazonaws.com/v1/repos/myrepo"
+        result = git_utils._redact_credentials_from_url(url)
+        assert "p%40ss%20word" not in result
+        assert "<credentials-redacted>" in result
+
+    def test_no_redaction_without_credentials(self):
+        """Test that URLs without credentials are unchanged."""
+        url = "https://github.com/user/repo.git"
+        result = git_utils._redact_credentials_from_url(url)
+        assert result == url
+
+    def test_no_redaction_for_ssh_url(self):
+        """Test that SSH URLs are not affected by redaction."""
+        url = "git@github.com:user/repo.git"
+        result = git_utils._redact_credentials_from_url(url)
+        assert result == url
+
+    def test_clone_failure_redacts_token(self):
+        """Test that CalledProcessError from a failed clone does not contain the token."""
+        token_url = "https://ghp_secret123@github.com/user/repo.git"
+        with patch(
+            "subprocess.check_call",
+            side_effect=subprocess.CalledProcessError(
+                128, ["git", "clone", token_url, "/tmp/dest"]
+            ),
+        ):
+            with pytest.raises(subprocess.CalledProcessError) as exc_info:
+                git_utils._run_clone_command(token_url, "/tmp/dest")
+
+            assert "ghp_secret123" not in str(exc_info.value)
+            assert "ghp_secret123" not in str(exc_info.value.cmd)
+            assert "<credentials-redacted>" in str(exc_info.value.cmd)
+
+    def test_clone_failure_redacts_username_password(self):
+        """Test that CalledProcessError from a failed clone does not contain username/password."""
+        cred_url = "https://admin:hunter2@github.com/org/repo.git"
+        with patch(
+            "subprocess.check_call",
+            side_effect=subprocess.CalledProcessError(
+                128, ["git", "clone", cred_url, "/tmp/dest"]
+            ),
+        ):
+            with pytest.raises(subprocess.CalledProcessError) as exc_info:
+                git_utils._run_clone_command(cred_url, "/tmp/dest")
+
+            assert "admin" not in str(exc_info.value.cmd)
+            assert "hunter2" not in str(exc_info.value.cmd)
+            assert "<credentials-redacted>" in str(exc_info.value.cmd)
+
+    def test_clone_failure_redacts_codecommit_credentials(self):
+        """Test that CodeCommit HTTPS credentials are redacted on failure."""
+        cc_url = "https://user:pass@git-codecommit.us-east-1.amazonaws.com/v1/repos/myrepo"
+        with patch(
+            "subprocess.check_call",
+            side_effect=subprocess.CalledProcessError(
+                128, ["git", "clone", cc_url, "/tmp/dest"]
+            ),
+        ):
+            with pytest.raises(subprocess.CalledProcessError) as exc_info:
+                git_utils._run_clone_command(cc_url, "/tmp/dest")
+
+            assert "user:pass" not in str(exc_info.value.cmd)
+            assert "<credentials-redacted>" in str(exc_info.value.cmd)
+
+    def test_clone_failure_suppresses_exception_chain(self):
+        """Test that the original exception chain is suppressed (from None)."""
+        token_url = "https://ghp_secret@github.com/user/repo.git"
+        with patch(
+            "subprocess.check_call",
+            side_effect=subprocess.CalledProcessError(
+                128, ["git", "clone", token_url, "/tmp/dest"]
+            ),
+        ):
+            with pytest.raises(subprocess.CalledProcessError) as exc_info:
+                git_utils._run_clone_command(token_url, "/tmp/dest")
+
+            assert exc_info.value.__cause__ is None
+
+    def test_clone_success_no_exception(self):
+        """Test that successful clone does not raise."""
+        url = "https://ghp_token@github.com/user/repo.git"
+        with patch("subprocess.check_call"):
+            git_utils._run_clone_command(url, "/tmp/dest")