Event collector integrations - Develop event collector integrations to fetch events and logs from external products to Cortex XSIAM. - Developer Guide - Cortex XSIAM - Cortex - Security Operations

To create an event collector integration use the demisto-sdk init --xsiam command. This creates a new content pack with all necessary Cortex XSIAM content items. The event collector integration can be found at Packs/Integrations/${VENDOR_NAME}EventCollector.

Event collector integration names (id, name, and display fields) should end with EventCollector so users can easily understand what the integration is used for.

Use the demisto-sdk init --xsiam command to automatically generate the necessary integration YAML configuration keys for an event collector. If you create the YAML manually, verify the following keys are set:

Example

script:
  isfetchevents: true
fromversion: 6.8.0
marketplaces:
- marketplacev2

For event collector integration instance settings, the event collector related parameters should be organized in one section by adding the Collect key to each relevant parameter in the integration YAML file:

sectionOrder:
- Connect
- Collect
configuration:
  # ...
- name: ...
  # ...
  section: Collect

For example, since the first_fetch and fetch_limit are event collector related parameters, add them to the Collect section:

- name: first_fetch
  defaultvalue: 3 days
  display: First fetch time
  required: false
  type: 0
  section: Collect # <--- Added
- name: fetch_limit
  defaultvalue: '1000'
  display: Fetch Limit
  additionalinfo: Maximum amount of detections to fetch. Audits API does not include a fetch limit therefore this configuration is only relevant to detections.
  required: false
  type: 0
  section: Collect # <--- Added

In the integration instance modal, it looks like this:

Every event collector integration supports at least three commands:

Call the send_events_to_xsiam() function from CommonServerPython when the fetch-events command is executed.

This command expects the following arguments:

Example: main() function from an event collector integration:

This example assumes the events are not in cef or leef formats, therefore the data_format argument is not used.

def main():
    params = demisto.params()

    client = Client(params.get('insecure'),
                    params.get('proxy'))

    command = demisto.command()
    demisto.info(f'Command being called is {command}')
    # Switch case
    try:
        if demisto.command() == 'fetch-events':
            events, last_run = fetch_events_command(client)
            # we submit the indicators in batches
            send_events_to_xsiam(events=events, vendor='MyVendor', product='MyProduct')
            demisto.setLastRun(next_fetch)
        else:
            results = get_events_command(client)
            return_results(results)
    except Exception as e:
        raise Exception(f'Error in {SOURCE_NAME} Integration [{e}]')

For more info on the send_events_to_xsiam() function, see the API reference.

Events with multiple types

If within fetch-events different API endpoints are called, then events may consist of multiple types with different structures. In this case, call send_events_to_xsiam() with an aggregated list of events from both endpoints. For example, for detections: List[Dict[str, Any]] and audits: List[Dict[str, Any]], send them as:

audits, detections, last_run = fetch_events_command(client)
send_events_to_xsiam(events=audits + detections, vendor='MyVendor', product='MyProduct')

For more details, see https://xsoar.pan.dev/docs/reference/api/common-server-python#send_events_to_xsiam.

When an integration runs for the first time, the last run time is not in the integration context. To set up the first run properly, use an if statement with a time that is specified in the integration settings.

It is best practice to specify in the integration settings how far back in time to fetch events for the first run.

Queries and parameters are configurable parameters in the integration settings that enable filtering events. For example, to import only certain event types into Cortex XSIAM, you need to query the API for only that specific event type.

The following example uses the First Run if statement and query.

# usually there will be some kind of query based on event creation date, 
    # or get all the events with id greater than X id and their status is New
    query = 'status=New'

    day_ago = datetime.now() - timedelta(days=1) 
    start_time = day_ago.time()
    if last_run and 'start_time' in last_run:
        start_time = last_run.get('start_time')

    # execute the query and get the events
    events = query_events(query, start_time)

When developing an event collector integration, you can implement parsing rules in the event collector code.Create Parsing Rules

The most common parsing rule is the _time system property, which indicates the event time from the remote system. For example, using the following events:

{   "id": "1234",
    "message": "New user added 'root2'",
    "type": "audit",
    "op": "add",
    "result": "success",
    "host_info": {
		"host": "prod-01",
		"os": "Windows"    },
    "created": "1676764803"  }

The created event property is a str representation of a timestamp (without milliseconds). However, the _time system property expects the result to be an str in format %Y-%m-%dT%H:%M:%S.000Z. Transform it using the timestamp_to_datestring function from CommonServerPython:

from datetime import datetime
from CommonServerPython import *
#  ...  
events: List[Dict[str, Any]] = get_events()  
for event in events:    
   event["_time"] = timestamp_to_datestring(float(event.get("created")) * 1000)
# ...

To ensure the parsing rule has been applied and is working as expected, run an XQL query to compare the _time and created fields:How to build XQL queries

dataset = "MyVendor_MyProduct_raw" | fields  _time,  created

After events are received by Cortex XSIAM, they are stored in a dataset in the structure of <vendor>_<product>_raw. If it's the first time fetching events, this dataset will be created. For more information about dataset management, see Dataset management.Dataset management

To see the events sent by the send_events_to_xsiam() function in your Cortex XSIAM instance: