Method download

python/composio/core/models/_files.py:580–622

(self, outdir: Path, chunk_size: int = _DEFAULT_CHUNK_SIZE)

    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