From d31c9ac5dd4db3c7c02b9231f93817583c8c215f Mon Sep 17 00:00:00 2001 From: John Chilton Date: Fri, 9 Sep 2022 18:42:35 -0400 Subject: [PATCH 1/2] Some typing for galaxy.util.path... --- lib/galaxy/util/compression_utils.py | 20 ++++---- lib/galaxy/util/path/__init__.py | 72 +++++++++++++++++++--------- 2 files changed, 61 insertions(+), 31 deletions(-) diff --git a/lib/galaxy/util/compression_utils.py b/lib/galaxy/util/compression_utils.py index 62dc82f027c..4fcc8adc6f4 100644 --- a/lib/galaxy/util/compression_utils.py +++ b/lib/galaxy/util/compression_utils.py @@ -20,7 +20,10 @@ from typing import ( from typing_extensions import Literal -from galaxy.util.path import safe_relpath +from galaxy.util.path import ( + safe_relpath, + StrPath, +) from .checkers import ( is_bz2, is_gzip, @@ -154,13 +157,14 @@ class CompressedFile: archive: Union[tarfile.TarFile, zipfile.ZipFile] @staticmethod - def can_decompress(file_path: str) -> bool: + def can_decompress(file_path: StrPath) -> bool: return tarfile.is_tarfile(file_path) or zipfile.is_zipfile(file_path) - def __init__(self, file_path: str, mode: str = "r") -> None: + def __init__(self, file_path: StrPath, mode: str = "r") -> None: + file_path_str = str(file_path) if tarfile.is_tarfile(file_path): self.file_type = "tar" - elif zipfile.is_zipfile(file_path) and not file_path.endswith(".jar"): + elif zipfile.is_zipfile(file_path) and not file_path_str.endswith(".jar"): self.file_type = "zip" self.file_name = os.path.splitext(os.path.basename(file_path))[0] if self.file_name.endswith(".tar"): @@ -199,7 +203,7 @@ class CompressedFile: common_prefix = "" return common_prefix - def extract(self, path: str) -> str: + def extract(self, path: StrPath) -> str: """Determine the path to which the archive should be extracted.""" contents = self.getmembers() extraction_path = path @@ -308,14 +312,14 @@ class CompressedFile: return True return False - def open_tar(self, filepath: str, mode: str) -> tarfile.TarFile: + def open_tar(self, filepath: StrPath, mode: str) -> tarfile.TarFile: return tarfile.open(filepath, mode, errorlevel=0) - def open_zip(self, filepath: str, mode: str) -> zipfile.ZipFile: + def open_zip(self, filepath: StrPath, mode: str) -> zipfile.ZipFile: mode = cast(Literal["a", "r", "w", "x"], mode) return zipfile.ZipFile(filepath, mode) - def zipfile_ok(self, path_to_archive: str) -> bool: + def zipfile_ok(self, path_to_archive: StrPath) -> bool: """ This function is a bit pedantic and not functionally necessary. It checks whether there is no file pointing outside of the extraction, because ZipFile.extractall() has some potential diff --git a/lib/galaxy/util/path/__init__.py b/lib/galaxy/util/path/__init__.py index 6e946ba3b7c..f6987a64dd4 100644 --- a/lib/galaxy/util/path/__init__.py +++ b/lib/galaxy/util/path/__init__.py @@ -12,6 +12,7 @@ from operator import getitem from os import ( extsep, makedirs, + PathLike, stat, walk, ) @@ -30,6 +31,15 @@ from os.path import ( sep as separator, ) from pathlib import Path +from typing import ( + AnyStr, + Iterator, + List, + Optional, + Tuple, + TYPE_CHECKING, + Union, +) try: from grp import getgrgid @@ -43,12 +53,26 @@ except ImportError: import galaxy.util +# Stable in Python 3.10 path types +if TYPE_CHECKING: + StrPath = Union[str, PathLike[str]] + BytesPath = Union[bytes, PathLike[bytes]] + GenericPath = Union[AnyStr, PathLike[AnyStr]] + StrOrBytesPath = Union[str, bytes, PathLike[str], PathLike[bytes]] +else: + StrPath = Union[str, PathLike] + BytesPath = Union[bytes, PathLike] + GenericPath = Union[AnyStr, PathLike[AnyStr]] + StrOrBytesPath = Union[str, bytes, PathLike, PathLike] + +AllowListT = Optional[List[GenericPath]] + WALK_MAX_DIRS = 10000 log = logging.getLogger(__name__) -def safe_path(path, allowlist=None): +def safe_path(path: GenericPath, allowlist: AllowListT = None): """Ensure that a the absolute location of the path (after following symlinks) is either itself or on the allowlist of acceptable locations. @@ -63,7 +87,7 @@ def safe_path(path, allowlist=None): return any(__contains(dirname(path), path, allowlist=allowlist)) -def safe_contains(prefix, path, allowlist=None, real=None): +def safe_contains(prefix: GenericPath, path: GenericPath, allowlist: AllowListT = None, real=None): """Ensure a path is contained within another path. Given any two filesystem paths, ensure that ``path`` is contained in ``prefix``. If ``path`` exists (either as an @@ -94,7 +118,7 @@ class _SafeContainsDirectoryChecker: self.prefix = prefix self.real_dirpath = realpath(join(prefix, dirpath)) - def check(self, filename): + def check(self, filename: GenericPath) -> bool: dirpath_path = join(self.real_dirpath, filename) if islink(dirpath_path): return safe_contains(self.prefix, filename, allowlist=self.allowlist) @@ -102,7 +126,7 @@ class _SafeContainsDirectoryChecker: return safe_contains(self.prefix, filename, allowlist=self.allowlist, real=dirpath_path) -def safe_makedirs(path): +def safe_makedirs(path: GenericPath) -> None: """Safely make a directory, do not fail if it already exists or is created during execution. :type path: string @@ -119,7 +143,7 @@ def safe_makedirs(path): raise -def safe_relpath(path): +def safe_relpath(path: GenericPath) -> bool: """Determine whether a relative path references a path outside its root. This is a path computation: the filesystem is not accessed to confirm the existence or nature of ``path``. @@ -152,7 +176,7 @@ def safe_walk(path, allowlist=None): if allowlist and i % WALK_MAX_DIRS == 0: raise RuntimeError( - "Breaking out of walk of %s after %s iterations (most likely infinite symlink recursion) at: %s" + "Breaking out of walk of %r after %s iterations (most likely infinite symlink recursion) at: %r" % (path, WALK_MAX_DIRS, dirpath) ) _prefix = partial(join, dirpath) @@ -176,7 +200,7 @@ def safe_walk(path, allowlist=None): yield (dirpath, dirnames, filenames) -def unsafe_walk(path, allowlist=None, username=None): +def unsafe_walk(path: GenericPath, allowlist: AllowListT = None, username: Optional[str] = None): """Walk a path and ensure that none of its contents are symlinks outside the path. It is assumed that ``path`` itself has already been validated e.g. with :func:`safe_relpath` or @@ -200,7 +224,7 @@ def unsafe_walk(path, allowlist=None, username=None): return unsafe_paths -def __path_permission_for_user(path, username): +def __path_permission_for_user(path: GenericPath, username: str) -> bool: """ :type path: string :param path: a directory or file to check @@ -227,7 +251,7 @@ def __path_permission_for_user(path, username): return False -def full_path_permission_for_user(prefix, path, username, skip_prefix=False): +def full_path_permission_for_user(prefix, path, username: str, skip_prefix=False): """ Assuming username is identical to the os username, this checks that the given user can read the specified path by checking the file permission @@ -260,7 +284,7 @@ def full_path_permission_for_user(prefix, path, username, skip_prefix=False): return can_read -def joinext(root, ext): +def joinext(root: str, ext: str) -> str: """ Roughly the reverse of os.path.splitext. @@ -274,7 +298,7 @@ def joinext(root, ext): return extsep.join((root.rstrip(extsep), ext.lstrip(extsep))) -def has_ext(path, ext, aliases=False, ignore=None): +def has_ext(path: AnyStr, ext: str, aliases=False, ignore=None): """ Determine whether ``path`` has extension ``ext`` @@ -297,7 +321,7 @@ def has_ext(path, ext, aliases=False, ignore=None): return _ext == ext -def get_ext(path, ignore=None, canonicalize=True): +def get_ext(path: AnyStr, ignore=None, canonicalize=True) -> str: """ Return the extension of ``path`` @@ -335,7 +359,7 @@ class Extensions(dict): return v raise KeyError(key) - def canonicalize(self, ext): + def canonicalize(self, ext: str) -> str: # shouldn't raise an IndexError because it should raise a KeyError first return self[ext][0] @@ -373,7 +397,7 @@ def external_chown(path, pwent, external_chown_script, description="file"): return False -def __listify(item): +def __listify(item) -> Union[list, tuple]: """A non-splitting version of :func:`galaxy.util.listify`.""" if not item: return [] @@ -386,7 +410,7 @@ def __listify(item): # helpers -def __walk(path): +def __walk(path: GenericPath) -> Iterator[GenericPath]: for dirpath, dirnames, filenames in walk(path): for name in dirnames: yield join(dirpath, name) @@ -394,7 +418,9 @@ def __walk(path): yield join(dirpath, name) -def __contains(prefix, path, allowlist=None, real=None): +def __contains( + prefix: GenericPath, path: GenericPath, allowlist: AllowListT = None, real: Optional[GenericPath] = None +): real = real or realpath(join(prefix, path)) yield not relpath(real, prefix).startswith(pardir) for aldir in allowlist or []: @@ -402,20 +428,20 @@ def __contains(prefix, path, allowlist=None, real=None): yield not relpath(real, aldir).startswith(pardir) -def __ext_strip_sep(ext): +def __ext_strip_sep(ext: str) -> str: return ext.lstrip(extsep) -def __splitext_no_sep(path): - path = galaxy.util.unicodify(path) - return (path.rsplit(extsep, 1) + [""])[0:2] +def __splitext_no_sep(path: AnyStr) -> List[str]: + path_as_str = galaxy.util.unicodify(path) + return (path_as_str.rsplit(extsep, 1) + [""])[0:2] -def __splitext_ignore(path, ignore=None): +def __splitext_ignore(path: AnyStr, ignore: Optional[Union[List[str], Tuple[str]]] = None) -> Tuple[str, str]: # note: unlike os.path.splitext this strips extsep from ext - ignore = map(__ext_strip_sep, __listify(ignore)) + ignore_map = map(__ext_strip_sep, __listify(ignore)) root, ext = __splitext_no_sep(path) - if ext in ignore: + if ext in ignore_map: new_path = path[0 : (-len(ext) - 1)] root, ext = __splitext_no_sep(new_path) From 2168933d7c1b4d0e2a5a81bdd418267fc68aed0e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bj=C3=B6rn=20Gr=C3=BCning?= Date: Sat, 10 Sep 2022 20:12:54 +0200 Subject: [PATCH 2/2] fix PathLike annotation --- lib/galaxy/util/path/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/galaxy/util/path/__init__.py b/lib/galaxy/util/path/__init__.py index f6987a64dd4..92e8d228980 100644 --- a/lib/galaxy/util/path/__init__.py +++ b/lib/galaxy/util/path/__init__.py @@ -62,7 +62,7 @@ if TYPE_CHECKING: else: StrPath = Union[str, PathLike] BytesPath = Union[bytes, PathLike] - GenericPath = Union[AnyStr, PathLike[AnyStr]] + GenericPath = Union[AnyStr, PathLike] StrOrBytesPath = Union[str, bytes, PathLike, PathLike] AllowListT = Optional[List[GenericPath]]