wehub-resource-sync
1192 lines
42 KiB
Python
1192 lines
42 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import logging
|
|
import os
|
|
import typing as t
|
|
from pathlib import Path
|
|
from urllib.parse import unquote, urlparse
|
|
import uuid
|
|
import datetime
|
|
|
|
import requests
|
|
import typing_extensions as te
|
|
from composio_client import BaseModel as _ComposioBaseModel
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
from composio.client import HttpClient
|
|
from composio.client.types import Tool
|
|
from composio.exceptions import (
|
|
ErrorDownloadingFile,
|
|
ErrorUploadingFile,
|
|
FileUploadAbortedError,
|
|
ResponseTooLargeError,
|
|
SDKFileNotFoundError,
|
|
)
|
|
from composio.utils import mimetypes
|
|
from composio.utils.json_schema import dereference_json_schema
|
|
from composio.utils.sensitive_file_upload_paths import (
|
|
assert_safe_local_file_upload_path,
|
|
)
|
|
from composio.utils.upload_dir_allowlist import (
|
|
assert_path_inside_upload_dirs,
|
|
)
|
|
from composio.utils.logging import WithLogger
|
|
|
|
if t.TYPE_CHECKING:
|
|
from .tools import ToolExecutionResponse
|
|
from ._modifiers import BeforeFileUploadContextCallable # noqa: F401
|
|
|
|
_DEFAULT_CHUNK_SIZE = 1024 * 1024
|
|
_FILE_UPLOAD = "/api/v3/files/upload/request"
|
|
_MAX_FILENAME_LENGTH = 100
|
|
"""
|
|
Maximum filename length to prevent issues with long URLs from public buckets.
|
|
Long filenames (containing hashes, UUIDs, or encoded metadata) are replaced
|
|
with timestamped filenames to match TypeScript SDK behavior.
|
|
"""
|
|
|
|
_MAX_RESPONSE_SIZE = 100 * 1024 * 1024 # 100 MB default limit
|
|
"""
|
|
Maximum response size in bytes when fetching files from URLs.
|
|
Prevents memory exhaustion attacks from malicious URLs pointing to large files.
|
|
"""
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
_CONNECT_TIMEOUT = 5 # seconds
|
|
_READ_TIMEOUT = 60 # seconds
|
|
"""
|
|
Separate connect and read timeouts for URL fetching.
|
|
Connect timeout is short to fail fast on unreachable hosts.
|
|
Read timeout is longer to allow for slower file transfers.
|
|
"""
|
|
|
|
_DELETE_VALUE: t.Final = object()
|
|
"""
|
|
Sentinel returned by the upload walker to signal that a value should be dropped
|
|
from its parent container. ``None`` and ``""`` are both legal payload values, so
|
|
the walker cannot use either to mean "remove this key/item".
|
|
"""
|
|
|
|
LOCAL_CACHE_DIRECTORY_NAME = ".composio"
|
|
"""
|
|
Local cache directory name for composio CLI
|
|
"""
|
|
|
|
ENV_LOCAL_CACHE_DIRECTORY = "COMPOSIO_CACHE_DIR"
|
|
"""
|
|
Environment to set the composio caching directory.
|
|
"""
|
|
|
|
LOCAL_CACHE_DIRECTORY = Path(
|
|
os.environ.get(
|
|
ENV_LOCAL_CACHE_DIRECTORY,
|
|
Path.home() / LOCAL_CACHE_DIRECTORY_NAME, # Fallback to user directory
|
|
)
|
|
)
|
|
"""
|
|
Path to local caching directory.
|
|
"""
|
|
|
|
try:
|
|
LOCAL_CACHE_DIRECTORY.mkdir(parents=True, exist_ok=True)
|
|
if not os.access(LOCAL_CACHE_DIRECTORY, os.W_OK):
|
|
raise OSError
|
|
except OSError as e:
|
|
raise RuntimeError(
|
|
f"Cache directory {LOCAL_CACHE_DIRECTORY} is not writable please "
|
|
f"provide a path that is writable using {ENV_LOCAL_CACHE_DIRECTORY} "
|
|
"environment variable."
|
|
) from e
|
|
|
|
|
|
LOCAL_OUTPUT_FILE_DIRECTORY = LOCAL_CACHE_DIRECTORY / "files"
|
|
"""
|
|
Default local directory into which files downloaded during tool execution are
|
|
written. Previously ``<cache>/outputs``; now ``<cache>/files`` for parity with
|
|
the TypeScript SDK. Override by passing ``file_download_dir=...`` to Composio,
|
|
or by setting ``outdir`` on ``FileHelper`` directly.
|
|
"""
|
|
|
|
|
|
def get_md5(file: Path) -> str:
|
|
"""Calculate MD5 hash of a file for integrity verification.
|
|
|
|
Note: MD5 is used here for file integrity checking and deduplication,
|
|
not for cryptographic security. The Composio API requires MD5 hashes
|
|
for file upload verification. For security-critical applications,
|
|
consider using SHA-256 for additional integrity checks.
|
|
|
|
Args:
|
|
file: Path to file to hash
|
|
|
|
Returns:
|
|
Hexadecimal MD5 hash string
|
|
"""
|
|
# `usedforsecurity=False` lets this run on FIPS-mode systems, where
|
|
# `hashlib.md5()` without the flag raises `ValueError: [digital envelope
|
|
# routines] unsupported`. We're hashing for integrity / deduplication,
|
|
# not security — the API just needs the digest for upload verification.
|
|
obj = hashlib.md5(usedforsecurity=False)
|
|
with file.open("rb") as fp:
|
|
while True:
|
|
line = fp.read(_DEFAULT_CHUNK_SIZE)
|
|
if not line:
|
|
break
|
|
obj.update(line)
|
|
return obj.hexdigest()
|
|
|
|
|
|
def upload(url: str, file: Path) -> bool:
|
|
"""Upload file to presigned S3 URL.
|
|
|
|
Args:
|
|
url: Presigned S3 upload URL
|
|
file: Path to file to upload
|
|
|
|
Returns:
|
|
True if upload succeeded (HTTP 200), False otherwise
|
|
"""
|
|
with file.open("rb") as data:
|
|
try:
|
|
response = requests.put(
|
|
url=url,
|
|
data=data,
|
|
timeout=(_CONNECT_TIMEOUT, _READ_TIMEOUT),
|
|
)
|
|
except requests.exceptions.RequestException as e:
|
|
_logger.debug(
|
|
"Upload to %s failed: %s",
|
|
_sanitize_url_for_logging(url),
|
|
type(e).__name__,
|
|
)
|
|
return False
|
|
return response.status_code == 200
|
|
|
|
|
|
class _FileUploadResponse(_ComposioBaseModel):
|
|
id: str
|
|
key: str
|
|
type: str
|
|
new_presigned_url: str
|
|
|
|
|
|
def _is_url(value: str) -> bool:
|
|
"""Check if a string is a valid HTTP/HTTPS URL."""
|
|
try:
|
|
parsed = urlparse(value)
|
|
return parsed.scheme in ("http", "https") and bool(parsed.netloc)
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _get_extension_from_mimetype(mimetype: str) -> str:
|
|
"""Get file extension from mimetype."""
|
|
mime_to_ext = {
|
|
"image/jpeg": ".jpg",
|
|
"image/png": ".png",
|
|
"image/gif": ".gif",
|
|
"image/webp": ".webp",
|
|
"image/svg+xml": ".svg",
|
|
"image/bmp": ".bmp",
|
|
"image/tiff": ".tiff",
|
|
"application/pdf": ".pdf",
|
|
"application/json": ".json",
|
|
"application/xml": ".xml",
|
|
"text/plain": ".txt",
|
|
"text/html": ".html",
|
|
"text/css": ".css",
|
|
"text/javascript": ".js",
|
|
"application/zip": ".zip",
|
|
"application/gzip": ".gz",
|
|
"audio/mpeg": ".mp3",
|
|
"audio/wav": ".wav",
|
|
"video/mp4": ".mp4",
|
|
"video/webm": ".webm",
|
|
}
|
|
return mime_to_ext.get(mimetype.lower(), "")
|
|
|
|
|
|
def _generate_timestamped_filename(extension: str) -> str:
|
|
"""Generate a unique filename with timestamp."""
|
|
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
unique_id = uuid.uuid4().hex[:8]
|
|
return f"file_{timestamp}_{unique_id}{extension}"
|
|
|
|
|
|
def _truncate_filename(filename: str, max_length: int = _MAX_FILENAME_LENGTH) -> str:
|
|
"""Truncate filename if it exceeds max length by generating a timestamped name.
|
|
|
|
Long filenames are common with public bucket URLs containing hashes or UUIDs.
|
|
These can cause issues, so we replace them with timestamped filenames while
|
|
preserving the extension.
|
|
|
|
Args:
|
|
filename: The original filename
|
|
max_length: Maximum allowed length for the filename
|
|
|
|
Returns:
|
|
The original filename if within limits, or a timestamped filename
|
|
with the extension preserved if the original is too long
|
|
"""
|
|
if len(filename) <= max_length:
|
|
return filename
|
|
|
|
# Extract extension
|
|
if "." in filename:
|
|
_, ext = filename.rsplit(".", 1)
|
|
ext = f".{ext}"
|
|
else:
|
|
ext = ""
|
|
|
|
# Generate a timestamped filename (matches TypeScript SDK behavior)
|
|
return _generate_timestamped_filename(ext)
|
|
|
|
|
|
def _sanitize_url_for_logging(url: str) -> str:
|
|
"""Sanitize URL for safe logging (remove sensitive query params).
|
|
|
|
Args:
|
|
url: URL to sanitize
|
|
|
|
Returns:
|
|
URL with query parameters hidden
|
|
"""
|
|
parsed = urlparse(url)
|
|
if parsed.query:
|
|
return f"{parsed.scheme}://{parsed.netloc}{parsed.path}?[REDACTED]"
|
|
return url
|
|
|
|
|
|
def _fetch_file_from_url(
|
|
url: str,
|
|
max_size: int = _MAX_RESPONSE_SIZE,
|
|
) -> t.Tuple[str, bytes, str]:
|
|
"""Fetch file content from a URL with security protections.
|
|
|
|
Security features:
|
|
- Response size limiting (prevents memory exhaustion)
|
|
- Redirects disabled (prevents redirect-based attacks)
|
|
- Separate connect/read timeouts
|
|
|
|
Args:
|
|
url: URL to fetch file from
|
|
max_size: Maximum response size in bytes (default: 100MB)
|
|
|
|
Returns:
|
|
Tuple of (filename, content_bytes, mimetype)
|
|
|
|
Raises:
|
|
ResponseTooLargeError: If response exceeds max_size
|
|
ErrorUploadingFile: If fetch fails for other reasons
|
|
"""
|
|
# Make request without following redirects
|
|
try:
|
|
response = requests.get(
|
|
url,
|
|
stream=True, # Enable streaming for size limiting
|
|
allow_redirects=False, # Disable redirects for security
|
|
timeout=(_CONNECT_TIMEOUT, _READ_TIMEOUT),
|
|
)
|
|
except requests.exceptions.Timeout:
|
|
raise ErrorUploadingFile(
|
|
f"Request timed out fetching URL: {_sanitize_url_for_logging(url)}"
|
|
)
|
|
except requests.exceptions.RequestException as e:
|
|
raise ErrorUploadingFile(
|
|
f"Failed to fetch file from URL: {_sanitize_url_for_logging(url)}. Error: {e}"
|
|
)
|
|
|
|
# Reject redirects - require direct URL to resource
|
|
if response.status_code in (301, 302, 303, 307, 308):
|
|
location = response.headers.get("Location", "unknown")
|
|
response.close()
|
|
raise ErrorUploadingFile(
|
|
f"URL returned redirect to {_sanitize_url_for_logging(location)}. "
|
|
f"Please provide a direct URL to the file."
|
|
)
|
|
|
|
# Check for successful response
|
|
if not response.ok:
|
|
response.close()
|
|
raise ErrorUploadingFile(
|
|
f"Failed to fetch file from URL: {_sanitize_url_for_logging(url)}. "
|
|
f"Status: {response.status_code}"
|
|
)
|
|
|
|
# Check Content-Length header first (early abort for oversized files)
|
|
content_length = response.headers.get("Content-Length")
|
|
if content_length and int(content_length) > max_size:
|
|
response.close()
|
|
raise ResponseTooLargeError(
|
|
f"File size ({int(content_length)} bytes) exceeds maximum allowed "
|
|
f"size ({max_size} bytes)"
|
|
)
|
|
|
|
# Stream response with size tracking
|
|
chunks: t.List[bytes] = []
|
|
total_bytes = 0
|
|
chunk_size = 8192 # 8 KB chunks
|
|
|
|
try:
|
|
for chunk in response.iter_content(chunk_size=chunk_size):
|
|
if chunk:
|
|
total_bytes += len(chunk)
|
|
if total_bytes > max_size:
|
|
response.close()
|
|
raise ResponseTooLargeError(
|
|
f"Response size exceeds maximum allowed size ({max_size} bytes)"
|
|
)
|
|
chunks.append(chunk)
|
|
finally:
|
|
response.close()
|
|
|
|
content = b"".join(chunks)
|
|
|
|
# Extract mimetype
|
|
mimetype = response.headers.get("content-type", "application/octet-stream")
|
|
# Handle mimetypes with charset or other parameters (e.g., "text/html; charset=utf-8")
|
|
mimetype = mimetype.split(";")[0].strip()
|
|
|
|
# Extract filename from URL (decode percent-encoded characters)
|
|
parsed_url = urlparse(url)
|
|
pathname = unquote(parsed_url.path)
|
|
filename = os.path.basename(pathname) if pathname else ""
|
|
|
|
# If no filename from URL or no extension, generate one
|
|
if not filename:
|
|
extension = _get_extension_from_mimetype(mimetype)
|
|
filename = _generate_timestamped_filename(extension)
|
|
else:
|
|
# If filename has no extension, try to add one from mimetype
|
|
if "." not in filename:
|
|
extension = _get_extension_from_mimetype(mimetype)
|
|
filename = _generate_timestamped_filename(extension)
|
|
|
|
# Truncate long filenames (common with public bucket URLs containing hashes)
|
|
filename = _truncate_filename(filename)
|
|
|
|
return filename, content, mimetype
|
|
|
|
|
|
def _upload_bytes_to_s3(
|
|
client: HttpClient,
|
|
filename: str,
|
|
content: bytes,
|
|
mimetype: str,
|
|
tool: str,
|
|
toolkit: str,
|
|
) -> str:
|
|
"""Upload bytes content to S3 and return the S3 key."""
|
|
md5_hash = hashlib.md5(content, usedforsecurity=False).hexdigest()
|
|
|
|
s3meta = client.post(
|
|
path=_FILE_UPLOAD,
|
|
body={
|
|
"md5": md5_hash,
|
|
"filename": filename,
|
|
"mimetype": mimetype,
|
|
"tool_slug": tool,
|
|
"toolkit_slug": toolkit,
|
|
},
|
|
cast_to=_FileUploadResponse,
|
|
)
|
|
|
|
# Upload the content directly to S3
|
|
try:
|
|
upload_response = requests.put(
|
|
url=s3meta.new_presigned_url,
|
|
data=content,
|
|
headers={"Content-Type": mimetype},
|
|
timeout=(_CONNECT_TIMEOUT, _READ_TIMEOUT),
|
|
)
|
|
except requests.exceptions.RequestException as e:
|
|
raise ErrorUploadingFile(
|
|
"Failed to upload to S3: "
|
|
f"{_sanitize_url_for_logging(s3meta.new_presigned_url)}. "
|
|
f"Error: {type(e).__name__}"
|
|
) from e
|
|
|
|
if upload_response.status_code != 200:
|
|
raise ErrorUploadingFile(
|
|
f"Failed to upload to S3. Status: {upload_response.status_code}. "
|
|
f"This may indicate an expired presigned URL or permission issue."
|
|
)
|
|
|
|
return s3meta.key
|
|
|
|
|
|
class FileUploadable(BaseModel):
|
|
model_config = ConfigDict(json_schema_extra={"file_uploadable": True})
|
|
|
|
name: str
|
|
mimetype: str
|
|
s3key: str
|
|
|
|
@classmethod
|
|
def from_url(
|
|
cls,
|
|
client: HttpClient,
|
|
url: str,
|
|
tool: str,
|
|
toolkit: str,
|
|
) -> te.Self:
|
|
"""Create a FileUploadable from a public URL.
|
|
|
|
Fetches the file content from the URL and uploads it to S3.
|
|
|
|
:param client: The HTTP client for API calls
|
|
:param url: The public URL to fetch the file from
|
|
:param tool: The tool slug
|
|
:param toolkit: The toolkit slug
|
|
:return: FileUploadable instance with S3 key
|
|
"""
|
|
filename, content, mimetype = _fetch_file_from_url(url)
|
|
|
|
s3key = _upload_bytes_to_s3(
|
|
client=client,
|
|
filename=filename,
|
|
content=content,
|
|
mimetype=mimetype,
|
|
tool=tool,
|
|
toolkit=toolkit,
|
|
)
|
|
|
|
return cls(name=filename, mimetype=mimetype, s3key=s3key)
|
|
|
|
@classmethod
|
|
def from_path(
|
|
cls,
|
|
client: HttpClient,
|
|
file: t.Union[str, Path],
|
|
tool: str,
|
|
toolkit: str,
|
|
*,
|
|
sensitive_file_upload_protection: bool = True,
|
|
file_upload_path_deny_segments: t.Optional[t.Sequence[str]] = None,
|
|
file_upload_allowlist: t.Optional[t.Sequence[Path]] = None,
|
|
before_file_upload: t.Optional["BeforeFileUploadContextCallable"] = None,
|
|
) -> te.Self:
|
|
"""Create a FileUploadable from a local file path or public URL.
|
|
|
|
If the file parameter is a URL (starts with http:// or https://),
|
|
it will fetch the file content from the URL and upload it to S3.
|
|
Otherwise, it treats it as a local file path.
|
|
|
|
:param client: The HTTP client for API calls
|
|
:param file: Local file path or public URL
|
|
:param tool: The tool slug
|
|
:param toolkit: The toolkit slug
|
|
:param sensitive_file_upload_protection: When True, block paths on the built-in denylist.
|
|
:param file_upload_path_deny_segments: Extra path segment names to merge with the built-in list.
|
|
:param file_upload_allowlist: When provided (not None), local paths must
|
|
resolve inside one of these directories on a component boundary.
|
|
Pass ``None`` to skip the check (e.g. manual upload APIs). URLs are
|
|
never checked against the allowlist. An empty sequence means
|
|
"no paths are allowed" (fail-closed).
|
|
:param before_file_upload: Optional context-form hook produced by
|
|
:func:`composio.core.models._modifiers.merge_before_file_upload`.
|
|
Receives ``{"path", "source", "tool", "toolkit"}`` where ``source``
|
|
is ``"url"`` for ``http(s)://...`` inputs and ``"path"`` for local
|
|
filesystem paths. Return a new string to substitute, or ``False``
|
|
to abort.
|
|
:return: FileUploadable instance with S3 key
|
|
"""
|
|
file_str = str(file) if isinstance(file, Path) else file
|
|
path_in = file_str
|
|
source: t.Literal["url", "path"] = (
|
|
"url" if isinstance(file_str, str) and _is_url(file_str) else "path"
|
|
)
|
|
|
|
if before_file_upload is not None:
|
|
out = before_file_upload(
|
|
{
|
|
"path": path_in,
|
|
"source": source,
|
|
"tool": tool,
|
|
"toolkit": toolkit,
|
|
}
|
|
)
|
|
if out is False:
|
|
raise FileUploadAbortedError(
|
|
"File upload was aborted because before_file_upload returned False."
|
|
)
|
|
if isinstance(out, str):
|
|
path_in = out
|
|
|
|
# Re-decide routing on the post-hook value: a URL-source hook may return
|
|
# a local path (and vice versa). Re-checking with `_is_url` keeps the
|
|
# URL fetch path and the local-file path properly separated, so a hook
|
|
# cannot, for example, smuggle `/etc/passwd` past the URL branch's
|
|
# missing allowlist/denylist by rewriting the URL into a path.
|
|
if isinstance(path_in, str) and _is_url(path_in):
|
|
return cls.from_url(client=client, url=path_in, tool=tool, toolkit=toolkit)
|
|
|
|
# Allowlist check runs BEFORE the denylist / existence checks when enabled,
|
|
# so the "configure file_upload_dirs" hint fires first for the common case
|
|
# (user has auto-upload on but hasn't configured dirs). Caller passes
|
|
# ``None`` to bypass the allowlist (manual upload APIs).
|
|
if file_upload_allowlist is not None:
|
|
assert_path_inside_upload_dirs(path_in, file_upload_allowlist)
|
|
|
|
assert_safe_local_file_upload_path(
|
|
path_in,
|
|
enabled=sensitive_file_upload_protection,
|
|
additional_deny_segments=file_upload_path_deny_segments,
|
|
)
|
|
|
|
# Handle as local file path
|
|
file = Path(path_in)
|
|
if not file.exists():
|
|
raise SDKFileNotFoundError(
|
|
f"File not found: {file}. Please provide a valid file path."
|
|
)
|
|
|
|
if not file.is_file():
|
|
raise SDKFileNotFoundError(
|
|
f"Not a file: {file}. Please provide a valid file path."
|
|
)
|
|
|
|
if not os.access(file, os.R_OK):
|
|
raise SDKFileNotFoundError(
|
|
f"File not readable: {file}. Please check the file permissions."
|
|
)
|
|
|
|
mimetype = mimetypes.guess(file=file)
|
|
s3meta = client.post(
|
|
path=_FILE_UPLOAD,
|
|
body={
|
|
"md5": get_md5(file=file),
|
|
"filename": file.name,
|
|
"mimetype": mimetype,
|
|
"tool_slug": tool,
|
|
"toolkit_slug": toolkit,
|
|
},
|
|
cast_to=_FileUploadResponse,
|
|
)
|
|
if not upload(url=s3meta.new_presigned_url, file=file):
|
|
raise ErrorUploadingFile(f"Error uploading file: {file}")
|
|
return cls(name=file.name, mimetype=mimetype, s3key=s3meta.key)
|
|
|
|
|
|
class FileDownloadable(BaseModel):
|
|
model_config = ConfigDict(json_schema_extra={"file_downloadable": True})
|
|
|
|
name: str = Field(..., description="Name of the file")
|
|
mimetype: str = Field(..., description="Mime type of the file.")
|
|
s3url: str = Field(..., description="URL of the file.")
|
|
|
|
def download(self, outdir: Path, chunk_size: int = _DEFAULT_CHUNK_SIZE) -> Path:
|
|
# SEC-316: `self.name` comes from the (potentially compromised or
|
|
# MITM'd) Composio API response. Strip directory components with
|
|
# `Path(...).name` so traversal sequences like `../../../foo` collapse
|
|
# to `foo`, then verify the resolved output stays under `outdir` so a
|
|
# name like `output_evil/foo` (sibling-prefix attack) is also rejected.
|
|
safe_name = Path(self.name).name
|
|
outfile = outdir / safe_name
|
|
if not outfile.resolve().is_relative_to(outdir.resolve()):
|
|
raise ErrorDownloadingFile(
|
|
f"Path traversal detected: filename '{self.name}' resolves "
|
|
"outside the intended output directory."
|
|
)
|
|
outdir.mkdir(exist_ok=True, parents=True)
|
|
try:
|
|
response = requests.get(
|
|
url=self.s3url,
|
|
stream=True,
|
|
timeout=(_CONNECT_TIMEOUT, _READ_TIMEOUT),
|
|
)
|
|
except requests.exceptions.RequestException as e:
|
|
raise ErrorDownloadingFile(
|
|
"Error downloading file: "
|
|
f"{_sanitize_url_for_logging(self.s3url)}. Error: {type(e).__name__}"
|
|
) from e
|
|
if response.status_code != 200:
|
|
response.close()
|
|
raise ErrorDownloadingFile(
|
|
f"Error downloading file: {_sanitize_url_for_logging(self.s3url)}"
|
|
)
|
|
|
|
try:
|
|
with outfile.open("wb") as fd:
|
|
for chunk in response.iter_content(chunk_size=chunk_size):
|
|
fd.write(chunk)
|
|
except requests.exceptions.RequestException as e:
|
|
raise ErrorDownloadingFile(
|
|
"Error downloading file: "
|
|
f"{_sanitize_url_for_logging(self.s3url)}. Error: {type(e).__name__}"
|
|
) from e
|
|
finally:
|
|
response.close()
|
|
return outfile
|
|
|
|
|
|
# Internal alias — ``FileHelper`` receives the already-adapted context-form
|
|
# callable produced by :func:`merge_before_file_upload`.
|
|
from composio.core.models._modifiers import ( # noqa: E402
|
|
BeforeFileUploadContextCallable as BeforeFileUpload,
|
|
)
|
|
|
|
|
|
class FileHelper(WithLogger):
|
|
def __init__(
|
|
self,
|
|
client: HttpClient,
|
|
outdir: t.Optional[str] = None,
|
|
*,
|
|
sensitive_file_upload_protection: bool = True,
|
|
file_upload_path_deny_segments: t.Optional[t.Sequence[str]] = None,
|
|
file_upload_allowlist: t.Optional[t.Sequence[Path]] = None,
|
|
) -> None:
|
|
"""
|
|
:param outdir: Where files downloaded during tool execution are written.
|
|
Defaults to ``~/.composio/files``.
|
|
:param file_upload_allowlist: Directories from which local files may be
|
|
auto-uploaded. ``None`` disables the allowlist check (used when
|
|
auto-upload is off, or for manual upload APIs). An empty list means
|
|
"no paths allowed" (fail-closed).
|
|
"""
|
|
super().__init__()
|
|
self._client = client
|
|
self._outdir = Path(outdir) if outdir else LOCAL_OUTPUT_FILE_DIRECTORY
|
|
self._sensitive_file_upload_protection = sensitive_file_upload_protection
|
|
self._file_upload_path_deny_segments = file_upload_path_deny_segments
|
|
self._file_upload_allowlist: t.Optional[t.Sequence[Path]] = (
|
|
list(file_upload_allowlist) if file_upload_allowlist is not None else None
|
|
)
|
|
|
|
def _has_file_property(
|
|
self, schema: t.Dict, property_name: str = "file_uploadable"
|
|
) -> bool:
|
|
"""Check if a schema (or any of its variants) contains a file property.
|
|
|
|
Recursively checks anyOf, oneOf, allOf, nested properties, and array items.
|
|
"""
|
|
if not isinstance(schema, dict):
|
|
return False
|
|
|
|
# Direct property check
|
|
if schema.get(property_name, False):
|
|
return True
|
|
|
|
# Check anyOf variants
|
|
if "anyOf" in schema:
|
|
for variant in schema["anyOf"]:
|
|
if self._has_file_property(variant, property_name):
|
|
return True
|
|
|
|
# Check oneOf variants
|
|
if "oneOf" in schema:
|
|
for variant in schema["oneOf"]:
|
|
if self._has_file_property(variant, property_name):
|
|
return True
|
|
|
|
# Check allOf variants
|
|
if "allOf" in schema:
|
|
for variant in schema["allOf"]:
|
|
if self._has_file_property(variant, property_name):
|
|
return True
|
|
|
|
# Check nested properties
|
|
if "properties" in schema:
|
|
for prop in schema["properties"].values():
|
|
if self._has_file_property(prop, property_name):
|
|
return True
|
|
|
|
# Check array items
|
|
if "items" in schema:
|
|
items = schema["items"]
|
|
if isinstance(items, list):
|
|
for item in items:
|
|
if self._has_file_property(item, property_name):
|
|
return True
|
|
elif isinstance(items, dict):
|
|
if self._has_file_property(items, property_name):
|
|
return True
|
|
|
|
return False
|
|
|
|
def _file_uploadable(self, schema: t.Dict) -> bool:
|
|
"""Check if a schema has file_uploadable property."""
|
|
return self._has_file_property(schema, "file_uploadable")
|
|
|
|
def _process_file_uploadable(self, schema: t.Dict) -> t.Dict:
|
|
return {
|
|
"type": "string",
|
|
"format": "path",
|
|
"description": schema.get("description", "Path to file."),
|
|
"title": schema.get("title"),
|
|
"file_uploadable": True,
|
|
}
|
|
|
|
def _transform_schema_for_file_upload(self, schema: t.Dict) -> t.Dict:
|
|
"""Recursively transform a schema, converting file_uploadable fields to path format.
|
|
|
|
Handles anyOf, oneOf, allOf, nested properties, and array items.
|
|
"""
|
|
if not isinstance(schema, dict):
|
|
return schema
|
|
|
|
# Direct file_uploadable - transform it
|
|
if schema.get("file_uploadable", False):
|
|
return self._process_file_uploadable(schema)
|
|
|
|
# Create a copy to avoid mutating the original
|
|
new_schema = dict(schema)
|
|
|
|
# Transform anyOf variants
|
|
if "anyOf" in schema:
|
|
new_schema["anyOf"] = [
|
|
self._transform_schema_for_file_upload(variant)
|
|
for variant in schema["anyOf"]
|
|
]
|
|
|
|
# Transform oneOf variants
|
|
if "oneOf" in schema:
|
|
new_schema["oneOf"] = [
|
|
self._transform_schema_for_file_upload(variant)
|
|
for variant in schema["oneOf"]
|
|
]
|
|
|
|
# Transform allOf variants
|
|
if "allOf" in schema:
|
|
new_schema["allOf"] = [
|
|
self._transform_schema_for_file_upload(variant)
|
|
for variant in schema["allOf"]
|
|
]
|
|
|
|
# Transform nested properties
|
|
if "properties" in schema:
|
|
new_schema["properties"] = {
|
|
key: self._transform_schema_for_file_upload(prop)
|
|
for key, prop in schema["properties"].items()
|
|
}
|
|
|
|
# Transform array items
|
|
if "items" in schema:
|
|
items = schema["items"]
|
|
if isinstance(items, list):
|
|
new_schema["items"] = [
|
|
self._transform_schema_for_file_upload(item) for item in items
|
|
]
|
|
elif isinstance(items, dict):
|
|
new_schema["items"] = self._transform_schema_for_file_upload(items)
|
|
|
|
return new_schema
|
|
|
|
def enhance_schema_descriptions(self, schema: t.Dict) -> t.Dict:
|
|
"""Add type hints and required notes to parameter descriptions.
|
|
|
|
This method enhances parameter descriptions by adding:
|
|
- Type hints ("Please provide a value of type...")
|
|
- Required notes ("This parameter is required.")
|
|
|
|
This is separate from file processing and should always run
|
|
regardless of `dangerously_allow_auto_upload_download_files`.
|
|
|
|
Schemas with no `properties` key (e.g. `{}` for tools that declare
|
|
no input parameters — common with MCP-backed toolkits) are returned
|
|
unchanged, matching the sibling `process_file_uploadable_schema`.
|
|
"""
|
|
if "properties" not in schema:
|
|
return schema
|
|
required = schema.get("required") or []
|
|
for _param, _schema in schema["properties"].items():
|
|
if _schema.get("type") in ["string", "integer", "number", "boolean"]:
|
|
ext = f"Please provide a value of type {_schema['type']}."
|
|
description = _schema.get("description", "").rstrip(".")
|
|
_schema["description"] = f"{description}. {ext}" if description else ext
|
|
|
|
if _param in required:
|
|
description = _schema.get("description")
|
|
_schema["description"] = (
|
|
(f"{description.rstrip('.')}. This parameter is required.")
|
|
if description
|
|
else "This parameter is required."
|
|
)
|
|
return schema
|
|
|
|
def process_file_uploadable_schema(self, schema: t.Dict) -> t.Dict:
|
|
"""Process file_uploadable fields in schema.
|
|
|
|
This method converts file_uploadable fields to path format.
|
|
Should only be called when the caller opted in via
|
|
`dangerously_allow_auto_upload_download_files=True`.
|
|
Recursively handles anyOf, oneOf, allOf, nested properties, and array items.
|
|
|
|
``$ref``/``$defs`` indirection is inlined first so file_uploadable flags
|
|
reachable only through a reference are visible to the transform below
|
|
(https://github.com/ComposioHQ/composio/issues/3506). Tool schemas come
|
|
from the API, which may emit a ``$ref`` without a ``$defs`` block
|
|
(https://github.com/ComposioHQ/composio/issues/3307); ``"sentinel"``
|
|
degrades that gracefully instead of raising.
|
|
"""
|
|
resolved = dereference_json_schema(schema, on_unresolved="sentinel")
|
|
if "properties" not in resolved:
|
|
return schema
|
|
|
|
# Preserve the original object's identity (callers such as
|
|
# ``process_schema_recursively`` mutate in place) while adopting the
|
|
# dereferenced, ``$defs``-stripped shape.
|
|
schema.clear()
|
|
schema.update(resolved)
|
|
schema["properties"] = {
|
|
key: self._transform_schema_for_file_upload(prop)
|
|
for key, prop in schema["properties"].items()
|
|
}
|
|
return schema
|
|
|
|
def process_schema_recursively(self, schema: t.Dict) -> t.Dict:
|
|
"""Process schema for both file handling and description enhancements.
|
|
|
|
This method is kept for backward compatibility. It calls both
|
|
process_file_uploadable_schema and enhance_schema_descriptions.
|
|
"""
|
|
self.process_file_uploadable_schema(schema)
|
|
self.enhance_schema_descriptions(schema)
|
|
return schema
|
|
|
|
def _schema_variants(self, schema: t.Dict) -> t.List[t.Dict]:
|
|
"""Return composed schema variants in the SDK's historical order."""
|
|
variants: t.List[t.Dict] = []
|
|
for key in ("anyOf", "oneOf", "allOf"):
|
|
schema_variants = schema.get(key)
|
|
if not isinstance(schema_variants, list):
|
|
continue
|
|
variants.extend(v for v in schema_variants if isinstance(v, dict))
|
|
return variants
|
|
|
|
def _json_schema_type_matches_value(self, schema: t.Dict, value: t.Any) -> bool:
|
|
"""Best-effort runtime shape match for selecting composed schemas."""
|
|
schema_type = schema.get("type")
|
|
if isinstance(schema_type, list):
|
|
return any(
|
|
self._json_schema_type_matches_value({**schema, "type": tpe}, value)
|
|
for tpe in schema_type
|
|
)
|
|
|
|
if schema_type is None:
|
|
if "properties" in schema:
|
|
return isinstance(value, dict)
|
|
if "items" in schema:
|
|
return isinstance(value, list)
|
|
return False
|
|
|
|
if schema_type == "array":
|
|
return isinstance(value, list)
|
|
if schema_type == "object":
|
|
return isinstance(value, dict)
|
|
if schema_type == "string":
|
|
return isinstance(value, str)
|
|
if schema_type == "integer":
|
|
return isinstance(value, int) and not isinstance(value, bool)
|
|
if schema_type == "number":
|
|
return isinstance(value, (int, float)) and not isinstance(value, bool)
|
|
if schema_type == "boolean":
|
|
return isinstance(value, bool)
|
|
if schema_type == "null":
|
|
return value is None
|
|
|
|
return False
|
|
|
|
def _find_schema_variant_with_file_property(
|
|
self,
|
|
schema: t.Dict,
|
|
property_name: str,
|
|
value: t.Any = None,
|
|
) -> t.Optional[t.Dict]:
|
|
"""Find the composed schema variant that should process a file value.
|
|
|
|
``anyOf``, ``oneOf`` and ``allOf`` are intentionally treated the same
|
|
here, matching the SDK's existing convention. When a runtime value is
|
|
available, prefer the file-bearing variant with the matching JSON Schema
|
|
shape; otherwise keep the historical first-match behavior.
|
|
"""
|
|
candidates = [
|
|
variant
|
|
for variant in self._schema_variants(schema)
|
|
if self._has_file_property(variant, property_name)
|
|
]
|
|
if not candidates:
|
|
return None
|
|
|
|
if value is not None:
|
|
for candidate in candidates:
|
|
if self._json_schema_type_matches_value(candidate, value):
|
|
return candidate
|
|
|
|
return candidates[0]
|
|
|
|
def _find_uploadable_schema_variant(
|
|
self, schema: t.Dict, value: t.Any = None
|
|
) -> t.Optional[t.Dict]:
|
|
"""Find a schema variant that contains file_uploadable properties."""
|
|
return self._find_schema_variant_with_file_property(
|
|
schema=schema,
|
|
property_name="file_uploadable",
|
|
value=value,
|
|
)
|
|
|
|
def _upload_file_value(
|
|
self,
|
|
value: t.Any,
|
|
tool: Tool,
|
|
before_file_upload: t.Optional[BeforeFileUpload],
|
|
) -> t.Any:
|
|
if value is None or value == "":
|
|
return _DELETE_VALUE
|
|
|
|
return FileUploadable.from_path(
|
|
client=self._client,
|
|
file=value,
|
|
tool=tool.slug,
|
|
toolkit=tool.toolkit.slug,
|
|
sensitive_file_upload_protection=self._sensitive_file_upload_protection,
|
|
file_upload_path_deny_segments=self._file_upload_path_deny_segments,
|
|
file_upload_allowlist=self._file_upload_allowlist,
|
|
before_file_upload=before_file_upload,
|
|
).model_dump()
|
|
|
|
def _substitute_file_upload_value(
|
|
self,
|
|
value: t.Any,
|
|
schema: t.Optional[t.Dict],
|
|
tool: Tool,
|
|
*,
|
|
before_file_upload: t.Optional[BeforeFileUpload] = None,
|
|
) -> t.Any:
|
|
"""Return ``value`` with file-uploadable leaves staged for execution."""
|
|
if not isinstance(schema, dict):
|
|
return value
|
|
|
|
if schema.get("file_uploadable", False):
|
|
return self._upload_file_value(
|
|
value=value,
|
|
tool=tool,
|
|
before_file_upload=before_file_upload,
|
|
)
|
|
|
|
uploadable_variant = self._find_uploadable_schema_variant(
|
|
schema=schema,
|
|
value=value,
|
|
)
|
|
if uploadable_variant is not None:
|
|
return self._substitute_file_upload_value(
|
|
value=value,
|
|
schema=uploadable_variant,
|
|
tool=tool,
|
|
before_file_upload=before_file_upload,
|
|
)
|
|
|
|
if isinstance(value, dict) and "properties" in schema:
|
|
processed: t.Dict[str, t.Any] = {}
|
|
properties = schema["properties"]
|
|
for key, item in value.items():
|
|
item_schema = properties.get(key)
|
|
processed_item = self._substitute_file_upload_value(
|
|
value=item,
|
|
schema=item_schema,
|
|
tool=tool,
|
|
before_file_upload=before_file_upload,
|
|
)
|
|
if processed_item is not _DELETE_VALUE:
|
|
processed[key] = processed_item
|
|
return processed
|
|
|
|
if isinstance(value, list) and "items" in schema:
|
|
items_schema = schema["items"]
|
|
if isinstance(items_schema, list):
|
|
items_schema = items_schema[0] if items_schema else None
|
|
|
|
processed_items: t.List[t.Any] = []
|
|
for item in value:
|
|
processed_item = self._substitute_file_upload_value(
|
|
value=item,
|
|
schema=items_schema,
|
|
tool=tool,
|
|
before_file_upload=before_file_upload,
|
|
)
|
|
if processed_item is not _DELETE_VALUE:
|
|
processed_items.append(processed_item)
|
|
return processed_items
|
|
|
|
return value
|
|
|
|
def _substitute_file_uploads_recursively(
|
|
self,
|
|
tool: Tool,
|
|
schema: t.Dict,
|
|
request: t.Dict,
|
|
*,
|
|
before_file_upload: t.Optional[BeforeFileUpload] = None,
|
|
) -> t.Dict:
|
|
processed = self._substitute_file_upload_value(
|
|
value=request,
|
|
schema=schema,
|
|
tool=tool,
|
|
before_file_upload=before_file_upload,
|
|
)
|
|
if processed is request:
|
|
return request
|
|
if isinstance(processed, dict):
|
|
request.clear()
|
|
request.update(processed)
|
|
return request
|
|
assert isinstance(processed, dict), (
|
|
"expected dict from _substitute_file_upload_value at the root; "
|
|
f"got {type(processed).__name__}"
|
|
)
|
|
return processed
|
|
|
|
def substitute_file_uploads(
|
|
self,
|
|
tool: Tool,
|
|
request: t.Dict,
|
|
*,
|
|
before_file_upload: t.Optional[BeforeFileUpload] = None,
|
|
) -> t.Dict:
|
|
"""Stage file-uploadable leaves in ``request`` and return it.
|
|
|
|
Mutation contract: the top-level ``request`` dict is mutated in place
|
|
and its identity is preserved (the return value is the same object).
|
|
Nested dicts inside ``request`` may be replaced with fresh dicts
|
|
rather than mutated, so callers should not retain references to
|
|
nested values across this call.
|
|
"""
|
|
return self._substitute_file_uploads_recursively(
|
|
tool=tool,
|
|
# Inline $ref/$defs once at the boundary so the walker sees
|
|
# file_uploadable flags hidden behind a reference. Non-mutating:
|
|
# ``tool.input_parameters`` is left untouched.
|
|
schema=dereference_json_schema(
|
|
tool.input_parameters, on_unresolved="sentinel"
|
|
),
|
|
request=request,
|
|
before_file_upload=before_file_upload,
|
|
)
|
|
|
|
def _is_file_downloadable(self, schema: t.Dict) -> bool:
|
|
"""Check if a schema has file_downloadable property."""
|
|
return self._has_file_property(schema, "file_downloadable")
|
|
|
|
def _find_downloadable_schema_variant(
|
|
self, schema: t.Dict, value: t.Any = None
|
|
) -> t.Optional[t.Dict]:
|
|
"""Find a schema variant that contains file_downloadable properties."""
|
|
return self._find_schema_variant_with_file_property(
|
|
schema=schema,
|
|
property_name="file_downloadable",
|
|
value=value,
|
|
)
|
|
|
|
def _download_file_value(self, value: t.Any, tool: Tool) -> t.Any:
|
|
if isinstance(value, dict) and "s3url" in value:
|
|
return str(
|
|
FileDownloadable(**value).download(
|
|
self._outdir / tool.toolkit.slug / tool.slug
|
|
)
|
|
)
|
|
return value
|
|
|
|
def _substitute_file_download_value(
|
|
self,
|
|
value: t.Any,
|
|
schema: t.Optional[t.Dict],
|
|
tool: Tool,
|
|
) -> t.Any:
|
|
"""Return ``value`` with file-downloadable leaves saved locally."""
|
|
if not isinstance(schema, dict):
|
|
return value
|
|
|
|
if schema.get("file_downloadable", False):
|
|
return self._download_file_value(value=value, tool=tool)
|
|
|
|
downloadable_variant = self._find_downloadable_schema_variant(
|
|
schema=schema,
|
|
value=value,
|
|
)
|
|
if downloadable_variant is not None:
|
|
return self._substitute_file_download_value(
|
|
value=value,
|
|
schema=downloadable_variant,
|
|
tool=tool,
|
|
)
|
|
|
|
if isinstance(value, dict) and "properties" in schema:
|
|
properties = schema["properties"]
|
|
return {
|
|
key: self._substitute_file_download_value(
|
|
value=item,
|
|
schema=properties.get(key),
|
|
tool=tool,
|
|
)
|
|
for key, item in value.items()
|
|
}
|
|
|
|
if isinstance(value, list) and "items" in schema:
|
|
items_schema = schema["items"]
|
|
if isinstance(items_schema, list):
|
|
items_schema = items_schema[0] if items_schema else None
|
|
return [
|
|
self._substitute_file_download_value(
|
|
value=item,
|
|
schema=items_schema,
|
|
tool=tool,
|
|
)
|
|
for item in value
|
|
]
|
|
|
|
return value
|
|
|
|
def _substitute_file_downloads_recursively(
|
|
self,
|
|
tool: Tool,
|
|
schema: t.Dict,
|
|
request: t.Dict,
|
|
) -> t.Dict:
|
|
processed = self._substitute_file_download_value(
|
|
value=request,
|
|
schema=schema,
|
|
tool=tool,
|
|
)
|
|
if processed is request:
|
|
return request
|
|
if isinstance(processed, dict):
|
|
request.clear()
|
|
request.update(processed)
|
|
return request
|
|
assert isinstance(processed, dict), (
|
|
"expected dict from _substitute_file_download_value at the root; "
|
|
f"got {type(processed).__name__}"
|
|
)
|
|
return processed
|
|
|
|
def substitute_file_downloads(
|
|
self,
|
|
tool: Tool,
|
|
response: ToolExecutionResponse,
|
|
) -> ToolExecutionResponse:
|
|
"""Materialize file-downloadable leaves in ``response`` and return it.
|
|
|
|
Mutation contract: the top-level ``response`` dict is mutated in
|
|
place and its identity is preserved (the return value is the same
|
|
object). Nested dicts inside ``response`` may be replaced with
|
|
fresh dicts rather than mutated, so callers should not retain
|
|
references to nested values across this call.
|
|
"""
|
|
return t.cast(
|
|
"ToolExecutionResponse",
|
|
self._substitute_file_downloads_recursively(
|
|
tool=tool,
|
|
# Inline $ref/$defs once at the boundary so the walker sees
|
|
# file_downloadable flags hidden behind a reference (e.g.
|
|
# GMAIL_GET_ATTACHMENT). Non-mutating: ``tool.output_parameters``
|
|
# is left untouched.
|
|
schema=dereference_json_schema(
|
|
tool.output_parameters, on_unresolved="sentinel"
|
|
),
|
|
request=t.cast(dict, response),
|
|
),
|
|
)
|