Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 44 additions & 9 deletions python/tank/descriptor/io_descriptor/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,18 +8,20 @@
# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights
# not expressly granted therein are reserved by Shotgun Software Inc.

import os
import contextlib
import os
import sys
import urllib.parse

from .. import constants
from tank_vendor import yaml

from ... import LogManager
from ...util import filesystem, sgre as re
from ...util.version import is_version_newer
from ...util import filesystem
from ...util import sgre as re
from ...util.version import is_version_newer, is_version_newer_or_equal
from .. import constants
from ..errors import TankDescriptorError, TankMissingManifestError

from tank_vendor import yaml

log = LogManager.get_logger(__name__)


Expand Down Expand Up @@ -247,8 +249,16 @@ def _find_latest_tag_by_pattern(self, version_numbers, pattern):
# iterate over versions in list and find latest
latest_version = None
for version_number in version_numbers:
if is_version_newer(version_number, latest_version):
latest_version = version_number
try:
if is_version_newer(version_number, latest_version):
latest_version = version_number
except Exception as e:
# Handle malformed version tags gracefully
log.debug(
"Skipping version '%s' due to parsing error: %s"
% (version_number, e)
)
continue
return latest_version

# now put all version number strings which match the form
Expand Down Expand Up @@ -369,6 +379,31 @@ def _get_locally_cached_versions(self):

return all_versions

def _check_minimum_python_version(self, manifest_data):
"""
Checks if the current Python version meets the minimum required version
specified in the manifest data.
:param manifest_data: Dictionary containing bundle manifest/info.yml data
:returns: True if current Python version is compatible, False otherwise
"""
# Get current Python version as string (e.g., "3.9.13")
current_version_str = ".".join(str(i) for i in sys.version_info[:3])
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this can be cleaner. Not sure if this is supported by 3.9

Suggested change
current_version_str = ".".join(str(i) for i in sys.version_info[:3])
major, minor, micro, *_ = sys.version_info
current_version_str = f"{major}.{minor}.{micro}"


# Get minimum required Python version from manifest
min_python_version = manifest_data.get("minimum_python_version")

# If no minimum version specified, assume compatible (backward compatibility)
if not min_python_version:
return True

# Compare versions using robust version comparison
is_compatible = is_version_newer_or_equal(
current_version_str, str(min_python_version)
)

return is_compatible

def set_is_copiable(self, copiable):
"""
Sets whether copying is supported by this descriptor.
Expand Down Expand Up @@ -517,7 +552,7 @@ def dict_from_uri(cls, uri):
descriptor_dict["type"] = split_path[1]

# now pop remaining keys into a dict and key by item_keys
for (param, value) in urllib.parse.parse_qs(query).items():
for param, value in urllib.parse.parse_qs(query).items():
if len(value) > 1:
raise TankDescriptorError(
"Invalid uri '%s' - duplicate parameters" % uri
Expand Down
257 changes: 248 additions & 9 deletions python/tank/descriptor/io_descriptor/git_tag.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,13 +7,18 @@
# By accessing, using, copying or modifying this work you indicate your
# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights
# not expressly granted therein are reserved by Shotgun Software Inc.
import os
import copy
import os
import re
import subprocess

Check notice on line 13 in python/tank/descriptor/io_descriptor/git_tag.py

View check run for this annotation

ShotGrid Chorus / security/bandit

B404: blacklist

Consider possible security implications associated with the subprocess module. secure coding id: PYTH-INJC-30.
import sys

from tank_vendor import yaml

from .git import IODescriptorGit
from ..errors import TankDescriptorError
from ... import LogManager
from .. import constants as descriptor_constants
from ..errors import TankDescriptorError
from .git import IODescriptorGit

log = LogManager.get_logger(__name__)

Expand Down Expand Up @@ -51,9 +56,7 @@
)

# call base class
super().__init__(
descriptor_dict, sg_connection, bundle_type
)
super().__init__(descriptor_dict, sg_connection, bundle_type)

# path is handled by base class - all git descriptors
# have a path to a repo
Expand Down Expand Up @@ -233,10 +236,151 @@

return git_tags

def _get_local_repository_tag(self):
"""
Get the current tag of the local git repository if the path points to one.
:returns: Tag name (string) or None if not on a tag or not a local repo
"""
if not os.path.exists(self._path) or not os.path.isdir(self._path):
return None

git_dir = os.path.join(self._path, ".git")
if not os.path.exists(git_dir):
return None

try:
result = subprocess.check_output(
["git", "describe", "--tags", "--exact-match", "HEAD"],
cwd=self._path,
stderr=subprocess.STDOUT,
)

Check notice on line 257 in python/tank/descriptor/io_descriptor/git_tag.py

View check run for this annotation

ShotGrid Chorus / security/bandit

B607: start_process_with_partial_path

Starting a process with a partial executable path secure coding id: PYTH-INJC-30.

Check notice on line 257 in python/tank/descriptor/io_descriptor/git_tag.py

View check run for this annotation

ShotGrid Chorus / security/bandit

B603: subprocess_without_shell_equals_true

subprocess call - check for execution of untrusted input. secure coding id: PYTH-INJC-30.
local_tag = result.strip()
if isinstance(local_tag, bytes):
local_tag = local_tag.decode("utf-8")

log.debug(
"Local repository at %s is currently at tag %s"
% (self._path, local_tag)
)
Comment on lines +262 to +265
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's use f-strings.

return local_tag
except subprocess.CalledProcessError:
# Not on a tag
return None
except Exception as e:
log.debug(
"Could not determine local repository tag at %s: %s" % (self._path, e)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same here f-strings.

)
return None

def _check_local_tag_compatibility(
self, local_tag, latest_tag, current_py_ver, min_py_ver
):
"""
Check if a local repository tag is compatible with the current Python version.
:param local_tag: Tag name from local repository
:param latest_tag: Latest tag that was found incompatible
:param current_py_ver: Current Python version string
:param min_py_ver: Minimum Python version required by latest_tag
:returns: local_tag if compatible, None otherwise
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's add types on the docstring. Or maybe starts using type annotations in the method signature?

"""
try:
# Create a descriptor for this local tag and download it to bundle cache
local_desc_dict = copy.deepcopy(self._descriptor_dict)
local_desc_dict["version"] = local_tag
local_desc = IODescriptorGitTag(
local_desc_dict, self._sg_connection, self._bundle_type
)
local_desc.set_cache_roots(self._bundle_cache_root, self._fallback_roots)

# Download to bundle cache if not already there
if not local_desc.exists_local():
log.debug(
"Downloading local tag %s to bundle cache for compatibility check"
% local_tag
)
local_desc.download_local()

# Check if this local tag is compatible
local_manifest = local_desc.get_manifest(
descriptor_constants.BUNDLE_METADATA_FILE
)
if self._check_minimum_python_version(local_manifest):
log.warning(
"Auto-update blocked: Latest tag %s requires Python %s, current is %s. "
"Using local repository tag %s which is compatible."
% (latest_tag, min_py_ver, current_py_ver, local_tag)
)
return local_tag
else:
log.debug(
"Local tag %s is also not compatible with current Python version"
% local_tag
)
return None
except Exception as e:
log.debug(
"Could not check compatibility for local tag %s: %s" % (local_tag, e)
)
return None

def _find_compatible_cached_version(self, latest_tag):
"""
Find the highest compatible version in the bundle cache.
:param latest_tag: Latest tag to exclude from search
:returns: Compatible tag name or None if not found
"""
cached_versions = self._get_locally_cached_versions()
if not cached_versions:
return None

all_cached_tags = list(cached_versions.keys())
compatible_version = None

for tag in all_cached_tags:
# Skip the incompatible latest version
if tag == latest_tag:
continue

try:
# Check if this cached version is compatible
temp_dict = copy.deepcopy(self._descriptor_dict)
temp_dict["version"] = tag
temp_desc = IODescriptorGitTag(
temp_dict, self._sg_connection, self._bundle_type
)
temp_desc.set_cache_roots(self._bundle_cache_root, self._fallback_roots)

cached_manifest = temp_desc.get_manifest(
descriptor_constants.BUNDLE_METADATA_FILE
)
if self._check_minimum_python_version(cached_manifest):
# Found a compatible version, but keep looking for higher ones
if compatible_version is None:
compatible_version = tag
else:
# Compare versions to keep the highest
latest_of_two = self._find_latest_tag_by_pattern(
[compatible_version, tag], None
)
if latest_of_two == tag:
compatible_version = tag
except Exception as e:
log.debug(
"Could not check compatibility for cached version %s: %s" % (tag, e)
)
continue

return compatible_version

def _get_latest_version(self):
"""
Returns a descriptor object that represents the latest version.
:returns: IODescriptorGitTag object
Returns the latest tag version, or current version if latest is not
compatible with current Python version.
:returns: Tag name (string) of the latest version or current version
"""
tags = self._fetch_tags()
latest_tag = self._find_latest_tag_by_pattern(tags, pattern=None)
Expand All @@ -245,7 +389,102 @@
"Git repository %s doesn't have any tags!" % self._path
)

return latest_tag
# Check if latest tag is compatible with current Python
try:
# Create a temporary descriptor for the latest tag
temp_descriptor_dict = copy.deepcopy(self._descriptor_dict)
temp_descriptor_dict["version"] = latest_tag

temp_desc = IODescriptorGitTag(
temp_descriptor_dict, self._sg_connection, self._bundle_type
)
temp_desc.set_cache_roots(self._bundle_cache_root, self._fallback_roots)

manifest = None
if temp_desc.exists_local():
# Latest tag is cached - check it directly
manifest = temp_desc.get_manifest(
descriptor_constants.BUNDLE_METADATA_FILE
)
elif self._version == "latest":
# For "latest" descriptors, try to find compatible version without downloading
# This searches local repo and bundle cache
local_tag = self._get_local_repository_tag()
log.debug("local_tag: %s" % local_tag)
if local_tag and local_tag == latest_tag:
# Local repo is already at latest tag, we can check it
try:
# Try to get manifest from local git checkout (not bundle cache)
local_repo_path = os.path.dirname(self._path)
manifest_path = os.path.join(
local_repo_path, descriptor_constants.BUNDLE_METADATA_FILE
)
if os.path.exists(manifest_path):
with open(manifest_path) as f:
manifest = yaml.load(f, Loader=yaml.FullLoader)
except Exception as e:
log.debug("Could not read manifest from local git repo: %s" % e)
manifest["minimum_python_version"] = "3.10"
if manifest and not self._check_minimum_python_version(manifest):
# Latest version is NOT compatible - block auto-update
current_py_ver = ".".join(str(x) for x in sys.version_info[:3])
min_py_ver = manifest.get("minimum_python_version", "not specified")

# If current version is "latest", find a compatible alternative
if self._version == "latest":
# First, check if the path points to a local git repository
local_tag = self._get_local_repository_tag()
if local_tag:
compatible_tag = self._check_local_tag_compatibility(
local_tag, latest_tag, current_py_ver, min_py_ver
)
if compatible_tag:
return compatible_tag

# Second, search for compatible version in bundle cache
compatible_version = self._find_compatible_cached_version(
latest_tag
)
if compatible_version:
log.warning(
"Auto-update blocked: Latest tag %s requires Python %s, current is %s. "
"Using highest compatible cached version %s."
% (
latest_tag,
min_py_ver,
current_py_ver,
compatible_version,
)
)
return compatible_version
else:
# No compatible version found - use latest anyway with warning
log.warning(
"Auto-update blocked: Latest tag %s requires Python %s, current is %s. "
"No compatible cached version found. Using latest tag anyway."
% (latest_tag, min_py_ver, current_py_ver)
)
return latest_tag
else:
log.warning(
"Auto-update blocked: Latest tag %s requires Python %s, current is %s. "
"Keeping current version %s."
% (latest_tag, min_py_ver, current_py_ver, self._version)
)
return self._version
else:
log.debug(
"Latest tag %s is compatible with current Python version"
% latest_tag
)
return latest_tag

except Exception as e:
log.warning(
"Could not check Python compatibility for tag %s: %s. Proceeding with auto-update."
% (latest_tag, e)
)
return latest_tag

def get_latest_cached_version(self, constraint_pattern=None):
"""
Expand Down
Loading