mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-27 05:30:39 +00:00
fix(paths): reject Windows reserved names on host-visible paths (#6102)
* fix(paths): reject Windows reserved names on host-visible paths Uploads, skill support files, and sandbox path checks now share one helper that rejects CON, PRN, AUX, NUL, COM1-COM9, and LPT1-LPT9 (including names like con.txt) and components that end in a dot or space. The check is not limited to Windows so a Linux host cannot store a name that breaks when the same tree is used on Windows. * fix(paths): validate quoted local bash paths as whole words The absolute-path scan stopped at whitespace, so a quoted upload name such as report. final.txt or CON notes.txt was rejected as a truncated segment. Local bash checks now use the shlex argument for traversal and Windows-incompatible segments. * fix(paths): do not block reads of existing host names * style(paths): format with ruff 0.15.12
This commit is contained in:
Li-john1021
GitHub
parent
9c75236ea1
commit
90c5d526e2
@ -53,6 +53,7 @@ from deerflow.sandbox.sandbox_provider import SandboxProvider, get_sandbox_provi
|
||||
from deerflow.sandbox.search import GrepMatch
|
||||
from deerflow.sandbox.security import LOCAL_HOST_BASH_DISABLED_MESSAGE, is_host_bash_allowed
|
||||
from deerflow.tools.types import Runtime
|
||||
from deerflow.utils.host_paths import windows_incompatible_segment
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@ -900,8 +901,80 @@ def mask_local_paths_in_output(output: str, thread_data: ThreadDataState | None)
|
||||
return result
|
||||
|
||||
|
||||
def _windows_incompatible_path_reason(path: str) -> str | None:
|
||||
"""Return why *path* is not portable, or None when every segment is fine."""
|
||||
normalised = path.replace("\\", "/")
|
||||
for segment in normalised.split("/"):
|
||||
reason = windows_incompatible_segment(segment)
|
||||
if reason:
|
||||
return reason
|
||||
return None
|
||||
|
||||
|
||||
def _host_path_exists(candidate: str) -> bool:
|
||||
"""Return True when *candidate* is already on the host filesystem.
|
||||
|
||||
``pathlib`` and the Win32 ANSI APIs strip a trailing space or dot, so a
|
||||
file created with those characters is checked again with the Windows
|
||||
extended-length path prefix. Virtual ``/mnt/...`` paths are not rewritten.
|
||||
"""
|
||||
try:
|
||||
if os.path.lexists(candidate):
|
||||
return True
|
||||
except OSError:
|
||||
pass
|
||||
if os.name != "nt":
|
||||
return False
|
||||
normalized = candidate.replace("/", "\\")
|
||||
if len(normalized) < 3 or normalized[1] != ":" or normalized[2] != "\\":
|
||||
return False
|
||||
if normalized.startswith("\\\\?\\"):
|
||||
return False
|
||||
try:
|
||||
return os.path.lexists("\\\\?\\" + normalized)
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
|
||||
def _stored_host_path_exists(path: str, thread_data: ThreadDataState | None) -> bool:
|
||||
"""Return True when *path* already names a host file or directory.
|
||||
|
||||
``/mnt/user-data`` paths are resolved through *thread_data*. The literal
|
||||
path is also checked so a name that exists inside the sandbox mount is
|
||||
recognized. A missing mapping does not count as stored.
|
||||
"""
|
||||
candidates: list[str] = []
|
||||
if path == VIRTUAL_PATH_PREFIX or path.startswith(f"{VIRTUAL_PATH_PREFIX}/"):
|
||||
if thread_data is not None:
|
||||
try:
|
||||
resolved = replace_virtual_path(path, thread_data)
|
||||
except Exception:
|
||||
resolved = None
|
||||
if resolved and resolved != path:
|
||||
candidates.append(resolved)
|
||||
candidates.append(path)
|
||||
else:
|
||||
candidates.append(path)
|
||||
return any(_host_path_exists(candidate) for candidate in candidates)
|
||||
|
||||
|
||||
def _reject_unstored_windows_incompatible_path(path: str, thread_data: ThreadDataState | None) -> None:
|
||||
"""Reject a non-portable path unless that exact path is already stored."""
|
||||
reason = _windows_incompatible_path_reason(path)
|
||||
if reason is None:
|
||||
return
|
||||
if _stored_host_path_exists(path, thread_data):
|
||||
return
|
||||
raise PermissionError(f"Access denied: {reason}")
|
||||
|
||||
|
||||
def _reject_path_traversal(path: str) -> None:
|
||||
"""Reject paths that contain '..' segments to prevent directory traversal."""
|
||||
"""Reject paths that contain '..' segments to prevent directory traversal.
|
||||
|
||||
Windows reserved names and trailing dots or spaces are intentionally not
|
||||
checked here. Portability is enforced when a path is created, not on every
|
||||
read of a name the host already stored.
|
||||
"""
|
||||
# Normalise to forward slashes, then check for '..' segments.
|
||||
normalised = path.replace("\\", "/")
|
||||
for segment in normalised.split("/"):
|
||||
@ -936,6 +1009,10 @@ def validate_local_tool_path(path: str, thread_data: ThreadDataState | None, *,
|
||||
raise SandboxRuntimeError("Thread data not available for local sandbox")
|
||||
|
||||
_reject_path_traversal(path)
|
||||
# Creating a non-portable name is rejected. Reading, and editing a file
|
||||
# that is already stored under that name, is not.
|
||||
if not read_only:
|
||||
_reject_unstored_windows_incompatible_path(path, thread_data)
|
||||
|
||||
# Skills paths — read-only access only
|
||||
if _is_skills_path(path):
|
||||
@ -1046,6 +1123,61 @@ def _split_shell_tokens(command: str) -> list[str]:
|
||||
return command.split()
|
||||
|
||||
|
||||
_SHELL_PATH_HARD_TERMINATORS = frozenset("\"'`;&|<>()")
|
||||
|
||||
|
||||
def _absolute_path_token_occurrences(command: str, tokens: list[str]) -> list[tuple[int, int, str]]:
|
||||
"""Locate contiguous occurrences of path-bearing shell tokens in *command*.
|
||||
|
||||
shlex strips quotes, so a quoted path with spaces is one token. Occurrences
|
||||
stay tied to their position so a later quoted ``.../CON notes.txt`` cannot
|
||||
excuse an earlier real ``/.../CON``. Tokens that are not a contiguous
|
||||
substring (newline normalization, escapes) yield no occurrence.
|
||||
"""
|
||||
occurrences: list[tuple[int, int, str]] = []
|
||||
for token in tokens:
|
||||
if "/" not in token:
|
||||
continue
|
||||
start = command.find(token)
|
||||
while start != -1:
|
||||
occurrences.append((start, start + len(token), token))
|
||||
start = command.find(token, start + 1)
|
||||
return occurrences
|
||||
|
||||
|
||||
def _extend_absolute_path_within_span(command: str, match_start: int, span_end: int) -> str:
|
||||
"""Continue a regex path through spaces inside one shell word.
|
||||
|
||||
Stop before quotes and shell metacharacters the absolute-path regex already
|
||||
treats as boundaries, so a trailing dot is not hidden by a closing quote.
|
||||
"""
|
||||
chars: list[str] = []
|
||||
for char in command[match_start:span_end]:
|
||||
if char in _SHELL_PATH_HARD_TERMINATORS:
|
||||
break
|
||||
chars.append(char)
|
||||
return "".join(chars)
|
||||
|
||||
|
||||
def _shell_absolute_path_candidate(command: str, occurrences: list[tuple[int, int, str]], match: re.Match[str]) -> str | None:
|
||||
"""Return the path to validate for one regex match, or None if already covered.
|
||||
|
||||
A match that starts a shell word is validated as that whole word. A later
|
||||
match inside that same absolute-path word is skipped. A match embedded in a
|
||||
larger word is extended through spaces only up to a hard terminator.
|
||||
"""
|
||||
covering = [item for item in occurrences if item[0] <= match.start() < item[1]]
|
||||
if not covering:
|
||||
return match.group()
|
||||
start, end, token = max(covering, key=lambda item: item[1] - item[0])
|
||||
if token.startswith("/") and start == match.start():
|
||||
return token
|
||||
if token.startswith("/") and start < match.start():
|
||||
return None
|
||||
extended = _extend_absolute_path_within_span(command, match.start(), end)
|
||||
return extended or match.group()
|
||||
|
||||
|
||||
def _is_shell_command_separator(token: str) -> bool:
|
||||
return token in _SHELL_COMMAND_SEPARATORS
|
||||
|
||||
@ -1061,29 +1193,41 @@ def _is_shell_assignment(token: str) -> bool:
|
||||
return bool(re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", name))
|
||||
|
||||
|
||||
def _is_allowed_local_bash_absolute_path(path: str, allowed_paths: list[str], *, allow_system_paths: bool) -> bool:
|
||||
def _reject_allowed_bash_path(path: str, thread_data: ThreadDataState | None) -> None:
|
||||
"""Traversal always; portability only when the host path is not stored yet."""
|
||||
_reject_path_traversal(path)
|
||||
_reject_unstored_windows_incompatible_path(path, thread_data)
|
||||
|
||||
|
||||
def _is_allowed_local_bash_absolute_path(
|
||||
path: str,
|
||||
allowed_paths: list[str],
|
||||
*,
|
||||
allow_system_paths: bool,
|
||||
thread_data: ThreadDataState | None,
|
||||
) -> bool:
|
||||
# Check for MCP filesystem server allowed paths
|
||||
if any(path.startswith(allowed_path) or path == allowed_path.rstrip("/") for allowed_path in allowed_paths):
|
||||
_reject_path_traversal(path)
|
||||
_reject_allowed_bash_path(path, thread_data)
|
||||
return True
|
||||
|
||||
if path == VIRTUAL_PATH_PREFIX or path.startswith(f"{VIRTUAL_PATH_PREFIX}/"):
|
||||
_reject_path_traversal(path)
|
||||
_reject_allowed_bash_path(path, thread_data)
|
||||
return True
|
||||
|
||||
# Allow skills container path (resolved by tools.py before passing to sandbox)
|
||||
if _is_skills_path(path):
|
||||
_reject_path_traversal(path)
|
||||
_reject_allowed_bash_path(path, thread_data)
|
||||
return True
|
||||
|
||||
# Allow ACP workspace path (path-traversal check only)
|
||||
if _is_acp_workspace_path(path):
|
||||
_reject_path_traversal(path)
|
||||
_reject_allowed_bash_path(path, thread_data)
|
||||
return True
|
||||
|
||||
# Allow custom mount container paths
|
||||
if _is_custom_mount_path(path):
|
||||
_reject_path_traversal(path)
|
||||
_reject_allowed_bash_path(path, thread_data)
|
||||
return True
|
||||
|
||||
if allow_system_paths and any(path == prefix.rstrip("/") or path.startswith(prefix) for prefix in _LOCAL_BASH_SYSTEM_PATH_PREFIXES):
|
||||
@ -1114,7 +1258,7 @@ def _next_cd_target(tokens: list[str], start_index: int) -> tuple[str | None, in
|
||||
return None, index
|
||||
|
||||
|
||||
def _validate_local_bash_cwd_target(command_name: str, target: str | None, allowed_paths: list[str]) -> None:
|
||||
def _validate_local_bash_cwd_target(command_name: str, target: str | None, allowed_paths: list[str], thread_data: ThreadDataState | None) -> None:
|
||||
if target is None or target == "-":
|
||||
raise PermissionError(f"Unsafe working directory change in command: {command_name}. Use paths under {VIRTUAL_PATH_PREFIX}")
|
||||
if target.startswith(("$", "`")):
|
||||
@ -1123,7 +1267,7 @@ def _validate_local_bash_cwd_target(command_name: str, target: str | None, allow
|
||||
raise PermissionError(f"Unsafe working directory change in command: {command_name} {target}. Use paths under {VIRTUAL_PATH_PREFIX}")
|
||||
if target.startswith("/"):
|
||||
_reject_path_traversal(target)
|
||||
if not _is_allowed_local_bash_absolute_path(target, allowed_paths, allow_system_paths=False):
|
||||
if not _is_allowed_local_bash_absolute_path(target, allowed_paths, allow_system_paths=False, thread_data=thread_data):
|
||||
raise PermissionError(f"Unsafe working directory change in command: {command_name} {target}. Use paths under {VIRTUAL_PATH_PREFIX}")
|
||||
|
||||
|
||||
@ -1144,13 +1288,11 @@ def _validate_local_bash_root_path_args(command_name: str, tokens: list[str], st
|
||||
index += 1
|
||||
|
||||
|
||||
def _validate_local_bash_shell_tokens(command: str, allowed_paths: list[str]) -> None:
|
||||
def _validate_local_bash_shell_tokens(command: str, allowed_paths: list[str], tokens: list[str], thread_data: ThreadDataState | None) -> None:
|
||||
"""Conservatively reject relative path escapes missed by absolute-path scanning."""
|
||||
if re.search(r"\$\([^)]*\b(?:cd|pushd)\b", command):
|
||||
raise PermissionError(f"Unsafe working directory change in command substitution. Use paths under {VIRTUAL_PATH_PREFIX}")
|
||||
|
||||
tokens = _split_shell_tokens(command)
|
||||
|
||||
for token in tokens:
|
||||
if _is_shell_command_separator(token) or _is_shell_redirection_operator(token):
|
||||
continue
|
||||
@ -1189,7 +1331,7 @@ def _validate_local_bash_shell_tokens(command: str, allowed_paths: list[str]) ->
|
||||
wrapped_name = tokens[index + 1].rsplit("/", 1)[-1]
|
||||
if wrapped_name in _LOCAL_BASH_CWD_COMMANDS:
|
||||
target, next_index = _next_cd_target(tokens, index + 2)
|
||||
_validate_local_bash_cwd_target(wrapped_name, target, allowed_paths)
|
||||
_validate_local_bash_cwd_target(wrapped_name, target, allowed_paths, thread_data)
|
||||
index = next_index
|
||||
continue
|
||||
_validate_local_bash_root_path_args(wrapped_name, tokens, index + 2)
|
||||
@ -1200,7 +1342,7 @@ def _validate_local_bash_shell_tokens(command: str, allowed_paths: list[str]) ->
|
||||
continue
|
||||
|
||||
target, next_index = _next_cd_target(tokens, index + 1)
|
||||
_validate_local_bash_cwd_target(command_name, target, allowed_paths)
|
||||
_validate_local_bash_cwd_target(command_name, target, allowed_paths, thread_data)
|
||||
index = next_index
|
||||
|
||||
|
||||
@ -1270,7 +1412,10 @@ def validate_local_bash_command_paths(command: str, thread_data: ThreadDataState
|
||||
config.yaml) are allowed (path-traversal checks only; write prevention
|
||||
for bash commands is not enforced here).
|
||||
A small allowlist of common system path prefixes is kept for executable
|
||||
and device references (e.g. /bin/sh, /dev/null).
|
||||
and device references (e.g. /bin/sh, /dev/null). Quoted shell words are
|
||||
validated whole, so a space inside quotes is not treated as a new segment.
|
||||
A windows-incompatible segment still rejects a path that is not already
|
||||
stored on the host; an existing host file may be read or renamed.
|
||||
"""
|
||||
if thread_data is None:
|
||||
raise SandboxRuntimeError("Thread data not available for local sandbox")
|
||||
@ -1282,7 +1427,9 @@ def validate_local_bash_command_paths(command: str, thread_data: ThreadDataState
|
||||
|
||||
unsafe_paths: list[str] = []
|
||||
allowed_paths = _get_mcp_allowed_paths()
|
||||
_validate_local_bash_shell_tokens(command, allowed_paths)
|
||||
tokens = _split_shell_tokens(command)
|
||||
_validate_local_bash_shell_tokens(command, allowed_paths, tokens, thread_data)
|
||||
path_token_occurrences = _absolute_path_token_occurrences(command, tokens)
|
||||
url_spans = _non_file_url_spans(command)
|
||||
|
||||
for match in _ABSOLUTE_PATH_PATTERN.finditer(command):
|
||||
@ -1291,10 +1438,13 @@ def validate_local_bash_command_paths(command: str, thread_data: ThreadDataState
|
||||
absolute_path = match.group()
|
||||
if _is_non_path_literal_fragment(absolute_path):
|
||||
continue
|
||||
if _is_allowed_local_bash_absolute_path(absolute_path, allowed_paths, allow_system_paths=True):
|
||||
candidate = _shell_absolute_path_candidate(command, path_token_occurrences, match)
|
||||
if candidate is None:
|
||||
continue
|
||||
if _is_allowed_local_bash_absolute_path(candidate, allowed_paths, allow_system_paths=True, thread_data=thread_data):
|
||||
continue
|
||||
|
||||
unsafe_paths.append(absolute_path)
|
||||
unsafe_paths.append(candidate)
|
||||
|
||||
if unsafe_paths:
|
||||
unsafe = ", ".join(sorted(dict.fromkeys(unsafe_paths)))
|
||||
|
||||
@ -12,6 +12,7 @@ from pathlib import Path
|
||||
|
||||
from deerflow.constants import DEFAULT_SKILLS_CONTAINER_PATH
|
||||
from deerflow.skills.types import SKILL_MD_FILE, Skill, SkillCategory # noqa: F401
|
||||
from deerflow.utils.host_paths import windows_incompatible_segment
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@ -134,13 +135,16 @@ class SkillStorage(ABC):
|
||||
return False
|
||||
return len(relative_parent.parts) == 1 and skill_file.parent.resolve().is_dir()
|
||||
|
||||
def ensure_safe_support_path(self, name: str, relative_path: str) -> Path:
|
||||
def ensure_safe_support_path(self, name: str, relative_path: str, *, enforce_portability: bool = True) -> Path:
|
||||
"""Validate and return the resolved absolute path for a support file.
|
||||
|
||||
The path must name a file *inside* one of the support directories: a
|
||||
bare ``assets`` resolves to its own allowed root and would otherwise
|
||||
pass the containment check, letting callers try to remove or overwrite
|
||||
a directory.
|
||||
|
||||
``enforce_portability`` is False only for removal, so a support file
|
||||
stored before this check can still be deleted. Traversal checks stay.
|
||||
"""
|
||||
_ALLOWED_SUPPORT_SUBDIRS = {"references", "templates", "scripts", "assets"}
|
||||
skill_dir = self.get_custom_skill_dir(self.validate_skill_name(name)).resolve()
|
||||
@ -151,6 +155,11 @@ class SkillStorage(ABC):
|
||||
raise ValueError("Supporting file path must be relative.")
|
||||
if any(part in {"..", ""} for part in relative.parts):
|
||||
raise ValueError("Supporting file path must not contain parent-directory traversal.")
|
||||
if enforce_portability:
|
||||
for part in relative.parts:
|
||||
reason = windows_incompatible_segment(part)
|
||||
if reason:
|
||||
raise ValueError(f"Supporting file path is not portable to Windows: {relative_path!r} ({reason})")
|
||||
top_level = relative.parts[0] if relative.parts else ""
|
||||
if top_level not in _ALLOWED_SUPPORT_SUBDIRS:
|
||||
raise ValueError(f"Supporting files must live under one of: {', '.join(sorted(_ALLOWED_SUPPORT_SUBDIRS))}.")
|
||||
@ -236,7 +245,7 @@ class SkillStorage(ABC):
|
||||
only feeds the history record, so it must never block the removal.
|
||||
A directory is rejected rather than removed.
|
||||
"""
|
||||
target = self.ensure_safe_support_path(name, relative_path)
|
||||
target = self.ensure_safe_support_path(name, relative_path, enforce_portability=False)
|
||||
if target.is_dir():
|
||||
raise ValueError(f"Supporting file path '{relative_path}' is a directory, not a file.")
|
||||
if not target.exists():
|
||||
|
||||
@ -14,6 +14,7 @@ from urllib.parse import quote
|
||||
|
||||
from deerflow.config.paths import VIRTUAL_PATH_PREFIX, get_paths
|
||||
from deerflow.runtime.user_context import get_effective_user_id
|
||||
from deerflow.utils.host_paths import windows_incompatible_segment
|
||||
from deerflow.utils.thread_id import validate_thread_id
|
||||
|
||||
|
||||
@ -73,6 +74,9 @@ def normalize_filename(filename: str) -> str:
|
||||
raise ValueError(f"Filename contains backslash: {filename!r}")
|
||||
if len(safe.encode("utf-8")) > _MAX_FILENAME_BYTES:
|
||||
raise ValueError(f"Filename too long: {len(safe)} chars")
|
||||
reason = windows_incompatible_segment(safe)
|
||||
if reason:
|
||||
raise ValueError(f"Filename is not portable to Windows: {filename!r} ({reason})")
|
||||
return safe
|
||||
|
||||
|
||||
|
||||
28
backend/packages/harness/deerflow/utils/host_paths.py
Normal file
28
backend/packages/harness/deerflow/utils/host_paths.py
Normal file
@ -0,0 +1,28 @@
|
||||
"""Portability checks for host-visible path segments."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
_WINDOWS_RESERVED_NAMES = frozenset({"CON", "PRN", "AUX", "NUL"} | {f"COM{i}" for i in range(1, 10)} | {f"LPT{i}" for i in range(1, 10)})
|
||||
|
||||
|
||||
def windows_incompatible_segment(segment: str) -> str | None:
|
||||
"""Return a reason when *segment* is unsafe to create as a host path component.
|
||||
|
||||
Reserved device names and trailing dots or spaces are rejected on every
|
||||
platform when a name is created: uploads, new custom-skill support files,
|
||||
and sandbox writes of a path that is not already stored. Reads, deletes of
|
||||
an existing support file, and access to a path that already exists on the
|
||||
host do not use this check. Those names alias devices or are silently
|
||||
stripped on Windows.
|
||||
|
||||
``.`` and ``..`` are ignored here; callers already reject traversal.
|
||||
Do not reject colons, quotes, or control characters here.
|
||||
"""
|
||||
if segment in {"", ".", ".."}:
|
||||
return None
|
||||
if segment.endswith((" ", ".")):
|
||||
return "trailing dot or space"
|
||||
stem = segment.split(".", 1)[0]
|
||||
if stem.upper() in _WINDOWS_RESERVED_NAMES:
|
||||
return "reserved Windows device name"
|
||||
return None
|
||||
@ -427,6 +427,21 @@ def test_reject_path_traversal_allows_normal_paths() -> None:
|
||||
_reject_path_traversal("/mnt/user-data/workspace/sub/dir/file.py")
|
||||
|
||||
|
||||
def test_reject_path_traversal_allows_windows_reserved_names() -> None:
|
||||
"""Reserved names are a portability concern, not directory traversal."""
|
||||
for path in (
|
||||
"/mnt/user-data/workspace/CON",
|
||||
"/mnt/user-data/workspace/dir/COM1.txt",
|
||||
"/mnt/user-data/workspace/file.txt.",
|
||||
"/mnt/user-data/workspace/file.txt ",
|
||||
):
|
||||
_reject_path_traversal(path)
|
||||
|
||||
|
||||
def test_reject_path_traversal_allows_names_that_only_look_reserved() -> None:
|
||||
_reject_path_traversal("/mnt/user-data/workspace/contour.txt")
|
||||
|
||||
|
||||
# ---------- validate_local_tool_path ----------
|
||||
|
||||
|
||||
@ -472,6 +487,22 @@ def test_validate_local_tool_path_allows_user_data_write() -> None:
|
||||
validate_local_tool_path(f"{VIRTUAL_PATH_PREFIX}/workspace/file.txt", _THREAD_DATA, read_only=False)
|
||||
|
||||
|
||||
def test_validate_local_tool_path_read_allows_nonportable_names() -> None:
|
||||
"""Reads must not fail solely because the name is not Windows-portable."""
|
||||
validate_local_tool_path(f"{VIRTUAL_PATH_PREFIX}/uploads/aux.pdf", _THREAD_DATA, read_only=True)
|
||||
validate_local_tool_path(f"{VIRTUAL_PATH_PREFIX}/uploads/notes.txt ", _THREAD_DATA, read_only=True)
|
||||
validate_local_tool_path(f"{VIRTUAL_PATH_PREFIX}/uploads/contour.txt", _THREAD_DATA, read_only=True)
|
||||
|
||||
|
||||
def test_validate_local_tool_path_write_rejects_new_nonportable_names() -> None:
|
||||
with pytest.raises(PermissionError, match="Access denied"):
|
||||
validate_local_tool_path(f"{VIRTUAL_PATH_PREFIX}/uploads/aux.pdf", _THREAD_DATA, read_only=False)
|
||||
with pytest.raises(PermissionError, match="Access denied"):
|
||||
validate_local_tool_path(f"{VIRTUAL_PATH_PREFIX}/uploads/notes.txt ", _THREAD_DATA, read_only=False)
|
||||
with pytest.raises(PermissionError, match="path traversal"):
|
||||
validate_local_tool_path(f"{VIRTUAL_PATH_PREFIX}/uploads/../../etc/passwd", _THREAD_DATA, read_only=False)
|
||||
|
||||
|
||||
def test_validate_local_tool_path_rejects_traversal_in_user_data() -> None:
|
||||
"""Path traversal via .. in user-data paths must be rejected."""
|
||||
with pytest.raises(PermissionError, match="path traversal"):
|
||||
@ -860,6 +891,79 @@ def test_validate_local_bash_command_paths_still_blocks_ascii_host_path_in_code(
|
||||
validate_local_bash_command_paths("python3 -c \"open('/etc/passwd').read()\"", _THREAD_DATA)
|
||||
|
||||
|
||||
def test_validate_local_bash_command_paths_allows_quoted_paths_with_spaces() -> None:
|
||||
"""Spaces inside quotes belong to the filename; they must not truncate a segment."""
|
||||
validate_local_bash_command_paths('cat "/mnt/user-data/uploads/report. final.txt"', _THREAD_DATA)
|
||||
validate_local_bash_command_paths('cat "/mnt/user-data/uploads/CON notes.txt"', _THREAD_DATA)
|
||||
|
||||
|
||||
def _windows_extended(path: Path) -> str:
|
||||
return (chr(92) * 2) + "?" + chr(92) + str(path)
|
||||
|
||||
|
||||
def _write_literal_host_file(path: Path, data: bytes) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
if os.name == "nt":
|
||||
fd = os.open(_windows_extended(path), os.O_CREAT | os.O_WRONLY | getattr(os, "O_BINARY", 0))
|
||||
try:
|
||||
os.write(fd, data)
|
||||
finally:
|
||||
os.close(fd)
|
||||
else:
|
||||
path.write_bytes(data)
|
||||
|
||||
|
||||
def _remove_literal_host_file(path: Path) -> None:
|
||||
if os.name == "nt":
|
||||
raw = _windows_extended(path)
|
||||
if os.path.lexists(raw):
|
||||
os.remove(raw)
|
||||
return
|
||||
if path.exists():
|
||||
path.unlink()
|
||||
|
||||
|
||||
def test_validate_local_bash_command_paths_allows_stored_nonportable_names(tmp_path: Path) -> None:
|
||||
"""cat/mv of a host file that already exists is not blocked for its name."""
|
||||
uploads = tmp_path / "uploads"
|
||||
workspace = tmp_path / "workspace"
|
||||
outputs = tmp_path / "outputs"
|
||||
for directory in (uploads, workspace, outputs):
|
||||
directory.mkdir()
|
||||
reserved = uploads / "aux.pdf"
|
||||
spaced = uploads / "notes.txt "
|
||||
_write_literal_host_file(reserved, b"pdf")
|
||||
_write_literal_host_file(spaced, b"sp")
|
||||
thread = {
|
||||
"workspace_path": str(workspace),
|
||||
"uploads_path": str(uploads),
|
||||
"outputs_path": str(outputs),
|
||||
}
|
||||
try:
|
||||
validate_local_bash_command_paths('cat "/mnt/user-data/uploads/aux.pdf"', thread)
|
||||
validate_local_bash_command_paths('mv "/mnt/user-data/uploads/aux.pdf" /mnt/user-data/workspace/renamed.pdf', thread)
|
||||
validate_local_bash_command_paths('cat "/mnt/user-data/uploads/notes.txt "', thread)
|
||||
validate_local_tool_path("/mnt/user-data/uploads/aux.pdf", thread, read_only=False)
|
||||
validate_local_tool_path("/mnt/user-data/uploads/notes.txt ", thread, read_only=True)
|
||||
finally:
|
||||
_remove_literal_host_file(reserved)
|
||||
_remove_literal_host_file(spaced)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"command",
|
||||
[
|
||||
'cat "/mnt/user-data/uploads/CON"',
|
||||
'cat "/mnt/user-data/uploads/report."',
|
||||
'cat "/mnt/user-data/uploads/foo bar/../../etc/passwd"',
|
||||
"cat /mnt/user-data/uploads/report. final.txt",
|
||||
],
|
||||
)
|
||||
def test_validate_local_bash_command_paths_rejects_quoted_reserved_or_trailing_dot_paths(command: str) -> None:
|
||||
with pytest.raises(PermissionError):
|
||||
validate_local_bash_command_paths(command, _THREAD_DATA)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"command",
|
||||
[
|
||||
|
||||
@ -11,6 +11,7 @@ member raised ``UnicodeDecodeError`` and was never removed, and a support
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
@ -73,6 +74,12 @@ def test_ensure_safe_support_path_requires_a_filename_below_the_subdir(storage,
|
||||
assert storage.ensure_safe_support_path("demo-skill", "assets/logo.png") == (skill_dir / "assets" / "logo.png").resolve()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("path", ["assets/CON", "assets/nul.txt", "assets/logo.png.", "assets/notes.md "])
|
||||
def test_ensure_safe_support_path_rejects_windows_incompatible_names(storage, skill_dir, path):
|
||||
with pytest.raises(ValueError, match="not portable to Windows"):
|
||||
storage.ensure_safe_support_path("demo-skill", path)
|
||||
|
||||
|
||||
def test_read_text_or_none_decodes_text_and_returns_none_for_binary(tmp_path: Path):
|
||||
text = tmp_path / "t.md"
|
||||
text.write_text("héllo", encoding="utf-8")
|
||||
@ -87,3 +94,41 @@ def test_read_text_or_none_only_masks_undecodable_content(tmp_path: Path):
|
||||
"""``None`` means "not text", never "could not read": IO errors still propagate."""
|
||||
with pytest.raises(FileNotFoundError):
|
||||
read_text_or_none(tmp_path / "missing.md")
|
||||
|
||||
|
||||
def _windows_extended(path: Path) -> str:
|
||||
return (chr(92) * 2) + "?" + chr(92) + str(path)
|
||||
|
||||
|
||||
def _write_literal(path: Path, data: bytes) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
if os.name == "nt":
|
||||
fd = os.open(_windows_extended(path), os.O_CREAT | os.O_WRONLY | getattr(os, "O_BINARY", 0))
|
||||
try:
|
||||
os.write(fd, data)
|
||||
finally:
|
||||
os.close(fd)
|
||||
else:
|
||||
path.write_bytes(data)
|
||||
|
||||
|
||||
def test_remove_preexisting_nonportable_support_file(storage, skill_dir):
|
||||
"""Deletion of a stored non-portable name succeeds; creating one still fails."""
|
||||
target = skill_dir / "scripts" / "AUX.py"
|
||||
_write_literal(target, b"print('aux')\n")
|
||||
with pytest.raises(ValueError, match="not portable to Windows"):
|
||||
storage.ensure_safe_support_path("demo-skill", "scripts/AUX.py")
|
||||
|
||||
previous = storage.remove_custom_skill_file("demo-skill", "scripts/AUX.py")
|
||||
|
||||
assert previous == "print('aux')\n"
|
||||
assert not target.exists()
|
||||
|
||||
if os.name != "nt":
|
||||
dotted = skill_dir / "assets" / "logo.png."
|
||||
_write_literal(dotted, b"dot")
|
||||
with pytest.raises(ValueError, match="not portable to Windows"):
|
||||
storage.ensure_safe_support_path("demo-skill", "assets/logo.png.")
|
||||
removed = storage.remove_custom_skill_file("demo-skill", "assets/logo.png.")
|
||||
assert removed == "dot"
|
||||
assert not dotted.exists()
|
||||
|
||||
@ -78,6 +78,21 @@ class TestNormalizeFilename:
|
||||
with pytest.raises(ValueError, match="NUL"):
|
||||
normalize_filename("report\x00.pdf")
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"filename",
|
||||
["CON", "con.txt", "PRN", "AUX", "NUL", "COM1", "COM9", "LPT1", "LPT9", "file.txt.", "file.txt ", "a.", "folder/CON"],
|
||||
)
|
||||
def test_rejects_windows_incompatible_names(self, filename):
|
||||
with pytest.raises(ValueError, match="not portable to Windows"):
|
||||
normalize_filename(filename)
|
||||
|
||||
@pytest.mark.parametrize("filename", ["report.pdf", "contour.txt", "console.log", "COM0", "COM10", ".gitignore"])
|
||||
def test_allows_portable_names(self, filename):
|
||||
assert normalize_filename(filename) == filename
|
||||
|
||||
def test_reserved_parent_is_stripped_with_the_directory(self):
|
||||
assert normalize_filename("CON/notes.md") == "notes.md"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# claim_unique_filename
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user