pycti.api.opencti_api_client

Classes

`File`	File object for OpenCTI file uploads.
`OpenCTIApiClient`	Main API client for OpenCTI

Functions

build_request_headers(token, custom_headers, ...)

Build request headers for OpenCTI API requests.

Module Contents

pycti.api.opencti_api_client.build_request_headers(token: str, custom_headers: str, app_logger, provider: str)[source]

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

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

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”

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”)

name[source]

data[source]

mime = 'text/plain'[source]

class pycti.api.opencti_api_client.OpenCTIApiClient(url: str, token: str, log_level: str = 'info', ssl_verify: bool | str = False, proxies: Dict[str, str] | None = None, json_logging: bool = False, bundle_send_to_queue: bool = True, cert: str | Tuple[str, str] | None = None, custom_headers: str | None = None, perform_health_check: bool = True, requests_timeout: int = 300, provider: str | None = None)[source]

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

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

bundle_send_to_queue = True[source]

ssl_verify = False[source]

cert = None[source]

proxies = None[source]

logger_class[source]

app_logger[source]

admin_logger[source]

api_token[source]

api_url[source]

provider = None[source]

request_headers[source]

session[source]

session_requests_timeout = 300[source]

work[source]

notification[source]

trash[source]

draft[source]

workspace[source]

public_dashboard[source]

playbook[source]

connector[source]

stix2[source]

pir[source]

internal_file[source]

file[source]

vocabulary[source]

label[source]

marking_definition[source]

external_reference[source]

kill_chain_phase[source]

opencti_stix_object_or_stix_relationship[source]

stix[source]

stix_domain_object[source]

stix_core_object[source]

stix_cyber_observable[source]

stix_core_relationship[source]

stix_sighting_relationship[source]

stix_nested_ref_relationship[source]

identity[source]

event[source]

location[source]

threat_actor[source]

threat_actor_group[source]

threat_actor_individual[source]

intrusion_set[source]

infrastructure[source]

campaign[source]

case_incident[source]

feedback[source]

case_rfi[source]

case_rft[source]

task[source]

incident[source]

malware[source]

malware_analysis[source]

tool[source]

channel[source]

narrative[source]

language[source]

vulnerability[source]

security_coverage[source]

attack_pattern[source]

course_of_action[source]

data_component[source]

data_source[source]

report[source]

note[source]

observed_data[source]

opinion[source]

grouping[source]

indicator[source]

capability[source]

role[source]

group[source]

user[source]

settings[source]

draft_id = ''[source]

set_applicant_id_header(applicant_id)[source]

Set the applicant ID header for impersonation.

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

set_playbook_id_header(playbook_id)[source]

Set the playbook ID header for tracking playbook execution.

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

set_event_id(event_id)[source]

Set the event ID header for event tracking.

Parameters:: event_id (str) – the ID of the event

get_draft_id()[source]

Get the current draft ID.

Returns:: the current draft ID or empty string if not set
Return type:: str

set_draft_id(draft_id)[source]

Set the draft ID header for draft mode operations.

Parameters:: draft_id (str) – the ID of the draft workspace

set_work_id(work_id)[source]

Set the work ID header for work validation

Parameters:: work_id (str) – the ID of the work

set_synchronized_upsert_header(synchronized)[source]

Set the synchronized upsert header.

Parameters:: synchronized (bool) – whether upsert should be synchronized

set_previous_standard_header(previous_standard)[source]

Set the previous standard header for update operations.

Parameters:: previous_standard (str) – the previous standard ID

get_request_headers(hide_token=True)[source]

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

set_retry_number(retry_number)[source]

Set the retry number header for tracking retries.

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

query(query, variables=None, disable_impersonate=False)[source]

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

fetch_opencti_file(fetch_uri, binary=False, serialize=False)[source]

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

health_check()[source]

Submit an example request to the OpenCTI API.

Returns:: returns True if the health check has been successful
Return type:: bool

get_logs_worker_config()[source]

Get the logs worker configuration from the OpenCTI platform.

Returns:: the logs worker configuration including Elasticsearch settings
Return type:: dict

not_empty(value)[source]

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: dict, with_pagination=False) → dict | list[source]

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_ids(data) → list[source]

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

process_multiple_fields(data)[source]

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

upload_file(**kwargs)[source]

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

create_draft(**kwargs)[source]

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

upload_pending_file(**kwargs)[source]

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

send_bundle_to_api(**kwargs)[source]

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

get_stix_content(id)[source]

Get the STIX content of any entity.

Parameters:: id (str) – the ID of the entity
Returns:: the STIX content in JSON
Return type:: dict

connector_jwt()[source]

static get_attribute_in_extension(key, stix_object) → Any[source]

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) → Any[source]

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