How to use the pycti.OpenCTIApiClient function in pycti
To help you get started, we’ve selected a few pycti examples, based on popular ways it is used in public projects.
Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
OpenCTI-Platform / opencti / opencti-worker / src / worker.py View on Github 
self.log_level = os.getenv('WORKER_LOG_LEVEL', 'info')
self.opencti_url = os.getenv('OPENCTI_URL', 'http://localhost:4000')
self.opencti_token = os.getenv('OPENCTI_TOKEN', 'ChangeMe')
self.rabbitmq_hostname = os.getenv('RABBITMQ_HOSTNAME', 'localhost')
self.rabbitmq_port = os.getenv('RABBITMQ_PORT', 5672)
self.rabbitmq_port_management = os.getenv('RABBITMQ_PORT_MANAGEMENT', 15672)
self.rabbitmq_management_ssl = os.getenv('RABBITMQ_MANAGEMENT_SSL', "false") == "true"
self.rabbitmq_username = os.getenv('RABBITMQ_USERNAME', 'guest')
self.rabbitmq_password = os.getenv('RABBITMQ_PASSWORD', 'guest')
# Check configuration
if len(self.opencti_token) == 0 or self.opencti_token == 'ChangeMe':
raise ValueError('Token configuration must be the same as APP__ADMIN__TOKEN')
# Check if openCTI is available
self.api = OpenCTIApiClient(self.opencti_url, self.opencti_token)
if not self.api.health_check():
raise ValueError('OpenCTI API seems down')
if self.type == 'import':
self.DEFAULT_QUEUE_NAME = 'import-platform'
self.DEFAULT_ROUTING_KEY = 'import.platform'
self.CONNECTOR_QUEUE_PREFIX = 'import-connectors-'
elif self.type == 'export':
self.DEFAULT_QUEUE_NAME = 'export-platform'
self.DEFAULT_ROUTING_KEY = 'export.platform'
self.CONNECTOR_QUEUE_PREFIX = 'export-connectors-'
# Configure logger
numeric_level = getattr(logging, self.log_level.upper(), None)
if not isinstance(numeric_level, int):
raise ValueError('Invalid log level: ' + self.log_level)self.rabbitmq_management_ssl = os.getenv('RABBITMQ_MANAGEMENT_SSL', "false") == "true"
self.rabbitmq_username = os.getenv('RABBITMQ_USERNAME', 'guest')
self.rabbitmq_password = os.getenv('RABBITMQ_PASSWORD', 'guest')
# Check configuration
if len(self.opencti_token) == 0 or self.opencti_token == '':
raise ValueError('Configuration not found')
# Configure logger
numeric_level = getattr(logging, self.log_level.upper(), None)
if not isinstance(numeric_level, int):
raise ValueError('Invalid log level: ' + self.log_level)
logging.basicConfig(level=numeric_level)
# Initialize OpenCTI client
self.opencti_api_client = OpenCTIApiClient(self.opencti_url, self.opencti_token)def __init__(self):
self.consumer_threads = {}
# Get configuration
config_file_path = os.path.dirname(os.path.abspath(__file__)) + '/config.yml'
config = yaml.load(open(config_file_path), Loader=yaml.FullLoader) if os.path.isfile(config_file_path) else {}
self.log_level = os.getenv('WORKER_LOG_LEVEL') or config['worker']['log_level']
self.opencti_url = os.getenv('OPENCTI_URL') or config['opencti']['url']
self.opencti_token = os.getenv('OPENCTI_TOKEN') or config['opencti']['token']
# Check if openCTI is available
self.api = OpenCTIApiClient(self.opencti_url, self.opencti_token)
# Configure logger
numeric_level = getattr(logging, self.log_level.upper(), None)
if not isinstance(numeric_level, int):
raise ValueError('Invalid log level: ' + self.log_level)
logging.basicConfig(level=numeric_level)
# Initialize variables
self.connectors = []
self.queues = []
