composiohq__composio/python/composio/utils/safe_path.py

"""Path containment primitives for untrusted input.

Every field of an API response is untrusted, including slugs, identifiers, and
filenames. A compromised toolkit or a MITM on the Composio API can set them to
anything. When such a value becomes part of a filesystem path, it must pass
through this module.

Two rules make the difference between a real check and a decorative one:

1. **Anchor on a constant.** Containment is only meaningful when the root is a
   trusted value. Checking a derived path against a directory that untrusted
   input helped build compares tainted against tainted and always passes. A
   directory built as ``root / untrusted_a / untrusted_b`` and then validated
   against that same built directory is checked against a reference the input
   was free to move.

2. **Validate before touching the filesystem.** No ``mkdir``, no ``open``, until
   the final path is known to be inside the root. Otherwise a rejected write
   still leaves attacker-chosen directories behind.

:func:`secure_join` enforces both. Prefer it over hand-rolled
``resolve() / is_relative_to()`` pairs — see ``python/AGENTS.md``.
"""

from __future__ import annotations

import os
import re
import sys
import typing as t
from pathlib import Path, PureWindowsPath

from composio.exceptions import UnsafePathComponentError

SAFE_COMPONENT_REGEX = re.compile(r"^[A-Za-z0-9_-]+$")
"""Characters permitted in a path component derived from untrusted input.

Deliberately identical to ``SLUG_REGEX`` in ``custom_tool_types.py``, which the
SDK already enforces on client-created custom tools. Backend-fetched tools were
never held to the same standard; they are now.

Every one of the 1070 toolkit slugs in the catalog snapshot at
``ts/packages/cli/src/generated/toolkit-slugs.ts`` conforms, so this is not
expected to reject anything legitimate. That snapshot is not the live catalog:
a production slug containing a dot or a non-ASCII character would hard-fail its
downloads rather than degrade.
"""

MAX_COMPONENT_LENGTH = 128
"""Upper bound on a single component, well under the 255-byte limit common to
ext4/APFS/NTFS. Keeps a long slug from raising ``OSError`` mid-write."""

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)}
    | {f"COM{i}" for i in "¹²³"}
    | {f"LPT{i}" for i in "¹²³"}
)
"""Reserved DOS device names. Writing to one on Windows targets the device
rather than a file. Rejected on every platform so behaviour does not diverge
between a POSIX developer machine and a Windows deployment."""


def is_inside_dir(child: Path, parent: Path) -> bool:
    """True iff ``child`` equals ``parent`` or is nested inside it on a path
    component boundary.

    ``/tmp/foo`` contains ``/tmp/foo/bar`` but NOT ``/tmp/foo-bar`` — a plain
    string prefix test gets that wrong. Comparison is case-insensitive on
    Windows, where ``C:\\Foo`` and ``c:\\foo`` are the same directory.

    Both arguments are assumed absolute and normalized.
    """
    try:
        child_str = str(child)
        parent_str = str(parent)
        if sys.platform == "win32":
            child_str = child_str.lower()
            parent_str = parent_str.lower()
        if child_str == parent_str:
            return True
        sep = os.sep
        parent_with_sep = parent_str if parent_str.endswith(sep) else parent_str + sep
        return child_str.startswith(parent_with_sep)
    except OSError:
        return False


def assert_safe_path_component(value: str, *, label: str = "path component") -> str:
    """Return ``value`` unchanged if it is safe to use as a single path
    component, else raise.

    Fails closed. Rejects traversal (``..``), separators of either platform,
    absolute paths, drive letters, NUL bytes, reserved device names, and
    anything outside :data:`SAFE_COMPONENT_REGEX`.

    :raises UnsafePathComponentError: when ``value`` is unsafe.
    """
    if not isinstance(value, str) or not value:
        raise UnsafePathComponentError(
            f"Refusing to build a path from an empty or non-string {label}: {value!r}"
        )

    # `PureWindowsPath` treats both `/` and `\` as separators, so a single check
    # catches `../x` and `..\x` regardless of the host platform. A slug crafted
    # for a Windows target must not slip through on a POSIX build machine.
    as_windows_path = PureWindowsPath(value)
    if len(as_windows_path.parts) != 1 or as_windows_path.anchor:
        raise UnsafePathComponentError(
            f"Refusing to build a path from a {label} containing path separators "
            f"or a drive letter: {value!r}"
        )

    if len(value) > MAX_COMPONENT_LENGTH:
        raise UnsafePathComponentError(
            f"Refusing to build a path from a {label} longer than "
            f"{MAX_COMPONENT_LENGTH} characters: {value[:32]!r}... "
            f"({len(value)} characters)"
        )

    # `.` and `..` are excluded by the regex (no `.` in the character class),
    # as are NUL bytes and every separator. The explicit checks above exist to
    # produce a precise error message rather than a generic pattern mismatch.
    #
    # `fullmatch`, not `match`: in a `match`, `$` also matches just before a
    # single trailing newline, so `"GMAIL\n"` would satisfy `^[A-Za-z0-9_-]+$`
    # and reach the filesystem with a control character in the name.
    if not SAFE_COMPONENT_REGEX.fullmatch(value):
        raise UnsafePathComponentError(
            f"Refusing to build a path from an unsafe {label}: {value!r}. "
            f"Expected only letters, digits, underscores, and hyphens "
            f"(pattern {SAFE_COMPONENT_REGEX.pattern})."
        )

    if value.upper() in WINDOWS_RESERVED_NAMES:
        raise UnsafePathComponentError(
            f"Refusing to build a path from a reserved device name as {label}: {value!r}"
        )

    return value


def safe_basename(name: str, *, label: str = "filename") -> str:
    """Collapse an untrusted filename to a bare, writable basename.

    Filenames need their own validator: :func:`assert_safe_path_component`
    forbids ``.``, which nearly every real filename contains. This applies the
    remaining checks — no separators, no traversal, no NUL, bounded length, no
    reserved device name — to the one component a server most directly controls.

    ``PureWindowsPath`` treats both ``/`` and ``\\`` as separators, so a name
    crafted for a Windows target (``..\\..\\evil``) is stripped even when the
    SDK runs on POSIX, where ``Path(...).name`` would return it intact.

    Names that leave no usable basename are refused rather than replaced with a
    generated one: a response that cannot name its own file is malformed or
    hostile, and inventing a name would hide that. ``.`` and the empty string
    both basename to ``""``, which makes an output path equal to its own
    directory and surfaces as ``IsADirectoryError`` at write time.

    :raises UnsafePathComponentError: when ``name`` yields no usable basename or
        is unsafe to write.
    """
    if not isinstance(name, str):
        raise UnsafePathComponentError(
            f"Refusing to write a non-string {label}: {name!r}"
        )

    raw_basename = PureWindowsPath(name).name
    if not raw_basename or not raw_basename.strip() or set(raw_basename) == {"."}:
        raise UnsafePathComponentError(
            f"Path traversal detected: {label} {name!r} leaves no usable "
            "basename to write to."
        )
    if "\x00" in raw_basename:
        raise UnsafePathComponentError(
            f"Refusing to write {label} containing a NUL byte: {name!r}"
        )
    if any(ord(char) < 32 or char in '<>:"|?*' for char in raw_basename):
        raise UnsafePathComponentError(
            f"Refusing to write {label} containing characters reserved by "
            f"Windows: {name!r}"
        )
    if raw_basename.endswith((" ", ".")):
        raise UnsafePathComponentError(
            f"Refusing to write {label} ending in a space or dot: {name!r}"
        )

    basename = raw_basename.strip()
    try:
        encoded_length = len(os.fsencode(basename))
    except UnicodeEncodeError as e:
        raise UnsafePathComponentError(
            f"Refusing to write {label} containing invalid Unicode: {name!r}"
        ) from e
    if encoded_length > MAX_COMPONENT_LENGTH:
        raise UnsafePathComponentError(
            f"Refusing to write {label} longer than {MAX_COMPONENT_LENGTH} bytes: "
            f"{basename[:32]!r}... ({encoded_length} bytes)"
        )
    # Compare everything before the first dot: on Windows `NUL.tar.gz` opens
    # the null device just as `NUL` does, so any number of extensions provides
    # no protection.
    device_name = basename.split(".", 1)[0].rstrip(" ").upper()
    if device_name in WINDOWS_RESERVED_NAMES:
        raise UnsafePathComponentError(
            f"Refusing to write {label} that is a reserved device name: {name!r}"
        )
    return basename


def resolve_root(root: t.Union[str, Path]) -> Path:
    """Normalize a trusted root to an absolute, symlink-resolved path.

    Every containment check must derive its anchor through this one function.
    When two call sites normalize differently — one expanding ``~`` and one not
    — the check compares mismatched paths and rejects legitimate writes while
    reporting them as attacks. Sharing the normalization is what keeps the two
    ends of a containment check comparable.
    """
    expanded = Path(root).expanduser()
    try:
        return expanded.resolve(strict=False)
    except OSError:
        return expanded


def secure_basename_join(
    base: t.Union[str, Path],
    name: str,
    *,
    root: t.Optional[t.Union[str, Path]] = None,
    label: str = "filename",
) -> Path:
    """Join a single untrusted filename under ``base``, contained within ``root``.

    The filename counterpart to :func:`secure_join`, which cannot be reused here
    because it forbids ``.`` — correct for a slug, wrong for ``report.pdf``.

    ``root`` defaults to ``base`` but is separate for the download path, where
    ``base`` is a per-tool subdirectory that untrusted slugs helped build and
    only the configured ``root`` above it is trusted. Anchoring on ``base``
    there would check the result against a directory those slugs had moved.

    :raises UnsafePathComponentError: when ``name`` is unsafe, or when the
        result escapes ``root``.
    """
    resolved_base = resolve_root(base)
    resolved_root = resolved_base if root is None else resolve_root(root)
    candidate = resolved_base / safe_basename(name, label=label)
    if not is_inside_dir(candidate.resolve(), resolved_root):
        raise UnsafePathComponentError(
            f"Path traversal detected: {label} {name!r} resolves to "
            f"{candidate.resolve()}, which is outside {resolved_root}."
        )
    return candidate


def secure_join(root: t.Union[str, Path], *components: str) -> Path:
    """Join untrusted ``components`` beneath the trusted ``root``.

    ``root`` is the sole anchor of trust and must not itself be derived from
    untrusted input — that is the whole point. Each component is validated by
    :func:`assert_safe_path_component`, then the joined result is resolved and
    re-checked against the resolved root. The second check is belt-and-braces:
    it catches a symlink inside ``root`` pointing outside it, which per-component
    validation cannot see.

    Performs no filesystem writes; the caller creates directories only after
    this returns.

    :raises UnsafePathComponentError: when any component is unsafe, or when the
        joined path escapes ``root``.
    """
    resolved_root = resolve_root(root)

    safe_components = [
        assert_safe_path_component(component) for component in components
    ]

    candidate = resolved_root.joinpath(*safe_components)
    try:
        resolved_candidate = candidate.resolve(strict=False)
    except OSError:
        resolved_candidate = candidate

    if not is_inside_dir(resolved_candidate, resolved_root):
        raise UnsafePathComponentError(
            f"Refusing to write outside the configured directory: "
            f"{components!r} resolves to {resolved_candidate}, "
            f"which is outside {resolved_root}."
        )

    return resolved_candidate