RecordedFutureEventCollector

This integration fetches alerts from Recorded Future.

Analytics & SIEM · Recorded Future Intelligence

Details

ID	`RecordedFutureEventCollector`
Provider	Mastercard
Category	Analytics & SIEM
From Version	6.8.0
Docker Image	`demisto/python3:3.12.13.10116658`
Supported Modules	Agentix XSIAM

README

This integration fetches alerts from Recorded Future.
This integration was integrated and tested with version 2 of the Recorded Future API.

This is the default integration for this content pack when configured by the Data Onboarder in Cortex XSIAM.

Configure Recorded Future Event Collector in Cortex

Parameter	Description	Required
API token	The API token to use for the connection.	True
Trust any certificate (not secure)	Use SSL secure connection or not.	False
Use system proxy settings	Use proxy settings for connection or not.	False
First fetch time	First fetch query `<number> <time unit>`, e.g., `7 days`. Default `3 days`.	False
Max fetch	The maximum number of events per fetch. Default and maximum is 1000.	False

Commands

You can execute these commands from the CLI, 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.

recorded-future-get-events

Gets events from Recorded Future.

Base Command

recorded-future-get-events

Input

Argument Name	Description	Required
should_push_events	If true, the command will create events, otherwise it will only display them. Possible values are: true, false. Default is false.	Required
limit	Maximum results to return. Default is 10.	Optional

Context Output

There is no context output for this command.

Configuration parameters

credentials — (required)
insecure — Trust any certificate (not secure)
proxy — Use system proxy settings
first_fetch — First fetch time
max_fetch — The maximum number of events per fetch. Default and maximum is 1000.

Commands (1)

recorded-future-get-events
Gets events from Recorded Future.

Source

RecordedFutureEventCollector.py RecordedFutureEventCollector_test.py RecordedFutureEventCollector.yml RecordedFutureEventCollector_description.md command_examples.txt

Download History Raw

from CommonServerPython import *
from RecordedFutureEventCollector import BASE_URL, DATE_FORMAT
from freezegun import freeze_time
import requests_mock
import pytest

""" HELPER FUNCTIONS """


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


def mock_send_events_to_xsiam(events, vendor, product):
    return events, vendor, product


""" TEST FUNCTIONS """


def test_module_succeed(mocker):
    from RecordedFutureEventCollector import Client, test_module

    client = Client(BASE_URL)
    mock_results = mocker.patch("RecordedFutureEventCollector.return_results")

    with requests_mock.Mocker() as m:
        m.get(f"{BASE_URL}/config/info", status_code=200, text='{"name": "admin"}')
        test_module(client)

    assert mock_results.call_args[0][0] == "ok"


def test_module_failed():
    from RecordedFutureEventCollector import Client, test_module

    client = Client(BASE_URL)

    with requests_mock.Mocker() as m:
        m.get(f"{BASE_URL}/config/info", status_code=401, text='{"error":{"status":401}}')
        with pytest.raises(DemistoException) as e:
            test_module(client)

    assert e.value.message == 'Failed due to - Error in API call [401] - None\n{"error": {"status": 401}}'


def test_get_events():
    from RecordedFutureEventCollector import Client, get_events

    client = Client(BASE_URL)

    with requests_mock.Mocker() as m:
        m.get(f"{BASE_URL}/alert/search", json=util_load_json("test_data/first_fetch.json"))

        mock_events = get_events(client, {"limit": 2})

    assert len(mock_events) == 2


@freeze_time("2023-12-24T01:02:03.000Z")
def test_fetch_events(mocker):
    from RecordedFutureEventCollector import Client, fetch_events, demisto

    client = Client(BASE_URL)

    with requests_mock.Mocker() as m:
        first_mock_request = m.get(f"{BASE_URL}/alert/search", json=util_load_json("test_data/first_fetch.json"))

        mock_events, next_last_run = fetch_events(client, limit=2, last_run=arg_to_datetime("3 days").strftime(DATE_FORMAT))

    assert len(mock_events) == 2
    assert next_last_run == {"last_run_ids": ["a12b-5"], "last_run_time": "2023-12-23T00:00:10"}
    assert first_mock_request.last_request.qs.get("triggered")[0].upper() == "[2023-12-21T01:02:03.000000Z,]"

    mocker.patch.object(demisto, "getLastRun", return_value=next_last_run)
    last_run_time = next_last_run.get("last_run_time")

    with requests_mock.Mocker() as m:
        second_mock_request = m.get(f"{BASE_URL}/alert/search", json=util_load_json("test_data/second_fetch.json"))

        mock_events, next_last_run = fetch_events(client, limit=2, last_run=last_run_time)

    assert len(mock_events) == 1
    assert next_last_run == {"last_run_ids": ["a12b-8"], "last_run_time": "2023-12-23T00:10:20"}
    assert second_mock_request.last_request.qs.get("triggered")[0].upper() == "[2023-12-23T00:00:10,]"


def test_main(mocker):
    from RecordedFutureEventCollector import main, VENDOR, PRODUCT

    mocker.patch.object(demisto, "command", return_value="recorded-future-get-events")
    mocker.patch.object(demisto, "args", return_value={"should_push_events": True, "limit": 2})
    events = mocker.patch("RecordedFutureEventCollector.send_events_to_xsiam", side_effect=mock_send_events_to_xsiam)

    with requests_mock.Mocker() as m:
        mock_request = m.get(f"{BASE_URL}/alert/search", json=util_load_json("test_data/first_fetch.json"))

        main()

    assert len(events.call_args[0][0]) == 2
    event = events.call_args[0][0][0]
    assert event.get("_time") == event.get("log", {}).get("triggered")
    assert events.call_args[1].get("vendor") == VENDOR
    assert events.call_args[1].get("product") == PRODUCT
    assert mock_request.last_request.query == "limit=2"