fix(artifacts): reject rooted, drive-qualified and traversing artifact filenames (v1)
FileArtifactService decided whether a caller-supplied filename was safe by joining it under the scope root, resolving the result, and checking that the resolved path was still inside the root. Two shapes got through. A drive-qualified name such as "C:\Windows\evil.txt" was converted to the relative "C:/Windows/evil.txt", which is not absolute, so it passed the guard; on Windows, joining it replaces the drive and the write lands outside the root. And "folder/../alias.txt" resolved back inside the root and was accepted, so two different filenames addressed one artifact. _resolve_scoped_artifact_path now rejects rooted, drive-qualified and parent-referencing filenames up front, before any joining or resolving. The whole function is replaced rather than patched, which also brings in the Windows path normalization that landed upstream separately; the two are entangled in the same few lines and splitting them would leave a form that is on neither branch. The file service's own private identifier validator gains the same drive-qualified check. It rejected "C:/x" only as a side effect of its separator rule and accepted a bare "C:evil", which on Windows escapes the root directory by the same join. That guard is not in the upstream commit; upstream had already deleted this validator in favour of the shared one, which this branch deliberately does not do. Behaviour change: a filename such as "folder/../alias.txt" that resolves back inside the scope is now rejected rather than silently aliased to "alias.txt". Anyone relying on that aliasing gets an InputValidationError. Port of the upstream fix to the v1 branch.
This commit is contained in:
@@ -33,6 +33,7 @@ from pydantic import Field
|
||||
from pydantic import ValidationError
|
||||
from typing_extensions import override
|
||||
|
||||
from . import artifact_util
|
||||
from ..errors.input_validation_error import InputValidationError
|
||||
from .base_artifact_service import ArtifactVersion
|
||||
from .base_artifact_service import BaseArtifactService
|
||||
@@ -85,14 +86,32 @@ def _to_posix_path(path_value: str) -> PurePosixPath:
|
||||
return PurePosixPath(path_value)
|
||||
|
||||
|
||||
def _is_rooted_or_drive_qualified(path_value: str) -> bool:
|
||||
"""Checks POSIX and Windows rooted or drive-qualified path forms."""
|
||||
# A Windows root covers POSIX absolute paths, UNC and device prefixes alike;
|
||||
# only the drive-relative form (`C:name`) has no root of its own.
|
||||
if artifact_util._is_drive_qualified(path_value):
|
||||
return True
|
||||
return bool(PureWindowsPath(path_value).root)
|
||||
|
||||
|
||||
def _has_parent_reference(path_value: str) -> bool:
|
||||
"""Checks parent traversal using either platform's separators."""
|
||||
return (
|
||||
".." in PurePosixPath(path_value).parts
|
||||
or ".." in PureWindowsPath(path_value).parts
|
||||
)
|
||||
|
||||
|
||||
def _resolve_scoped_artifact_path(
|
||||
scope_root: Path, filename: str
|
||||
) -> tuple[Path, Path]:
|
||||
"""Returns the absolute artifact directory and its relative path.
|
||||
|
||||
The caller is expected to pass the scope root directory (user or session).
|
||||
This helper joins the filename under that root, resolves traversal segments,
|
||||
and guards against paths that escape the scope root.
|
||||
Filenames that are rooted, drive-qualified, or contain a parent reference are
|
||||
rejected outright, including parent references that would resolve back inside
|
||||
the scope root. Whatever remains is joined under the scope root.
|
||||
|
||||
Args:
|
||||
scope_root: Directory that defines the storage scope.
|
||||
@@ -103,17 +122,23 @@ def _resolve_scoped_artifact_path(
|
||||
to `scope_root`.
|
||||
|
||||
Raises:
|
||||
InputValidationError: If `filename` resolves outside of `scope_root`.
|
||||
InputValidationError: If `filename` is rooted, drive-qualified, contains a
|
||||
parent reference, or otherwise resolves outside of `scope_root`.
|
||||
"""
|
||||
stripped = _strip_user_namespace(filename).strip()
|
||||
pure_path = _to_posix_path(stripped)
|
||||
|
||||
if _is_rooted_or_drive_qualified(stripped):
|
||||
raise InputValidationError(
|
||||
f"Rooted or drive-qualified artifact filename {filename!r} is not "
|
||||
"permitted; provide a path relative to the storage scope."
|
||||
)
|
||||
if _has_parent_reference(stripped):
|
||||
raise InputValidationError(
|
||||
f"Artifact filename {filename!r} must not contain parent traversal."
|
||||
)
|
||||
|
||||
scope_root_resolved = scope_root.resolve(strict=False)
|
||||
if pure_path.is_absolute():
|
||||
raise InputValidationError(
|
||||
f"Absolute artifact filename {filename!r} is not permitted; "
|
||||
"provide a path relative to the storage scope."
|
||||
)
|
||||
pure_path = _to_posix_path(stripped)
|
||||
candidate = scope_root_resolved / Path(pure_path)
|
||||
|
||||
candidate = candidate.resolve(strict=False)
|
||||
@@ -147,7 +172,7 @@ def _validate_path_segment(value: str, field_name: str) -> None:
|
||||
|
||||
Raises:
|
||||
InputValidationError: If the value contains path separators, traversal
|
||||
segments, or null bytes.
|
||||
segments, null bytes, or is drive-qualified.
|
||||
"""
|
||||
if not value:
|
||||
raise InputValidationError(f"{field_name} must not be empty.")
|
||||
@@ -157,6 +182,10 @@ def _validate_path_segment(value: str, field_name: str) -> None:
|
||||
raise InputValidationError(
|
||||
f"{field_name} {value!r} must not contain path separators."
|
||||
)
|
||||
if artifact_util._is_drive_qualified(value):
|
||||
raise InputValidationError(
|
||||
f"{field_name} {value!r} must not be drive-qualified."
|
||||
)
|
||||
if value in (".", "..") or ".." in value.split("/"):
|
||||
raise InputValidationError(
|
||||
f"{field_name} {value!r} must not contain traversal segments."
|
||||
|
||||
@@ -18,7 +18,7 @@ from __future__ import annotations
|
||||
class InputValidationError(ValueError):
|
||||
"""Represents an error raised when user input fails validation."""
|
||||
|
||||
def __init__(self, message="Invalid input."):
|
||||
def __init__(self, message: str = "Invalid input.") -> None:
|
||||
"""Initializes the InputValidationError exception.
|
||||
|
||||
Args:
|
||||
|
||||
@@ -723,9 +723,21 @@ async def test_file_list_artifact_versions(tmp_path, artifact_service_factory):
|
||||
("filename", "session_id"),
|
||||
[
|
||||
("../escape.txt", "sess123"),
|
||||
(r"..\escape.txt", "sess123"),
|
||||
("folder/../alias.txt", "sess123"),
|
||||
(r"folder\..\alias.txt", "sess123"),
|
||||
(r"folder/..\alias.txt", "sess123"),
|
||||
("user:../escape.txt", "sess123"),
|
||||
(r"user:..\escape.txt", "sess123"),
|
||||
(r"user:folder\..\alias.txt", "sess123"),
|
||||
("/absolute/path.txt", "sess123"),
|
||||
("user:/absolute/path.txt", None),
|
||||
(r"C:\absolute\path.txt", "sess123"),
|
||||
("C:/absolute/path.txt", "sess123"),
|
||||
("C:drive-relative.txt", "sess123"),
|
||||
(r"\\server\share\file.txt", "sess123"),
|
||||
("//server/share/file.txt", "sess123"),
|
||||
(r"\rooted\file.txt", "sess123"),
|
||||
],
|
||||
)
|
||||
async def test_file_save_artifact_rejects_out_of_scope_paths(
|
||||
@@ -756,6 +768,8 @@ async def test_file_save_artifact_rejects_out_of_scope_paths(
|
||||
".",
|
||||
"has/slash",
|
||||
"back\\slash",
|
||||
"C:evil",
|
||||
"C:",
|
||||
"null\x00byte",
|
||||
"",
|
||||
],
|
||||
@@ -787,6 +801,8 @@ async def test_file_save_artifact_rejects_traversal_in_user_id(
|
||||
".",
|
||||
"has/slash",
|
||||
"back\\slash",
|
||||
"C:evil",
|
||||
"C:",
|
||||
"null\x00byte",
|
||||
"",
|
||||
],
|
||||
|
||||
Reference in New Issue
Block a user