CyberArk Identity Event Collector

This integration collects events from the Idaptive Next-Gen Access (INGA) using REST APIs.

Analytics & SIEM · CyberArk Identity

Details

IDCyberArk Identity Event Collector
ProviderCyberArk
CategoryAnalytics & SIEM
From Version6.8.0
Docker Imagedemisto/py3-tools:1.0.0.10120494
Supported ModulesAgentix XSIAM

README

CyberArk Identity log event collector integration for Cortex XSIAM.
This integration was integrated and tested with version 22.4 of CyberArk Identity Event Collector.

Configure CyberArk Identity Event Collector in Cortex

Parameter Description Required
Server URL The CyberArk Identity URL to get the logs from. For example, https://{{tenant}}.my.idaptive.app. True
App ID The application ID to fetch the logs from. True
User name The user that was created in CyberArk for the XSIAM integration. For example, admin@example.com. True
Password The password for the user that was created in CyberArk for the XSIAM integration. True
First fetch time The period to retrieve events for.
Format: <number> <time unit>, for example 12 hours, 1 day, 3 months.
Default is 3 days.		 True
Maximum number of events per fetch The maximum number of items to retrieve per request from CyberArk’s API. True
Trust any certificate (not secure) When selected, certificates are not checked. False
Use system proxy settings Runs the integration instance using the proxy server (HTTP or HTTPS) that you defined in the server configuration. False

Commands

You can execute these commands from the Cortex XSIAM Alerts War Room as part of an automation, or in a playbook.
After you successfully execute a command, a DBot message appears in the War Room with the command details.

cyberarkidentity-get-events

Returns a list of events

Base Command

cyberarkidentity-get-events

Input

Argument Name Description Required
should_push_events Set this argument to True to create events, otherwise events will only be displayed. Default is False. Required
limit The maximum number of events per fetch. Default is 1000. Optional
from The first fetch time (<number> <time unit>, for example 12 hours, 1 day, 3 months). Default is 3 days. Optional

Context Output

There is no context output for this command.

Command example

!cyberarkidentity-get-events should_push_events=false limit=10 from="3 days"

Human Readable Output

CyberArkIdentity RedRock records

Auth Method Directory Service Uuid From IP Address ID Level Normalized User Request Device OS Request Host Name Request Is Mobile Device Tenant User Guid When Logged When Occurred _ Table Name
None 123456abcdef.123456.abcdef 1.1.1.1 123456abcdef.123456.abcdef Info admin@example.com.11 Unknown 1.1.1.1 false AAM4730 123456abcdef.123456.abcdef /Date(1652376432605)/ /Date(1652376432605)/ events
None 123456abcdef.123456.abcdef 1.1.1.1 123456abcdef.123456.abcdeg Info admin@example.com.11 Unknown 1.1.1.1 false AAM4730 123456abcdef.123456.abcdef /Date(1652376492682)/ /Date(1652376492682)/ events
None 123456abcdef.123456.abcdef 1.1.1.1 123456abcdef.123456.abcdeh Info admin@example.com.11 Unknown 1.1.1.1 false AAM4730 123456abcdef.123456.abcdef /Date(1652376552546)/ /Date(1652376552546)/ events

Configuration parameters

  • url — Server URL (required)
  • app_id — App ID (required)
  • credentials — User name (required)
  • from — First fetch time (required)
  • limit — Maximum number of events per fetch (required)
  • insecure — Trust any certificate (not secure)
  • proxy — Use system proxy settings

Commands (1)

  • cyberarkidentity-get-events

    Returns a list of events

Source

CyberArkIdentityEventCollector.py CyberArkIdentityEventCollector_test.py CyberArkIdentityEventCollector.yml CyberArkIdentityEventCollector_description.md command_examples.txt
Download History Raw 
import json

import demistomock as demisto
import requests_mock

URL = "https://example.my.idaptive.app/"
DEMISTO_PARAMS = {
    "url": URL,
    "credentials": {
        "identifier": "admin@example.com.11",
        "password": "123456",
    },
    "from": "3 days",
    "app_id": "test_app",
    "limit": 100,
}


def util_load_json(path):
    with open(path, encoding="utf-8") as f:
        return json.loads(f.read())


def mock_set_last_run(last_run):
    return last_run


# @freeze_time('2022-05-12T00:00:00Z')
def test_fetch_events_few_events(mocker):
    """
    Given
        - 3 events was created in CyberArk side in the last 3 days.
    When
        - fetch-events is running (with limit set to 100).
    Then
        - Verify that all 3 events were created in XSIAM.
        - Verify last_run was set as expected.
    """

    params = mocker.patch.object(demisto, "params", return_value=DEMISTO_PARAMS)
    args = mocker.patch.object(demisto, "args", return_value={"should_push_events": True})
    mock_last_run = mocker.patch.object(demisto, "setLastRun", side_effect=mock_set_last_run)
    results = mocker.patch.object(demisto, "results")
    mocker.patch("CyberArkIdentityEventCollector.send_events_to_xsiam")

    with requests_mock.Mocker() as m:
        m.post(f"{URL}oauth2/platformtoken", json={"access_token": "123456abc"})
        m.post(f"{URL}RedRock/Query", json=util_load_json("test_data/events.json"))

        from CyberArkIdentityEventCollector import main

        main("cyberarkidentity-get-events", params.return_value | args.return_value)

    events = results.call_args[0][0]["Contents"]
    last_run = mock_last_run.call_args[0][0]
    assert last_run.get("from") == "2022-05-15T13:35:26.645000"
    assert len(last_run.get("ids")) == len(events) == 3


def test_fetch_events_no_events(mocker):
    """
    Given
        - 3 events was created in CyberArk side in the last 3 days.
    When
        - fetch-events is running (with limit set to 100).
    Then
        - Make sure no events was created in XSIAM.
        - Make sure last_run was set as expected.
    """

    params = mocker.patch.object(demisto, "params", return_value=DEMISTO_PARAMS)
    args = mocker.patch.object(demisto, "args", return_value={"should_push_events": True})
    mock_last_run = mocker.patch.object(demisto, "setLastRun", side_effect=mock_set_last_run)
    results = mocker.patch.object(demisto, "results")
    mocker.patch("CyberArkIdentityEventCollector.send_events_to_xsiam")

    with requests_mock.Mocker() as m:
        m.post(f"{URL}oauth2/platformtoken", json={"access_token": "123456abc"})
        m.post(f"{URL}RedRock/Query", json={"Result": {}})

        from CyberArkIdentityEventCollector import main

        main("cyberarkidentity-get-events", params.return_value | args.return_value)

    events = results.call_args[0][0]["Contents"]
    last_run = mock_last_run.call_args
    assert not last_run
    assert not events


def test_fetch_events_limit_set_to_one(mocker):
    """
    Given
        - 3 events was created in CyberArk side in the last 3 days.
    When
        - fetch-events is running (with limit set to 1).
    Then
        - Verify that only 1 event were created in XSIAM.
        - Verify last_run was set as expected.
    """

    demisto_params = DEMISTO_PARAMS
    demisto_params["limit"] = 1
    params = mocker.patch.object(demisto, "params", return_value=demisto_params)
    args = mocker.patch.object(demisto, "args", return_value={"should_push_events": True})
    mock_last_run = mocker.patch.object(demisto, "setLastRun", side_effect=mock_set_last_run)
    results = mocker.patch.object(demisto, "results")
    mocker.patch("CyberArkIdentityEventCollector.send_events_to_xsiam")

    with requests_mock.Mocker() as m:
        m.post(f"{URL}oauth2/platformtoken", json={"access_token": "123456abc"})
        m.post(f"{URL}RedRock/Query", json=util_load_json("test_data/events.json"))

        from CyberArkIdentityEventCollector import main

        main("cyberarkidentity-get-events", params.return_value | args.return_value)

    events = results.call_args[0][0]["Contents"]
    last_run = mock_last_run.call_args[0][0]
    assert last_run.get("from") == "2022-05-15T13:35:03.570000"
    assert len(last_run.get("ids")) == len(events) == 1