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

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_logs_worker_config()

Get the logs worker configuration from the OpenCTI platform.

Returns:

the logs worker configuration including Elasticsearch settings

Return type:

dict

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