diff --git a/.github/workflows/windows_smb_test.yml b/.github/workflows/windows_smb_test.yml new file mode 100644 index 000000000..ab2cdf99a --- /dev/null +++ b/.github/workflows/windows_smb_test.yml @@ -0,0 +1,66 @@ +name: Windows SMB Path Test + +# Validates UNC path containment against a real SMB share, which lexical ntpath modeling +# cannot do. Regression coverage for issue #1321. +# +# Not part of Code Quality: creating a share needs administrator rights, and the loopback +# share is slower and more environment-dependent than a unit test. +on: + workflow_dispatch: + workflow_call: + inputs: + tag: + description: Git ref (tag/branch/SHA) to test. Defaults to the triggering ref. + required: false + type: string + default: '' + +jobs: + test: + name: UNC Containment (real SMB) + runs-on: windows-latest + timeout-minutes: 15 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ inputs.tag || github.ref }} + + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: '3.12' + + - name: Confirm SMB prerequisites + # Fail with a clear message here rather than having every test skip itself, + # which would look like a pass. + shell: pwsh + run: | + $admin = ([Security.Principal.WindowsPrincipal] ` + [Security.Principal.WindowsIdentity]::GetCurrent() + ).IsInRole([Security.Principal.WindowsBuiltInRole]::Administrator) + if (-not $admin) { throw "Administrator rights are required to create an SMB share." } + Get-Service LanmanServer, LanmanWorkstation | Format-Table -AutoSize + Start-Service LanmanServer + Start-Service LanmanWorkstation + # Developer Mode lets a non-elevated process create symlinks; the escape + # test needs one and skips itself otherwise. + $key = 'HKLM:\SOFTWARE\Microsoft\Windows\CurrentVersion\AppModelUnlock' + New-Item -Path $key -Force | Out-Null + Set-ItemProperty -Path $key -Name AllowDevelopmentWithoutDevLicense -Value 1 -Type DWord + + - name: Install + run: | + python -m pip install --upgrade pip hatch + + - name: Run the SMB path tests + shell: pwsh + run: | + hatch run pytest test/integ/windows_smb -v --no-cov -p no:randomly + + - name: Report skips + # A skipped SMB test is indistinguishable from a passing one in the summary, + # so surface the count explicitly. + if: always() + shell: pwsh + run: | + hatch run pytest test/integ/windows_smb --no-cov -q -rs 2>&1 | + Select-String -Pattern 'SKIPPED|passed|failed' diff --git a/pyproject.toml b/pyproject.toml index e9a373419..b137e7746 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -123,9 +123,29 @@ line-length = 100 # E402 (imports must be at the top of the file) is pinned explicitly here so it # can't silently regress if a future ruff drops it from the defaults; it is # build-failing via `hatch run lint`. -extend-select = ["RUF022", "E402"] +# TID251 bans path helpers that raise on Windows UNC paths; see banned-api below. +extend-select = ["RUF022", "E402", "TID251"] ignore = ["E501"] +[tool.ruff.lint.flake8-tidy-imports.banned-api] +# commonpath raises ValueError on Windows UNC paths (issue #1321): callers that catch it +# silently reject valid paths, callers that don't crash. ntpath/posixpath are banned too +# because this codebase passes explicit path modules around. +"os.path.commonpath".msg = "Use deadline.client._path_utils.is_path_contained for containment checks, or deadline.client._path_summary.common_ancestor for a displayed summary; commonpath raises ValueError on Windows UNC paths. See issue #1321." +"ntpath.commonpath".msg = "Use deadline.client._path_utils.is_path_contained for containment checks, or deadline.client._path_summary.common_ancestor for a displayed summary; commonpath raises ValueError on Windows UNC paths. See issue #1321." +"posixpath.commonpath".msg = "Use deadline.client._path_utils.is_path_contained for containment checks, or deadline.client._path_summary.common_ancestor for a displayed summary; commonpath raises ValueError on Windows UNC paths. See issue #1321." +# commonprefix compares strings, not path components, so it reports '\\host\share2' as +# sharing a prefix with '\\host\share'. It is never the right containment primitive. +"os.path.commonprefix".msg = "Use deadline.client._path_utils.is_path_contained or deadline.client._path_summary.common_ancestor; commonprefix is a string-prefix match, not a path-component match." + +[tool.ruff.lint.per-file-ignores] +# The sanctioned wrappers, and the one place allowed to reach for what they replace. +"src/deadline/client/_path_utils.py" = ["TID251"] +# These compare the wrappers against the stdlib behavior they replace. +"test/unit/deadline_client/test_path_utils.py" = ["TID251"] +"test/unit/deadline_client/test_path_summary.py" = ["TID251"] +"test/unit/deadline_client/api/test_job_bundle_submission_asset_refs.py" = ["TID251"] + [tool.ruff.lint.isort] known-first-party = ["deadline"] diff --git a/src/deadline/client/_path_summary.py b/src/deadline/client/_path_summary.py new file mode 100644 index 000000000..f15126f80 --- /dev/null +++ b/src/deadline/client/_path_summary.py @@ -0,0 +1,74 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + +""" +Summarizing a group of paths for display. + +Kept out of ``_path_utils`` because this is presentation rather than a trust decision, and +it carries cases that only a displayed string cares about -- unresolved ``..`` runs and +preserving the caller's spelling -- which a reader auditing containment should not have to +read past. + +Like the containment helpers, this is purely lexical and never raises. +""" + +from __future__ import annotations + +import os +from typing import Any, Sequence + +from ._path_utils import _PARDIR, _UNC_ANCHOR, _split_anchored + +__all__ = [ + "common_ancestor", +] + + +def _leading_pardir_count(parts: list[str]) -> int: + """Count the leading '..' run that ``normpath`` could not resolve. + + Counted on parts rather than whole components because an anchor can precede the run + ('C:..\\x' is the parent of the working directory on drive C:). + """ + count = 0 + for part in parts: + if part != _PARDIR: + break + count += 1 + return count + + +def common_ancestor(paths: Sequence[Any], *, path_module: Any = os.path) -> str: + """Return the deepest directory containing every path in ``paths``. + + This is ``os.path.commonpath`` without the exceptions: paths in unrelated spaces return + ``""`` rather than raising, and a UNC host is a valid answer for paths on different + shares of one server. The result keeps the first path's spelling and, like + ``commonpath``, is purely lexical. + """ + if not paths: + return "" + + split = [_split_anchored(p, path_module, normalize_case=True) for p in paths] + normalized = [([a] if a else []) + parts for a, parts in split] + anchor, spelled_parts = _split_anchored(paths[0], path_module, normalize_case=False) + spelled = ([anchor] if anchor else []) + spelled_parts + + # '..' and '../..' are rooted at different unknown places, so runs of differing depth + # share nothing. Comparing them positionally would return the shallower path, which is + # not an ancestor of the deeper one -- os.path.commonpath has that bug. + if len({_leading_pardir_count(parts) for _, parts in split}) > 1: + return "" + + shared = min(len(components) for components in normalized) + while shared > 0 and any(other[:shared] != normalized[0][:shared] for other in normalized): + shared -= 1 + if shared == 0: + return "" + # Matching only the bare anchor means different servers, so no shared directory. + if shared == 1 and normalized[0][0] == _UNC_ANCHOR: + return "" + + # The anchor carries its own separator, so it abuts the first part directly. + if anchor: + return anchor + path_module.sep.join(spelled[1:shared]) + return path_module.sep.join(spelled[:shared]) diff --git a/src/deadline/client/_path_utils.py b/src/deadline/client/_path_utils.py new file mode 100644 index 000000000..b6bfceb87 --- /dev/null +++ b/src/deadline/client/_path_utils.py @@ -0,0 +1,234 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + +""" +Path containment helpers that understand Windows UNC paths. + +``os.path.commonpath`` raises ``ValueError`` rather than comparing a host-level UNC path +(``\\\\server``) with a path under one of its shares: ``splitdrive`` reports no drive for +the former and ``\\\\server\\share`` for the latter. It raises the same way for two shares +on one host. Callers that read that exception as "not contained" reject valid paths. + +These helpers compare paths component by component instead, so a UNC host is an ordinary +ancestor of its shares. Every function takes an explicit ``path_module`` +(``ntpath``/``posixpath``), so Windows semantics stay testable on non-Windows hosts. + +Comparisons are lexical -- pass ``realpath`` output in if symlinks must be resolved -- and +never raise. Anything unresolvable fails closed, since callers use containment to decide +whether a path is trusted. +""" + +from __future__ import annotations + +import os +import string +from typing import Any, Iterable + +__all__ = [ + "is_absolute_path", + "is_any_path_contained", + "is_path_contained", + "normalized_path", + "path_components", +] + +# Anchors the UNC path space. It names no server on its own, so unlike POSIX '/' it is not +# a directory and contains nothing. +_UNC_ANCHOR = "\\\\" + +# Spelled out rather than taken from os.* because these helpers parse Windows paths on any +# host, where os.sep is '/'. +_EXTENDED_PREFIX = "\\\\?\\" +_DEVICE_PREFIX = "\\\\.\\" +_EXTENDED_UNC_MARKER = "UNC" + +_PARDIR = ".." + + +def _splitroot(text: str, path_module: Any) -> tuple[str, str, str]: + """``path_module.splitroot``, backported for Python < 3.12. + + ``(drive, root)`` is what distinguishes one path space from another: ``('', '\\\\')`` + (rooted but driveless) and ``('\\\\\\\\', '')`` (UNC) both consist only of separators. + """ + splitroot = getattr(path_module, "splitroot", None) + if splitroot is not None: + return splitroot(text) + + drive, rest = path_module.splitdrive(text) + separators = path_module.sep + (getattr(path_module, "altsep", None) or "") + if rest[:1] not in separators or not rest: + return drive, "", rest + leading = len(rest) - len(rest.lstrip(separators)) + # POSIX gives '//' its own root spelling, but collapses three or more. + root_length = 2 if (leading == 2 and path_module.sep == "/") else 1 + return drive, rest[:root_length], rest[root_length:] + + +def _denotes_drive(text: str) -> bool: + """True for a bare drive spelling such as ``'C:'``.""" + return len(text) == 2 and text[1] == ":" and text[0] in string.ascii_letters + + +def _fold_extended_length_prefix(text: str) -> str: + """Rewrite an extended-length path as the plain path it denotes. + + ``\\\\?\\`` only turns off Win32 normalization; it names the same location as the plain + spelling. Folding it keeps one location from having two sets of components, which would + report a prefixed path outside a root that plainly contains it. Forms with no plain + spelling (``Volume{GUID}``, ``GLOBALROOT``, and the ``\\\\.\\`` device namespace) are + left alone, so they keep a path space of their own and alias nothing. + """ + if not text.startswith(_EXTENDED_PREFIX): + return text + denoted = text[len(_EXTENDED_PREFIX) :] + head = denoted.split("\\", 1)[0] + if head.upper() == _EXTENDED_UNC_MARKER: + # '\\?\UNC\server\share' is '\\server\share'. 'UNC' alone names no server, so it + # folds to the bare anchor, which contains nothing. + return _UNC_ANCHOR + denoted[len(_EXTENDED_UNC_MARKER) :].lstrip("\\") + if _denotes_drive(head): + return denoted + return text + + +def _split_anchored(path: Any, path_module: Any, normalize_case: bool) -> tuple[str, list[str]]: + """Return ``(anchor, parts)``, where ``anchor + sep.join(parts)`` reconstructs ``path``. + + The anchor names the path space and carries its own trailing separator. A UNC anchor is + the bare ``\\\\`` marker, leaving the server and share as ordinary parts -- which is what + lets a host-level root contain the shares beneath it. + """ + text = str(path) + windows = path_module.sep == "\\" + if windows: + text = text.replace("/", "\\") + # Folded before normpath, which leaves '..' alone inside a '\\?\' path before 3.11 + # and collapses it after. Folding first makes the result the same on every + # supported interpreter. + text = _fold_extended_length_prefix(text) + # Read off the text, not off splitdrive's drive: before Python 3.11 splitdrive reports + # no drive at all for a UNC path that names no share, which would put a host-level root + # in the rooted-driveless space and stop it containing its own shares -- the bug this + # module exists to fix. + in_unc_space = ( + windows + and text.startswith(_UNC_ANCHOR) + and not text.startswith(_EXTENDED_PREFIX) + and not text.startswith(_DEVICE_PREFIX) + ) + text = path_module.normpath(text) + if in_unc_space and not text.startswith(_UNC_ANCHOR): + # Those same versions collapse the leading pair itself ('\\host' -> '\host'). + text = _UNC_ANCHOR + text.lstrip(path_module.sep) + if normalize_case: + text = path_module.normcase(text) + + drive, root, tail = _splitroot(text, path_module) + parts = [part for part in tail.split(path_module.sep) if part] + + if not windows: + # '//foo' and '/foo' are the same file on the platforms this client targets. + return (path_module.sep if root else ""), parts + + if drive.startswith(_EXTENDED_PREFIX) or drive.startswith(_DEVICE_PREFIX): + # Whatever reaches here has no plain spelling to fold to, so the drive is a whole + # anchor occupying its own space. The anchor carries its own trailing separator, so + # a share root and the files under it -- which differ only by it -- still match. + return drive + path_module.sep, parts + if in_unc_space: + # The server and share are ordinary parts beneath the bare anchor, which is what + # lets a host-level root contain the shares under it. + return _UNC_ANCHOR, [p for p in text[len(_UNC_ANCHOR) :].split(path_module.sep) if p] + return drive + root, parts + + +def path_components( + path: Any, + *, + path_module: Any = os.path, + normalize_case: bool = True, +) -> list[str]: + """Split ``path`` into the components used for ancestor comparisons. + + ``..`` segments are resolved first. The first component is the path space (``'/'``, + ``'C:\\'``, ``'C:'`` for drive-relative, ``'\\\\'`` for UNC, absent when relative) and + the rest are the path's parts, so comparing these lists component-wise confuses neither + one path space for another nor a string prefix for a directory prefix. + + An extended-length prefix folds to the plain path it denotes, so ``\\\\?\\C:\\a`` and + ``C:\\a`` yield the same components. Prefixed forms that denote no plain path keep a + space of their own. + + ``normalize_case`` lowercases components on Windows to match the filesystem. + """ + anchor, parts = _split_anchored(path, path_module, normalize_case) + return ([anchor] if anchor else []) + parts + + +def is_absolute_path(path: Any, *, path_module: Any = os.path) -> bool: + """Return True iff ``path`` names a location without consulting the working directory. + + ``path_module.isabs`` cannot be used before Python 3.11: it tests what ``splitdrive`` + leaves behind, and for a UNC path that names a share ``splitdrive`` consumes the whole + string, so ``isabs(r'\\\\host\\share')`` is False there. Callers use this to decide + whether a path may be trusted as a root or accepted as a parameter value, so a UNC + share silently reading as relative drops valid roots and rejects valid values. + + A drive-relative path (``C:x``, meaning ``x`` under the working directory on ``C:``) is + not absolute, because resolving it needs the working directory -- which is the thing the + known-root hardening must never let a caller supply implicitly. A rooted, driveless path + (``\\x``) is absolute: it names the current drive's root, not the working directory. + + That second answer is why this cannot just call ``path_module.isabs`` on the newest + interpreters either -- ``ntpath.isabs`` returns True for ``\\x`` through 3.12 and False + from 3.13. Answering from the anchor keeps the verdict the same on every version. + """ + anchor, _ = _split_anchored(path, path_module, normalize_case=True) + return bool(anchor) and not _denotes_drive(anchor) + + +def normalized_path(path: Any, *, path_module: Any = os.path) -> str: + """Return ``path`` with ``..``, ``.``, repeated separators and separator style resolved. + + ``path_module.normpath`` with the version differences handled: before Python 3.11 it + collapses the leading pair on a UNC path that names no share (``\\\\host`` -> ``\\host``), + moving a host-level root out of the UNC space so it matches none of its own shares. + Case is preserved, unlike the components used for comparison. + """ + anchor, parts = _split_anchored(path, path_module, normalize_case=False) + return anchor + path_module.sep.join(parts) + + +def is_path_contained( + path: Any, + root: Any, + *, + path_module: Any = os.path, +) -> bool: + """Return True iff ``path`` equals or is a descendant of ``root``. + + Containment is anchored on whole components, so a sibling that merely shares a string + prefix (root ``/trusted/project`` vs path ``/trusted/project-secret``) is outside the + root. Paths in unrelated spaces -- different drives, different UNC hosts, one relative + and one absolute -- are not contained. + """ + root_components = path_components(root, path_module=path_module) + candidate_components = path_components(path, path_module=path_module) + # A bare '\\' root names no server, so it is an ancestor of nothing -- otherwise it + # would prefix, and so trust, every reachable share. ntpath.isabs lets it reach here. + if root_components == [_UNC_ANCHOR]: + return candidate_components == [_UNC_ANCHOR] + if candidate_components[: len(root_components)] != root_components: + return False + # Shared leading '..' belongs to the root; one below it could climb back out. + return _PARDIR not in candidate_components[len(root_components) :] + + +def is_any_path_contained( + path: Any, + roots: Iterable[Any], + *, + path_module: Any = os.path, +) -> bool: + """Return True iff ``path`` is contained by any root in ``roots``.""" + return any(is_path_contained(path, root, path_module=path_module) for root in roots) diff --git a/src/deadline/client/api/_submit_job_bundle.py b/src/deadline/client/api/_submit_job_bundle.py index 691f32891..baf5b5df9 100644 --- a/src/deadline/client/api/_submit_job_bundle.py +++ b/src/deadline/client/api/_submit_job_bundle.py @@ -71,6 +71,12 @@ summarize_path_list, ) from ...job_attachments.api._hashing import _hash_attachments +from .._path_utils import ( + is_absolute_path, + is_any_path_contained, + normalized_path, + path_components, +) logger = logging.getLogger(__name__) @@ -83,22 +89,12 @@ def hashing_telemetry_callback(hashing_summary: SummaryStatistics): def _is_known_path(path: Path | str, known_roots: Iterable[Path | str]) -> bool: """Return True iff ``path`` equals or is a descendant of any root in ``known_roots``. - Containment is anchored via ``os.path.commonpath`` equality (the same idiom as - loader.py): a path is contained only when it shares a whole-component prefix with a - root, so a sibling that merely shares a string prefix (root ``/trusted/project`` vs - candidate ``/trusted/project-secret``) is outside the root. + Containment is anchored on whole components, so a sibling that merely shares a + string prefix (root ``/trusted/project`` vs candidate ``/trusted/project-secret``) + is outside the root. """ - norm_candidate = os.path.normpath(str(path)) - for known_path in known_roots: - norm_root = os.path.normpath(str(known_path)) - try: - if os.path.commonpath([norm_root, norm_candidate]) == norm_root: - return True - except ValueError: - # commonpath raises for mixed absolute/relative paths or different Windows - # drives; such paths are not contained. - continue - return False + # Passed explicitly, and read at call time, so tests can patch it for another platform. + return is_any_path_contained(path, known_roots, path_module=os.path) def _summarize_asset_paths( @@ -294,22 +290,49 @@ def _filter_redundant_known_paths(known_asset_paths: Iterable[str]) -> list[str] This algorithm identifies any paths that have a different path as a prefix, and removes them from the list. Pseudo-code is: - 1. Sort the paths from shortest to longest, so any prefix of a path has + 1. Sort the paths from fewest to most components, so any prefix of a path has to happen before that path. 2. For each path, split it into parts (i.e. '/mnt/prod/project' becomes - ['/', 'mnt', 'prod', 'project']), and then insert it part by part into + ['', 'mnt', 'prod', 'project']), and then insert it part by part into a nested dict called dir_tree organized as a TRIE. The value True in the TRIE indicates that a path with that as its final part is in the list. 3. While inserting a path into the TRIE, detect whether another path already had a prefix of the parts, and filter out the path when that occurs. + + Components come from ``path_components`` rather than ``Path.parts`` so a Windows UNC + host is an ancestor of its shares (``Path.parts`` collapses '\\\\server\\share' into one + atom) and case variants of one location dedupe on Windows. + + Roots are expanded for '~' (the config file and the CLI submitter's default data + directory supply one unexpanded) and dropped unless absolute. A non-absolute root + matches no candidate anyway, but dropping it here means a future caller cannot turn it + into a trusted tree by resolving it -- ``os.path.abspath("")`` is the whole working + directory, which would suppress the unknown-path warning and let a non-interactive + submit upload undesignated files. An empty root arrives from a PATH parameter whose + allowedValues suppressed absolutization, and from ``--known-asset-path``/MCP input. """ + # Passed explicitly, and read at call time, so tests can patch it for another platform. + expanded = (os.path.expanduser(path) for path in known_asset_paths if path) + # normalized_path, not abspath: dedupes equivalent spellings without consulting the cwd. + # Not os.path.normpath, which before Python 3.11 collapses the leading pair on a + # host-level UNC root ('\\host' -> '\host'), moving it out of the UNC space so it then + # matches none of its own shares -- and this list is what _is_known_path compares. + ordered = list( + dict.fromkeys( + normalized_path(path, path_module=os.path) + for path in expanded + if is_absolute_path(path, path_module=os.path) + ) + ) + components = {path: path_components(path, path_module=os.path) for path in ordered} # This directory tree gets filled with the known asset paths, with # a True value as a marker for the last part of already seen paths. dir_tree: dict[str, Any] = {} filtered_paths: list[str] = [] - # Process the paths from shortest to longest, so that prefixes are always seen first - for path in sorted(known_asset_paths, key=len): - parts = Path(path).parts + # Fewest components first, so prefixes are seen first. Ties keep input order, so of two + # spellings of one location the caller's first -- highest precedence -- is retained. + for path in sorted(ordered, key=lambda p: (len(components[p]), ordered.index(p))): + parts = components[path] current: Optional[dict[str, Any]] = dir_tree for part in parts[:-1]: # If we see a True value, another path is a prefix so we can skip it. @@ -791,7 +814,10 @@ def _path_parameter_known_paths(resolved_parameters): bundle_parameter_types.get(name) == "PATH" and isinstance(value, str) and value != "" - and not os.path.isabs(value) + # Not os.path.isabs: before Python 3.11 it reads a UNC path naming a + # share as relative, rejecting a valid value on the very setup #1321 + # reports. + and not is_absolute_path(value, path_module=os.path) ): raise DeadlineOperationError( f"Pre-submission hook emitted a relative PATH value for parameter " diff --git a/src/deadline/client/cli/_groups/job_group.py b/src/deadline/client/cli/_groups/job_group.py index 21519cef9..d3c3a8ef9 100644 --- a/src/deadline/client/cli/_groups/job_group.py +++ b/src/deadline/client/cli/_groups/job_group.py @@ -41,6 +41,7 @@ from ... import api from ...config import config_file +from ..._path_summary import common_ancestor from ...exceptions import DeadlineOperationError, DeadlineOperationTimedOut from .._common import ( _OUTPUT_FORMAT_HELP, @@ -910,7 +911,7 @@ def _get_summary_of_files_to_download_message( return _get_json_line(JSON_MSG_TYPE_PRESUMMARY, output_paths_by_root) else: paths_message_joined = " " + "\n ".join( - f"{os.path.commonpath([os.path.join(directory, p) for p in output_paths])} ({len(output_paths)} file{'s' if len(output_paths) > 1 else ''})" + f"{common_ancestor([os.path.join(directory, p) for p in output_paths], path_module=os.path)} ({len(output_paths)} file{'s' if len(output_paths) > 1 else ''})" for directory, output_paths in output_paths_by_root.items() ) return f"\nSummary of files to download:\n{paths_message_joined}\n" diff --git a/src/deadline/client/job_bundle/loader.py b/src/deadline/client/job_bundle/loader.py index 5094a92e7..eaaacc935 100644 --- a/src/deadline/client/job_bundle/loader.py +++ b/src/deadline/client/job_bundle/loader.py @@ -16,6 +16,7 @@ import yaml +from .._path_utils import is_path_contained from ..exceptions import DeadlineOperationError @@ -34,8 +35,7 @@ def validate_directory_symlink_containment(job_bundle_dir: str) -> None: for path in chain(dir_names, file_names): norm_path = os.path.normpath(os.path.join(root_dir, path)) resolved_path = os.path.realpath(norm_path) - common_path = os.path.commonpath([resolved_root, resolved_path]) - if common_path != resolved_root: + if not is_path_contained(resolved_path, resolved_root, path_module=os.path): raise DeadlineOperationError( f"Job bundle cannot contain a path that resolves outside of the resolved bundle directory:\n{resolved_root}\n\nPath in bundle:\n{norm_path}\nResolves to:\n{resolved_path}" ) diff --git a/src/deadline/client/job_bundle/parameters.py b/src/deadline/client/job_bundle/parameters.py index 4721eb721..40ec94729 100644 --- a/src/deadline/client/job_bundle/parameters.py +++ b/src/deadline/client/job_bundle/parameters.py @@ -23,6 +23,7 @@ NotRequired = object TypedDict = object +from .._path_utils import is_absolute_path, is_path_contained from ..exceptions import DeadlineOperationError from .loader import read_yaml_or_json_object @@ -793,14 +794,16 @@ def read_job_bundle_parameters(bundle_dir: str) -> list[JobParameter]: ): default = parameter.get("default") if default: - if os.path.isabs(default): + # Not os.path.isabs, which before Python 3.11 reads a UNC path naming a + # share as relative -- such a default reached the containment check below + # and failed there, reporting the wrong reason. + if is_absolute_path(default, path_module=os.path): raise DeadlineOperationError( f"Job Template for job bundle {bundle_dir}:\nDefault PATH '{default}' for parameter '{name}' is absolute.\nPATH values must be relative, and must resolve within the Job Bundle directory." ) bundle_real_path = os.path.realpath(bundle_dir) default_real_path = os.path.realpath(os.path.join(bundle_real_path, default)) - common_path = os.path.commonpath([bundle_real_path, default_real_path]) - if common_path != bundle_real_path: + if not is_path_contained(default_real_path, bundle_real_path, path_module=os.path): raise DeadlineOperationError( f"Job Template for job bundle {bundle_dir}:\nDefault PATH '{default_real_path}' for parameter '{name}' specifies files outside of Job Bundle directory '{bundle_real_path}'.\nPATH values must be relative, and must resolve within the Job Bundle directory." ) diff --git a/test/integ/windows_smb/test_unc_path_containment.py b/test/integ/windows_smb/test_unc_path_containment.py new file mode 100644 index 000000000..14acad6a8 --- /dev/null +++ b/test/integ/windows_smb/test_unc_path_containment.py @@ -0,0 +1,223 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + +""" +Real-SMB validation of UNC path containment. + +Every other test of the path helpers models Windows lexically through ``ntpath``, which +is faithful to Windows' string rules but says nothing about SMB: whether a host-level UNC +path can be listed, whether ``realpath`` rewrites a mapped drive back to UNC form, or +whether a share walks like a directory. These run against a loopback share, so the +verdicts are checked against a real redirector. + +Requires Windows and administrator rights; see .github/workflows/windows_smb_test.yml. +Regression coverage for https://github.com/aws-deadline/deadline-cloud/issues/1321. +""" + +from __future__ import annotations + +import os +import socket +import subprocess +import sys +import uuid +from pathlib import Path +from typing import Iterator + +import pytest + +from deadline.client._path_summary import common_ancestor +from deadline.client._path_utils import is_path_contained +from deadline.client.api._submit_job_bundle import ( + _filter_redundant_known_paths, + _is_known_path, +) +from deadline.client.job_bundle.loader import validate_directory_symlink_containment +from deadline.client.exceptions import DeadlineOperationError + +pytestmark = [ + pytest.mark.integ, + pytest.mark.skipif(sys.platform != "win32", reason="SMB shares require Windows."), +] + + +def _run(*args: str) -> subprocess.CompletedProcess: + return subprocess.run(args, capture_output=True, text=True, check=False) + + +@pytest.fixture(scope="module") +def smb_share(tmp_path_factory) -> Iterator[tuple[str, Path]]: + """Share a local directory over SMB and yield ``(unc_root, local_path)``. + + ``unc_root`` is the share path (``\\\\\\``); the host-level root is + derived from it by the tests that need one. + """ + local_path = tmp_path_factory.mktemp("smb_export") + share_name = f"dltest{uuid.uuid4().hex[:8]}" + + created = _run("net", "share", f"{share_name}={local_path}", "/GRANT:Everyone,FULL") + if created.returncode != 0: + pytest.skip( + f"could not create an SMB share: {created.stdout.strip()} {created.stderr.strip()}" + ) + + # The loopback host name matters: 'localhost' and '127.0.0.1' are both valid UNC + # hosts, but the machine name is what a real farm would use. + unc_root = rf"\\{socket.gethostname()}\{share_name}" + try: + # Fail fast and clearly if the redirector cannot reach the new share, rather + # than letting every assertion below fail with a confusing error. + if not os.path.isdir(unc_root): + pytest.skip(f"SMB share {unc_root} is not reachable from this host") + yield unc_root, local_path + finally: + _run("net", "share", share_name, "/DELETE", "/Y") + + +def test_host_level_root_contains_share_contents(smb_share): + """The reported bug: a '\\\\server' root must contain files on its shares.""" + unc_root, local_path = smb_share + host_root = unc_root.rsplit("\\", 1)[0] + + asset = Path(unc_root) / "assets" / "scene.c4d" + asset.parent.mkdir(parents=True, exist_ok=True) + asset.write_text("scene", encoding="utf8") + + assert os.path.isfile(asset), f"{asset} was not written through the share" + assert is_path_contained(asset, host_root) + assert is_path_contained(asset, unc_root) + assert _is_known_path(asset, [host_root]) + assert _is_known_path(asset, [unc_root]) + + +def test_host_level_root_survives_redundancy_filtering(smb_share): + """A host-level root must reach the containment check intact. + + It is absolute per ``os.path.isabs`` and must subsume its own shares, so the + filter keeps the host and drops the share. + """ + unc_root, _ = smb_share + host_root = unc_root.rsplit("\\", 1)[0] + + assert _filter_redundant_known_paths([host_root]) == [host_root] + assert _filter_redundant_known_paths([host_root, unc_root]) == [host_root] + + +def test_neighbouring_host_is_not_contained(smb_share): + """A different host must not be contained, even one sharing a string prefix.""" + unc_root, _ = smb_share + host_root = unc_root.rsplit("\\", 1)[0] + + assert not is_path_contained(rf"{host_root}2\share\file", host_root) + assert not is_path_contained(r"\\other-host\share\file", host_root) + + +def test_realpath_of_share_content_stays_contained(smb_share): + """``realpath`` output must still be recognized as inside the share. + + Both containment guards resolve their operands first, so any rewriting by the + redirector would make them silently reject valid paths. + """ + unc_root, _ = smb_share + host_root = unc_root.rsplit("\\", 1)[0] + + nested = Path(unc_root) / "resolve_probe" / "file.txt" + nested.parent.mkdir(parents=True, exist_ok=True) + nested.write_text("probe", encoding="utf8") + + resolved = os.path.realpath(nested) + assert is_path_contained(resolved, os.path.realpath(unc_root)) + assert is_path_contained(resolved, host_root), ( + f"realpath rewrote {nested} to {resolved}, which no longer resolves under {host_root}" + ) + + +def test_bundle_on_share_passes_symlink_containment(smb_share): + """A job bundle living on a share must validate, including at the share root. + + ``\\\\server\\share`` vs its own files is one of the pairs ``os.path.commonpath`` + rejected outright. + """ + unc_root, _ = smb_share + + bundle = Path(unc_root) / "bundle" + bundle.mkdir(parents=True, exist_ok=True) + (bundle / "template.yaml").write_text( + "specificationVersion: jobtemplate-2023-09\n", encoding="utf8" + ) + validate_directory_symlink_containment(str(bundle)) + + # And a bundle that IS the share root. + root_bundle = Path(unc_root) / "root_bundle" + root_bundle.mkdir(parents=True, exist_ok=True) + (root_bundle / "template.yaml").write_text( + "specificationVersion: jobtemplate-2023-09\n", encoding="utf8" + ) + validate_directory_symlink_containment(str(root_bundle)) + + +def test_symlink_escaping_the_share_is_rejected(smb_share): + """A symlink out of a bundle on a share must still be caught. + + This is the security direction: the lexical tests assert it, but only a real + filesystem exercises the ``realpath`` resolution the guard depends on. + """ + unc_root, local_path = smb_share + + bundle = Path(unc_root) / "escape_bundle" + bundle.mkdir(parents=True, exist_ok=True) + outside = local_path / "outside_secret.txt" + outside.write_text("secret", encoding="utf8") + + link = bundle / "escape.txt" + try: + os.symlink(outside, link) + except OSError as exc: # pragma: no cover - depends on runner privileges + pytest.skip(f"cannot create a symlink on this share: {exc}") + + with pytest.raises(DeadlineOperationError): + validate_directory_symlink_containment(str(bundle)) + + +def test_mapped_drive_resolves_and_compares(smb_share): + """A mapped drive letter is a distinct path space from the UNC path it points at. + + Studios commonly map a share to a drive letter. Whichever spelling ``realpath`` + reports, containment must agree with it rather than mixing the two. + """ + unc_root, _ = smb_share + host_root = unc_root.rsplit("\\", 1)[0] + + for letter in ("Y:", "Z:"): + if _run("net", "use", letter, unc_root).returncode == 0: + drive = letter + break + else: + pytest.skip("no free drive letter to map the share onto") + + try: + asset = Path(drive + "\\") / "mapped_probe.txt" + asset.write_text("mapped", encoding="utf8") + + resolved = os.path.realpath(asset) + # Whatever spelling realpath returns, it must be contained by the matching + # root and not by the other path space. + if resolved.startswith("\\\\"): + assert is_path_contained(resolved, host_root) + else: + assert is_path_contained(resolved, drive + "\\") + assert not is_path_contained(resolved, host_root) + finally: + _run("net", "use", drive, "/DELETE", "/Y") + + +def test_common_ancestor_across_shares_on_one_host(smb_share): + """Files on two shares of one host share only the host. + + ``os.path.commonpath`` raises ``ValueError: Paths don't have the same drive`` for + this pair, which is what made the download summary crash. + """ + unc_root, _ = smb_share + host_root = unc_root.rsplit("\\", 1)[0] + + ancestor = common_ancestor([rf"{host_root}\share1\a.exr", rf"{host_root}\share2\b.exr"]) + assert ancestor.rstrip("\\").lower() == host_root.lower(), ancestor diff --git a/test/unit/deadline_client/cli/test_cli_bundle_submit_known_paths.py b/test/unit/deadline_client/cli/test_cli_bundle_submit_known_paths.py index 4a827f437..dfc4d2929 100644 --- a/test/unit/deadline_client/cli/test_cli_bundle_submit_known_paths.py +++ b/test/unit/deadline_client/cli/test_cli_bundle_submit_known_paths.py @@ -4,6 +4,7 @@ Tests for the known asset paths functionality in the bundle_submit CLI command. """ +import ntpath import os import json import tempfile @@ -16,6 +17,7 @@ from deadline.client import config from deadline.client.cli import main +from deadline.client.api import _submit_job_bundle as sjb from deadline.client.api._submit_job_bundle import ( _filter_redundant_known_paths, _generate_message_for_asset_paths, @@ -44,14 +46,19 @@ ], ) def test_filter_redundant_known_paths(input, expected): + if os.name == "nt": + # The filter normalizes its roots, so a '/a' root comes back spelled '\a' here. + # Redundancy filtering is what these cases pin; the separator is os.path's business. + expected = [path.replace("/", "\\") for path in expected] assert sorted(_filter_redundant_known_paths(input)) == expected if os.name == "nt": - assert sorted(_filter_redundant_known_paths(path.replace("/", "\\") for path in input)) == [ - path.replace("/", "\\") for path in expected - ] + assert ( + sorted(_filter_redundant_known_paths(path.replace("/", "\\") for path in input)) + == expected + ) assert sorted( _filter_redundant_known_paths("C:" + path.replace("/", "\\") for path in input) - ) == ["C:" + path.replace("/", "\\") for path in expected] + ) == ["C:" + path for path in expected] @pytest.mark.parametrize( @@ -146,6 +153,146 @@ def test_is_known_path(path, roots, expected): assert _is_known_path(path, roots) is expected +@pytest.mark.parametrize( + "path, roots, expected", + [ + # Regression for https://github.com/aws-deadline/deadline-cloud/issues/1321: + # a host-level UNC root must contain paths under any of its shares. + ( + r"\\192.168.20.20\projects\assets\FA_Anim\260304_FA_Anim.c4d", + [r"\\192.168.20.20"], + True, + ), + (r"\\host\share\file", [r"\\host"], True), + (r"\\host\share\file", ["\\\\host\\"], True), + (r"\\host\share\file", [r"\\host\share"], True), + # Neither a different nor a prefix-sharing host is contained. + (r"\\other\share\file", [r"\\host"], False), + (r"\\host2\share\file", [r"\\host"], False), + (r"\\host\share2\file", [r"\\host\share"], False), + # A UNC candidate is not contained by a drive-letter root, and vice versa. + (r"\\host\share\file", [r"C:\trusted"], False), + (r"C:\trusted\file", [r"\\host\share"], False), + # Contained by the second of several roots, including a mismatched-drive first root. + (r"\\host\share\file", [r"D:\other", r"\\host"], True), + # A bare UNC anchor names no server, so it must not trust every reachable share. It + # passes the isabs filter, so '--known-asset-path \\' reaches here as a root. + (r"\\corp\finance\salaries.xlsx", ["\\\\"], False), + (r"\\corp\finance\salaries.xlsx", ["//"], False), + (r"\\corp\finance\salaries.xlsx", ["\\\\?\\UNC\\"], False), + # A useless root must not shadow a real one that follows it. + (r"\\host\share\file", ["\\\\", r"\\host"], True), + ], +) +def test_is_known_path_windows_semantics(path, roots, expected): + """Windows path semantics, exercised via ntpath so the cases run on every platform.""" + with patch.object(sjb.os, "path", ntpath): + assert _is_known_path(path, roots) is expected + + +@pytest.mark.parametrize( + "input, expected", + [ + # A host-level root makes its shares redundant. + ([r"\\host", r"\\host\share"], [r"\\host"]), + ([r"\\host\share", r"\\host"], [r"\\host"]), + ([r"\\host\share\a", r"\\host"], [r"\\host"]), + # Distinct hosts and shares are all kept. + ([r"\\host\s1", r"\\host\s2"], [r"\\host\s1", r"\\host\s2"]), + ([r"\\host1", r"\\host2"], [r"\\host1", r"\\host2"]), + # A host sharing a string prefix is not made redundant. + ([r"\\host", r"\\host2\share"], [r"\\host", r"\\host2\share"]), + # Case variants of the same location are redundant on Windows. + ([r"\\host\Share", r"\\HOST\share\sub"], [r"\\host\Share"]), + ([r"C:\proj", r"c:\PROJ\sub"], [r"C:\proj"]), + # Drive-letter roots stay separate from UNC roots. + ([r"C:\proj", r"\\host\share"], [r"C:\proj", r"\\host\share"]), + ], +) +def test_filter_redundant_known_paths_windows_semantics(input, expected): + # abspath is left native so the already-absolute inputs pass through unchanged. + with patch.object(sjb.os.path, "abspath", lambda p: p), patch.object(sjb.os, "path", ntpath): + assert _filter_redundant_known_paths(input) == expected + + +def test_filter_redundant_known_paths_expands_user_paths(): + """ + A '~'-prefixed root has to be expanded to match an absolute candidate. Such a root + reaches here from the config file and the CLI job submitter's default data + directory, neither of which goes through shell expansion. + """ + home_root = os.path.join("~", "projects") + expected_home = os.path.join(os.path.expanduser("~"), "projects") + + assert _filter_redundant_known_paths([home_root]) == [expected_home] + assert _is_known_path(os.path.join(expected_home, "scene.ma"), [expected_home]) is True + + # Expanding must not defeat redundancy filtering: '~/projects' and its subdirectory + # name the same tree, so only the ancestor survives. + assert _filter_redundant_known_paths([home_root, os.path.join(home_root, "sub")]) == [ + expected_home + ] + + +@pytest.mark.parametrize( + "known_path", + [ + # An empty known path reaches this code from `--known-asset-path ""`, from the + # MCP tool's unvalidated JSON array, and from a PATH/FILE job parameter whose + # allowedValues suppressed absolutization (os.path.dirname("scene.ma") == ""). + "", + # Relative roots, including the Windows root-relative and drive-relative forms. + "assets", + os.path.join("..", "shared"), + "\\projects", + "C:rel", + ], +) +def test_filter_redundant_known_paths_drops_unanchored_paths(known_path): + """ + A root that names no absolute location must be dropped, not resolved against the cwd. + + It matches no candidate either way, but dropping it at the boundary means a future + caller cannot turn it into a trusted tree: os.path.abspath("") is the whole working + directory, which would suppress the unknown-asset-path warning and let a + non-interactive submit upload undesignated files. + """ + assert _filter_redundant_known_paths([known_path]) == [] + + # A real root alongside an unanchored one still survives. + real_root = os.path.abspath(os.path.join(os.sep, "trusted", "project")) + assert _filter_redundant_known_paths([known_path, real_root]) == [real_root] + + +def test_filter_redundant_known_paths_unanchored_path_does_not_trust_cwd(): + """The working directory must not become a known root via an empty path.""" + cwd_file = os.path.join(os.getcwd(), "unrelated_secret.txt") + assert _is_known_path(cwd_file, _filter_redundant_known_paths([""])) is False + + +def test_generate_message_for_asset_paths_unc_host_root_is_known(): + """ + Regression for issue #1321: files on a share under a host-level UNC known root + must not trigger the unknown-path warning. + """ + known_root = r"\\192.168.20.20" + inside_file = r"\\192.168.20.20\projects\assets\FA_Anim\260304_FA_Anim.c4d" + + upload_group = AssetUploadGroup( + asset_groups=[AssetRootGroup(root_path=r"\\192.168.20.20\projects", inputs={inside_file})], # type: ignore[arg-type] + total_input_files=1, + total_input_bytes=12, + ) + + with patch("deadline.client.api._submit_job_bundle.os.path", ntpath): + message, no_warnings = _generate_message_for_asset_paths( + upload_group, storage_profile=None, known_asset_paths=[known_root] + ) + + assert no_warnings is True, message + assert "WARNING: Files were specified outside of known asset paths." not in message, message + + def test_generate_message_for_asset_paths_sibling_prefix_is_unknown(): """ Security regression test: a known root must NOT "contain" a sibling path that diff --git a/test/unit/deadline_client/cli/test_cli_job.py b/test/unit/deadline_client/cli/test_cli_job.py index 61b51e8d9..d7ba97057 100644 --- a/test/unit/deadline_client/cli/test_cli_job.py +++ b/test/unit/deadline_client/cli/test_cli_job.py @@ -7,6 +7,7 @@ from datetime import timezone import datetime import json +import ntpath import os from typing import Dict, List import pytest @@ -913,6 +914,38 @@ def test_get_summary_of_files_to_download_message_windows( ) +@pytest.mark.parametrize( + "output_paths_by_root, expected_result", + [ + # A root under a UNC share summarizes to the shared subdirectory. + ( + {r"\\host\share": ["renders/image1.png", "renders/image2.png"]}, + "\nSummary of files to download:\n \\\\host\\share\\renders (2 files)\n", + ), + # Files directly at a UNC share root summarize to the share itself. os.path.commonpath + # returns '\\\\host\\share\\' here, leaving a stray trailing separator in the message. + ( + {r"\\host\share": ["image1.png", "image2.png"]}, + "\nSummary of files to download:\n \\\\host\\share (2 files)\n", + ), + ( + {r"\\host\share": ["only.png"]}, + "\nSummary of files to download:\n \\\\host\\share\\only.png (1 file)\n", + ), + ], +) +def test_get_summary_of_files_to_download_message_unc_paths( + output_paths_by_root: Dict[str, List[str]], + expected_result: str, +): + """UNC path summaries, exercised via ntpath so the cases run on every platform.""" + with patch.object(job_group.os, "path", ntpath): + assert ( + _get_summary_of_files_to_download_message(output_paths_by_root, is_json_format=False) + == expected_result + ) + + def test_cli_job_wait_succeeded(fresh_deadline_config): """ Test that job wait command returns exit code 0 when job succeeds. diff --git a/test/unit/deadline_client/job_bundle/test_job_bundle_loader.py b/test/unit/deadline_client/job_bundle/test_job_bundle_loader.py index 01858a774..08f5e3bce 100644 --- a/test/unit/deadline_client/job_bundle/test_job_bundle_loader.py +++ b/test/unit/deadline_client/job_bundle/test_job_bundle_loader.py @@ -7,13 +7,17 @@ """ import json +import ntpath import os import sys +from contextlib import contextmanager +from unittest.mock import patch import pytest import yaml from deadline.client.exceptions import DeadlineOperationError +from deadline.client.job_bundle import loader from deadline.client.job_bundle.loader import ( parse_yaml_or_json_content, read_yaml_or_json, @@ -241,6 +245,87 @@ def test_validate_directory_symlink_containment_fail(tmpdir): validate_directory_symlink_containment(str(test_root)) +class TestSymlinkContainmentWindowsPaths: + """ + Windows path semantics for validate_directory_symlink_containment, exercised through + a simulated ntpath filesystem so the cases run on every platform. + + os.path.commonpath raises ValueError for a bundle located at a UNC share root + ('\\\\host\\share' vs '\\\\host\\share\\template.yaml' -> "Can't mix absolute and + relative paths"), and for a symlink escaping a drive-letter bundle onto a UNC share + ('C:\\bundle' vs '\\\\host\\share\\x' -> "Paths don't have the same drive"). Neither + exception is caught, so both would surface as a raw ValueError rather than a + containment verdict. + """ + + @contextmanager + def _simulated_windows_bundle(self, bundle_dir, entries, resolves_to): + """Simulate an ntpath filesystem holding ``entries`` under ``bundle_dir``. + + ``resolves_to`` maps a normalized path to the location it resolves to, standing + in for a symlink target. + """ + + class _WindowsPath: + def __getattr__(self, name): + return getattr(ntpath, name) + + @staticmethod + def isdir(path): + return path == bundle_dir + + @staticmethod + def realpath(path): + return resolves_to.get(ntpath.normpath(path), ntpath.normpath(path)) + + def walk(top): + yield top, [], list(entries) + + with patch.object(loader.os, "walk", walk), patch.object(loader.os, "path", _WindowsPath()): + yield + + def test_bundle_at_unc_share_root_is_valid(self): + """A bundle directory that is itself a UNC share root contains its own files.""" + bundle_dir = r"\\host\share" + with self._simulated_windows_bundle(bundle_dir, ["template.yaml"], {}): + validate_directory_symlink_containment(bundle_dir) + + def test_bundle_under_unc_share_is_valid(self): + bundle_dir = r"\\host\share\bundle" + with self._simulated_windows_bundle(bundle_dir, ["template.yaml"], {}): + validate_directory_symlink_containment(bundle_dir) + + def test_symlink_escaping_unc_share_root_is_rejected(self): + bundle_dir = r"\\host\share" + with self._simulated_windows_bundle( + bundle_dir, + ["escape.yaml"], + {r"\\host\share\escape.yaml": r"\\host\other\secret.yaml"}, + ): + with pytest.raises(DeadlineOperationError): + validate_directory_symlink_containment(bundle_dir) + + def test_symlink_from_drive_bundle_onto_unc_share_is_rejected(self): + bundle_dir = r"C:\bundle" + with self._simulated_windows_bundle( + bundle_dir, + ["escape.yaml"], + {r"C:\bundle\escape.yaml": r"\\host\share\secret.yaml"}, + ): + with pytest.raises(DeadlineOperationError): + validate_directory_symlink_containment(bundle_dir) + + def test_symlink_to_sibling_prefix_directory_is_rejected(self): + bundle_dir = r"C:\bundle" + with self._simulated_windows_bundle( + bundle_dir, + ["escape.yaml"], + {r"C:\bundle\escape.yaml": r"C:\bundle-secret\secret.yaml"}, + ): + with pytest.raises(DeadlineOperationError): + validate_directory_symlink_containment(bundle_dir) + + class TestHiddenParameterValidation: """Tests for hidden parameter validation in read_job_bundle_parameters.""" diff --git a/test/unit/deadline_client/job_bundle/test_job_parameters.py b/test/unit/deadline_client/job_bundle/test_job_parameters.py index 0e960c802..f44b86280 100644 --- a/test/unit/deadline_client/job_bundle/test_job_parameters.py +++ b/test/unit/deadline_client/job_bundle/test_job_parameters.py @@ -7,6 +7,11 @@ from __future__ import annotations +import ntpath +from contextlib import contextmanager +from copy import deepcopy +from unittest.mock import patch + import pytest from deadline.client.job_bundle import parameters @@ -686,3 +691,86 @@ def test_ui_control_for_parameter_definition_errors(parameter_def): def test_parameter_definition_difference(parameter1, parameter2, expected_difference): """Test that parameter_definition_difference returns expected differences.""" assert parameters.parameter_definition_difference(parameter1, parameter2) == expected_difference + + +class TestPathDefaultContainmentWindowsPaths: + """ + Windows path semantics for the PATH-default containment check in + read_job_bundle_parameters, exercised through a simulated ntpath filesystem so the + cases run on every platform. + + os.path.commonpath raises ValueError when the bundle sits at a UNC share root + ('\\\\host\\share' vs '\\\\host\\share\\sub' -> "Can't mix absolute and relative + paths"). That exception is not caught, so a valid template would fail to load with a + raw ValueError instead of resolving its default. + """ + + TEMPLATE = { + "specificationVersion": "jobtemplate-2023-09", + "name": "PathDefault", + "parameterDefinitions": [ + { + "name": "OutDir", + "type": "PATH", + "objectType": "DIRECTORY", + "dataFlow": "OUT", + "default": "output", + } + ], + } + + @contextmanager + def _simulated_windows_bundle(self, bundle_dir, resolves_to=None): + resolves_to = resolves_to or {} + + class _WindowsPath: + def __getattr__(self, name): + return getattr(ntpath, name) + + @staticmethod + def realpath(path): + return resolves_to.get(ntpath.normpath(path), ntpath.normpath(path)) + + def read_yaml_or_json_object(bundle_dir, filename, required): + # Deep-copied because read_job_bundle_parameters sets 'value' on the + # parameter definitions in place. + return deepcopy(self.TEMPLATE) if filename == "template" else None + + with ( + patch.object(parameters.os, "path", _WindowsPath()), + patch.object(parameters, "read_yaml_or_json_object", read_yaml_or_json_object), + ): + yield + + def _out_dir_value(self, result): + return next(p for p in result if p["name"] == "OutDir")["value"] + + def test_bundle_at_unc_share_root_resolves_default(self): + bundle_dir = r"\\host\share" + with self._simulated_windows_bundle(bundle_dir): + result = parameters.read_job_bundle_parameters(bundle_dir) + assert self._out_dir_value(result) == r"\\host\share\output" + + def test_bundle_under_unc_share_resolves_default(self): + bundle_dir = r"\\host\share\bundle" + with self._simulated_windows_bundle(bundle_dir): + result = parameters.read_job_bundle_parameters(bundle_dir) + assert self._out_dir_value(result) == r"\\host\share\bundle\output" + + def test_default_resolving_outside_unc_share_is_rejected(self): + bundle_dir = r"\\host\share\bundle" + with self._simulated_windows_bundle( + bundle_dir, + {r"\\host\share\bundle\output": r"\\host\other\secret"}, + ): + with pytest.raises(exceptions.DeadlineOperationError): + parameters.read_job_bundle_parameters(bundle_dir) + + def test_default_resolving_from_drive_bundle_onto_unc_share_is_rejected(self): + bundle_dir = r"C:\bundle" + with self._simulated_windows_bundle( + bundle_dir, + {r"C:\bundle\output": r"\\host\share\secret"}, + ): + with pytest.raises(exceptions.DeadlineOperationError): + parameters.read_job_bundle_parameters(bundle_dir) diff --git a/test/unit/deadline_client/test_path_summary.py b/test/unit/deadline_client/test_path_summary.py new file mode 100644 index 000000000..514577e7f --- /dev/null +++ b/test/unit/deadline_client/test_path_summary.py @@ -0,0 +1,109 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + +""" +Tests for the path-group summary helper. + +Windows semantics go through an explicit ``ntpath`` so these run on every platform. +""" + +import ntpath +import posixpath + +import pytest + +from deadline.client._path_summary import common_ancestor +from deadline.client._path_utils import is_path_contained + + +@pytest.mark.parametrize("path_module", [ntpath, posixpath]) +def test_common_ancestor_contains_its_inputs(path_module): + """A non-empty common_ancestor must contain every path it was derived from.""" + paths = ( + [ + r"\\host\share\a", + r"\\host\s2\b", + r"C:\a\b", + r"C:\a\c", + "C:foo", + r"..\a\b", + r"..\a\c", + # Drive-relative '..' puts the run behind an anchor, where a guard counting + # from index zero would miss it. + r"C:..\x", + r"C:..\..\x", + r"C:..\a\y", + r"\\?\C:", + r"\\?\C:\a", + ] + if path_module is ntpath + else ["/a/b", "/a/c", "//a/d", "../a/b", "../a/c", "../../a/b", "rel/f", "rel/g"] + ) + for first in paths: + for second in paths: + ancestor = common_ancestor([first, second], path_module=path_module) + if not ancestor: + continue + assert is_path_contained(first, ancestor, path_module=path_module), (first, ancestor) + assert is_path_contained(second, ancestor, path_module=path_module), (second, ancestor) + + +@pytest.mark.parametrize( + "paths, path_module, expected", + [ + # The common ancestor of paths under one share, spelled with its real case. + ( + [r"\\host\Share\Proj\a.txt", r"\\host\Share\Proj\sub\b.txt"], + ntpath, + r"\\host\Share\Proj", + ), + # Different shares on one host share only the host. os.path.commonpath raises + # ValueError for this pair. + ([r"\\host\s1\a", r"\\host\s2\b"], ntpath, r"\\host"), + # Different hosts share nothing. There is no location above a UNC host, so the + # bare '\\\\' that their leading components have in common is not an answer. + ([r"\\host1\s\a", r"\\host2\s\b"], ntpath, ""), + ([r"\\host1", r"\\host2"], ntpath, ""), + # Different drives share nothing. + ([r"C:\a\b", r"D:\a\b"], ntpath, ""), + ([r"C:\a\b", r"\\host\share\b"], ntpath, ""), + ([r"C:\proj\a", r"C:\proj\b"], ntpath, r"C:\proj"), + ([r"C:\proj\a"], ntpath, r"C:\proj\a"), + (["/a/b/c", "/a/b/d"], posixpath, "/a/b"), + (["/a/b", "/c/d"], posixpath, "/"), + # A doubled POSIX root is the same space as '/', so these behave like ordinary + # absolute paths rather than a separate namespace. + (["//a/b", "//c/d"], posixpath, "/"), + (["//a/b", "//a/c"], posixpath, "/a"), + (["/a", "/b"], posixpath, "/"), + (["/a/b"], posixpath, "/a/b"), + (["a/b", "/c/d"], posixpath, ""), + ([], posixpath, ""), + # Paths whose unresolved leading '..' runs differ in depth are rooted at + # different unknown directories, so they share none. Positional comparison + # would wrongly read the shared '..' as one directory and return '..', which + # is not an ancestor of '../../up'. os.path.commonpath has that bug. + (["../up", "../../up"], posixpath, ""), + (["../../up", "../up"], posixpath, ""), + ([r"..\up", r"..\..\up"], ntpath, ""), + # The '..' run can sit behind an anchor, where a guard counting from index 0 + # would not see it. 'C:..' is the cwd's parent on C:, 'C:..\..' its grandparent. + ([r"C:..\x", r"C:..\..\x"], ntpath, ""), + ([r"C:..", r"C:..\.."], ntpath, ""), + # Equal depth behind an anchor is still comparable. + ([r"C:..\a\x", r"C:..\a\y"], ntpath, r"C:..\a"), + # Equal '..' depth is comparable again. + (["../a/x", "../a/y"], posixpath, "../a"), + (["../../a/x", "../../a/y"], posixpath, "../../a"), + # Relative inputs keep their own spelling and gain no leading separator. A + # Windows-style path read under posixpath semantics is one of these, since + # 'C:' is an ordinary component there rather than a drive. + (["a/b/c", "a/b/d"], posixpath, "a/b"), + ( + ["C:/Users/u/renders/i1.png", "C:/Users/u/renders/i2.png"], + posixpath, + "C:/Users/u/renders", + ), + ], +) +def test_common_ancestor(paths, path_module, expected): + assert common_ancestor(paths, path_module=path_module) == expected diff --git a/test/unit/deadline_client/test_path_utils.py b/test/unit/deadline_client/test_path_utils.py new file mode 100644 index 000000000..1797d457b --- /dev/null +++ b/test/unit/deadline_client/test_path_utils.py @@ -0,0 +1,615 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + +""" +Tests for the path containment helpers. + +Windows semantics go through an explicit ``ntpath`` so these run on every platform. UNC +paths cannot be built with ``os.path.join(os.sep, ...)``, so tests written against the +native path module silently skip them on POSIX. +""" + +import itertools +import ntpath +import posixpath +import sys +from pathlib import PurePosixPath, PureWindowsPath +from typing import Any + +import pytest + +from deadline.client._path_utils import ( + _splitroot, + is_absolute_path, + is_any_path_contained, + is_path_contained, + normalized_path, + path_components, +) + + +@pytest.mark.parametrize( + "candidate, root, expected", + [ + # Regression for https://github.com/aws-deadline/deadline-cloud/issues/1321: + # a host-level UNC root contains the shares beneath it. os.path.commonpath + # rejects this pair because it reads '\\192.168.20.20' as having no drive but + # '\\192.168.20.20\projects' as being a drive. + ( + r"\\192.168.20.20\projects\assets\FA_Anim\260304_FA_Anim.c4d", + r"\\192.168.20.20", + True, + ), + (r"\\host\share\file", r"\\host", True), + (r"\\host\share", r"\\host", True), + (r"\\host", r"\\host", True), + # A trailing separator on a host-level root does not change containment. + (r"\\host\share\file", "\\\\host\\", True), + # A different host is not contained. + (r"\\other\share\file", r"\\host", False), + # A host that merely shares a string prefix is not contained. + (r"\\host2\share\file", r"\\host", False), + # Share-level roots behave like directories. + (r"\\host\share\file", r"\\host\share", True), + (r"\\host\share", r"\\host\share", True), + (r"\\host\share2\file", r"\\host\share", False), + # The host is not contained by one of its shares. + (r"\\host", r"\\host\share", False), + # Forward slashes are accepted on Windows. + ("//host/share/file", r"\\host", True), + (r"\\host\share\file", "//host/share", True), + # Case-insensitive, matching the filesystem. + (r"\\HOST\Share\File", r"\\host\share", True), + # Drive letters. + (r"C:\trusted\project\sub\file", r"C:\trusted\project", True), + (r"C:\trusted\project", r"C:\trusted\project", True), + (r"C:\trusted\project-secret\file", r"C:\trusted\project", False), + (r"C:\trusted\projectextra", r"C:\trusted\project", False), + (r"C:\trusted", r"C:\trusted\project", False), + (r"c:\trusted\project\file", r"C:\TRUSTED\Project", True), + # '..' is resolved before comparing. + (r"C:\trusted\project\..\project-secret\f", r"C:\trusted\project", False), + (r"C:\trusted\project\sub\..\f", r"C:\trusted\project", True), + (r"\\host\share\a\..\..\b\f", r"\\host\share\a", False), + # Windows clamps '..' at a share root, so this stays inside the share. + (r"\\host\share\sub\..\..\other\f", r"\\host\share", True), + # A '..' that normpath cannot resolve (there is no share to clamp against) + # fails closed rather than being read as a component named '..'. + (r"\\host\..\other\share\f", r"\\host", False), + # Mismatched drives are simply not contained; no exception. + (r"D:\trusted\project\file", r"C:\trusted\project", False), + (r"\\host\share\file", r"C:\trusted\project", False), + (r"C:\trusted\project\file", r"\\host\share", False), + # A drive-relative path ('C:file' means 'file' relative to the cwd on C:) + # cannot be resolved here, so it fails closed. + ("C:file", "C:\\", False), + # Relative paths are not contained by absolute roots and vice versa. + (r"relative\file", r"C:\trusted", False), + (r"C:\trusted\file", r"relative", False), + # An extended-length prefix only turns off Win32 normalization; it denotes an + # ordinary location, so it folds to the plain spelling and compares equal to it in + # either direction. job-attachments carries the '\\?\' form through its internals + # and strips it only at display boundaries, so a prefixed path does reach here. + (r"\\?\C:\trusted\project\file", r"C:\trusted\project", True), + (r"C:\trusted\project\file", r"\\?\C:\trusted\project", True), + (r"\\?\UNC\host\share\file", r"\\host\share", True), + (r"\\host\share\file", r"\\?\UNC\host\share", True), + (r"\\?\UNC\host\share\file", r"\\host", True), + (r"\\?\C:\trusted\project\file", r"\\?\C:\trusted\project", True), + (r"\\?\UNC\host\share\file", r"\\?\UNC\host\share", True), + # Folding does not weaken component anchoring: a sibling that merely shares a + # string prefix is still outside the root. + (r"\\?\C:\trusted\project-secret\f", r"\\?\C:\trusted\project", False), + # A rooted, driveless root ('\') is a different path space than the UNC + # namespace, so it must not contain remote paths -- nor they it. + (r"\\attacker\share\evil", "\\", False), + ("\\x", "\\\\", False), + ("\\x", "\\", True), + # The bare anchor names no server, so it is an ancestor of nothing -- treating it as + # POSIX '/' would trust every reachable share. It counts as fully qualified, so a + # caller filtering roots on is_absolute_path lets it reach here; '//' and + # '\\?\UNC\' normalize to it. + (r"\\host\share\file", "\\\\", False), + (r"\\host\share\file", "\\\\\\\\", False), + (r"\\host\share\file", "//", False), + (r"\\host\share\file", "\\\\?\\UNC\\", False), + (r"\\host", "\\\\", False), + # The anchor is still reflexive, and a root naming an actual server still works. + ("\\\\", "\\\\", True), + (r"\\host\share\file", r"\\host", True), + # 'C:' means the cwd on drive C:, so it contains drive-relative paths but not + # the drive root's absolute contents. + (r"C:\Windows", "C:", False), + ("C:foo", "C:", True), + (r"C:\a", "C:\\", True), + # A prefixed drive with no plain spelling keeps its own space: a device path + # must not alias the drive it resembles, in either direction. + (r"\\.\C:\secret", "C:\\", False), + (r"C:\secret", r"\\.\C:", False), + (r"\\?\Volume{abc}\trusted\f", r"Volume{abc}\trusted", False), + (r"\\?\Volume{abc}\trusted\f", r"\\?\Volume{abc}\trusted", True), + # '\\?\C:' folds to the drive-relative 'C:' space and '\\?\C:\' to the drive root, + # so each behaves as the plain spelling it denotes -- including keeping those two + # spaces apart, which is why the last two disagree. + ("C:foo", r"\\?\C:", True), + (r"C:\a", r"\\?\C:", False), + (r"C:\a\f", "\\\\?\\C:\\", True), + (r"\\?\C:\a", "\\\\?\\C:\\", True), + (r"\\?\C:\a", r"\\?\C:", False), + # A relative path is contained in itself even when normpath leaves a leading + # '..' it cannot cancel; only a '..' below the root can climb back out. + (r"..\a", r"..\a", True), + (r"..\a\b", r"..\a", True), + (r"..\a\..\b", r"..\a", False), + ], +) +def test_is_path_contained_windows(candidate, root, expected): + assert is_path_contained(candidate, root, path_module=ntpath) is expected + + +@pytest.mark.parametrize( + "candidate, root, expected", + [ + ("/trusted/project", "/trusted/project", True), + ("/trusted/project/sub/file", "/trusted/project", True), + ("/trusted/project/file", "/trusted/project/", True), + ("/trusted/project-secret/f", "/trusted/project", False), + ("/trusted/projectextra", "/trusted/project", False), + ("/trusted", "/trusted/project", False), + ("/somewhere/else", "/trusted/project", False), + ("/trusted/project/../project-secret/f", "/trusted/project", False), + ("/trusted/project/sub/../f", "/trusted/project", True), + ("relative/file", "/trusted/project", False), + ("/trusted/file", "relative", False), + # Everything absolute is contained by the root directory. + ("/trusted/project", "/", True), + # POSIX paths are case-sensitive, so a case variant fails closed. + ("/Trusted/Project/f", "/trusted/project", False), + # A backslash is an ordinary filename character on POSIX, so a Windows-style + # UNC string is just a relative filename and matches nothing. + (r"\\host\share\file", r"\\host", False), + # A doubled root names the same file, so containment does not depend on how + # many leading slashes either side was spelled with. + ("//mnt/shared/f", "/mnt/shared", True), + ("///mnt/shared/f", "/mnt/shared", True), + ("/mnt/shared/f", "//mnt/shared", True), + # Reflexive, and tolerant of a leading '..' shared with the root. + ("../a", "../a", True), + ("../a/b", "../a", True), + ("../a/../b", "../a", False), + ("..", "..", True), + # A root of '.' is not the working directory: normpath renders it as a lone '.' + # component, which prefixes nothing. Unreachable (callers pass absolute roots) and + # fails closed, so it is pinned as a known limitation rather than fixed. + ("rel", ".", False), + ("rel/f", ".", False), + ("/abs/f", ".", False), + (".", ".", True), + (".", "rel", False), + ], +) +def test_is_path_contained_posix(candidate, root, expected): + assert is_path_contained(candidate, root, path_module=posixpath) is expected + + +@pytest.mark.parametrize("path_module", [ntpath, posixpath]) +def test_is_path_contained_is_reflexive(path_module): + """Every path contains itself, whatever space it is in.""" + paths = ( + [ + r"\\host", + r"\\host\share\a", + "C:", + "C:\\", + r"C:\a", + "\\", + r"..\a", + r"rel\f", + ".", + r"C:..\x", + r"C:..\..\x", + r"\\?\C:", + ] + if path_module is ntpath + else ["/", "//", "/a/b", "../a", "../../a", "rel/f", "."] + ) + for path in paths: + assert is_path_contained(path, path, path_module=path_module) is True, path + + +@pytest.mark.parametrize( + "root, contained", + [ + # The reported case: a host-level root and a file on one of its shares. Before + # Python 3.11 both normpath and splitdrive strip a share-less UNC path down to a + # rooted-driveless one ('\\host' -> '\host', splitdrive -> no drive), which put the + # root in a different path space than the candidate and left #1321 unfixed on 3.9 + # and 3.10. + (r"\\host", True), + ("\\\\host\\", True), + (r"\\host\share", True), + # A different server, and the bare anchor that names none, must not contain it. + (r"\\host2", False), + ("\\\\", False), + ("\\", False), + ], +) +def test_host_level_unc_root_containment_is_version_independent(root, contained): + """Issue #1321 on every supported interpreter, not just 3.10+.""" + assert is_path_contained(r"\\host\share\f", root, path_module=ntpath) is contained + + +class _PreThreeElevenNtpath: + """``ntpath`` as it behaved before Python 3.11 for a UNC path that names no share. + + Both ``normpath`` and ``splitdrive`` stripped such a path down to a rooted, driveless + one. Injecting this exercises that branch on any interpreter, rather than only on the + 3.9 and 3.10 jobs -- the same reason the rest of this file injects ``ntpath``. + """ + + # Forces the _splitroot backport, which is what those versions had. + splitroot = None + + @staticmethod + def _is_shareless_unc(text: str) -> bool: + return text.startswith("\\\\") and "\\" not in text[2:] + + @staticmethod + def normpath(text: str) -> str: + result = ntpath.normpath(text) + if _PreThreeElevenNtpath._is_shareless_unc(result): + return result[1:] + return result + + @staticmethod + def splitdrive(text: str): + if _PreThreeElevenNtpath._is_shareless_unc(text): + return "", text + return ntpath.splitdrive(text) + + def __getattr__(self, name): + return getattr(ntpath, name) + + +def test_host_level_unc_root_survives_pre_3_11_normpath(): + """A host-level root stays in the UNC space even when normpath collapses its anchor.""" + legacy: Any = _PreThreeElevenNtpath() + # Confirm the proxy actually reproduces the old behavior, so this cannot pass vacuously. + assert legacy.normpath("\\\\host") == "\\host" + assert legacy.splitdrive("\\\\host") == ("", "\\\\host") + + assert path_components(r"\\host", path_module=legacy) == ["\\\\", "host"] + assert is_path_contained(r"\\host\share\f", r"\\host", path_module=legacy) is True + assert normalized_path(r"\\host", path_module=legacy) == r"\\host" + # A rooted, driveless path must not be promoted into the UNC space by the restore. + assert path_components(r"\host", path_module=legacy) == ["\\", "host"] + assert is_path_contained(r"\\host\share\f", "\\", path_module=legacy) is False + + +@pytest.mark.parametrize( + "prefixed, plain", + [ + (r"\\?\C:\proj\a.txt", r"C:\proj\a.txt"), + (r"\\?\UNC\host\share\a.txt", r"\\host\share\a.txt"), + ], +) +def test_extended_length_prefix_agrees_with_plain_spelling(prefixed, plain): + """A prefixed path is contained by exactly the roots its plain spelling is. + + job-attachments carries the '\\\\?\\' form through its internals and strips it only at + display boundaries, so a prefixed path can reach a containment check. Treating it as its + own path space would report it outside a root that plainly contains it. + """ + roots = [ + plain, + ntpath.dirname(plain), + r"C:\proj", + "C:\\", + r"\\host\share", + r"\\host", + r"D:\other", + ] + for root in roots: + assert is_path_contained(prefixed, root, path_module=ntpath) is is_path_contained( + plain, root, path_module=ntpath + ), root + # A prefixed *root* folds the same way, so it behaves like its plain spelling. + assert is_path_contained(plain, root, path_module=ntpath) is is_path_contained( + plain, _prefixed_form(root), path_module=ntpath + ), root + + +def _prefixed_form(path: str) -> str: + """Spell ``path`` in extended-length form.""" + if path.startswith("\\\\"): + return "\\\\?\\UNC" + path[1:] + return "\\\\?\\" + path + + +def test_extended_length_prefix_resolves_dot_segments_uniformly(): + """normpath leaves '..' alone inside a '\\\\?\\' path before 3.10 and collapses it after. + + Folding to the plain spelling first makes the components the same on every supported + interpreter, so containment does not depend on the running Python. + """ + assert path_components(r"\\?\C:\a\..\b", path_module=ntpath) == ["c:\\", "b"] + assert path_components(r"\\?\UNC\host\share\a\..\b", path_module=ntpath) == [ + "\\\\", + "host", + "share", + "b", + ] + + +@pytest.mark.parametrize("path_module", [ntpath, posixpath]) +def test_splitroot_backport_matches_stdlib(path_module): + """The Python < 3.12 shim must agree with splitroot on every path space. + + Python 3.12 added ``splitroot``; this project supports 3.9, so on older + interpreters the shim is what distinguishes one path space from another. Hiding + ``splitroot`` exercises the shim on any interpreter. + """ + if not hasattr(path_module, "splitroot"): + pytest.skip("stdlib splitroot unavailable, nothing to compare against") + + cases = ( + [ + "\\\\", + "\\", + r"\\srv", + r"\\srv\share", + r"\\srv\share\a", + "C:", + "C:\\", + "C:foo", + r"C:\a", + "C:\\\\a", + r"\\?\C:\a", + r"\\?\UNC\srv\sh\a", + r"\\?\Volume{abc}\a", + r"\\.\C:\a", + "", + r"rel\f", + "\\\\\\srv", + ] + if path_module is ntpath + else ["/", "//", "///", "////", "/a", "//a/b", "///a", "rel", "rel/f", ""] + ) + + class _NoSplitroot: + """Proxy that hides splitroot so the backport path is taken.""" + + splitroot = None + + def __getattr__(self, name): + return getattr(path_module, name) + + for case in cases: + assert _splitroot(case, _NoSplitroot()) == path_module.splitroot(case), case + + +@pytest.mark.skipif( + sys.version_info < (3, 12), + reason="pathlib parses a host-only UNC path as drive-less before 3.12, so it is not a" + " usable oracle there.", +) +@pytest.mark.parametrize("path_module", [ntpath, posixpath]) +def test_agrees_with_pathlib_except_for_unc_hosts(path_module): + """Differential check against ``PurePath.is_relative_to`` as an independent oracle. + + pathlib folds a UNC server and share into one atom, so it cannot see a host-level root + as an ancestor of its shares -- that gap is issue #1321 and the only sanctioned + disagreement. Elsewhere pathlib is the reference. It does not resolve '..', so the + corpus avoids inputs needing normalization; this supplements the explicit cases above + rather than replacing them. + """ + if path_module is ntpath: + flavour: Any = PureWindowsPath + # Spans every path space, including the three where earlier versions of this + # module wrongly reported containment: bare roots, the device namespace, and + # drive-relative paths. + corpus = [ + r"\\srv", + r"\\srv\share", + r"\\srv\share\a", + r"\\srv\other\b", + r"\\srv2\share", + "C:\\", + r"C:\a", + r"C:\a\b", + r"C:\a-secret", + r"D:\a", + "rel", + r"rel\f", + "\\", + "\\\\", + r"\x", + "C:", + "C:foo", + r"\\.\C:\a", + r"\\?\Volume{abc}\a", + ] + else: + flavour = PurePosixPath + corpus = ["/", "/a", "/a/b", "/a-secret", "rel", "rel/f"] + + for candidate, root in itertools.permutations(corpus, 2): + ours = is_path_contained(candidate, root, path_module=path_module) + pathlibs = flavour(candidate).is_relative_to(flavour(root)) + if ours == pathlibs: + continue + # The only sanctioned disagreement: a UNC root that pathlib cannot see as an + # ancestor because it folds the server and share into one atom. We may only be + # more permissive than pathlib here, never elsewhere and never in reverse. + assert path_module is ntpath, (candidate, root, ours, pathlibs) + assert ours is True and pathlibs is False, (candidate, root, ours, pathlibs) + # The root must name an actual server. The bare '\\\\' anchor names none, so it + # is not a sanctioned disagreement -- pathlib is right to contain nothing there. + assert flavour(root).drive.startswith("\\\\"), (candidate, root) + assert str(root) != "\\\\", (candidate, root) + assert flavour(candidate).drive.startswith("\\\\"), (candidate, root) + assert not flavour(root).parts[1:], (candidate, root) + + +def test_is_any_path_contained(): + assert is_any_path_contained("/a/f", ["/b", "/a"]) is True + assert is_any_path_contained("/c/f", ["/b", "/a"]) is False + # No roots means nothing is contained. + assert is_any_path_contained("/a/f", []) is False + assert ( + is_any_path_contained(r"\\host\share\f", [r"D:\other", r"\\host"], path_module=ntpath) + is True + ) + + +@pytest.mark.parametrize( + "path, path_module, expected", + [ + # The first component is the path space; a UNC server and share are ordinary + # parts beneath the '\\\\' anchor, which is what lets a host-level root + # contain them. + (r"\\host", ntpath, ["\\\\", "host"]), + ("\\\\host\\", ntpath, ["\\\\", "host"]), + (r"\\host\share", ntpath, ["\\\\", "host", "share"]), + (r"\\host\share\a\b", ntpath, ["\\\\", "host", "share", "a", "b"]), + ("C:\\", ntpath, ["c:\\"]), + (r"C:\a", ntpath, ["c:\\", "a"]), + # 'C:' (drive-relative, meaning the cwd on C:) is a different space than 'C:\'. + ("C:", ntpath, ["c:"]), + ("C:foo", ntpath, ["c:", "foo"]), + # A rooted, driveless path is its own space, distinct from the UNC anchor. + ("\\", ntpath, ["\\"]), + ("\\\\", ntpath, ["\\\\"]), + # An extended-length prefix only turns off Win32 normalization: it denotes the + # same location, so it folds to the plain spelling rather than occupying a space + # of its own. Otherwise a prefixed path reads as outside a root that plainly + # contains it, which is the same false negative as issue #1321. + (r"\\?\C:\a", ntpath, ["c:\\", "a"]), + (r"\\?\c:\a", ntpath, ["c:\\", "a"]), + (r"\\?\C:", ntpath, ["c:"]), + ("//?/C:/a", ntpath, ["c:\\", "a"]), + (r"\\?\UNC\host\share", ntpath, ["\\\\", "host", "share"]), + (r"\\?\UNC\host\share\f", ntpath, ["\\\\", "host", "share", "f"]), + (r"\\?\unc\host\share\f", ntpath, ["\\\\", "host", "share", "f"]), + (r"\\?\UNC\host", ntpath, ["\\\\", "host"]), + # 'UNC' alone names no server, so it folds to the bare anchor, which contains + # nothing rather than prefixing every reachable share. + (r"\\?\UNC", ntpath, ["\\\\"]), + # These denote no plain path, so they keep their prefix and a space of their own + # and cannot alias the drive or UNC path they resemble. + (r"\\?\Volume{abc}\a", ntpath, ["\\\\?\\volume{abc}\\", "a"]), + (r"\\?\GLOBALROOT\Device\X\f", ntpath, ["\\\\?\\globalroot\\", "device", "x", "f"]), + (r"\\.\C:\a", ntpath, ["\\\\.\\c:\\", "a"]), + ("/", posixpath, ["/"]), + ("/a/b", posixpath, ["/", "a", "b"]), + ("/a/b/", posixpath, ["/", "a", "b"]), + ("a/b", posixpath, ["a", "b"]), + # '//foo' and '/foo' are the same file on every supported platform, so a + # doubled root collapses rather than forming a separate namespace. + ("//a/b", posixpath, ["/", "a", "b"]), + ("///a/b", posixpath, ["/", "a", "b"]), + ], +) +def test_path_components(path, path_module, expected): + assert path_components(path, path_module=path_module) == expected + + +@pytest.mark.parametrize( + "path, path_module, expected", + [ + # The reason this exists rather than calling isabs directly: before Python 3.11 + # ntpath.isabs tests what splitdrive leaves behind, and for a UNC path naming a + # share splitdrive consumes the whole string -- so isabs(r"\\host\s1") is False + # there. Callers gate trust on this, so a valid root was dropped and a valid + # PATH parameter value rejected. + (r"\\host\s1", ntpath, True), + (r"\\host\s1\f", ntpath, True), + ("\\\\host\\", ntpath, True), + (r"\\host", ntpath, True), + # The bare anchor names no server but is still fully qualified, so it reaches the + # containment check -- which rejects it, as test_is_path_contained_windows pins. + ("\\\\", ntpath, True), + (r"C:\a", ntpath, True), + ("C:\\", ntpath, True), + (r"\\?\C:\a", ntpath, True), + (r"\\?\UNC\host\share", ntpath, True), + (r"\\.\C:\a", ntpath, True), + # A rooted, driveless path names the current drive's root, not the working + # directory, so it is absolute. ntpath.isabs agrees through 3.12 and disagrees from + # 3.13; answering from the anchor keeps the verdict version-independent, and a + # cross-platform caller passing '/a' roots on Windows depends on this. + ("\\", ntpath, True), + (r"\x", ntpath, True), + ("/a", ntpath, True), + ("/", ntpath, True), + # Drive-relative needs the working directory on that drive, which is exactly what + # the known-root hardening must not let a caller supply implicitly. + ("C:", ntpath, False), + ("C:foo", ntpath, False), + ("rel", ntpath, False), + (r"rel\f", ntpath, False), + (r"..\a", ntpath, False), + ("", ntpath, False), + (".", ntpath, False), + ("/a", posixpath, True), + ("//a", posixpath, True), + ("rel/f", posixpath, False), + ("../a", posixpath, False), + ("", posixpath, False), + ], +) +def test_is_absolute_path(path, path_module, expected): + assert is_absolute_path(path, path_module=path_module) is expected + + +def test_is_absolute_path_never_accepts_a_working_directory_relative_path(): + """The property the known-root hardening depends on. + + ``ntpath.isabs`` disagrees with itself across supported versions in two places -- a UNC + path naming a share (False before 3.11) and a rooted, driveless path (True through 3.12) + -- so it cannot be the reference. What must hold on every version is narrower: a path + that needs the working directory to resolve is never absolute, because such a root would + let the directory the shell happens to be in become trusted. + """ + relative = { + ntpath: ["rel", r"rel\f", r"..\a", ".", "", "C:", "C:foo", r"C:..\x"], + posixpath: ["rel", "rel/f", "../a", ".", ""], + } + for path_module, paths in relative.items(): + for path in paths: + assert is_absolute_path(path, path_module=path_module) is False, (path_module, path) + + +@pytest.mark.parametrize( + "path, path_module, expected", + [ + # The reason this exists rather than calling normpath directly: before Python 3.11 + # normpath collapses the leading pair on a UNC path that names no share, which moves + # a host-level known-asset root out of the UNC space so it matches none of its own + # shares. _filter_redundant_known_paths feeds its output to _is_known_path. + (r"\\host", ntpath, r"\\host"), + ("\\\\host\\", ntpath, r"\\host"), + ("\\\\", ntpath, "\\\\"), + (r"\\host\share\a\..\b", ntpath, r"\\host\share\b"), + # Case is preserved, unlike the components used for comparison. + (r"\\Host\Share", ntpath, r"\\Host\Share"), + (r"C:\A\.\b\..\c", ntpath, r"C:\A\c"), + ("C:/a/b", ntpath, r"C:\a\b"), + # An extended-length prefix folds to the plain path it denotes. + (r"\\?\C:\a", ntpath, r"C:\a"), + (r"\\?\UNC\host\share\f", ntpath, r"\\host\share\f"), + ("/a/", posixpath, "/a"), + ("/a/b/../c", posixpath, "/a/c"), + ("/", posixpath, "/"), + ], +) +def test_normalized_path(path, path_module, expected): + assert normalized_path(path, path_module=path_module) == expected + + +def test_path_components_preserves_case_when_asked(): + assert path_components(r"\\Host\Share\File", path_module=ntpath, normalize_case=False) == [ + "\\\\", + "Host", + "Share", + "File", + ]