diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index c2b5b9f98..f855eba5b 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -54,7 +54,7 @@ repos: exclude: "scripts\/tank_cmd.bat|setup\/root_binaries\/tank.bat" # Sort imports and lint. Must run before ruff-format so formatting is final. - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.16.0 + rev: v0.16.5 hooks: - id: ruff-check args: [--fix] diff --git a/python/tank/descriptor/io_descriptor/git.py b/python/tank/descriptor/io_descriptor/git.py index 6b119fafd..59081eb86 100644 --- a/python/tank/descriptor/io_descriptor/git.py +++ b/python/tank/descriptor/io_descriptor/git.py @@ -8,9 +8,12 @@ # agreement to the Shotgun Pipeline Toolkit Source Code License. All rights # not expressly granted therein are reserved by Shotgun Software Inc. import os +import shlex import subprocess import tempfile +import urllib.parse import uuid +from typing import Optional, Union from ... import LogManager from ...util import filesystem, is_windows @@ -34,6 +37,152 @@ def _check_output(*args, **kwargs): return subprocess_check_output(*args, **kwargs) +def _sanitize_url(url: Optional[str]) -> Optional[str]: + """ + Sanitizes a git URL by removing embedded credentials (username, password, or token). + + Examples: + https://ghp_token123@github.com/org/repo.git + -> https://***@github.com/org/repo.git + + https://user:pass@example.com/repo.git + -> https://***@example.com/repo.git + + git@github.com:org/repo.git + -> git@github.com:org/repo.git (no change for SSH URLs) + + :param url: Git URL that may contain embedded credentials + :return: Sanitized URL with credentials replaced by *** + """ + if not url: + return url + + try: + parsed = urllib.parse.urlparse(url) + + # If the URL has a username or password, replace them with *** + if parsed.username or parsed.password: + # Reconstruct the netloc with sanitized credentials + sanitized_netloc = "***@" + parsed.hostname + if parsed.port: + sanitized_netloc += ":" + str(parsed.port) + + # Rebuild the URL with the sanitized netloc + sanitized_url = urllib.parse.urlunparse( + ( + parsed.scheme, + sanitized_netloc, + parsed.path, + parsed.params, + parsed.query, + parsed.fragment, + ) + ) + return sanitized_url + except Exception: + # Best-effort sanitization for malformed URLs that still contain userinfo + if "://" in url: + scheme, rest = url.split("://", 1) + if "@" in rest: + # Only sanitize if '@' appears before any '/' + at_pos = rest.find("@") + slash_pos = rest.find("/") + if slash_pos == -1 or at_pos < slash_pos: + after_at = rest.split("@", 1)[1] + return "%s://***@%s" % (scheme, after_at) + + return url + + +def _sanitize_command(cmd: Union[str, list]) -> Union[str, list]: + """ + Sanitizes a git command (string or list) by replacing credentials in any URLs. + + :param cmd: Command as a string or list of arguments + :return: Sanitized command in the same format as input + """ + if isinstance(cmd, list): + return [_sanitize_url(arg) if isinstance(arg, str) else arg for arg in cmd] + elif isinstance(cmd, str): + # For string commands, we need to be more careful + # Split on spaces but preserve quoted strings + + try: + # Try to parse as shell command + parts = shlex.split(cmd) + sanitized_parts = [_sanitize_url(part) for part in parts] + # Rebuild with proper quoting + return " ".join( + '"%s"' % part if " " in part else part for part in sanitized_parts + ) + except Exception: + # If parsing fails, do simple replacement + # This is a fallback for malformed commands + words = cmd.split() + return " ".join(_sanitize_url(word) for word in words) + return cmd + + +def _sanitize_exception( + exc: SubprocessCalledProcessError, url_to_sanitize: Optional[str] = None +) -> SubprocessCalledProcessError: + """ + Sanitizes a SubprocessCalledProcessError by replacing credentials in the command and output. + + :param exc: SubprocessCalledProcessError exception + :param url_to_sanitize: Optional URL to specifically sanitize (if known) + :return: New exception with sanitized command and output + """ + if not isinstance(exc, SubprocessCalledProcessError): + return exc + + sanitized_cmd = _sanitize_command(exc.cmd) + + # Sanitize the output as well, as it may contain URLs with credentials + sanitized_output = exc.output + if exc.output: + if isinstance(exc.output, bytes): + try: + output_str = exc.output.decode("utf-8") + # Sanitize any URLs in the output + if url_to_sanitize: + output_str = output_str.replace( + url_to_sanitize, _sanitize_url(url_to_sanitize) + ) + # Also try to find and sanitize any URL patterns + import re + + output_str = re.sub( + r"https?://[^@\s]+@[^\s]+", + lambda m: _sanitize_url(m.group(0)), + output_str, + ) + sanitized_output = output_str.encode("utf-8") + except (UnicodeDecodeError, AttributeError): + sanitized_output = exc.output + elif isinstance(exc.output, str): + output_str = exc.output + if url_to_sanitize: + output_str = output_str.replace( + url_to_sanitize, _sanitize_url(url_to_sanitize) + ) + # Also try to find and sanitize any URL patterns + import re + + output_str = re.sub( + r"https?://[^@\s]+@[^\s]+", + lambda m: _sanitize_url(m.group(0)), + output_str, + ) + sanitized_output = output_str + + # Create a new exception with the sanitized command and output + new_exc = SubprocessCalledProcessError( + exc.returncode, sanitized_cmd, output=sanitized_output + ) + return new_exc + + class TankGitError(TankError): """ Errors related to git communication @@ -68,6 +217,18 @@ def __init__(self, descriptor_dict, sg_connection, bundle_type): if self._path.endswith("/") or self._path.endswith("\\"): self._path = self._path[:-1] + def __repr__(self): + """ + Low level representation with sanitized credentials. + """ + class_name = self.__class__.__name__ + # Create a sanitized copy of the descriptor dict with credentials removed + sanitized_dict = self._descriptor_dict.copy() + if "path" in sanitized_dict: + sanitized_dict["path"] = _sanitize_url(sanitized_dict["path"]) + sanitized_uri = self.uri_from_dict(sanitized_dict) + return "<%s %s>" % (class_name, sanitized_uri) + @LogManager.log_timing def _clone_then_execute_git_commands( self, target_path, commands, depth=None, ref=None, is_latest_commit=None @@ -112,8 +273,8 @@ def _clone_then_execute_git_commands( log.debug("Checking that git exists and can be executed...") try: output = _check_output(["git", "--version"]) - except Exception: - log.exception("Unexpected error:") + except Exception as e: + log.exception("Unexpected error: %s: %s", e.__class__.__name__, e) raise TankGitError( "Cannot execute the 'git' command. Please make sure that git is " "installed on your system and that the git executable has been added to the PATH." @@ -144,7 +305,10 @@ def _clone_then_execute_git_commands( # If we can't there's no point doing all of this and we should just use # os.system. if is_windows(): - log.debug("Executing command '%s' using subprocess module." % cmd) + log.debug( + "Executing command '%s' using subprocess module." + % _sanitize_command(cmd) + ) try: # It's important to pass GIT_TERMINAL_PROMPT=0 or the git subprocess will # just hang waiting for credentials to be entered on the missing terminal. @@ -158,12 +322,14 @@ def _clone_then_execute_git_commands( # If that works, we're done and we don't need to use os.system. run_with_os_system = False status = 0 - except SubprocessCalledProcessError: - log.debug("Subprocess call failed.") + except SubprocessCalledProcessError as e: + # Sanitize the exception to remove credentials + sanitized_exc = _sanitize_exception(e, self._path) + log.debug("Subprocess call failed: %s" % sanitized_exc) if run_with_os_system: # Make sure path and repo path are quoted. - log.debug("Executing command '%s' using os.system" % cmd) + log.debug("Executing command '%s' using os.system" % _sanitize_command(cmd)) log.debug( "Note: in a terminal environment, this may prompt for authentication" ) @@ -173,7 +339,7 @@ def _clone_then_execute_git_commands( if status != 0: raise TankGitError( "Error executing git operation. The git command '%s' " - "returned error code %s." % (cmd, status) + "returned error code %s." % (_sanitize_command(cmd), status) ) log.debug("Git clone into '%s' successful." % target_path) @@ -195,9 +361,11 @@ def _clone_then_execute_git_commands( output = output.strip().strip("'") except SubprocessCalledProcessError as e: + # Sanitize the exception to remove any potential credentials + sanitized_exc = _sanitize_exception(e, self._path) raise TankGitError( - f"Error executing GIT operation '{full_command}': {e.output}" - f" (Return code {e.returncode}). " + f"Error executing GIT operation '{_sanitize_command(full_command)}': {sanitized_exc.output}" + f" (Return code {sanitized_exc.returncode}). " " Supported GIT version: 1.9+." ) log.debug("Execution successful. stderr/stdout: '%s'" % output) @@ -253,6 +421,9 @@ def has_remote_access(self): self._tmp_clone_then_execute_git_commands([], depth=1) log.debug("...connection established") except Exception as e: + # Sanitize any credentials that might be in the exception + if isinstance(e, SubprocessCalledProcessError): + e = _sanitize_exception(e, self._path) log.debug("...could not establish connection: %s" % e) can_connect = False return can_connect diff --git a/python/tank/descriptor/io_descriptor/git_branch.py b/python/tank/descriptor/io_descriptor/git_branch.py index bfb779ed1..f0097673d 100644 --- a/python/tank/descriptor/io_descriptor/git_branch.py +++ b/python/tank/descriptor/io_descriptor/git_branch.py @@ -11,8 +11,15 @@ import os from ... import LogManager +from ...util.process import SubprocessCalledProcessError from ..errors import TankDescriptorError -from .git import IODescriptorGit, TankGitError, _check_output +from .git import ( + IODescriptorGit, + TankGitError, + _check_output, + _sanitize_exception, + _sanitize_url, +) log = LogManager.get_logger(__name__) @@ -77,7 +84,11 @@ def __str__(self): Human readable representation """ # git@github.com:manneohrstrom/tk-hiero-publish.git, branch master, commit 12313123 - return "%s, Branch %s, Commit %s" % (self._path, self._branch, self._version) + return "%s, Branch %s, Commit %s" % ( + _sanitize_url(self._path), + self._branch, + self._version, + ) def _get_bundle_cache_path(self, bundle_cache_root): """ @@ -115,8 +126,23 @@ def _is_latest_commit(self, version, branch): log.debug("Checking if the version is pointing to the latest commit...") try: output = _check_output(["git", "ls-remote", self._path, branch]) - except Exception: - log.exception("Unexpected error:") + except SubprocessCalledProcessError as e: + # Sanitize the exception to remove credentials from the command + sanitized_exc = _sanitize_exception(e, self._path) + # Log the sanitized exception manually (don't use log.exception() as it logs + # the original exception from the context) + log.exception( + "Unexpected error:\n%s: %s", + sanitized_exc.__class__.__name__, + sanitized_exc, + ) + # Use exception chaining to attach the sanitized exception + raise TankGitError( + "Cannot execute the 'git' command. Please make sure that git is " + "installed on your system and that the git executable has been added to the PATH." + ) from sanitized_exc + except Exception as e: + log.exception("Unexpected error: %s: %s", e.__class__.__name__, e) raise TankGitError( "Cannot execute the 'git' command. Please make sure that git is " "installed on your system and that the git executable has been added to the PATH." diff --git a/python/tank/descriptor/io_descriptor/git_tag.py b/python/tank/descriptor/io_descriptor/git_tag.py index 130a317b1..96452e5b1 100644 --- a/python/tank/descriptor/io_descriptor/git_tag.py +++ b/python/tank/descriptor/io_descriptor/git_tag.py @@ -12,8 +12,9 @@ import re from ... import LogManager +from ...util.process import SubprocessCalledProcessError from ..errors import TankDescriptorError -from .git import IODescriptorGit +from .git import IODescriptorGit, _sanitize_exception, _sanitize_url log = LogManager.get_logger(__name__) @@ -64,7 +65,7 @@ def __str__(self): Human readable representation """ # git@github.com:manneohrstrom/tk-hiero-publish.git, tag v1.2.3 - return "%s, Tag %s" % (self._path, self._version) + return "%s, Tag %s" % (_sanitize_url(self._path), self._version) def _get_bundle_cache_path(self, bundle_cache_root): """ @@ -142,8 +143,12 @@ def _download_local(self, destination_path): destination_path, [], depth=1, ref=self._version ) except Exception as e: + # Sanitize any credentials that might be in the exception or path + if isinstance(e, SubprocessCalledProcessError): + e = _sanitize_exception(e, self._path) raise TankDescriptorError( - "Could not download %s, tag %s: %s" % (self._path, self._version, e) + "Could not download %s, tag %s: %s" + % (_sanitize_url(self._path), self._version, e) ) def get_latest_version(self, constraint_pattern=None): @@ -220,13 +225,16 @@ def _fetch_tags(self): git_tags.append(m.group(1)) except Exception as e: + # Sanitize any credentials that might be in the exception + if isinstance(e, SubprocessCalledProcessError): + e = _sanitize_exception(e, self._path) raise TankDescriptorError( - "Could not get list of tags for %s: %s" % (self._path, e) + "Could not get list of tags for %s: %s" % (_sanitize_url(self._path), e) ) if len(git_tags) == 0: raise TankDescriptorError( - "Git repository %s doesn't have any tags!" % self._path + "Git repository %s doesn't have any tags!" % _sanitize_url(self._path) ) return git_tags @@ -240,7 +248,7 @@ def _get_latest_version(self): latest_tag = self._find_latest_tag_by_pattern(tags, pattern=None) if latest_tag is None: raise TankDescriptorError( - "Git repository %s doesn't have any tags!" % self._path + "Git repository %s doesn't have any tags!" % _sanitize_url(self._path) ) return latest_tag diff --git a/tests/descriptor_tests/test_git.py b/tests/descriptor_tests/test_git.py index 84ee8d37f..e7bbe2643 100644 --- a/tests/descriptor_tests/test_git.py +++ b/tests/descriptor_tests/test_git.py @@ -244,3 +244,298 @@ def test_fail(self): with self.assertRaises(sgtk.descriptor.errors.TankDescriptorError): self._create_desc(location_dict, True) + + def test_credential_sanitization(self): + """ + Test that credentials in git URLs are properly sanitized in string representations. + """ + from sgtk.descriptor.io_descriptor.git import _sanitize_url + + # Test GitHub PAT token + url_with_pat = ( + "https://ghp_1234567890abcdefghijklmnopqrstuv@github.com/org/repo.git" + ) + sanitized = _sanitize_url(url_with_pat) + self.assertEqual(sanitized, "https://***@github.com/org/repo.git") + self.assertNotIn("ghp_", sanitized) + + # Test username:password format + url_with_userpass = "https://user:password@example.com/repo.git" + sanitized = _sanitize_url(url_with_userpass) + self.assertEqual(sanitized, "https://***@example.com/repo.git") + self.assertNotIn("user", sanitized) + self.assertNotIn("password", sanitized) + + # Test URL with port + url_with_port = "https://token@github.enterprise.com:8443/org/repo.git" + sanitized = _sanitize_url(url_with_port) + self.assertEqual( + sanitized, "https://***@github.enterprise.com:8443/org/repo.git" + ) + self.assertNotIn("token", sanitized) + + # Test SSH URL (should not be modified) + ssh_url = "git@github.com:org/repo.git" + sanitized = _sanitize_url(ssh_url) + self.assertEqual(sanitized, ssh_url) + + # Test local path (should not be modified) + local_path = "/path/to/local/repo.git" + sanitized = _sanitize_url(local_path) + self.assertEqual(sanitized, local_path) + + # Test URL without credentials (should not be modified) + url_no_creds = "https://github.com/org/repo.git" + sanitized = _sanitize_url(url_no_creds) + self.assertEqual(sanitized, url_no_creds) + + # Test None value + sanitized = _sanitize_url(None) + self.assertIsNone(sanitized) + + # Test empty string + sanitized = _sanitize_url("") + self.assertEqual(sanitized, "") + + @skip_if_git_missing + def test_descriptor_repr_sanitization(self): + """ + Test that descriptor __repr__ and __str__ methods sanitize credentials. + """ + # Test git_branch descriptor with PAT token + location_dict_with_token = { + "type": "git_branch", + "path": "https://ghp_secret123@github.com/org/repo.git", + "branch": "master", + "version": "abc1234", + } + + desc = self._create_desc(location_dict_with_token) + + # Check that repr doesn't contain the token + desc_repr = repr(desc) + self.assertNotIn("ghp_secret123", desc_repr) + # The repr may URL-encode *** as %2A%2A%2A + self.assertTrue( + "***" in desc_repr or "%2A%2A%2A" in desc_repr, + "Sanitization marker not found in repr", + ) + + # Check that str doesn't contain the token + # Note: str(desc) uses Descriptor.__str__() which returns "system_name version" + # and doesn't include the URL, so we just verify no credentials leak + desc_str = str(desc) + self.assertNotIn("ghp_secret123", desc_str) + + # Check that the IO descriptor's str representation sanitizes credentials + io_desc_str = str(desc._io_descriptor) + self.assertNotIn("ghp_secret123", io_desc_str) + self.assertIn("***", io_desc_str) + + # Test git descriptor (tag-based) with credentials + location_dict_git = { + "type": "git", + "path": "https://user:pass@example.com/repo.git", + "version": "v1.0.0", + } + + desc_git = self._create_desc(location_dict_git) + + # Check that repr doesn't contain credentials + desc_repr = repr(desc_git) + self.assertNotIn("user", desc_repr) + # Note: "pass" might appear in "sgtk:descriptor:git?pass=..." so we check more carefully + # In the sanitized version, the credentials should be replaced with *** + self.assertTrue( + "***" in desc_repr or "%2A%2A%2A" in desc_repr, + "Sanitization marker not found in repr", + ) + + # Check that the IO descriptor's str representation sanitizes credentials + io_desc_str_git = str(desc_git._io_descriptor) + self.assertNotIn("user", io_desc_str_git) + self.assertNotIn("pass", io_desc_str_git) + self.assertIn("***", io_desc_str_git) + + def test_exception_sanitization(self): + """ + Test that SubprocessCalledProcessError exceptions are sanitized. + """ + from sgtk.descriptor.io_descriptor.git import ( + _sanitize_command, + _sanitize_exception, + ) + from tank.util.process import SubprocessCalledProcessError + + # Test sanitization of command list + cmd_list = [ + "git", + "ls-remote", + "https://ghp_secret123@github.com/org/repo.git", + "master", + ] + sanitized_list = _sanitize_command(cmd_list) + self.assertNotIn("ghp_secret123", str(sanitized_list)) + self.assertIn("***", str(sanitized_list)) + + # Test sanitization of command string + cmd_string = 'git clone "https://user:pass@example.com/repo.git" /tmp/repo' + sanitized_string = _sanitize_command(cmd_string) + self.assertNotIn("user", sanitized_string) + self.assertNotIn("pass", sanitized_string) + self.assertIn("***", sanitized_string) + + # Test sanitization of SubprocessCalledProcessError with list command + exc = SubprocessCalledProcessError(128, cmd_list, output=b"some error") + sanitized_exc = _sanitize_exception(exc) + exc_str = str(sanitized_exc) + self.assertNotIn("ghp_secret123", exc_str) + self.assertIn("***", exc_str) + self.assertEqual(sanitized_exc.returncode, 128) + + # Test sanitization of SubprocessCalledProcessError with string command + exc_str_cmd = SubprocessCalledProcessError(128, cmd_string, output=b"error") + sanitized_exc_str = _sanitize_exception(exc_str_cmd) + exc_str_repr = str(sanitized_exc_str) + self.assertNotIn("user", exc_str_repr) + self.assertNotIn("pass", exc_str_repr) + self.assertIn("***", exc_str_repr) + + def test_exception_chain_sanitization(self): + """ + Test that exception __cause__ and __context__ are sanitized to prevent + credential leaks in exception chains. + """ + from sgtk.descriptor.io_descriptor.git import _sanitize_exception + from tank.util.process import SubprocessCalledProcessError + + # Create an exception with credentials in the command + cmd_with_creds = [ + "git", + "ls-remote", + "https://ghp_secret123@github.com/org/repo.git", + "master", + ] + original_exc = SubprocessCalledProcessError(128, cmd_with_creds) + + # Sanitize it + sanitized_exc = _sanitize_exception(original_exc) + + # Verify the sanitized exception doesn't contain credentials + self.assertNotIn("ghp_secret123", str(sanitized_exc)) + self.assertIn("***", str(sanitized_exc)) + + # Verify __cause__ is sanitized (if set) + if sanitized_exc.__cause__ is not None: + self.assertNotIn("ghp_secret123", str(sanitized_exc.__cause__)) + + # Verify __context__ is sanitized (if set) + if sanitized_exc.__context__ is not None: + self.assertNotIn("ghp_secret123", str(sanitized_exc.__context__)) + + @skip_if_git_missing + def test_git_branch_error_handling_sanitizes_credentials(self): + """ + Integration test: Verify that when git_branch descriptor fails with + credentials in the URL, the error and exception chain are sanitized. + """ + import logging + from io import StringIO + + # Create a descriptor with credentials that will fail + location_dict = { + "type": "git_branch", + "path": "https://ghp_secret123@github.com/fake/nonexistent.git", + "branch": "master", + "version": "abc1234", + } + + # Set up log capture to check what gets logged + log_stream = StringIO() + handler = logging.StreamHandler(log_stream) + handler.setLevel(logging.DEBUG) + logger = logging.getLogger("sgtk.core.descriptor.io_descriptor.git_branch") + original_level = logger.level + logger.setLevel(logging.DEBUG) + logger.addHandler(handler) + + try: + desc = self._create_desc(location_dict) + + # Try to check if it's the latest commit - this should fail + # because the repo doesn't exist + try: + desc._is_latest_commit("abc1234", "master") + self.fail("Expected TankGitError to be raised") + except Exception as e: + # Verify the exception message doesn't contain credentials + exc_str = str(e) + self.assertNotIn("ghp_secret123", exc_str) + + # Check the entire exception chain + current_exc = e + while current_exc is not None: + self.assertNotIn( + "ghp_secret123", + str(current_exc), + "Credentials found in exception chain: %s" % type(current_exc), + ) + # Check both __cause__ and __context__ + if current_exc.__cause__ is not None: + current_exc = current_exc.__cause__ + elif current_exc.__context__ is not None: + current_exc = current_exc.__context__ + else: + break + + # Check that nothing was logged with credentials + log_contents = log_stream.getvalue() + self.assertNotIn( + "ghp_secret123", + log_contents, + "Credentials found in log output:\n%s" % log_contents, + ) + + finally: + logger.removeHandler(handler) + logger.setLevel(original_level) + + def test_git_tag_exception_sanitization(self): + """ + Test that git_tag.py properly sanitizes exceptions in error handlers. + """ + from sgtk.descriptor.io_descriptor.git_tag import IODescriptorGitTag + from tank.descriptor.errors import TankDescriptorError + + # Create a git tag descriptor with credentials + location_dict = { + "type": "git", + "path": "https://token123@github.com/fake/nonexistent.git", + "version": "v1.0.0", + } + + desc = IODescriptorGitTag(location_dict, None, None) + + # Mock _tmp_clone_then_execute_git_commands to raise an error + from unittest.mock import patch + + from tank.util.process import SubprocessCalledProcessError + + cmd_with_creds = [ + "git", + "clone", + "https://token123@github.com/fake/nonexistent.git", + ] + mock_exc = SubprocessCalledProcessError(128, cmd_with_creds) + + with patch.object( + desc, "_tmp_clone_then_execute_git_commands", side_effect=mock_exc + ): + try: + desc._fetch_tags() + self.fail("Expected TankDescriptorError to be raised") + except TankDescriptorError as e: + # Verify credentials are not in the error message + error_msg = str(e) + self.assertNotIn("token123", error_msg) + self.assertIn("***", error_msg)