Skip to content

Download files

You can download files from the sandbox using a sha256 hash.

Code example
from ptsandbox import Sandbox, SandboxKey

async def example() -> None:
    async with Sandbox(key=SandboxKey(...)) as sandbox:
        data = await sandbox.get_file("...")
        with open("./file", "wb") as fd:
            fd.write(data)

Use streaming if you don't want to load the entire file into memory:

Code example
import aiofiles
from ptsandbox import Sandbox, SandboxKey

async def example() -> None:
    async with Sandbox(key=SandboxKey(...)) as sandbox:
        async with aiofiles.open("./file", "wb") as fd:
            async for chunk in sandbox.get_file_stream("..."):
                await fd.write(chunk)

ptsandbox.sandbox.sandbox.Sandbox.get_file async

get_file(hash: str, read_timeout: int = 120) -> bytes

Download file from the sandbox by hash

Parameters:

  • hash (str) –

    sha256 hash of the file

  • read_timeout (int, default: 120 ) –

    response waiting time in seconds

Returns:

  • bytes

    file data

Raises:

  • SandboxException

    if the hash type cannot be determined

  • ClientResponseError

    if the server returns an error status (404 if the file is not found)

  • ClientError

    on connection or transport errors

ptsandbox.sandbox.sandbox.Sandbox.get_file_stream async

get_file_stream(
    hash: str, read_timeout: int = 120
) -> AsyncIterator[bytes]

Download file from the sandbox by hash

Parameters:

  • hash (str) –

    sha256 hash of the file

  • read_timeout (int, default: 120 ) –

    response waiting time in seconds

Returns:

  • AsyncIterator[bytes]

    streaming file data

Raises:

  • SandboxException

    if the hash type cannot be determined

  • ClientResponseError

    if the server returns an error status (404 if the file is not found)

  • ClientError

    on connection or transport errors

Low-level API

Under the hood, Sandbox.get_file uses download_artifact with a sha256:<hash> file URI. You can use these methods directly to download by any file URI (e.g. from a task report):

Direct API usage
from ptsandbox import Sandbox, SandboxKey

async with Sandbox(SandboxKey(...)) as sandbox:
    # Download as bytes
    data = await sandbox.api.download_artifact("sha256:abc123...")

    # Or stream to avoid loading into memory
    async for chunk in sandbox.api.download_artifact_stream("sha256:abc123..."):
        ...

ptsandbox.sandbox.api._storage.StorageMixin.download_artifact async

download_artifact(
    file_uri: str, read_timeout: int = 120
) -> bytes

Download file from the sandbox by hash

Parameters:

  • file_uri (str) –

    Permanent file identifier in the sha256:<sha256_hex> format.

    This is not the same as the temporary file_uri returned by :meth:upload_file (sfm:stream_v1...), which is only valid for use with scan creation endpoints.

  • read_timeout (int, default: 120 ) –

    response waiting time in seconds

Returns:

  • bytes

    File data

Raises:

  • ClientResponseError

    if the server returns an error status (404 if the file is not found)

  • ClientError

    on connection or transport errors

ptsandbox.sandbox.api._storage.StorageMixin.download_artifact_stream async

download_artifact_stream(
    file_uri: str, read_timeout: int = 120
) -> AsyncIterator[bytes]

Download file from the sandbox by hash

Parameters:

  • file_uri (str) –

    Permanent file identifier in the sha256:<sha256_hex> format.

    This is not the same as the temporary file_uri returned by :meth:upload_file (sfm:stream_v1...), which is only valid for use with scan creation endpoints.

  • read_timeout (int, default: 120 ) –

    response waiting time in seconds

Returns:

  • AsyncIterator[bytes]

    streaming file data

Raises:

  • ClientResponseError

    if the server returns an error status (404 if the file is not found)

  • ClientError

    on connection or transport errors

Download all files from a task
import asyncio
import sys
from pathlib import Path
from typing import Any, Coroutine
from uuid import UUID

import aiofiles

from ptsandbox import Sandbox, SandboxKey
from ptsandbox.models import ArtifactType

semaphore = asyncio.Semaphore(12)


async def save_file(sandbox: Sandbox, file: Path, hash: str) -> None:
    file.parent.mkdir(parents=True, exist_ok=True)

    async with semaphore:
        async with aiofiles.open(f"{file}.{hash}", "wb") as fd:
            async for chunk in sandbox.get_file_stream(hash):
                await fd.write(chunk)

    print(f"saved {file}")


async def main(task_id: UUID) -> None:
    async with Sandbox(
        key=SandboxKey(
            name="test-key-1",
            key="<TOKEN_FROM_SANDBOX>",
            host="10.10.10.10",
        ),
    ) as sandbox:
        await download_artifacts(sandbox, task_id)


async def download_artifacts(sandbox: Sandbox, task_id: UUID) -> None:
    result = await sandbox.get_report(task_id)
    if (report := result.get_long_report()) is None:
        print("Can't get full report")
        return

    tasks: list[Coroutine[Any, Any, None]] = []
    for artifact in report.artifacts:
        if not (sandbox_result := artifact.find_sandbox_result()):
            continue

        if not sandbox_result.details:
            continue

        if not sandbox_result.details.sandbox:
            continue

        if not sandbox_result.details.sandbox.artifacts:
            continue

        for file in sandbox_result.details.sandbox.artifacts:
            if not file.file_info:
                continue

            if file.type != ArtifactType.FILE:
                continue

            tasks.append(
                save_file(
                    sandbox,
                    Path("artifacts") / Path(file.file_info.file_path.removeprefix("/")),
                    file.file_info.sha256,
                )
            )

    await asyncio.gather(*tasks)


if __name__ == "__main__":
    asyncio.run(main(UUID(sys.argv[1])))

Restrictions

The sandbox doesn't let you view a task report created with a different token, so you can only download your own files.

Get task report

To download all files from a task, you first need the full report which contains the list of artifacts:

Code example
from uuid import UUID
from ptsandbox import Sandbox, SandboxKey

async with Sandbox(SandboxKey(...)) as sandbox:
    report = await sandbox.get_report(UUID("..."))
    if (long_report := report.get_long_report()) is not None:
        for artifact in long_report.artifacts:
            print(artifact)

ptsandbox.sandbox.sandbox.Sandbox.get_report async

get_report(task_id: str | UUID) -> SandboxBaseTaskResponse

Getting the full task scan report

The check was completed successfully. The results are in the message body. If the scan result is not ready yet, the result and artifacts keys are missing.

⚠ The results will be returned only for the key that the analysis was started with. Sandbox restrictions for now.

Parameters:

  • task_id (str | UUID) –

    ID of the task to check

Returns:

  • SandboxBaseTaskResponse

    The response from the sandbox is either with partial information (when using async_result), or with full information.

Raises:

  • SandboxException

    if the passed task_id is not in UUID format

  • ClientResponseError

    if the server returns an error status

  • ClientError

    on connection or transport errors

  • ValidationError

    if the response body does not match the expected model