wehub-resource-sync
240 lines
8.6 KiB
Python
240 lines
8.6 KiB
Python
"""Allowlist enforcement for automatic file upload during tool execution.
|
|
|
|
This is only consulted when
|
|
``dangerously_allow_auto_upload_download_files=True``. Manual upload APIs are
|
|
not subject to the allowlist (parity with the TypeScript SDK).
|
|
|
|
- A local path is accepted iff its symlink-resolved absolute path is inside one
|
|
of the allowed directories on a path-component boundary. ``/tmp/foo`` allows
|
|
``/tmp/foo/bar`` but NOT ``/tmp/foo-bar``.
|
|
- URLs never hit this check.
|
|
- User-provided ``file_upload_dirs`` fully REPLACES the default
|
|
``[~/.composio/temp]``.
|
|
- On Windows, path comparisons are case-insensitive.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import typing as t
|
|
from pathlib import Path
|
|
|
|
from composio.exceptions import (
|
|
FileUploadPathNotAllowedError,
|
|
SDKFileNotFoundError,
|
|
)
|
|
|
|
UPLOAD_TEMP_DIRECTORY_NAME = "temp"
|
|
|
|
|
|
def get_default_upload_dir() -> t.Optional[Path]:
|
|
"""Absolute path of the default upload staging directory
|
|
(``<home>/.composio/temp``), or None when a home directory cannot be
|
|
determined."""
|
|
try:
|
|
home = Path.home()
|
|
except (RuntimeError, OSError):
|
|
return None
|
|
return (home / ".composio" / UPLOAD_TEMP_DIRECTORY_NAME).resolve()
|
|
|
|
|
|
def ensure_dir_exists(p: Path) -> None:
|
|
"""Best-effort directory creation. Allowlist enforcement still works if
|
|
creation fails (e.g. read-only filesystem)."""
|
|
try:
|
|
p.mkdir(parents=True, exist_ok=True)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
FileUploadDirs = t.Union[t.Sequence[str], t.Literal[False], None]
|
|
"""User-facing type for ``file_upload_dirs``.
|
|
|
|
- ``None`` (default) -> use ``[<home>/.composio/temp]``.
|
|
- ``False`` -> reject every local path during auto-upload (URLs / bytes still
|
|
work).
|
|
- ``[]`` -> same as ``False`` (kept as an alias; prefer ``False``).
|
|
- ``Sequence[str]`` (non-empty) -> allowlist replacing the default.
|
|
"""
|
|
|
|
|
|
def resolve_effective_upload_allowlist(
|
|
user_dirs: FileUploadDirs,
|
|
) -> t.List[Path]:
|
|
"""Resolve the effective allowlist.
|
|
|
|
- ``None`` -> ``[<default>]`` when available, else ``[]`` (fail-closed).
|
|
- ``False`` -> ``[]`` (explicit "reject all local paths"). URLs and
|
|
in-memory bytes still work because they aren't path-checked.
|
|
- ``Sequence[str]`` (including ``[]``) REPLACES the default and is returned
|
|
verbatim after ``~``-expansion and ``resolve()``. ``[]`` is treated as
|
|
equivalent to ``False``.
|
|
- Empty / blank / non-string entries are skipped.
|
|
- Duplicates are removed (case-insensitive on Windows).
|
|
"""
|
|
if user_dirs is None:
|
|
default = get_default_upload_dir()
|
|
return [default] if default is not None else []
|
|
if user_dirs is False:
|
|
return []
|
|
|
|
seen: t.Set[str] = set()
|
|
out: t.List[Path] = []
|
|
for raw in user_dirs:
|
|
if not isinstance(raw, (str, os.PathLike)):
|
|
continue
|
|
s = str(raw).strip()
|
|
if not s:
|
|
continue
|
|
abs_path = Path(s).expanduser().resolve()
|
|
key = str(abs_path).lower() if sys.platform == "win32" else str(abs_path)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
out.append(abs_path)
|
|
return out
|
|
|
|
|
|
def _is_inside_dir(child: Path, parent: Path) -> bool:
|
|
"""True iff ``child`` equals ``parent`` or is nested inside on a component
|
|
boundary. Assumes both are 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 _format_allowlist(dirs: t.Sequence[Path]) -> str:
|
|
if not dirs:
|
|
return " (none configured — all local paths are blocked)"
|
|
return "\n".join(f" - {d}" for d in dirs)
|
|
|
|
|
|
def _build_help_footer(allowlist: t.Sequence[Path]) -> str:
|
|
return "\n".join(
|
|
[
|
|
"",
|
|
"Allowed upload directories (file_upload_dirs):",
|
|
_format_allowlist(allowlist),
|
|
"",
|
|
"Fix one of the following:",
|
|
"",
|
|
" 1. Recommended — move the file under an allowed directory, or add your",
|
|
" directory to the allowlist:",
|
|
" Composio(",
|
|
" file_upload_dirs=['/abs/path/to/uploads', '~/.composio/temp'],",
|
|
" )",
|
|
" Note: user-provided `file_upload_dirs` REPLACES the default",
|
|
" `~/.composio/temp`. Include it explicitly if you want staged uploads",
|
|
" to keep working.",
|
|
"",
|
|
" 2. Pass the file by URL (http://... / https://...) instead of a",
|
|
" filesystem path. URLs are never path-checked.",
|
|
"",
|
|
" 3. (Dangerous) Bypass the allowlist entirely by opting into automatic",
|
|
" file upload/download from any readable path:",
|
|
" Composio(dangerously_allow_auto_upload_download_files=True, ...)",
|
|
" WARNING: This lets a prompt-injected or misbehaving tool read ANY",
|
|
" file the process can access. Only enable it in trusted, sandboxed",
|
|
" environments. The built-in sensitive-path denylist (.ssh, .aws,",
|
|
" .env, private SSH keys, etc.) still applies unless you also set",
|
|
" `sensitive_file_upload_protection=False`.",
|
|
]
|
|
)
|
|
|
|
|
|
def assert_path_inside_upload_dirs(
|
|
file_path: t.Union[str, Path],
|
|
allowlist: t.Sequence[Path],
|
|
) -> None:
|
|
"""Raise an elaborate error if the path is missing or outside the allowlist.
|
|
|
|
:raises SDKFileNotFoundError: when ``file_path`` does not exist on disk.
|
|
:raises FileUploadPathNotAllowedError: when resolved path is outside every
|
|
entry of ``allowlist`` (including when allowlist is empty).
|
|
"""
|
|
attempted = str(file_path)
|
|
abs_path = Path(attempted).expanduser()
|
|
try:
|
|
abs_path = abs_path.resolve(strict=False)
|
|
except OSError:
|
|
pass
|
|
|
|
if not abs_path.exists():
|
|
cwd = Path.cwd()
|
|
parent = abs_path.parent
|
|
parent_exists = parent.exists()
|
|
raise SDKFileNotFoundError(
|
|
"\n".join(
|
|
[
|
|
f'Refusing to auto-upload "{attempted}": the file does not exist on disk.',
|
|
"",
|
|
f"Path attempted: {attempted}",
|
|
f"Resolved to: {abs_path}",
|
|
f"Process cwd: {cwd}",
|
|
f"Parent exists: {'yes (' + str(parent) + ')' if parent_exists else 'no (' + str(parent) + ')'}",
|
|
"",
|
|
"Common causes:",
|
|
" - Typo in the filename passed to the tool.",
|
|
" - Relative path resolved against the wrong working directory",
|
|
" (relative paths use os.getcwd() at the moment of upload).",
|
|
" - File was deleted between the tool being called and the upload starting.",
|
|
"",
|
|
_build_help_footer(allowlist),
|
|
]
|
|
)
|
|
)
|
|
|
|
try:
|
|
real_path = abs_path.resolve(strict=True)
|
|
except OSError:
|
|
real_path = abs_path
|
|
|
|
if not allowlist:
|
|
raise FileUploadPathNotAllowedError(
|
|
"\n".join(
|
|
[
|
|
f'Refusing to auto-upload "{attempted}": no upload directories are configured.',
|
|
"",
|
|
f"Path attempted: {attempted}",
|
|
f"Resolved to: {real_path}",
|
|
"",
|
|
"Automatic file upload during tool execution is locked down by default",
|
|
"to prevent a prompt-injected tool from exfiltrating server files",
|
|
"(source code, .env, SSH keys, etc.).",
|
|
_build_help_footer(allowlist),
|
|
]
|
|
)
|
|
)
|
|
|
|
for dir_entry in allowlist:
|
|
try:
|
|
real_dir = Path(dir_entry).expanduser().resolve(strict=False)
|
|
except OSError:
|
|
real_dir = Path(dir_entry)
|
|
if _is_inside_dir(real_path, real_dir):
|
|
return
|
|
|
|
raise FileUploadPathNotAllowedError(
|
|
"\n".join(
|
|
[
|
|
f'Refusing to auto-upload "{attempted}": resolved path is not inside any',
|
|
"directory in the configured `file_upload_dirs` allowlist.",
|
|
"",
|
|
f"Path attempted: {attempted}",
|
|
f"Resolved to: {real_path}",
|
|
_build_help_footer(allowlist),
|
|
]
|
|
)
|
|
)
|