Skip to content

Tasks

List tasks, export them to CSV, get task summaries, and browse artifact trees.

Get all tasks

Code example
import asyncio
from ptsandbox import Sandbox, SandboxKey

async def main():
    async with Sandbox(...) as sandbox:
        await sandbox.ui.authorize()

        tasks = await sandbox.ui.get_tasks()
        print(tasks)

asyncio.run(main())

ptsandbox.sandbox.ui._tasks.TasksMixin.get_tasks async 

get_tasks(
    query: str = "",
    limit: int = 20,
    offset: int = 0,
    utc_offset_seconds: int = 0,
    next_cursor: str | None = None,
) -> SandboxUITasksResponse

Get tasks listing

Parameters:

  • query (str, default: '' ) –

    filtering using the query language. For the syntax, see the user documentation.

    age < 30d AND (task.correlated.state != UNKNOWN ) ORDER BY start desc
  • limit (int, default: 20 ) –

    limit on the number of records to be returned

  • offset (int, default: 0 ) –

    the offset of the returned records. If the next cursor is specified, the offset is counted from the cursor.

  • utc_offset_seconds (int, default: 0 ) –

    the offset of the user's time from UTC, which will be used for the time in QL queries

  • next_cursor (str | None, default: None ) –

    the value from the previous request

Returns:

Raises:

  • SandboxException –

    if not authorized or token refresh fails

  • 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

Export in csv

Code example
import asyncio
import aiofiles
from ptsandbox import Sandbox, SandboxKey

async def main():
    async with Sandbox(...) as sandbox:
        await sandbox.ui.authorize()

        async with aiofiles.open("./tasks.csv", "wb") as fd:
            async for chunk in sandbox.ui.get_tasks_csv():
                await fd.write(chunk)

asyncio.run(main())

ptsandbox.sandbox.ui._tasks.TasksMixin.get_tasks_csv async 

get_tasks_csv(
    query: str = "",
    columns: list[
        Literal[
            "action",
            "behavioralAnalysis",
            "fromTo",
            "priority",
            "processedTime",
            "quarantine",
            "source",
            "status",
            "taskName",
            "time",
            "verdict",
            "verdictTime",
        ]
    ]
    | None = None,
    utc_offset_seconds: int = 0,
) -> AsyncIterator[bytes]

Export a tasks listing to CSV

Parameters:

  • query (str, default: '' ) –

    filtering using the query language. For the syntax, see the user documentation.

  • columns (list[Literal['action', 'behavioralAnalysis', 'fromTo', 'priority', 'processedTime', 'quarantine', 'source', 'status', 'taskName', 'time', 'verdict', 'verdictTime']] | None, default: None ) –

    the list of csv columns to be exported.

  • utc_offset_seconds (int, default: 0 ) –

    the offset of the user's time from UTC, which will be used for the time in QL queries

Returns:

  • AsyncIterator[bytes] –

    AsyncIterator with chunks of CSV file

Raises:

  • SandboxException –

    if not authorized or token refresh fails

  • ClientResponseError –

    if the server returns an error status

  • ClientError –

    on connection or transport errors

Get filter values

Code example
import asyncio
from ptsandbox import Sandbox, SandboxKey

async def main():
    async with Sandbox(...) as sandbox:
        await sandbox.ui.authorize()

        values = await sandbox.ui.get_tasks_filter_values()
        print(values)

asyncio.run(main())

ptsandbox.sandbox.ui._tasks.TasksMixin.get_tasks_filter_values async 

get_tasks_filter_values(
    from_: str = "",
    to: str = "",
    scan_id: UUID | None = None,
) -> SandboxTasksFilterValuesResponse

Get possible values for filters based on sources and validation results

Parameters:

  • from_ (str, default: '' ) –

    for which period possible values are being searched: minimum time

  • to (str, default: '' ) –

    for which period possible values are being searched: maximum time

  • scan_id (UUID | None, default: None ) –

    filter by task ID

Returns:

Raises:

  • SandboxException –

    if not authorized or token refresh fails

  • 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

Task

Summary

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

async def main():
    async with Sandbox(...) as sandbox:
        await sandbox.ui.authorize()

        summary = await sandbox.ui.get_task_summary(UUID("..."))
        print(summary)

asyncio.run(main())

ptsandbox.sandbox.ui._tasks.TasksMixin.get_task_summary async 

get_task_summary(
    scan_id: UUID,
) -> SandboxTasksSummaryResponse

Get information about a specific task

Parameters:

  • scan_id (UUID) –

    task id

Returns:

Raises:

  • SandboxException –

    if not authorized or token refresh fails

  • 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

Get a tree of artifacts for a specific task

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

async def main():
    async with Sandbox(...) as sandbox:
        await sandbox.ui.authorize()

        summary = await sandbox.ui.get_task_tree(UUID("..."))
        print(summary)

asyncio.run(main())

ptsandbox.sandbox.ui._artifacts.ArtifactsMixin.get_task_tree async 

get_task_tree(
    scan_id: UUID,
    *,
    parent_path: list[int] | None = None,
    filtered_by_ids: list[int] | None = None,
    limit: int = 1000,
    offset: int = 0,
    max_tree_level: int = 3,
    sort_mode: Literal[
        "DANGEROUS", "ALPHABETICAL"
    ] = "DANGEROUS",
) -> SandboxTreeResponse

Get a tree of artifacts for a specific task

Parameters:

  • scan_id (UUID) –

    ...

  • parent_path (list[int] | None, default: None ) –

    the full path to the parent to start loading the tree from. For example: [0, 2, 10]

  • filtered_by_ids (list[int] | None, default: None ) –

    a list of IDs of specific nodes to be returned, for example: [0, 2, 10, 11]

  • limit (int, default: 1000 ) –

    limit on the number of records to be returned

  • offset (int, default: 0 ) –

    the indentation from which the records are returned, used for pagination

  • max_tree_level (int, default: 3 ) –

    the maximum depth (relative to the parent) to be returned

  • sort_mode (Literal['DANGEROUS', 'ALPHABETICAL'], default: 'DANGEROUS' ) –

    the sorting method. First, the dangerous ones are 'DANGEROUS' or just alphabetically 'ALPHABETIC'

Returns:

Raises:

  • SandboxException –

    if not authorized or token refresh fails

  • 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

Download all the artifacts of the task

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

async def main():
    async with Sandbox(...) as sandbox:
        await sandbox.ui.authorize()

        async with aiofiles.open("artifacts.zip", "wb") as fd:
            async for chunk in sandbox.ui.get_task_artifacts(UUID("...")):
                await fd.write(chunk)

asyncio.run(main())

ptsandbox.sandbox.ui._artifacts.ArtifactsMixin.get_task_artifacts async 

get_task_artifacts(
    scan_id: UUID,
    *,
    query: str = "",
    include_sandbox_logs: Literal["true", "false"] = "true",
    skip_data_files: Literal["true", "false"] = "false",
) -> AsyncIterator[bytes]

Download all the artifacts of the task

Parameters:

  • scan_id (UUID) –

    ...

  • query (str, default: '' ) –

    filtering using the query language. For the syntax, see the user documentation.

  • include_sandbox_logs (Literal['true', 'false'], default: 'true' ) –

    whether to include BA logs as a result

  • skip_data_files (Literal['true', 'false'], default: 'false' ) –

    whether to include data files in the result

Returns:

  • AsyncIterator[bytes] –

    Sandbox returns an encrypted zip archive (password - infected), so we just export a set of bytes.

  • AsyncIterator[bytes] –

    If necessary, you can use pyzipper to unpack

Raises:

  • SandboxException –

    if not authorized or token refresh fails

  • ClientResponseError –

    if the server returns an error status

  • ClientError –

    on connection or transport errors

Get scan result for a specific artifact

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

async def main():
    async with Sandbox(...) as sandbox:
        await sandbox.ui.authorize()

        scan_id = UUID("...")
        tree = await sandbox.ui.get_task_tree(scan_id)
        for children in tree.children:
            scan = await sandbox.ui.get_task_artifact_scans(scan_id, children.node_id)

asyncio.run(main())

ptsandbox.sandbox.ui._artifacts.ArtifactsMixin.get_task_artifact_scans async 

get_task_artifact_scans(
    scan_id: UUID, node_id: int
) -> SandboxScansResponse

Getting scan results for a specific artifact

Parameters:

  • scan_id (UUID) –

    ...

  • node_id (int) –

    ...

Returns:

Raises:

  • SandboxException –

    if not authorized or token refresh fails

  • 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