fix: sanitize git clone repo input url

For commit: aws/sagemaker-python-sdk-staging@ed143b7

Author: zhaoqizqwang

sagemaker-core/src/sagemaker/core/git_utils.py

@@ -20,7 +20,69 @@
 import warnings
 import six
 from six.moves import urllib
+import re
+from pathlib import Path
+from urllib.parse import urlparse
+
+def _sanitize_git_url(repo_url):
+    """Sanitize Git repository URL to prevent URL injection attacks.
+
+    Args:
+        repo_url (str): The Git repository URL to sanitize
 
+    Returns:
+        str: The sanitized URL
+
+    Raises:
+        ValueError: If the URL contains suspicious patterns that could indicate injection
+    """
+    at_count = repo_url.count("@")
+
+    if repo_url.startswith("git@"):
+        # git@ format requires exactly one @
+        if at_count != 1:
+            raise ValueError("Invalid SSH URL format: git@ URLs must have exactly one @ symbol")
+    elif repo_url.startswith("ssh://"):
+        # ssh:// format can have 0 or 1 @ symbols
+        if at_count > 1:
+            raise ValueError("Invalid SSH URL format: multiple @ symbols detected")
+    elif repo_url.startswith("https://") or repo_url.startswith("http://"):
+        # HTTPS format allows 0 or 1 @ symbols
+        if at_count > 1:
+            raise ValueError("Invalid HTTPS URL format: multiple @ symbols detected")
+
+        # Check for invalid characters in the URL before parsing
+        # These characters should not appear in legitimate URLs
+        invalid_chars = ["<", ">", "[", "]", "{", "}", "\\", "^", "`", "|"]
+        for char in invalid_chars:
+            if char in repo_url:
+                raise ValueError("Invalid characters in hostname")
+
+        try:
+            parsed = urlparse(repo_url)
+
+            # Check for suspicious characters in hostname that could indicate injection
+            if parsed.hostname:
+                # Check for URL-encoded characters that might be used for obfuscation
+                suspicious_patterns = ["%25", "%40", "%2F", "%3A"]  # encoded %, @, /, :
+                for pattern in suspicious_patterns:
+                    if pattern in parsed.hostname.lower():
+                        raise ValueError(f"Suspicious URL encoding detected in hostname: {pattern}")
+
+                # Validate that the hostname looks legitimate
+                if not re.match(r"^[a-zA-Z0-9.-]+$", parsed.hostname):
+                    raise ValueError("Invalid characters in hostname")
+
+        except Exception as e:
+            if isinstance(e, ValueError):
+                raise
+            raise ValueError(f"Failed to parse URL: {str(e)}")
+    else:
+        raise ValueError(
+            "Unsupported URL scheme: only https://, http://, git@, and ssh:// are allowed"
+        )
+
+    return repo_url
 
 def git_clone_repo(git_config, entry_point, source_dir=None, dependencies=None):
     """Git clone repo containing the training code and serving code.
@@ -87,6 +149,10 @@ def git_clone_repo(git_config, entry_point, source_dir=None, dependencies=None):
     if entry_point is None:
         raise ValueError("Please provide an entry point.")
     _validate_git_config(git_config)
+
+    # SECURITY: Sanitize the repository URL to prevent injection attacks
+    git_config["repo"] = _sanitize_git_url(git_config["repo"])
+
     dest_dir = tempfile.mkdtemp()
     _generate_and_run_clone_command(git_config, dest_dir)
 

sagemaker-core/tests/unit/test_git_utils.py

@@ -14,6 +14,7 @@
 
 import pytest
 
+from sagemaker.core import git_utils
 from sagemaker.core.git_utils import _validate_git_config
 
 
@@ -135,3 +136,200 @@ def test_validate_git_config_repo_none():
 
     with pytest.raises(ValueError, match="'repo' must be a string"):
         _validate_git_config(git_config)
+
+
+class TestGitUrlSanitization:
+    """Test cases for Git URL sanitization to prevent injection attacks."""
+
+    def test_sanitize_git_url_valid_https_urls(self):
+        """Test that valid HTTPS URLs pass sanitization."""
+        valid_urls = [
+            "https://github.com/user/repo.git",
+            "https://gitlab.com/user/repo.git",
+            "https://token@github.com/user/repo.git",
+            "https://user:pass@github.com/user/repo.git",
+            "http://internal-git.company.com/repo.git",
+        ]
+
+        for url in valid_urls:
+            result = git_utils._sanitize_git_url(url)
+            assert result == url
+
+    def test_sanitize_git_url_valid_ssh_urls(self):
+        """Test that valid SSH URLs pass sanitization."""
+        valid_urls = [
+            "git@github.com:user/repo.git",
+            "git@gitlab.com:user/repo.git",
+            "ssh://git@github.com/user/repo.git",
+            "ssh://git-codecommit.us-west-2.amazonaws.com/v1/repos/test-repo/",
+            "git@internal-git.company.com:repo.git",
+        ]
+
+        for url in valid_urls:
+            result = git_utils._sanitize_git_url(url)
+            assert result == url
+
+    def test_sanitize_git_url_blocks_multiple_at_https(self):
+        """Test that HTTPS URLs with multiple @ symbols are blocked."""
+        malicious_urls = [
+            "https://user@attacker.com@github.com/repo.git",
+            "https://token@evil.com@gitlab.com/user/repo.git",
+            "https://a@b@c@github.com/repo.git",
+            "https://user@malicious-host@github.com/legit/repo.git",
+        ]
+
+        for url in malicious_urls:
+            with pytest.raises(ValueError) as error:
+                git_utils._sanitize_git_url(url)
+            assert "multiple @ symbols detected" in str(error.value)
+
+    def test_sanitize_git_url_blocks_multiple_at_ssh(self):
+        """Test that SSH URLs with multiple @ symbols are blocked."""
+        malicious_urls = [
+            "git@attacker.com@github.com:repo.git",
+            "git@evil@gitlab.com:user/repo.git",
+            "ssh://git@malicious@github.com/repo.git",
+            "git@a@b@c:repo.git",
+        ]
+
+        for url in malicious_urls:
+            with pytest.raises(ValueError) as error:
+                git_utils._sanitize_git_url(url)
+            assert any(
+                phrase in str(error.value)
+                for phrase in ["multiple @ symbols detected", "exactly one @ symbol"]
+            )
+
+    def test_sanitize_git_url_blocks_invalid_schemes_and_git_at_format(self):
+        """Test that invalid schemes and git@ format violations are blocked."""
+        unsupported_scheme_urls = [
+            "git-github.com:user/repo.git",
+        ]
+
+        for url in unsupported_scheme_urls:
+            with pytest.raises(ValueError) as error:
+                git_utils._sanitize_git_url(url)
+            assert "Unsupported URL scheme" in str(error.value)
+
+        invalid_git_at_urls = [
+            "git@github.com@evil.com:repo.git",
+        ]
+
+        for url in invalid_git_at_urls:
+            with pytest.raises(ValueError) as error:
+                git_utils._sanitize_git_url(url)
+            assert "exactly one @ symbol" in str(error.value)
+
+    def test_sanitize_git_url_blocks_url_encoding_obfuscation(self):
+        """Test that URL-encoded obfuscation attempts are blocked."""
+        obfuscated_urls = [
+            "https://github.com%25evil.com/repo.git",
+            "https://user@github.com%40attacker.com/repo.git",
+            "https://github.com%2Fevil.com/repo.git",
+            "https://github.com%3Aevil.com/repo.git",
+        ]
+
+        for url in obfuscated_urls:
+            with pytest.raises(ValueError) as error:
+                git_utils._sanitize_git_url(url)
+            assert any(
+                phrase in str(error.value)
+                for phrase in ["Suspicious URL encoding detected", "Invalid characters in hostname"]
+            )
+
+    def test_sanitize_git_url_blocks_invalid_hostname_chars(self):
+        """Test that hostnames with invalid characters are blocked."""
+        invalid_urls = [
+            "https://github<script>.com/repo.git",
+            "https://github>.com/repo.git",
+            "https://github[].com/repo.git",
+            "https://github{}.com/repo.git",
+        ]
+
+        for url in invalid_urls:
+            with pytest.raises(ValueError) as error:
+                git_utils._sanitize_git_url(url)
+            assert any(
+                phrase in str(error.value)
+                for phrase in [
+                    "Invalid characters in hostname",
+                    "Failed to parse URL",
+                    "does not appear to be an IPv4 or IPv6 address",
+                ]
+            )
+
+    def test_sanitize_git_url_blocks_unsupported_schemes(self):
+        """Test that unsupported URL schemes are blocked."""
+        unsupported_urls = [
+            "ftp://github.com/repo.git",
+            "file:///local/repo.git",
+            "javascript:alert('xss')",
+            "data:text/html,<script>alert('xss')</script>",
+        ]
+
+        for url in unsupported_urls:
+            with pytest.raises(ValueError) as error:
+                git_utils._sanitize_git_url(url)
+            assert "Unsupported URL scheme" in str(error.value)
+
+    def test_git_clone_repo_blocks_malicious_https_url(self):
+        """Test that git_clone_repo blocks malicious HTTPS URLs."""
+        malicious_git_config = {
+            "repo": "https://user@attacker.com@github.com/legit/repo.git",
+            "branch": "main",
+        }
+        entry_point = "train.py"
+
+        with pytest.raises(ValueError) as error:
+            git_utils.git_clone_repo(malicious_git_config, entry_point)
+        assert "multiple @ symbols detected" in str(error.value)
+
+    def test_git_clone_repo_blocks_malicious_ssh_url(self):
+        """Test that git_clone_repo blocks malicious SSH URLs."""
+        malicious_git_config = {
+            "repo": "git@OBVIOUS@github.com:sage-maker/temp-sev2.git",
+            "branch": "main",
+        }
+        entry_point = "train.py"
+
+        with pytest.raises(ValueError) as error:
+            git_utils.git_clone_repo(malicious_git_config, entry_point)
+        assert "exactly one @ symbol" in str(error.value)
+
+    def test_git_clone_repo_blocks_url_encoded_attack(self):
+        """Test that git_clone_repo blocks URL-encoded attacks."""
+        malicious_git_config = {
+            "repo": "https://github.com%40attacker.com/repo.git",
+            "branch": "main",
+        }
+        entry_point = "train.py"
+
+        with pytest.raises(ValueError) as error:
+            git_utils.git_clone_repo(malicious_git_config, entry_point)
+        assert "Suspicious URL encoding detected" in str(error.value)
+
+    def test_sanitize_git_url_comprehensive_attack_scenarios(self):
+        attack_scenarios = [
+            "https://USER@YOUR_NGROK_OR_LOCALHOST/malicious.git@github.com%25legit%25repo.git",
+            "https://user@malicious-host@github.com/legit/repo.git",
+            "git@attacker.com@github.com:user/repo.git",
+            "ssh://git@evil.com@github.com/repo.git",
+            "https://github.com%40evil.com/repo.git",
+            "https://user@github.com%2Fevil.com/repo.git",
+        ]
+
+        entry_point = "train.py"
+
+        for malicious_url in attack_scenarios:
+            git_config = {"repo": malicious_url}
+            with pytest.raises(ValueError) as error:
+                git_utils.git_clone_repo(git_config, entry_point)
+            assert any(
+                phrase in str(error.value)
+                for phrase in [
+                    "multiple @ symbols detected",
+                    "exactly one @ symbol",
+                    "Suspicious URL encoding detected",
+                    "Invalid characters in hostname",
+                ]
+            )