diff --git a/backend/packages/harness/deerflow/sandbox/tools.py b/backend/packages/harness/deerflow/sandbox/tools.py index 0262d0f9c5..af297cab01 100644 --- a/backend/packages/harness/deerflow/sandbox/tools.py +++ b/backend/packages/harness/deerflow/sandbox/tools.py @@ -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))) diff --git a/backend/packages/harness/deerflow/skills/storage/skill_storage.py b/backend/packages/harness/deerflow/skills/storage/skill_storage.py index 15814ba2d1..d4cf4edeb8 100644 --- a/backend/packages/harness/deerflow/skills/storage/skill_storage.py +++ b/backend/packages/harness/deerflow/skills/storage/skill_storage.py @@ -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(): diff --git a/backend/packages/harness/deerflow/uploads/manager.py b/backend/packages/harness/deerflow/uploads/manager.py index ea3992e253..a65a269988 100644 --- a/backend/packages/harness/deerflow/uploads/manager.py +++ b/backend/packages/harness/deerflow/uploads/manager.py @@ -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 diff --git a/backend/packages/harness/deerflow/utils/host_paths.py b/backend/packages/harness/deerflow/utils/host_paths.py new file mode 100644 index 0000000000..9550446a7d --- /dev/null +++ b/backend/packages/harness/deerflow/utils/host_paths.py @@ -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 diff --git a/backend/tests/test_sandbox_tools_security.py b/backend/tests/test_sandbox_tools_security.py index 67a9983f71..22649193cd 100644 --- a/backend/tests/test_sandbox_tools_security.py +++ b/backend/tests/test_sandbox_tools_security.py @@ -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", [ diff --git a/backend/tests/test_skill_storage_support_files.py b/backend/tests/test_skill_storage_support_files.py index 56a5eb4a86..09a3eca30f 100644 --- a/backend/tests/test_skill_storage_support_files.py +++ b/backend/tests/test_skill_storage_support_files.py @@ -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() diff --git a/backend/tests/test_uploads_manager.py b/backend/tests/test_uploads_manager.py index 65c6bdbf05..408724d981 100644 --- a/backend/tests/test_uploads_manager.py +++ b/backend/tests/test_uploads_manager.py @@ -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