mirror of
https://github.com/agentscope-ai/ReMe.git
synced 2026-08-28 05:25:04 +00:00
* fix(file_io): auto appending suffix for all related steps * fix(file_io): covering boundary cases of potential directory input
179 lines
6.3 KiB
Python
179 lines
6.3 KiB
Python
"""Path resolution, validation, and filetype gating for CRUD steps."""
|
|
|
|
import re
|
|
from pathlib import Path
|
|
|
|
from ...utils import get_logger
|
|
|
|
logger = get_logger(log_to_file=False)
|
|
|
|
NON_MD_WARNING = (
|
|
"non-markdown file detected; CRUD operations are recommended on markdown files. "
|
|
"Operating in compatibility mode may carry risks of errors."
|
|
)
|
|
|
|
NON_IMAGE_WARNING = (
|
|
"non-image file detected; CRUD image operations are recommended on standard image formats. "
|
|
"Operating in compatibility mode may carry risks of errors."
|
|
)
|
|
|
|
IMAGE_MIME_BY_EXT: dict[str, str] = {
|
|
".png": "image/png",
|
|
".jpg": "image/jpeg",
|
|
".jpeg": "image/jpeg",
|
|
".webp": "image/webp",
|
|
".gif": "image/gif",
|
|
".bmp": "image/bmp",
|
|
".tiff": "image/tiff",
|
|
".heic": "image/heic",
|
|
}
|
|
|
|
_INVALID_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
|
|
_RESERVED_NAMES = {
|
|
"CON",
|
|
"PRN",
|
|
"AUX",
|
|
"NUL",
|
|
*(f"COM{i}" for i in range(1, 10)),
|
|
*(f"LPT{i}" for i in range(1, 10)),
|
|
}
|
|
|
|
|
|
# pylint: disable=too-many-return-statements
|
|
def validate_filename_component(name: str, *, kind: str = "filename") -> str | None:
|
|
"""Return an error message, or ``None`` when ``name`` is a safe filename component."""
|
|
if not name:
|
|
return f"{kind} is required"
|
|
if name in (".", ".."):
|
|
return f"{kind} cannot be '.' or '..': {name!r}"
|
|
if name != name.strip():
|
|
return f"{kind} cannot have leading or trailing whitespace: {name!r}"
|
|
if _INVALID_CHARS.search(name):
|
|
return f'{kind} contains invalid characters (one of < > : " / \\ | ? * or a control char): {name!r}'
|
|
if name.endswith("."):
|
|
return f"{kind} cannot end with '.': {name!r}"
|
|
if name.split(".", 1)[0].upper() in _RESERVED_NAMES:
|
|
return f"{kind} is a Windows-reserved device name: {name!r}"
|
|
return None
|
|
|
|
|
|
def is_relative_to(path: Path, parent: Path) -> bool:
|
|
"""Return True when ``path`` is equal to or nested under ``parent``."""
|
|
try:
|
|
path.relative_to(parent)
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def display_path(workspace_path: Path, target: Path) -> str:
|
|
"""Return ``target`` as a workspace-relative POSIX path for user-facing messages.
|
|
|
|
Falls back to the absolute string when ``target`` is not under ``workspace_path``.
|
|
"""
|
|
try:
|
|
return target.relative_to(workspace_path).as_posix()
|
|
except ValueError:
|
|
return str(target)
|
|
|
|
|
|
# pylint: disable=too-many-return-statements
|
|
def resolve_path(
|
|
workspace_path: Path,
|
|
raw: str,
|
|
*,
|
|
allow_empty: bool = False,
|
|
) -> tuple[Path | None, str | None]:
|
|
"""Resolve a `path=` argument against ``workspace_path``.
|
|
|
|
Returns ``(abs_path, None)`` on success, or ``(None, error_message)`` on failure.
|
|
"""
|
|
if not raw or not str(raw).strip():
|
|
if allow_empty:
|
|
return workspace_path.resolve(), None
|
|
return None, "`path` is required"
|
|
s = str(raw).strip()
|
|
# ``~`` paths refer to the user's home directory rather than the workspace.
|
|
# They are deliberately unsupported: do not expand them (which could make a
|
|
# home-relative path look like a workspace escape), and report them as a
|
|
# missing target to callers.
|
|
if s.startswith("~"):
|
|
return None, f"file {s!r} does not exist"
|
|
p = Path(s)
|
|
workspace = workspace_path.resolve()
|
|
if p.is_absolute():
|
|
if Path(s).is_absolute():
|
|
logger.info("absolute path detected, recommending relative paths")
|
|
target = p.resolve()
|
|
else:
|
|
for part in p.parts:
|
|
err = validate_filename_component(part, kind="path component")
|
|
if err:
|
|
return None, err
|
|
target = (workspace / p).resolve()
|
|
if not is_relative_to(target, workspace):
|
|
return None, "`path` must stay inside the workspace"
|
|
return target, None
|
|
|
|
|
|
def _check_path_permission(workspace_path: Path, target: Path, allowed_paths) -> bool:
|
|
"""Return whether ``target`` is covered by the optional allowed-path scope.
|
|
|
|
``None`` leaves access unrestricted. Existing files allow that exact path;
|
|
existing directories allow their descendants. Missing entries are treated
|
|
as directory-like path prefixes, so a scoped ``write`` may create the path
|
|
itself or a descendant. Containment is path-component based, never a string
|
|
prefix. Entries beginning with ``~`` are ignored. Invalid constraints and
|
|
workspace-escaping entries fail closed.
|
|
"""
|
|
if allowed_paths is None:
|
|
return True
|
|
if isinstance(allowed_paths, (str, Path)):
|
|
allowed_paths = [allowed_paths]
|
|
if not isinstance(allowed_paths, (list, tuple)) or not allowed_paths:
|
|
logger.warning("invalid _allowed_paths constraint; denying access (fail closed)")
|
|
return False
|
|
|
|
allowed_files: list[Path] = []
|
|
allowed_dirs: list[Path] = []
|
|
for raw in allowed_paths:
|
|
raw_string = str(raw).strip()
|
|
if raw_string.startswith("~"):
|
|
logger.warning(f"home-relative _allowed_paths entry {raw_string!r}; skipping")
|
|
continue
|
|
resolved, err = resolve_path(workspace_path, raw_string)
|
|
if err or resolved is None:
|
|
logger.warning(f"invalid _allowed_paths entry {raw_string!r} ({err}); denying access (fail closed)")
|
|
return False
|
|
if resolved.is_file():
|
|
allowed_files.append(resolved)
|
|
elif resolved.is_dir():
|
|
allowed_dirs.append(resolved)
|
|
else:
|
|
allowed_dirs.append(resolved)
|
|
|
|
resolved_target = target.resolve()
|
|
return resolved_target in allowed_files or any(
|
|
is_relative_to(resolved_target, directory) for directory in allowed_dirs
|
|
)
|
|
|
|
|
|
def gate_md(target: Path) -> tuple[Path, bool]:
|
|
"""Markdown gate with compatibility fallback.
|
|
|
|
Returns ``(path, is_md)``:
|
|
- No suffix -> auto-append `.md`, ``is_md=True``.
|
|
- `.md` suffix -> ``is_md=True``.
|
|
- Any other suffix -> ``is_md=False`` (caller handles degraded mode).
|
|
"""
|
|
if target.suffix == "":
|
|
return target.with_suffix(".md"), True
|
|
if target.suffix.lower() != ".md":
|
|
return target, False
|
|
return target, True
|
|
|
|
|
|
def gate_image(target: Path) -> tuple[Path, bool, str | None]:
|
|
"""Image gate with compatibility fallback. Returns ``(path, is_image, mime)``."""
|
|
mime = IMAGE_MIME_BY_EXT.get(target.suffix.lower())
|
|
return target, mime is not None, mime
|