TrendMicroVisionOneEventCollector

Palo Alto Networks TrendAI Vision One™ Event Collector integration for Cortex XSIAM collects the Workbench, Observed Attack Techniques, Search Detections and Audit logs.

Data Enrichment & Threat Intelligence · TrendAI Vision One™

Details

IDTrendMicroVisionOneEventCollector
ProviderTrendAI™
CategoryData Enrichment & Threat Intelligence
From Version6.10.0
Docker Imagedemisto/python3:3.12.13.10116658
Supported ModulesAgentix XSIAM

README

Palo Alto Networks TrendAI Vision One™ Event Collector integration for Cortex XSIAM collects the Workbench, Observed Attack Techniques, Search Detections and Audit logs.
TrendAI Vision One™ is a purpose-built threat defense platform that provides added value and new benefits beyond XDR solutions, allowing you to see more and respond faster. Providing deep and broad extended detection and response (XDR) capabilities that collect and automatically correlate data across multiple security layers—email, endpoints, servers, cloud workloads, and networks—TrendAI Vision One™ prevents the majority of attacks with automated protection.

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

Configure TrendAI Vision One™ Event Collector on Cortex XSIAM

  1. Navigate to Settings > Configurations > Data Collection > Automation & Feed Integrations.
  2. Search for TrendAI Vision One™ Event Collector.
  3. Click Add instance to create and configure a new integration instance.
Parameter Description Required
Your server URL The API endpoint to the TrendAI Vision One™ instance, see domains list: https://automation.trendmicro.com/xdr/Guides/First-Steps-Toward-Using-the-APIs True
TrendAI Vision One™ API Key The TrendAI Vision One™ API Key. Refer to the help section or to the information below on how to retrieve the API key. False
The maximum number of events per fetch The maximum number of events to fetch every time fetch is executed for each specified log type (Workbench, Observed Attack Techniques, Search Detections and Audit logs). False
Observed attack techniques and Search detections logs date-time range (hours) Defines the date-range (hours) in each API call that retrieves observed attack techniques and search detections logs. Used mainly to prevent timeouts by looping over large amount of logs. Default is 24 hours. False
First fetch timestamp (<number> <time unit>, e.g., 12 hours, 7 days) The first fetch time, since when to start fetching events False
Log types to fetch To maximize scalability, it is recommended to configure one type per instance. Default value is Workbench, Observed Attack Techniques, Search Detections, Audit. False
Fetch events   False
Trust any certificate (not secure)   False
Use system proxy settings   False
  1. Click Test to validate the URLs, token, and connection.

This integration fetches the following logs/alerts from TrendAI Vision One™ and requires the following permissions:

Log Type Action Role Permission Required API Documentation
Workbench Logs Workbench https://automation.trendmicro.com/xdr/api-v3#tag/Workbench
Observed Attack Techniques Logs Observed Attack Techniques https://automation.trendmicro.com/xdr/api-v3#tag/Observed-Attack-Techniques
Search Detection Logs Search https://automation.trendmicro.com/xdr/api-v3#tag/Search/paths/~1v3.0~1search~1detections/get
Audit Logs Audit Logs https://automation.trendmicro.com/xdr/api-v3#tag/Audit-Logs

You can then create a user account and generate an API key to be used for the Cortex XSIAM integration by following these steps in TrendAI Vision One™.

  1. Navigate to Administration > User Accounts.
  2. Click Add Account.
  3. Fill in the Add Account details assigning the role you created in the previous step and choosing APIs only as the access level.
  4. Complete the account creation process by following the steps in the email you receive.
  5. This will generate an Authentication token that can then be used to configure the Cortex XSIAM integration.

Built-in Roles:
TrendAI Vision One™ has built-in roles with fixed permissions that Master Administrators can assign to accounts.

The following table provides a brief description of each role.

Role Description
Master Administrator Can access all apps and TrendAI Vision One™ features.
Operator (formerly Administrator) Can configure system settings and connect products.
Auditor Has “View” access to specific TrendAI Vision One™ apps and features.
Senior Analyst Can investigate XDR alerts, take response actions, approve managed XDR requests, and manage detection models.
Analyst Can investigate XDR alerts and take response actions.

API Limitations

  • You cannot retrieve audit logs that are older than 180 days. Therefore, if setting a first fetch that is more than 180 days, for audit logs it will be a maximum of 180 days.
  • For API rate limits, refer here
  • Observed Attack Techniques Logs and Search Detection Logs are fetched from the newest to the oldest as its the logs are returned in descending order from the api.
  • For Observed Attack Techniques Logs and Search Detection Logs it is possible that the limit will be exceeded due to api limitations.

Commands

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

trend-micro-vision-one-get-events

Returns a list of logs.

Base Command

trend-micro-vision-one-get-events

Input

Argument Name Description Required
limit The maximum number of logs to retrieve. Default is 50. Optional
from_time From which time to retrieve the log(s) in ISO8601 format. For example yyyy-MM-ddThh:mm:ssZ in UTC. Required
to_time To which time to retrieve the log(s) in ISO8601 format. Defaults to the current time if not provided. For example yyyy-MM-ddThh:mm:ssZ in UTC. Optional
should_push_events Whether to push the fetched events to Cortex XSIAM or not. Possible values are: false, true. Default is false. Optional
log_type Comma-separated list of log-types to retrieve, options are audit, observed_attack_techniques, search_detections and workbench. Default is to retrieve all of them. Optional

Context Output

Path Type Description
TrendMicroVisionOne.Events Unknown TrendAI Vision One™ events.

Configuration parameters

  • url — Server URL (required)
  • credentials — (required)
  • max_fetch — The maximum number of events per fetch.
  • first_fetch — First fetch timestamp (<number> <time unit>, e.g., 12 hours, 7 days)
  • log_types — Log types to fetch
  • isFetchEvents — Fetch events
  • insecure — Trust any certificate (not secure)
  • proxy — Use system proxy settings

Commands (1)

  • trend-micro-vision-one-get-events

    Returns a list of logs.

Source

TrendMicroVisionOneEventCollector.py TrendMicroVisionOneEventCollector_test.py TrendMicroVisionOneEventCollector.yml TrendMicroVisionOneEventCollector_description.md
Download History Raw 
from urllib.parse import parse_qs, urlparse

import demistomock as demisto
import pytest
import pytz
from CommonServerPython import *  # noqa # pylint: disable=unused-wildcard-import
from freezegun import freeze_time
from TrendMicroVisionOneEventCollector import (
    DATE_FORMAT,
    DEFAULT_MAX_LIMIT,
    Client,
    LastRunLogsStartTimeFields,
    LogTypes,
    UrlSuffixes,
)

from CommonServerUserPython import *  # noqa

BASE_URL = "https://api.xdr.trendmicro.com"


@pytest.fixture()
def client() -> Client:
    return Client(base_url=BASE_URL, api_key="api-key", proxy=False, verify=True)


def get_url_params(url: str) -> Dict[str, str]:
    parsed_url = urlparse(url)
    query_parameters = parse_qs(parsed_url.query)
    return {key: value[0] for key, value in query_parameters.items()}


def create_any_type_logs(
    start: int,
    end: int,
    created_time_field: str,
    id_field_name: str,
    last_event_fetch_time: str | None,
    extra_seconds: int = 0,
    ascending_order: bool = False,
):
    """
    Create mocks of any type of log based on multiple parameters
    """
    last_event_fetch_time = last_event_fetch_time or (datetime.now() - timedelta(minutes=1)).strftime(DATE_FORMAT)
    if ascending_order:
        return [
            {
                id_field_name: i,
                created_time_field: (  # type: ignore
                    dateparser.parse(last_event_fetch_time) + timedelta(seconds=i + extra_seconds)  # type: ignore
                ).strftime(DATE_FORMAT)
                if created_time_field != "eventTime"
                else int(
                    (dateparser.parse(last_event_fetch_time) + timedelta(seconds=i + extra_seconds)).timestamp()  # type: ignore
                )
                * 1000,
            }
            for i in range(start + 1, end + 1)
        ]
    # descending order
    return [
        {
            id_field_name: end - i + 1,
            created_time_field: (datetime.now(tz=pytz.utc) - timedelta(seconds=i + start + extra_seconds)).strftime(DATE_FORMAT)
            if created_time_field != "eventTime"
            else int((datetime.now(tz=pytz.utc) - timedelta(seconds=i + start + extra_seconds)).timestamp()) * 1000,
        }
        for i in range(start + 1, end + 1)
    ]


def create_logs_mocks(
    url: str,
    num_of_events: int,
    created_time_field: str,
    id_field_name: str,
    last_event_fetch_time: str | None,
    url_suffix,
    top: int = 10,
    extra_seconds: int = 0,
) -> Dict:
    """
    Create mocks of any type of log, then returns the logs and pagination link if needed to proceed.
    """
    url_params = get_url_params(url)
    top = arg_to_number(url_params.get("top")) or top
    fetched_amount_of_events = arg_to_number(url_params.get("fetchedAmountOfEvents")) or 0

    if fetched_amount_of_events >= num_of_events:
        return {"items": []}

    logs = create_any_type_logs(
        start=fetched_amount_of_events,
        end=min(fetched_amount_of_events + top, num_of_events),
        created_time_field=created_time_field,
        id_field_name=id_field_name,
        last_event_fetch_time=last_event_fetch_time,
        extra_seconds=extra_seconds,
        ascending_order=url_suffix in [UrlSuffixes.AUDIT.value, UrlSuffixes.WORKBENCH.value],
    )
    fetched_amount_of_events += len(logs)

    response = {
        "items": logs,
    }

    if fetched_amount_of_events < num_of_events:
        response["nextLink"] = f"{BASE_URL}/v3.0{url_suffix}?top={top}&fetchedAmountOfEvents={fetched_amount_of_events}"

    return response


def _http_request_side_effect_decorator(
    last_workbench_time: str | None = None,
    last_oat_time: str | None = None,
    last_search_detection_logs: str | None = None,
    last_audit_log_time: str | None = None,
    num_of_workbench_logs: int = 0,
    num_of_oat_logs: int = 0,
    num_of_search_detection_logs: int = 0,
    num_of_audit_logs: int = 0,
):
    """
    general side effect function for creating logs from any type.
    """

    def _http_request_side_effect(**kwargs):
        full_url = kwargs.get("full_url") or ""
        params = kwargs.get("params") or {}
        if UrlSuffixes.WORKBENCH.value in full_url:
            return create_logs_mocks(
                url=full_url,
                num_of_events=num_of_workbench_logs,
                url_suffix=UrlSuffixes.WORKBENCH.value,
                created_time_field="createdDateTime",
                id_field_name="id",
                top=10,
                last_event_fetch_time=last_workbench_time,
            )
        if UrlSuffixes.OBSERVED_ATTACK_TECHNIQUES.value in full_url:
            return create_logs_mocks(
                url=full_url,
                num_of_events=num_of_oat_logs,
                url_suffix=UrlSuffixes.OBSERVED_ATTACK_TECHNIQUES.value,
                created_time_field="detectedDateTime",
                id_field_name="uuid",
                top=params.get("top") or 200,
                last_event_fetch_time=last_oat_time,
            )
        if UrlSuffixes.SEARCH_DETECTIONS.value in full_url:
            return create_logs_mocks(
                url=full_url,
                num_of_events=num_of_search_detection_logs,
                url_suffix=UrlSuffixes.SEARCH_DETECTIONS.value,
                created_time_field="eventTime",
                id_field_name="uuid",
                top=params.get("top") or DEFAULT_MAX_LIMIT,
                last_event_fetch_time=last_search_detection_logs,
            )
        else:
            return create_logs_mocks(
                url=full_url,
                num_of_events=num_of_audit_logs,
                url_suffix=UrlSuffixes.AUDIT.value,
                created_time_field="loggedDateTime",
                id_field_name="loggedUser",
                top=params.get("top") or 200,
                last_event_fetch_time=last_audit_log_time,
            )

    return _http_request_side_effect


def start_freeze_time(timestamp):
    _start_freeze_time = freeze_time(timestamp)
    _start_freeze_time.start()


class TestFetchEvents:
    def test_fetch_events_main_flow_no_new_logs(self, mocker):
        """
        Given:
           - no logs from any kind
           - last_run = {
                'workbench_start_time': '2023-01-01T14:00:00Z',
                'found_workbench_logs': [],
                'oat_detection_start_time': '2023-01-01T14:00:00Z',
                'dedup_found_oat_logs': [],
                'pagination_found_oat_logs': [],
                'oat_detection_next_link': '',
                'search_detection_start_time': '2023-01-01T14:00:00Z',
                'dedup_found_search_detection_logs': [],
                'pagination_found_search_detection_logs': [],
                'search_detection_next_link': '',
                'audit_start_time': '2023-01-01T14:00:00Z',
                'found_audit_logs': []
            }
           - max_fetch = 1000

        When:
           - running fetch-events through the main flow

        Then:
           - make sure no events are returned
           - make sure no last run time remains the same for every log.

        """
        from TrendMicroVisionOneEventCollector import main

        start_freeze_time("2023-01-01T15:00:00Z")

        mocker.patch.object(demisto, "params", return_value={"max_fetch": 1000})
        mocker.patch.object(demisto, "command", return_value="fetch-events")
        mocker.patch.object(
            demisto,
            "getLastRun",
            return_value={
                "workbench_start_time": "2023-01-01T14:00:00Z",
                "found_workbench_logs": [],
                "oat_detection_start_time": "2023-01-01T14:00:00Z",
                "dedup_found_oat_logs": [],
                "pagination_found_oat_logs": [],
                "oat_detection_next_link": "",
                "search_detection_start_time": "2023-01-01T14:00:00Z",
                "dedup_found_search_detection_logs": [],
                "pagination_found_search_detection_logs": [],
                "search_detection_next_link": "",
                "audit_start_time": "2023-01-01T14:00:00Z",
                "found_audit_logs": [],
            },
        )
        mocker.patch.object(
            BaseClient,
            "_http_request",
            side_effect=_http_request_side_effect_decorator(
                last_workbench_time="2023-01-01T14:00:00Z",
                last_audit_log_time="2023-01-01T14:00:00Z",
                last_oat_time="2023-01-01T14:00:00Z",
                last_search_detection_logs="2023-01-01T14:00:00Z",
            ),
        )

        set_last_run_mocker = mocker.patch.object(demisto, "setLastRun")
        send_events_to_xsiam_mocker = mocker.patch("TrendMicroVisionOneEventCollector.send_events_to_xsiam")
        main()

        assert set_last_run_mocker.call_args.args[0] == {
            "workbench_start_time": "2023-01-01T14:00:01Z",
            "found_workbench_logs": [],
            "oat_detection_start_time": "2023-01-01T14:00:01Z",
            "dedup_found_oat_logs": [],
            "pagination_found_oat_logs": [],
            "oat_detection_next_link": "",
            "search_detection_start_time": "2023-01-01T14:00:01Z",
            "dedup_found_search_detection_logs": [],
            "pagination_found_search_detection_logs": [],
            "search_detection_next_link": "",
            "audit_start_time": "2023-01-01T14:00:01Z",
            "found_audit_logs": [],
        }

        assert send_events_to_xsiam_mocker.call_count == 1
        # 1000 workbench + 1000 oat + 500 search detections + 500 audit logs
        assert len(send_events_to_xsiam_mocker.call_args.kwargs["events"]) == 0

    def test_fetch_events_main_flow_no_last_run(self, mocker):
        """
        Given:
           - 1000 workbench + 1000 oat + 500 search detections + 500 audit logs
           - no last run
           - max_fetch = 1000

        When:
           - running fetch-events through the main flow

        Then:
           - make sure last run is correct
           - make sure 3000 logs were sent

        """
        from TrendMicroVisionOneEventCollector import main

        start_freeze_time("2023-01-01T15:00:00Z")

        mocker.patch.object(demisto, "params", return_value={"max_fetch": 1000})
        mocker.patch.object(demisto, "command", return_value="fetch-events")
        mocker.patch.object(demisto, "getLastRun", return_value={})
        mocker.patch.object(
            BaseClient,
            "_http_request",
            side_effect=_http_request_side_effect_decorator(
                num_of_workbench_logs=1500,
                num_of_oat_logs=1500,
                num_of_search_detection_logs=500,
                num_of_audit_logs=500,
                last_workbench_time="2023-01-01T14:00:00Z",
                last_audit_log_time="2023-01-01T14:00:00Z",
            ),
        )

        set_last_run_mocker = mocker.patch.object(demisto, "setLastRun")
        send_events_to_xsiam_mocker = mocker.patch("TrendMicroVisionOneEventCollector.send_events_to_xsiam")
        main()

        assert set_last_run_mocker.call_args.args[0] == {
            "workbench_start_time": "2023-01-01T14:16:40Z",
            "found_workbench_logs": [1000],
            "oat_detection_start_time": "2023-01-01T14:59:59Z",
            "dedup_found_oat_logs": [1000],
            "pagination_found_oat_logs": [1000],
            "oat_detection_next_link": "https://api.xdr.trendmicro.com/v3.0/oat/detections?top=1000&fetchedAmountOfEvents=1000",
            "search_detection_start_time": "2023-01-01T14:59:59Z",
            "dedup_found_search_detection_logs": [500],
            "pagination_found_search_detection_logs": [],
            "search_detection_next_link": "",
            "audit_start_time": "2023-01-01T14:08:20Z",
            "found_audit_logs": ["77b363584231085e7909d48e0e103a07b6c10127e00da6e4739f07248eee7682"],
        }

        assert send_events_to_xsiam_mocker.call_count == 1
        # 1000 workbench + 1000 oat + 500 search detections + 500 audit logs
        assert len(send_events_to_xsiam_mocker.call_args.kwargs["events"]) == 3000

    def test_fetch_events_main_flow_timeout(self, mocker):
        """
        Given:
           - get_observed_attack_techniques_logs times out
           - last_run = {"oat_detection_start_time": "2023-01-01T14:59:59Z", "dedup_found_oat_logs": [1000]}
           - max_fetch = 1000 and log_types = "Observed Attack Techniques"

        When:
           - running fetch-events through the main flow

        Then:
           - Assert last run updated with half max_fetch and nextTrigger 30
           - Assert no events were sent to XSIAM
        """
        from TrendMicroVisionOneEventCollector import main

        mocker.patch.object(demisto, "params", return_value={"max_fetch": 1000, "log_types": ["Observed Attack Techniques"]})
        mocker.patch.object(demisto, "command", return_value="fetch-events")

        last_run = {"oat_detection_start_time": "2023-01-01T14:59:59Z", "dedup_found_oat_logs": [1000]}
        mocker.patch.object(demisto, "getLastRun", return_value=last_run)

        get_observed_attack_techniques_logs_mocker = mocker.patch(
            "TrendMicroVisionOneEventCollector.get_observed_attack_techniques_logs",
            side_effect=SignalTimeoutError,  # assume logic times out and timeout handler in ExecutionTimeout class throws error
        )

        set_last_run_mocker = mocker.patch.object(demisto, "setLastRun")
        send_events_to_xsiam_mocker = mocker.patch("TrendMicroVisionOneEventCollector.send_events_to_xsiam")

        main()

        assert get_observed_attack_techniques_logs_mocker.call_count == 1

        assert set_last_run_mocker.call_args.args[0] == {**last_run, "max_fetch": 500, "nextTrigger": "30"}

        assert send_events_to_xsiam_mocker.call_count == 1
        assert send_events_to_xsiam_mocker.call_args.kwargs["events"] == []

    def test_get_workbench_logs_no_last_run(self, mocker, client: Client):
        """
        Given:
         - no last run
         - limit = 500
         - 1000 workbench events

        When:
         - running get_workbench_logs function

        Then:
         - make sure only 500 events are returned
         - make sure latest workbench event is saved in the workbench_logs_time without adding 1 second to it.
         - make sure in the cache we will have event with 500 id as its the event that happened in the last second.
        """
        from TrendMicroVisionOneEventCollector import get_workbench_logs

        start_freeze_time("2023-01-01T15:00:00Z")

        mocker.patch.object(
            BaseClient, "_http_request", side_effect=_http_request_side_effect_decorator(num_of_workbench_logs=1000)
        )

        workbench_logs, updated_last_run = get_workbench_logs(client=client, first_fetch="1 month ago", last_run={}, limit=500)

        assert len(workbench_logs) == 500
        assert updated_last_run == {
            LastRunLogsStartTimeFields.WORKBENCH.value: "2023-01-01T15:07:20Z",
            "found_workbench_logs": [500],
        }
        assert workbench_logs[-1]["_time"] == "2023-01-01T15:07:20Z"

    def test_get_workbench_logs_with_last_run(self, mocker, client: Client):
        """
        Given:
         - last_run={'workbench_logs_time': '2023-01-01T14:00:00Z', 'found_workbench_logs': [1, 2, 3, 4, 5, 6, 7, 8]}
         - limit = 500
         - 200 workbench events

        When:
         - running get_workbench_logs function

        Then:
         - make sure only 192 events are returned as there are 8 events in cache from last run
         - make sure latest workbench event is saved in the workbench_logs_time without adding 1 second to it.
        """
        from TrendMicroVisionOneEventCollector import get_workbench_logs

        start_freeze_time("2023-01-01T15:00:00Z")

        mocker.patch.object(
            BaseClient,
            "_http_request",
            side_effect=_http_request_side_effect_decorator(
                num_of_workbench_logs=200, last_workbench_time="2023-01-01T14:00:00Z"
            ),
        )

        workbench_logs, updated_last_run = get_workbench_logs(
            client=client,
            first_fetch="1 month ago",
            last_run={
                LastRunLogsStartTimeFields.WORKBENCH.value: "2023-01-01T14:00:00Z",
                "found_workbench_logs": [1, 2, 3, 4, 5, 6, 7, 8],
            },
            limit=500,
        )

        assert len(workbench_logs) == 192
        assert updated_last_run == {
            LastRunLogsStartTimeFields.WORKBENCH.value: "2023-01-01T14:03:20Z",
            "found_workbench_logs": [200],
        }
        assert workbench_logs[-1]["_time"] == "2023-01-01T14:03:20Z"

    def test_get_observed_attack_techniques_logs_no_last_run(self, mocker, client: Client):
        """
        Given:
         - no last run
         - limit = 500
         - 1000 oat events

        When:
         - running get_observed_attack_techniques_logs function

        Then:
         - make sure only 500 events are returned
         - make sure latest observed attack technique event is saved in the oat_detection_logs_time without adding 1 second to it.
        """
        from TrendMicroVisionOneEventCollector import get_observed_attack_techniques_logs

        start_freeze_time("2023-01-01T15:00:00Z")

        mocker.patch.object(BaseClient, "_http_request", side_effect=_http_request_side_effect_decorator(num_of_oat_logs=1000))

        observed_attack_techniques_logs, updated_last_run = get_observed_attack_techniques_logs(
            client=client, first_fetch="1 month ago", last_run={}, limit=500
        )

        assert len(observed_attack_techniques_logs) == 500
        assert updated_last_run == {
            "oat_detection_start_time": "2023-01-01T14:59:59Z",
            "dedup_found_oat_logs": [1000],
            "pagination_found_oat_logs": [],
            "oat_detection_next_link": "",
        }

    def test_get_observed_attack_techniques_logs_with_last_run(self, mocker, client: Client):
        """
        Given:
         - last_run={
                'oat_detection_start_time': '2023-01-01T14:00:00Z',
                'dedup_found_oat_logs': [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15],
                'pagination_found_oat_logs': [],
                'oat_detection_next_link': f'{BASE_URL}/v3.0{UrlSuffixes.OBSERVED_ATTACK_TECHNIQUES.value}'
                                           f'?top=1000&fetchedAmountOfEvents=100'
            }
         - limit = 500
         - 300 oat events

        When:
         - running get_observed_attack_techniques_logs function

        Then:
         - make sure only 185 events are returned
         - make sure latest observed attack technique event is saved in the oat_detection_logs_time without adding 1 second to it.
        """
        from TrendMicroVisionOneEventCollector import get_observed_attack_techniques_logs

        start_freeze_time("2023-01-01T15:00:00Z")

        mocker.patch.object(
            BaseClient,
            "_http_request",
            side_effect=_http_request_side_effect_decorator(num_of_oat_logs=300, last_oat_time="2023-01-01T14:00:00Z"),
        )

        observed_attack_techniques_logs, updated_last_run = get_observed_attack_techniques_logs(
            client=client,
            first_fetch="1 month ago",
            last_run={
                "oat_detection_start_time": "2023-01-01T14:00:00Z",
                "dedup_found_oat_logs": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15],
                "pagination_found_oat_logs": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15],
                "oat_detection_next_link": f"{BASE_URL}/v3.0{UrlSuffixes.OBSERVED_ATTACK_TECHNIQUES.value}"
                f"?top=1000&fetchedAmountOfEvents=100",
            },
            limit=500,
        )

        assert len(observed_attack_techniques_logs) == 185
        assert updated_last_run == {
            "oat_detection_start_time": "2023-01-01T14:00:00Z",
            "dedup_found_oat_logs": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15],
            "pagination_found_oat_logs": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15],
            "oat_detection_next_link": "",
        }

    def test_get_search_detection_logs_no_last_run(self, mocker, client: Client):
        """
        Given:
         - no last run
         - limit = 500
         - 1000 search detection events

        When:
         - running get_search_detection_logs function

        Then:
         - make sure only 500 events are returned
         - make sure latest observed attack technique event is saved in
           the search_detection_logs_time without adding 1 second to it.
        """
        from TrendMicroVisionOneEventCollector import get_search_detection_logs

        start_freeze_time("2023-01-01T15:00:00Z")

        mocker.patch.object(
            BaseClient, "_http_request", side_effect=_http_request_side_effect_decorator(num_of_search_detection_logs=1000)
        )

        search_detection_logs, updated_last_run = get_search_detection_logs(
            client=client, first_fetch="1 month ago", last_run={}, limit=500
        )

        assert len(search_detection_logs) == 500
        assert updated_last_run == {
            "search_detection_start_time": "2023-01-01T14:59:59Z",
            "dedup_found_search_detection_logs": [500],
            "pagination_found_search_detection_logs": [500],
            "search_detection_next_link": "https://api.xdr.trendmicro.com/v3.0/search/detections?top=500"
            "&fetchedAmountOfEvents=500",
        }

    def test_get_search_detection_logs_with_last_run(self, mocker, client: Client):
        """
        Given:
         - last_run = {
                'search_detection_start_time': '2023-01-01T14:00:00Z',
                'dedup_found_search_detection_logs': [1, 2, 3, 4, 5, 6, 7],
                'pagination_found_search_detection_logs': [],
                'search_detection_next_link': 'https://api.xdr.trendmicro.com/v3.0/search/detections?top=200'
                                              '&fetchedAmountOfEvents=500'
            }
         - limit = 500
         - 200 detection events

        When:
         - running get_search_detection_logs function

        Then:
         - make sure only 193 events are returned
         - make sure latest search detection log event time is saved in
           the search_detection_logs_time without adding 1 second to it.
        """
        from TrendMicroVisionOneEventCollector import get_search_detection_logs

        start_freeze_time("2023-01-01T15:00:00Z")

        mocker.patch.object(
            BaseClient,
            "_http_request",
            side_effect=_http_request_side_effect_decorator(
                num_of_search_detection_logs=700, last_search_detection_logs="2023-01-01T14:00:00Z"
            ),
        )

        search_detection_logs, updated_last_run = get_search_detection_logs(
            client=client,
            first_fetch="1 month ago",
            last_run={
                "search_detection_start_time": "2023-01-01T14:00:00Z",
                "dedup_found_search_detection_logs": [1, 2, 3, 4, 5, 6, 7],
                "pagination_found_search_detection_logs": [1, 2, 3, 4, 5, 6, 7],
                "search_detection_next_link": "https://api.xdr.trendmicro.com/v3.0/search/detections?top=200"
                "&fetchedAmountOfEvents=500",
            },
            limit=500,
        )

        assert len(search_detection_logs) == 193
        assert updated_last_run == {
            "search_detection_start_time": "2023-01-01T14:00:00Z",
            "dedup_found_search_detection_logs": [1, 2, 3, 4, 5, 6, 7],
            "pagination_found_search_detection_logs": [1, 2, 3, 4, 5, 6, 7],
            "search_detection_next_link": "",
        }

    def test_get_audit_logs_no_last_run(self, mocker, client: Client):
        """
        Given:
         - no last run
         - limit = 500
         - 200 audit logs

        When:
         - running get_audit_logs function

        Then:
         - make sure only 500 events are returned
         - make sure in the cache we will have event with hash of the 500 id as its the event that happened in the last second.
        """
        from TrendMicroVisionOneEventCollector import get_audit_logs

        start_freeze_time("2023-01-01T15:00:00Z")

        mocker.patch.object(
            BaseClient,
            "_http_request",
            side_effect=_http_request_side_effect_decorator(num_of_audit_logs=1000, last_audit_log_time="2023-01-01T14:00:00Z"),
        )

        audit_logs, updated_last_run = get_audit_logs(client=client, first_fetch="1 month ago", last_run={}, limit=500)

        assert len(audit_logs) == 500
        assert updated_last_run == {
            LastRunLogsStartTimeFields.AUDIT.value: "2023-01-01T14:08:20Z",
            "found_audit_logs": ["77b363584231085e7909d48e0e103a07b6c10127e00da6e4739f07248eee7682"],
        }

    def test_get_audit_logs_with_last_run(self, mocker, client: Client):
        """
        Given:
         - last_run={'audit_logs_time': '2023-01-01T14:00:00Z'}
         - limit = 500
         - 200 audit logs

        When:
         - running get_audit_logs function

        Then:
         - make sure only 200 events are returned
         - make sure in the cache we will have event with hash of the 200 id as its the event that happened in the last second.
        """
        from TrendMicroVisionOneEventCollector import get_audit_logs

        start_freeze_time("2023-01-01T15:00:00Z")

        mocker.patch.object(
            BaseClient,
            "_http_request",
            side_effect=_http_request_side_effect_decorator(num_of_audit_logs=200, last_audit_log_time="2023-01-01T14:00:00Z"),
        )

        audit_logs, updated_last_run = get_audit_logs(
            client=client,
            first_fetch="1 month ago",
            last_run={LastRunLogsStartTimeFields.AUDIT.value: "2023-01-01T14:00:00Z"},
            limit=500,
        )

        assert len(audit_logs) == 200
        assert updated_last_run == {
            LastRunLogsStartTimeFields.AUDIT.value: "2023-01-01T14:03:20Z",
            "found_audit_logs": ["4410bac4975e15bc234ee627129e46665744349bb59830968a2e4769fe0afc0e"],
        }


@pytest.mark.parametrize(
    "last_run_time, first_fetch, log_type_time_field_name, start_time, expected_start_and_end_date_times",
    [
        (
            None,
            "3 years",
            LastRunLogsStartTimeFields.WORKBENCH.value,
            "2023-01-01T15:20:45Z",
            ("2020-01-01T15:20:45Z", "2023-01-01T15:20:45Z"),
        ),
        (
            None,
            "3 years",
            LastRunLogsStartTimeFields.AUDIT.value,
            "2023-01-01T15:20:45Z",
            ("2022-07-05T15:20:45Z", "2023-01-01T15:20:45Z"),
        ),
        (
            None,
            "3 years ago",
            LastRunLogsStartTimeFields.OBSERVED_ATTACK_TECHNIQUES.value,
            "2023-01-01T15:20:45Z",
            ("2020-01-01T15:20:45Z", "2020-12-31T15:20:45Z"),
        ),
        (
            None,
            "1 month ago",
            LastRunLogsStartTimeFields.OBSERVED_ATTACK_TECHNIQUES.value,
            "2023-01-01T15:20:45Z",
            ("2022-12-01T15:20:45Z", "2023-01-01T15:20:45Z"),
        ),
        (
            "2023-01-01T15:00:00Z",
            "1 month ago",
            LastRunLogsStartTimeFields.SEARCH_DETECTIONS.value,
            "2023-01-01T15:20:45Z",
            ("2023-01-01T15:00:00Z", "2023-01-01T15:20:45Z"),
        ),
    ],
)
def test_get_datetime_range(
    last_run_time: str | None,
    first_fetch: str,
    log_type_time_field_name: str,
    start_time: str,
    expected_start_and_end_date_times: tuple[str, str],
):
    """
    Given:
        - Case A: last_run_time=None, first_fetch=3 years ago, log_type_time_field_name=workbench_logs_time,
                  start_time=2023-01-01T15:20:45Z
        - Case B: last_run_time=None, first_fetch=3 years ago, log_type_time_field_name=audit_logs_time,
                  start_time=2023-01-01T15:20:45Z
        - Case C: last_run_time=None, first_fetch=3 years ago, log_type_time_field_name=oat_detection_logs_time,
                  start_time=2023-01-01T15:20:45Z
        - Case D: last_run_time=None, first_fetch=1 month ago, log_type_time_field_name=oat_detection_logs_time,
                  start_time=2023-01-01T15:20:45Z
        - Case E: last_run_time=2023-01-01T15:00:00Z, first_fetch=1 month ago,
                  log_type_time_field_name=search_detection_logs_time, start_time=2023-01-01T15:20:45Z
    When:
        - running get datetime range
    Then:
        - Case A: make sure start time is 3 years ago and end time is "now".
        - Case B: make sure the start time is 180 days ago and end time is "now"
        - Case C: make sure the start time is 3 years ago and the time is 1 day after.
        - Case D: make sure the start time is 1 month ago and end time is "now"
        - Case E: make sure the start time is the last_run_time and end time is "now"
    """
    from TrendMicroVisionOneEventCollector import get_datetime_range

    start_freeze_time(start_time)

    assert (
        get_datetime_range(
            last_run_time=last_run_time, first_fetch=first_fetch, log_type_time_field_name=log_type_time_field_name
        )
        == expected_start_and_end_date_times
    )


def test_module_main_flow(mocker):
    """
    Given:
        - 1 log of each type
    When:
        - test-module through main
    Then:
        - make sure that test-module returns 'ok'
        - make sure send_events_to_xsiam function was not called
    """
    from TrendMicroVisionOneEventCollector import main

    start_freeze_time("2023-01-01T15:20:45Z")

    mocker.patch.object(demisto, "params", return_value={"first_fetch": "1 year ago"})
    mocker.patch.object(demisto, "command", return_value="test-module")
    mocker.patch.object(
        BaseClient,
        "_http_request",
        side_effect=_http_request_side_effect_decorator(
            num_of_workbench_logs=1, num_of_oat_logs=1, num_of_search_detection_logs=1, num_of_audit_logs=1
        ),
    )

    return_results_mocker = mocker.patch("TrendMicroVisionOneEventCollector.return_results")
    send_events_to_xsiam_mocker = mocker.patch("TrendMicroVisionOneEventCollector.send_events_to_xsiam")

    main()

    assert not send_events_to_xsiam_mocker.called
    assert return_results_mocker.call_args.args[0] == "ok"


@pytest.mark.parametrize(
    "args, expected_outputs",
    [
        (
            {
                "from_time": "2023-01-01T15:00:45Z",
                "to_time": "2023-01-01T15:20:45Z",
                "log_type": f"{LogTypes.AUDIT.value},{LogTypes.SEARCH_DETECTIONS.value},"
                f"{LogTypes.OBSERVED_ATTACK_TECHNIQUES.value},{LogTypes.WORKBENCH.value}",
            },
            [
                {"Id": 1, "Time": "2023-01-01T15:19:46Z", "Type": "Workbench"},
                {"Id": 1, "Time": "2023-01-01T15:20:44Z", "Type": "Observed Attack Technique"},
                {"Id": 1, "Time": "2023-01-01T15:20:44Z", "Type": "Search Detection"},
                {"Id": 1, "Time": "2023-01-01T15:19:46Z", "Type": "Audit"},
            ],
        ),
        (
            {"from_time": "2023-01-01T15:00:45Z", "to_time": "2023-01-01T15:20:45Z", "log_type": LogTypes.AUDIT.value},
            [{"Id": 1, "Time": "2023-01-01T15:19:46Z", "Type": "Audit"}],
        ),
        (
            {
                "from_time": "2023-01-01T15:00:45Z",
                "to_time": "2023-01-01T15:20:45Z",
                "log_type": LogTypes.OBSERVED_ATTACK_TECHNIQUES.value,
            },
            [{"Id": 1, "Time": "2023-01-01T15:20:44Z", "Type": "Observed Attack Technique"}],
        ),
        (
            {
                "from_time": "2023-01-01T15:00:45Z",
                "to_time": "2023-01-01T15:20:45Z",
                "log_type": LogTypes.SEARCH_DETECTIONS.value,
            },
            [{"Id": 1, "Time": "2023-01-01T15:20:44Z", "Type": "Search Detection"}],
        ),
        (
            {"from_time": "2023-01-01T15:00:45Z", "to_time": "2023-01-01T15:20:45Z", "log_type": LogTypes.WORKBENCH.value},
            [{"Id": 1, "Time": "2023-01-01T15:19:46Z", "Type": "Workbench"}],
        ),
    ],
)
def test_get_events_command_main_flow(mocker, args: Dict, expected_outputs: List[Dict]):
    """
    Given:
        - Case A: log_type=all
        - Case B: log_type=audit_logs
        - Case C: log_type=oat_detection_logs
        - Case D: log_type=search_detection_logs
        - Case E: log_type=workbench_logs
    When:
        - running trend-micro-vision-one-get-events through main
    Then:
        - make sure when log_type=all, all events are returned.
        - make sure for each log type only the correct log will be returned.
    """
    from TrendMicroVisionOneEventCollector import main

    start_freeze_time("2023-01-01T15:20:45Z")

    mocker.patch.object(demisto, "params", return_value={"first_fetch": "1 year ago"})
    mocker.patch.object(demisto, "args", return_value=args)
    mocker.patch.object(demisto, "command", return_value="trend-micro-vision-one-get-events")
    mocker.patch.object(
        BaseClient,
        "_http_request",
        side_effect=_http_request_side_effect_decorator(
            num_of_workbench_logs=1, num_of_oat_logs=1, num_of_search_detection_logs=1, num_of_audit_logs=1
        ),
    )

    return_results_mocker = mocker.patch("TrendMicroVisionOneEventCollector.return_results")

    main()

    assert return_results_mocker.call_args.args[0].outputs == expected_outputs
    assert "events for log_types=" in return_results_mocker.call_args.args[0].readable_output