pycti.api.opencti_api_client

class pycti.api.opencti_api_client.File(name, data, mime='text/plain')

File object for OpenCTI file uploads.

Represents a file to be uploaded via the OpenCTI API.

Parameters:

• name (str) – the filename

• data (str or bytes) – the file content (string or bytes)

• mime (str, optional) – the MIME type of the file, defaults to “text/plain”

__init__(name, data, mime='text/plain')

Initialize the File instance.

Parameters:

• name (str) – the filename

• data (str or bytes) – the file content

• mime (str) – the MIME type of the file (default: “text/plain”)

class pycti.api.opencti_api_client.OpenCTIApiClient(url, token, log_level='info', ssl_verify=False, proxies=None, json_logging=False, bundle_send_to_queue=True, cert=None, custom_headers=None, perform_health_check=True, requests_timeout=300, provider=None)

Main API client for OpenCTI

Parameters:

• url (str) – OpenCTI API url

• token (str) – OpenCTI API token

• log_level (str, optional) – log level for the client

• ssl_verify (bool, str, optional) – Requiring the requests to verify the TLS certificate at the server.

• proxies (dict, optional) – proxy configuration with “http” and “https” keys (e.g., {“http”: “http://my_proxy:8080”, “https”: “http://my_proxy:8080”})

• json_logging (bool, optional) – format the logs as json if set to True

• bundle_send_to_queue (bool, optional) – if bundle will be sent to queue

• cert (str, tuple, optional) – If String, file path to pem file. If Tuple, a (‘path_to_cert.crt’, ‘path_to_key.key’) pair representing the certificate and the key.

• custom_headers (str, optional must in the format header01:value;header02:value) – Add custom headers to use with the graphql queries

• perform_health_check (bool, optional) – if client init must check the api access

• requests_timeout (int, optional) – define the timeout for API requests in seconds

• provider (string, optional) – define client provider, and is used to specify it in requests user agent header

__init__(url, token, log_level='info', ssl_verify=False, proxies=None, json_logging=False, bundle_send_to_queue=True, cert=None, custom_headers=None, perform_health_check=True, requests_timeout=300, provider=None)

Initialize the OpenCTIApiClient instance.

Parameters:

• url (str) – OpenCTI platform URL

• token (str) – OpenCTI API authentication token

• log_level (str) – logging level (default: “info”)

• ssl_verify (Union[bool, str]) – SSL certificate verification setting

• proxies (Dict[str, str] or None) – proxy configuration dictionary with “http” and “https” keys

• json_logging (bool) – whether to format logs as JSON (default: False)

• bundle_send_to_queue (bool) – whether bundles are sent to queue (default: True)

• cert (str, tuple, or None) – client certificate path or tuple of (cert, key) paths

• custom_headers (str or None) – custom headers in format “header01:value;header02:value”

• perform_health_check (bool) – whether to check API access on init (default: True)

• requests_timeout (int) – timeout for API requests in seconds (default: 300)

• provider (str or None) – client provider for User-Agent header (format: provider/version)

Raises:

ValueError – If URL or token is missing or invalid

create_draft(**kwargs)

Create a draft in OpenCTI API.

Parameters:

• draft_name (str) – the name of the draft to create (required)

• entity_id (str, optional) – the entity ID to associate with the draft

Returns:

returns the draft workspace ID

Return type:

str

fetch_opencti_file(fetch_uri, binary=False, serialize=False)

Get file from the OpenCTI API.

Parameters:

• fetch_uri (str) – download URI to use

• binary (bool, optional) – if True, returns raw bytes; if False, returns text, defaults to False

• serialize (bool, optional) – if True, returns base64-encoded content, defaults to False

Returns:

returns either the file content as text, bytes, base64-encoded string, or None on failure

Return type:

str, bytes, or None

fetch_opencti_file_by_id(file_id, binary=False, serialize=False)

Get file from the OpenCTI API using its file id.

Builds the download URL from the platform base URL, avoiding the need for callers to craft the storage URL themselves.

Parameters:

• file_id (str) – id of the file to fetch (e.g. an importFiles entry id)

• binary (bool, optional) – if True, returns raw bytes; if False, returns text, defaults to False

• serialize (bool, optional) – if True, returns base64-encoded content, defaults to False

Returns:

returns either the file content as text, bytes, base64-encoded string, or None on failure

Return type:

str, bytes, or None

static get_attribute_in_extension(key, stix_object)

Get an attribute value from OpenCTI STIX extensions.

Searches for the key in OpenCTI extension definitions, or falls back to the object’s top-level attributes.

Parameters:

• key (str) – the attribute key to retrieve

• stix_object (dict) – the STIX object containing extensions

Returns:

the attribute value if found, None otherwise

Return type:

Any

static get_attribute_in_mitre_extension(key, stix_object)

Get an attribute value from MITRE ATT&CK STIX extension.

Parameters:

• key (str) – the attribute key to retrieve

• stix_object (dict) – the STIX object containing extensions

Returns:

the attribute value if found, None otherwise

Return type:

Any

get_draft_id()

Get the current draft ID.

Returns:

the current draft ID or empty string if not set

Return type:

str

get_request_headers(hide_token=True)

Get a copy of current request headers.

Parameters:

hide_token (bool) – if True, masks the Authorization token with asterisks

Returns:

copy of request headers

Return type:

dict

get_stix_content(id)

Get the STIX content of any entity.

Parameters:

id (str) – the ID of the entity

Returns:

the STIX content in JSON

Return type:

dict

health_check()

Submit an example request to the OpenCTI API.

Returns:

returns True if the health check has been successful

Return type:

bool

not_empty(value)

Check if a value is empty for str, list and int.

Parameters:

value (str or list or int or float or bool or datetime.date) – value to check

Returns:

returns True if the value is one of the supported types and not empty

Return type:

bool

process_multiple(data, with_pagination=False)

Process data returned by the OpenCTI API with multiple entities.

Parameters:

• data (dict) – data to process

• with_pagination (bool, optional) – whether to use pagination with the API, defaults to False

Returns:

returns either a dict or list with the processed entities

Return type:

dict or list

process_multiple_fields(data)

Process data returned by the OpenCTI API with multiple fields.

Parameters:

data (dict) – data to process

Returns:

returns the data dict with all fields processed

Return type:

dict

process_multiple_ids(data)

Process data returned by the OpenCTI API with multiple ids.

Parameters:

data (list) – data to process

Returns:

returns a list of ids

Return type:

list

query(query, variables=None, disable_impersonate=False)

Submit a query to the OpenCTI GraphQL API.

Parameters:

• query (str) – GraphQL query string

• variables (dict, optional) – GraphQL query variables, defaults to {}

• disable_impersonate (bool, optional) – removes impersonate header if set to True, defaults to False

Returns:

returns the response JSON content

Return type:

dict

Raises:

ValueError – if the API returns an error or non-200 status code

send_bundle_to_api(**kwargs)

Push a bundle to a queue through OpenCTI API.

Parameters:

• connector_id (str) – the connector ID (required)

• bundle (str) – the STIX bundle to push (required)

• work_id (str, optional) – the work ID to associate with the bundle

Returns:

returns the query response for the bundle push

Return type:

dict

set_applicant_id_header(applicant_id)

Set the applicant ID header for impersonation.

Parameters:

applicant_id (str) – the ID of the user to impersonate

set_draft_id(draft_id)

Set the draft ID header for draft mode operations.

Parameters:

draft_id (str) – the ID of the draft workspace

set_event_id(event_id)

Set the event ID header for event tracking.

Parameters:

event_id (str) – the ID of the event

set_playbook_id_header(playbook_id)

Set the playbook ID header for tracking playbook execution.

Parameters:

playbook_id (str) – the ID of the playbook being executed

set_previous_standard_header(previous_standard)

Set the previous standard header for update operations.

Parameters:

previous_standard (str) – the previous standard ID

set_retry_number(retry_number)

Set the retry number header for tracking retries.

Parameters:

retry_number (int or None) – the current retry attempt number, or None to clear

set_synchronized_upsert_header(synchronized)

Set the synchronized upsert header.

Parameters:

synchronized (bool) – whether upsert should be synchronized

set_work_id(work_id)

Set the work ID header for work validation

Parameters:

work_id (str) – the ID of the work

upload_file(**kwargs)

upload a file to OpenCTI API

Parameters:

**kwargs – arguments for file upload (required: file_name and data)

Returns:

returns the query response for the file upload

Return type:

dict

upload_pending_file(**kwargs)

Upload a pending file to OpenCTI API.

Parameters:

• file_name (str) – the name of the file to upload (required)

• data (str or bytes, optional) – the file content, defaults to reading from file_name path

• mime_type (str, optional) – the MIME type of the file, defaults to “text/plain”

• entity_id (str, optional) – the entity ID to associate with the file

• file_markings (list, optional) – list of marking definition IDs to apply

Returns:

returns the query response for the file upload

Return type:

dict

pycti.api.opencti_api_client.build_request_headers(token, custom_headers, app_logger, provider)

Build request headers for OpenCTI API requests.

Parameters:

• token (str) – the API authentication token

• custom_headers (str) – custom headers in format “header01:value;header02:value”

• app_logger (logging.Logger) – the application logger instance

• provider (str) – the provider string for User-Agent header

Returns:

dictionary of request headers

Return type:

dict