iManageThreatManager

iManage Threat Manager protects privileged information against internal and external threat actors using machine learning and user behavior analytics.

Analytics & SIEM · iManage Threat Manager

Details

IDiManageThreatManager
ProvideriManage
CategoryAnalytics & SIEM
From Version8.4.0
Docker Imagedemisto/fastapi:0.125.0.10158186
Supported ModulesXSIAM

README

This integration fetches alerts from iManage Threat Manager, an AI-driven security solution that uses machine learning to detect unusual user behavior, prevent data loss, and ensure compliance.

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

Configure iManage Threat Manager in Cortex

Parameter Description Required
Server URL Should be in format https://<your-instance>.tm-cloudimanage.com True
User Name Username for user sign-in authentication. Required for Addressable Alerts and Detect And Protect Alerts. Conditional
Password Password for user sign-in authentication. Required for Addressable Alerts and Detect And Protect Alerts. Conditional
Token Application token for API token authentication. Required for Behavior Analytics alerts. Conditional
Secret Application secret for API token authentication. Required for Behavior Analytics alerts. Conditional
Fetch events Whether to fetch events. False
Events types to fetch Select which event types to fetch: Behavior Analytics alerts, Addressable Alerts, Detect And Protect Alerts. Default is Behavior Analytics alerts. False
Maximum number of events per type Default and maximum is 900 events to fetch for each type. False
Trust any certificate (not secure) Use SSL secure connection or not. False
Use system proxy settings Use proxy settings for connection or not. False

iManage Threat Manager Authentication

The integration supports two authentication methods:

Application Token Authentication (for Behavior Analytics alerts)

To generate an application token and secret from the Threat Manager admin console:

  1. In iManage Threat Manager, browse to Configuration > System > Application Tokens for Utility Access.
  2. Select New Token. The New Token dialog opens.
  3. In the Token Name field, enter a unique name for this application token.
  4. Select the Export Alert List permission.
  5. In Token Expiry Time in minutes, enter the number of minutes before this token becomes invalid.
    • By default, application tokens expire after 1400 minutes (1 day). The maximum value is 525600 (365 days).
  6. Select Generate Token.
  7. The New Token dialog shows the generated application token and secret.

Note: The Integrations Manager role is required to generate an application token. If a user with the Integration Manager role is made inactive or the role is removed, all existing application tokens created by that user become inactive.

User Sign-in Authentication (for Addressable Alerts and Detect And Protect Alerts)

Use your iManage Threat Manager username and password. This provides a similar level of access to what the user would have in the admin console.

Important: These alert types cannot be accessed through application token authentication and require user credentials.

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.

imanage-threat-manager-get-events

Fetches events from iManage Threat Manager. This manual command is used for developing or debugging and should be used with caution, as it can create events, leading to event duplication and exceeding API request limits.

Base Command

imanage-threat-manager-get-events

Input

Argument Name Description Required
should_push_events If true, the command creates events; otherwise, it only displays them. Possible values are true and false. The default value is false. Required
event_type Type of events to fetch. Possible values are: Behavior Analytics alerts, Addressable Alerts, Detect And Protect Alerts. Default is Behavior Analytics alerts. Optional
limit Maximum number of results to return. Default is 50. Optional
from_date Start date from which to get events. Supports ISO format or natural language (e.g., “7 days ago”, “1 hour ago”). Default is 1 hour ago. Optional
to_date End date until which to get events. Supports ISO format or natural language (e.g., “now”, “30 minutes ago”). Default is now. Optional

Context Output

There is no context output for this command.

Additional Information

  • Timezone: All timestamps are in UTC

Configuration parameters

  • url — Server URL (required)
  • credentials_user — User Name
  • credentials_token — Token
  • insecure — Trust any certificate (not secure)
  • proxy — Use system proxy settings
  • isFetchEvents — Fetch events
  • event_types — Events types to fetch
  • max_events_per_type — Maximum number of events per type

Commands (1)

  • imanage-threat-manager-get-events

    Fetches events from iManage Threat Manager.

Source

iManageThreatManager.py iManageThreatManager_test.py iManageThreatManager.yml iManageThreatManager_description.md
Download History Raw 
import base64
import json

import pytest
from freezegun import freeze_time
from iManageThreatManager import (
    Client,
    get_events_command,
    fetch_events_command,
    validate_credentials_for_event_types,
    _add_fields_to_events,
    _deduplicate_events,
    _fetch_events_with_pagination,
    _update_next_run_state,
    BEHAVIOR_ANALYTICS,
    ADDRESSABLE_ALERTS,
    DETECT_AND_PROTECT_ALERTS,
    EVENT_TYPE_CONFIG,
)
from CommonServerPython import DemistoException


BASE_URL = "https://test-instance.tm-cloudimanage.com"


@pytest.fixture(autouse=True)
def mock_content_client_init(mocker):
    """Mock ContentClient.__init__ to avoid httpx/anyio initialization in tests."""
    mocker.patch("iManageThreatManager.ContentClient.__init__", return_value=None)


@pytest.fixture
def client():
    """
    Given:
        - Base URL and authentication credentials
    When:
        - Creating a Client instance
    Then:
        - Ensure the client is properly initialized
    """
    c = Client(
        base_url=BASE_URL,
        verify=True,
        proxy=False,
        username="test_user",
        password="test_password",
        token="test_token",
        secret="test_secret",
    )
    # Set attributes that ContentClient.__init__ would normally set
    c._base_url = BASE_URL
    c._verify = True
    return c


class TestValidateCredentialsForEventTypes:
    """Tests for validate_credentials_for_event_types function"""

    def test_validate_credentials_behavior_analytics_success(self, client):
        """
        Given:
            - Client with valid token and secret
            - Behavior Analytics event type
        When:
            - Calling validate_credentials_for_event_types
        Then:
            - Ensure no exception is raised
        """
        validate_credentials_for_event_types(client, [BEHAVIOR_ANALYTICS])

    def test_validate_credentials_addressable_alerts_success(self, client):
        """
        Given:
            - Client with valid username and password
            - Addressable Alerts event type
        When:
            - Calling validate_credentials_for_event_types
        Then:
            - Ensure no exception is raised
        """
        validate_credentials_for_event_types(client, [ADDRESSABLE_ALERTS])

    def test_validate_credentials_missing_token_secret(self):
        """
        Given:
            - Client without token and secret
            - Behavior Analytics event type
        When:
            - Calling validate_credentials_for_event_types
        Then:
            - Ensure DemistoException is raised
        """
        client_no_token = Client(base_url=BASE_URL, verify=True, proxy=False, username="user", password="pass")
        with pytest.raises(DemistoException, match="Token and Secret"):
            validate_credentials_for_event_types(client_no_token, [BEHAVIOR_ANALYTICS])

    def test_validate_credentials_missing_username_password(self):
        """
        Given:
            - Client without username and password
            - Addressable Alerts event type
        When:
            - Calling validate_credentials_for_event_types
        Then:
            - Ensure DemistoException is raised
        """
        client_no_creds = Client(base_url=BASE_URL, verify=True, proxy=False, token="token", secret="secret")
        with pytest.raises(DemistoException, match="Username and Password"):
            validate_credentials_for_event_types(client_no_creds, [ADDRESSABLE_ALERTS])

    def test_validate_credentials_multiple_event_types(self, client):
        """
        Given:
            - Client with all credentials
            - Multiple event types
        When:
            - Calling validate_credentials_for_event_types
        Then:
            - Ensure no exception is raised
        """
        validate_credentials_for_event_types(client, [BEHAVIOR_ANALYTICS, ADDRESSABLE_ALERTS, DETECT_AND_PROTECT_ALERTS])

    @pytest.mark.parametrize(
        "client_kwargs, event_types, expected_errors",
        [
            pytest.param(
                {"token": "token", "secret": "secret"},
                [BEHAVIOR_ANALYTICS, ADDRESSABLE_ALERTS, DETECT_AND_PROTECT_ALERTS],
                ["Addressable Alerts", "Detect And Protect Alerts"],
                id="token_only-missing_user_password_for_two_types",
            ),
            pytest.param(
                {"username": "user", "password": "pass"},
                [BEHAVIOR_ANALYTICS, ADDRESSABLE_ALERTS],
                ["Token and Secret"],
                id="user_only-missing_token_secret",
            ),
            pytest.param(
                {},
                [BEHAVIOR_ANALYTICS, ADDRESSABLE_ALERTS],
                ["Token and Secret", "Username and Password"],
                id="no_credentials-all_missing",
            ),
        ],
    )
    def test_validate_credentials_mixed_partial(self, client_kwargs, event_types, expected_errors):
        """
        Given:
            - Client with partial credentials
            - Multiple event types where some require missing credentials
        When:
            - Calling validate_credentials_for_event_types
        Then:
            - Ensure DemistoException is raised listing all missing credentials
        """
        client = Client(base_url=BASE_URL, verify=True, proxy=False, **client_kwargs)
        with pytest.raises(DemistoException) as exc_info:
            validate_credentials_for_event_types(client, event_types)
        error_msg = str(exc_info.value)
        for expected in expected_errors:
            assert expected in error_msg


class TestDeduplicateEvents:
    """Tests for _deduplicate_events function"""

    def test_deduplicate_events_no_duplicates(self):
        """
        Given:
            - Events with unique IDs
            - Empty last_run_ids
        When:
            - Calling _deduplicate_events
        Then:
            - Ensure all events are returned
        """
        events = [
            {"id": "1", "alert_time": 1000},
            {"id": "2", "alert_time": 900},
            {"id": "3", "alert_time": 800},
        ]
        result = _deduplicate_events(events, [], 700)
        assert len(result) == 3

    def test_deduplicate_events_with_duplicates(self):
        """
        Given:
            - Events with some duplicate IDs from last run
            - All events are newer than last_fetch_time
        When:
            - Calling _deduplicate_events
        Then:
            - Ensure all events are returned (newer events bypass ID check)
        """
        events = [
            {"id": "1", "alert_time": 1000},
            {"id": "2", "alert_time": 900},
            {"id": "3", "alert_time": 800},
        ]
        last_run_ids = ["2"]
        # All events are newer than 700, so ID check is bypassed
        result = _deduplicate_events(events, last_run_ids, 700)
        assert len(result) == 3

    def test_deduplicate_events_with_duplicates_same_time(self):
        """
        Given:
            - Events at or before last_fetch_time
            - Some IDs are duplicates from last run
        When:
            - Calling _deduplicate_events
        Then:
            - Ensure duplicates are removed
        """
        events = [
            {"id": "1", "alert_time": 700},
            {"id": "2", "alert_time": 700},
            {"id": "3", "alert_time": 700},
        ]
        last_run_ids = ["2"]
        result = _deduplicate_events(events, last_run_ids, 700)
        assert len(result) == 2
        assert result[0]["id"] == "1"
        assert result[1]["id"] == "3"

    def test_deduplicate_events_newer_than_last_fetch(self):
        """
        Given:
            - Events newer than last_fetch_time
        When:
            - Calling _deduplicate_events
        Then:
            - Ensure all newer events are returned without checking IDs
        """
        events = [
            {"id": "1", "alert_time": 1000},
            {"id": "2", "alert_time": 900},
        ]
        result = _deduplicate_events(events, ["1", "2"], 800)
        assert len(result) == 2

    def test_deduplicate_events_without_ids(self):
        """
        Given:
            - Events without ID field
        When:
            - Calling _deduplicate_events
        Then:
            - Ensure events are kept (cannot deduplicate)
        """
        events = [
            {"alert_time": 1000},
            {"alert_time": 900},
        ]
        result = _deduplicate_events(events, [], 700)
        assert len(result) == 2


class TestUpdateNextRunState:
    """Tests for _update_next_run_state function"""

    def test_no_new_events_keeps_old_state(self):
        """
        Given:
            - Empty events list
            - Previous fetch time and IDs
        When:
            - Calling _update_next_run_state
        Then:
            - Ensure old timestamp and IDs are preserved
        """
        events = []
        last_fetch_time = 1000
        last_run_ids = ["id1", "id2"]

        new_time, new_ids = _update_next_run_state(events, last_fetch_time, last_run_ids)

        assert new_time == 1000
        assert new_ids == ["id1", "id2"]

    def test_new_events_with_newer_timestamp_replaces_ids(self):
        """
        Given:
            - Events with timestamp newer than last_fetch_time
            - Previous fetch time and IDs
        When:
            - Calling _update_next_run_state
        Then:
            - Ensure timestamp is updated and old IDs are replaced with new ones
        """
        events = [
            {"id": "new1", "alert_time": 2000},
            {"id": "new2", "alert_time": 2000},
            {"id": "old1", "alert_time": 1500},
        ]
        last_fetch_time = 1000
        last_run_ids = ["id1", "id2"]

        new_time, new_ids = _update_next_run_state(events, last_fetch_time, last_run_ids)

        assert new_time == 2000
        assert set(new_ids) == {"new1", "new2"}
        assert "id1" not in new_ids
        assert "id2" not in new_ids

    def test_new_events_with_same_timestamp_combines_ids(self):
        """
        Given:
            - Events with same timestamp as last_fetch_time
            - Previous fetch time and IDs
        When:
            - Calling _update_next_run_state
        Then:
            - Ensure timestamp stays same and old and new IDs are combined
        """
        events = [
            {"id": "new1", "alert_time": 1000},
            {"id": "new2", "alert_time": 1000},
        ]
        last_fetch_time = 1000
        last_run_ids = ["id1", "id2"]

        new_time, new_ids = _update_next_run_state(events, last_fetch_time, last_run_ids)

        assert new_time == 1000
        assert set(new_ids) == {"id1", "id2", "new1", "new2"}

    def test_new_events_with_same_timestamp_avoids_duplicate_ids(self):
        """
        Given:
            - Events with same timestamp as last_fetch_time
            - Some IDs overlap with previous run
        When:
            - Calling _update_next_run_state
        Then:
            - Ensure IDs are combined without duplicates
        """
        events = [
            {"id": "id1", "alert_time": 1000},  # Duplicate
            {"id": "new1", "alert_time": 1000},
        ]
        last_fetch_time = 1000
        last_run_ids = ["id1", "id2"]

        new_time, new_ids = _update_next_run_state(events, last_fetch_time, last_run_ids)

        assert new_time == 1000
        assert set(new_ids) == {"id1", "id2", "new1"}
        assert len(new_ids) == 3  # No duplicates

    def test_older_timestamp_combines_ids(self):
        """
        Given:
            - Events with timestamp older than last_fetch_time (API doesn't filter by start_date)
            - Previous fetch time and IDs
        When:
            - Calling _update_next_run_state
        Then:
            - Ensure timestamp is preserved but IDs are combined to prevent duplicate fetches
        """
        events = [
            {"id": "old1", "alert_time": 500},
        ]
        last_fetch_time = 1000
        last_run_ids = ["id1", "id2"]

        new_time, new_ids = _update_next_run_state(events, last_fetch_time, last_run_ids)

        assert new_time == 1000
        assert set(new_ids) == {"id1", "id2", "old1"}


class TestAddFieldsToEvents:
    """Tests for _add_fields_to_events function"""

    def test_add_fields_to_events_with_update_time(self):
        """
        Given:
            - Events with update_time
            - Source log type parameter
        When:
            - Calling _add_fields_to_events
        Then:
            - Ensure _time and _source_log_type fields are added with correct format
        """
        events = [
            {"update_time": 1609459200000, "id": "1"},  # 2021-01-01 00:00:00 UTC
            {"update_time": 1609545600000, "id": "2"},  # 2021-01-02 00:00:00 UTC
        ]
        _add_fields_to_events(events, "BehaviorAnalytics")
        assert "_time" in events[0]
        assert "_time" in events[1]
        assert events[0]["_time"] == "2021-01-01T00:00:00Z"
        assert events[1]["_time"] == "2021-01-02T00:00:00Z"
        assert events[0]["_source_log_type"] == "BehaviorAnalytics"
        assert events[1]["_source_log_type"] == "BehaviorAnalytics"

    def test_add_fields_to_events_entry_status_new(self):
        """
        Given:
            - Events where update_time equals alert_time
        When:
            - Calling _add_fields_to_events
        Then:
            - Ensure _ENTRY_STATUS is set to 'new'
        """
        events = [
            {"update_time": 1609459200000, "alert_time": 1609459200000, "id": "1"},
        ]
        _add_fields_to_events(events, "BehaviorAnalytics")
        assert events[0]["_ENTRY_STATUS"] == "new"

    def test_add_fields_to_events_entry_status_modified(self):
        """
        Given:
            - Events where update_time is greater than alert_time
        When:
            - Calling _add_fields_to_events
        Then:
            - Ensure _ENTRY_STATUS is set to 'modified'
        """
        events = [
            {"update_time": 1609545600000, "alert_time": 1609459200000, "id": "1"},
        ]
        _add_fields_to_events(events, "BehaviorAnalytics")
        assert events[0]["_ENTRY_STATUS"] == "modified"

    def test_add_fields_to_events_without_update_time(self):
        """
        Given:
            - Events without update_time field
            - Source log type parameter
        When:
            - Calling _add_fields_to_events
        Then:
            - Ensure _time field is not added but _source_log_type is added
        """
        events = [{"id": "1"}]
        _add_fields_to_events(events, "AddressableAlerts")
        assert events[0].get("_time") is None
        assert events[0]["_source_log_type"] == "AddressableAlerts"

    def test_add_fields_to_events_empty_list(self):
        """
        Given:
            - Empty events list
            - Source log type parameter
        When:
            - Calling _add_fields_to_events
        Then:
            - Ensure no error is raised
        """
        events = []
        _add_fields_to_events(events, "BehaviorAnalytics")
        assert events == []


class TestFetchEventsWithPagination:
    """Tests for _fetch_events_with_pagination function"""

    def test_pagination_single_page(self, client, mocker):
        """
        Given:
            - Limit of 50 events
            - API returns 30 events (less than page size)
        When:
            - Calling _fetch_events_with_pagination
        Then:
            - Ensure pagination stops after first page
        """
        mocker.patch.object(
            client,
            "_fetch_alerts",
            return_value=[{"id": str(i), "alert_time": 1000 - i} for i in range(30)],
        )

        events = _fetch_events_with_pagination(client, BEHAVIOR_ANALYTICS, 500, 1000, 50)
        assert len(events) == 30

    def test_pagination_multiple_pages(self, client, mocker):
        """
        Given:
            - Limit of 200 events
        When:
            - Calling _fetch_events_with_pagination
        Then:
            - Ensure _fetch_alerts is called 3 times with page_size: 90, 90, 20
        """
        # Mock _fetch_alerts to return dummy data
        mock_fetch = mocker.patch.object(
            client,
            "_fetch_alerts",
            side_effect=[
                [{"id": f"1-{i}", "alert_time": 1000 - i} for i in range(90)],
                [{"id": f"2-{i}", "alert_time": 910 - i} for i in range(90)],
                [{"id": f"3-{i}", "alert_time": 820 - i} for i in range(90)],
            ],
        )

        _fetch_events_with_pagination(client, BEHAVIOR_ANALYTICS, 500, 1000, 200)

        assert mock_fetch.call_count == 3
        # Verify page_size parameter in each call
        assert mock_fetch.call_args_list[0][0][3] == 90  # First call
        assert mock_fetch.call_args_list[1][0][3] == 90  # Second call
        assert mock_fetch.call_args_list[2][0][3] == 20  # Third call

    def test_pagination_with_deduplication_at_boundary(self, client, mocker):
        """
        Given:
            - Events with same alert_time at page boundary
        When:
            - Calling _fetch_events_with_pagination
        Then:
            - Ensure duplicates at boundary are removed
        """
        # First page: last 3 events have alert_time=910
        page1 = [{"id": f"1-{i}", "alert_time": 1000 - i} for i in range(87)]
        page1.extend(
            [
                {"id": "dup-1", "alert_time": 910},
                {"id": "dup-2", "alert_time": 910},
                {"id": "dup-3", "alert_time": 910},
            ]
        )

        # Second page: includes the same 3 duplicate events plus new ones
        page2 = [
            {"id": "dup-1", "alert_time": 910},
            {"id": "dup-2", "alert_time": 910},
            {"id": "dup-3", "alert_time": 910},
        ]
        page2.extend([{"id": f"2-{i}", "alert_time": 909 - i} for i in range(50)])

        mocker.patch.object(client, "_fetch_alerts", side_effect=[page1, page2])

        events = _fetch_events_with_pagination(client, BEHAVIOR_ANALYTICS, 500, 1000, 200)

        # Should have 90 from page1 + 50 new from page2 = 140 (3 duplicates removed)
        assert len(events) == 140
        # Verify no duplicate IDs
        event_ids = [e["id"] for e in events]
        assert len(event_ids) == len(set(event_ids))

    def test_pagination_boundary_all_identical_alert_time_and_id(self, client, mocker):
        """
        Given:
            - All events at page boundary have identical alert_time and id
        When:
            - Calling _fetch_events_with_pagination
        Then:
            - Ensure no duplicate events are created
        """
        # Page 1: all events at the boundary share the same alert_time
        page1 = [{"id": f"1-{i}", "alert_time": 1000 - i} for i in range(85)]
        page1.extend([{"id": f"boundary-{i}", "alert_time": 915} for i in range(5)])

        # Page 2: starts with the same boundary events (duplicates) plus new events
        page2 = [{"id": f"boundary-{i}", "alert_time": 915} for i in range(5)]
        page2.extend([{"id": f"2-{i}", "alert_time": 914 - i} for i in range(30)])

        mocker.patch.object(client, "_fetch_alerts", side_effect=[page1, page2])

        events = _fetch_events_with_pagination(client, BEHAVIOR_ANALYTICS, 500, 1000, 200)

        # Verify no duplicate IDs
        event_ids = [e["id"] for e in events]
        assert len(event_ids) == len(
            set(event_ids)
        ), f"Found duplicate IDs: {[eid for eid in event_ids if event_ids.count(eid) > 1]}"
        # Should have 90 from page1 + 30 new from page2 = 120 (5 duplicates removed)
        assert len(events) == 120

    def test_pagination_stops_when_limit_reached(self, client, mocker):
        """
        Given:
            - Limit of 100 events
        When:
            - Calling _fetch_events_with_pagination
        Then:
            - Ensure pagination stops at limit with page_size: 90, 10
        """
        mock_fetch = mocker.patch.object(
            client,
            "_fetch_alerts",
            side_effect=[
                [{"id": f"1-{i}", "alert_time": 1000 - i} for i in range(90)],  # Page 1: 90 events
                [{"id": f"2-{i}", "alert_time": 910 - i} for i in range(10)],  # Page 2: 10 events
            ],
        )

        events = _fetch_events_with_pagination(client, BEHAVIOR_ANALYTICS, 500, 1000, 100)

        assert mock_fetch.call_count == 2
        assert mock_fetch.call_args_list[0][0][3] == 90  # First call: page_size=90
        assert mock_fetch.call_args_list[1][0][3] == 10  # Second call: page_size=10 (remaining)
        assert len(events) == 100


class TestClient:
    """Tests for Client class methods"""

    def test_get_access_token_from_token_secret(self, client, mocker):
        """
        Given:
            - Valid token and secret
        When:
            - Calling get_access_token_from_token_secret
        Then:
            - Ensure access token is returned
        """
        # Mock to bypass caching
        mocker.patch.object(client, "_get_cached_token", return_value=None)
        mocker.patch.object(client, "_cache_token")

        mocker.patch.object(client, "_http_request", return_value={"access_token": "test_access_token"})
        token = client.get_access_token_from_token_secret()
        assert token == "test_access_token"
        assert client._access_token == "test_access_token"

    def test_get_access_token_force_new(self, client, mocker):
        """
        Given:
            - Client with existing cached token
            - force_new=True parameter
        When:
            - Calling get_access_token_from_token_secret with force_new=True
        Then:
            - Ensure new token is generated bypassing cache
        """
        # Set existing token
        client._access_token = "old_token"
        mocker.patch.object(client, "_cache_token")

        mocker.patch.object(client, "_http_request", return_value={"access_token": "new_token"})
        token = client.get_access_token_from_token_secret(force_new=True)
        assert token == "new_token"
        assert client._access_token == "new_token"

    def test_get_access_token_from_username_password(self, client, mocker):
        """
        Given:
            - Valid username and password
        When:
            - Calling get_access_token_from_username_password
        Then:
            - Ensure access token is returned
        """
        mocker.patch.object(client, "_http_request", return_value={"access_token": "test_user_access_token"})
        token = client.get_access_token_from_username_password()
        assert token == "test_user_access_token"
        assert client._user_access_token == "test_user_access_token"

    def test_fetch_alerts_with_retry_on_429(self, client, mocker):
        """
        Given:
            - API returns HTTP 429 (Too Many Requests) on first attempt
            - API succeeds on second attempt
        When:
            - Calling _fetch_alerts
        Then:
            - Ensure retry mechanism works and events are returned
        """
        mocker.patch("time.sleep")  # Mock sleep to speed up test
        mocker.patch.object(client, "_get_cached_token", return_value=None)
        mocker.patch.object(client, "_cache_token")

        call_count = [0]

        def http_request_side_effect(**kwargs):
            url_suffix = kwargs.get("url_suffix", "")
            if "login" in url_suffix:
                return {"access_token": f"token_{call_count[0]}"}
            # Alert endpoint
            call_count[0] += 1
            if call_count[0] == 1:
                raise DemistoException("Error in API call [429] - Too Many Requests")
            return {"results": [{"id": "1", "alert_time": 1000}]}

        mocker.patch.object(client, "_http_request", side_effect=http_request_side_effect)

        alerts = client._fetch_alerts(BEHAVIOR_ANALYTICS, 500, 1000, 10)
        assert len(alerts) == 1
        assert call_count[0] == 2  # Should have made 2 attempts

    def test_fetch_alerts_with_retry_on_401(self, client, mocker):
        """
        Given:
            - API returns HTTP 401 (Unauthorized - token expired) on first attempt
            - API succeeds on second attempt with new token
        When:
            - Calling _fetch_alerts
        Then:
            - Ensure token is regenerated and request succeeds
        """
        mocker.patch("time.sleep")
        mocker.patch.object(client, "_get_cached_token", return_value=None)
        mocker.patch.object(client, "_cache_token")

        call_count = [0]

        def http_request_side_effect(**kwargs):
            url_suffix = kwargs.get("url_suffix", "")
            if "login" in url_suffix:
                return {"access_token": f"token_{call_count[0]}"}
            # Alert endpoint
            call_count[0] += 1
            if call_count[0] == 1:
                raise DemistoException("Error in API call [401] - Unauthorized")
            return {"results": [{"id": "1", "alert_time": 1000}]}

        mocker.patch.object(client, "_http_request", side_effect=http_request_side_effect)

        alerts = client._fetch_alerts(BEHAVIOR_ANALYTICS, 500, 1000, 10)
        assert len(alerts) == 1
        assert call_count[0] == 2

    def test_fetch_alerts_retry_exhausted(self, client, mocker, capfd):
        """
        Given:
            - API returns HTTP 429 on all attempts
        When:
            - Calling _fetch_alerts
        Then:
            - Ensure exception is raised after max retries
        """
        mocker.patch("time.sleep")
        mocker.patch.object(client, "_get_cached_token", return_value=None)
        mocker.patch.object(client, "_cache_token")

        def http_request_side_effect(**kwargs):
            url_suffix = kwargs.get("url_suffix", "")
            if "login" in url_suffix:
                return {"access_token": "token"}
            raise DemistoException("Error in API call [429] - Too Many Requests")

        mocker.patch.object(client, "_http_request", side_effect=http_request_side_effect)

        with capfd.disabled(), pytest.raises(DemistoException):
            client._fetch_alerts(BEHAVIOR_ANALYTICS, 500, 1000, 10)

    def test_fetch_alerts_no_retry_on_other_errors(self, client, mocker):
        """
        Given:
            - API returns HTTP 500 (Server Error)
        When:
            - Calling _fetch_alerts
        Then:
            - Ensure no retry is attempted and exception is raised immediately
        """
        mocker.patch.object(client, "_get_cached_token", return_value=None)

        def http_request_side_effect(**kwargs):
            url_suffix = kwargs.get("url_suffix", "")
            if "login" in url_suffix:
                return {"access_token": "token"}
            raise DemistoException("Error in API call [500] - Server Error")

        mocker.patch.object(client, "_http_request", side_effect=http_request_side_effect)

        with pytest.raises(DemistoException):
            client._fetch_alerts(BEHAVIOR_ANALYTICS, 500, 1000, 10)


class TestJwtExtraction:
    """Tests for _extract_jwt_expiration edge cases"""

    @pytest.mark.parametrize(
        "jwt_token, expected_behavior",
        [
            pytest.param(
                "only.two_parts",
                "default",
                id="malformed_jwt_wrong_number_of_parts",
            ),
            pytest.param(
                "a.b.c.d",
                "default",
                id="malformed_jwt_too_many_parts",
            ),
            pytest.param(
                "header.!!!invalid_base64!!!.signature",
                "default",
                id="invalid_base64_encoding",
            ),
        ],
    )
    @freeze_time("2021-01-10T00:00:00Z")
    def test_extract_jwt_expiration_malformed(self, client, jwt_token, expected_behavior):
        """
        Given:
            - A malformed JWT token (wrong parts count, invalid base64, or missing exp)
        When:
            - Calling _extract_jwt_expiration
        Then:
            - Ensure default 30-minute expiry is returned
        """
        result = client._extract_jwt_expiration(jwt_token)
        # Default is current time + 1800 seconds (30 minutes)
        # Frozen time: 2021-01-10T00:00:00Z = 1610236800
        expected_default = 1610236800 + 1800
        assert result == expected_default

    @freeze_time("2021-01-10T00:00:00Z")
    def test_extract_jwt_expiration_missing_exp_field(self, client):
        """
        Given:
            - A valid JWT structure but payload missing 'exp' field
        When:
            - Calling _extract_jwt_expiration
        Then:
            - Ensure default 30-minute expiry is returned
        """
        # Create a valid JWT with no 'exp' field
        payload = base64.urlsafe_b64encode(json.dumps({"sub": "user123"}).encode()).decode().rstrip("=")
        jwt_token = f"header.{payload}.signature"
        result = client._extract_jwt_expiration(jwt_token)
        expected_default = 1610236800 + 1800
        assert result == expected_default

    def test_extract_jwt_expiration_valid(self, client):
        """
        Given:
            - A valid JWT with 'exp' field
        When:
            - Calling _extract_jwt_expiration
        Then:
            - Ensure the correct expiration timestamp is returned
        """
        exp_time = 1610240400  # Some future timestamp
        payload = base64.urlsafe_b64encode(json.dumps({"exp": exp_time}).encode()).decode().rstrip("=")
        jwt_token = f"header.{payload}.signature"
        result = client._extract_jwt_expiration(jwt_token)
        assert result == exp_time


class TestTokenCaching:
    """Tests for _cache_token and _get_cached_token edge cases"""

    @freeze_time("2021-01-10T00:00:00Z")
    def test_get_cached_token_expired(self, client, mocker):
        """
        Given:
            - A cached token that has already expired
        When:
            - Calling _get_cached_token
        Then:
            - Ensure None is returned (token is not used)
        """
        # Frozen time: 1610236800. Token expired at 1610236700 (in the past)
        mocker.patch(
            "demistomock.getIntegrationContext",
            return_value={
                "api_access_token": "expired_token",
                "api_token_expiry": 1610236700,
            },
        )
        result = client._get_cached_token("api_access_token", "api_token_expiry")
        assert result is None

    @freeze_time("2021-01-10T00:00:00Z")
    def test_get_cached_token_expires_within_buffer(self, client, mocker):
        """
        Given:
            - A cached token that expires within the 5-minute buffer (< 300 seconds from now)
        When:
            - Calling _get_cached_token
        Then:
            - Ensure None is returned (token is considered expired)
        """
        # Frozen time: 1610236800. Token expires at 1610237000 (200 seconds from now, within 300s buffer)
        mocker.patch(
            "demistomock.getIntegrationContext",
            return_value={
                "api_access_token": "almost_expired_token",
                "api_token_expiry": 1610237000,
            },
        )
        result = client._get_cached_token("api_access_token", "api_token_expiry")
        assert result is None

    @freeze_time("2021-01-10T00:00:00Z")
    def test_get_cached_token_valid(self, client, mocker):
        """
        Given:
            - A cached token that is still valid (expires well beyond the 5-minute buffer)
        When:
            - Calling _get_cached_token
        Then:
            - Ensure the cached token is returned
        """
        # Frozen time: 1610236800. Token expires at 1610238800 (2000 seconds from now, well beyond 300s buffer)
        mocker.patch(
            "demistomock.getIntegrationContext",
            return_value={
                "api_access_token": "valid_token",
                "api_token_expiry": 1610238800,
            },
        )
        result = client._get_cached_token("api_access_token", "api_token_expiry")
        assert result == "valid_token"

    @freeze_time("2021-01-10T00:00:00Z")
    def test_get_cached_token_non_integer_expiry(self, client, mocker):
        """
        Given:
            - A cached token with a non-integer expiry value in integration context
        When:
            - Calling _get_cached_token
        Then:
            - Ensure None is returned (invalid expiry treated as expired)
        """
        mocker.patch(
            "demistomock.getIntegrationContext",
            return_value={
                "api_access_token": "some_token",
                "api_token_expiry": "not_an_integer",
            },
        )
        result = client._get_cached_token("api_access_token", "api_token_expiry")
        assert result is None

    def test_cache_token_empty_token(self, client, mocker):
        """
        Given:
            - An empty token string
        When:
            - Calling _cache_token
        Then:
            - Ensure the token is not cached (setIntegrationContext is not called)
        """
        mock_set_context = mocker.patch("demistomock.setIntegrationContext")
        mocker.patch("demistomock.getIntegrationContext", return_value={})
        client._cache_token("", "api_access_token", "api_token_expiry")
        mock_set_context.assert_not_called()


class TestTestModuleCommand:
    """Tests for test_module_command function"""

    @freeze_time("2021-01-10T00:00:00Z")
    def test_test_module_success_behavior_analytics(self, client, mocker):
        """
        Given:
            - Valid client with token and secret
            - Behavior Analytics event type
        When:
            - Calling test_module_command
        Then:
            - Ensure 'ok' is returned
        """
        from iManageThreatManager import test_module_command

        mocker.patch.object(client, "_fetch_alerts", return_value=[])

        result = test_module_command(client, {}, [BEHAVIOR_ANALYTICS])
        assert result == "ok"

    @freeze_time("2021-01-10T00:00:00Z")
    def test_test_module_success_addressable_alerts(self, client, mocker):
        """
        Given:
            - Valid client with username and password
            - Addressable Alerts event type
        When:
            - Calling test_module_command
        Then:
            - Ensure 'ok' is returned
        """
        from iManageThreatManager import test_module_command

        mocker.patch.object(client, "_fetch_alerts", return_value=[])

        result = test_module_command(client, {}, [ADDRESSABLE_ALERTS])
        assert result == "ok"

    def test_test_module_auth_error(self, client, mocker, capfd):
        """
        Given:
            - Invalid credentials
        When:
            - Calling test_module_command
        Then:
            - Ensure authorization error message is returned
        """
        from iManageThreatManager import test_module_command

        mocker.patch.object(
            client,
            "_fetch_alerts",
            side_effect=DemistoException("Error in API call [401] - Unauthorized"),
        )

        with capfd.disabled():
            result = test_module_command(client, {}, [BEHAVIOR_ANALYTICS])
            assert "Authorization Error" in result

    @freeze_time("2021-01-10T00:00:00Z")
    def test_test_module_calls_fetch_alerts_without_retries(self, client, mocker):
        """
        Given:
            - Valid client with credentials
            - Behavior Analytics event type
        When:
            - Calling test_module_command
        Then:
            - Ensure _fetch_alerts is called with enable_retries=False
              to avoid long retry delays that cause platform timeouts
        """
        from iManageThreatManager import test_module_command

        mock_fetch = mocker.patch.object(client, "_fetch_alerts", return_value=[])

        test_module_command(client, {}, [BEHAVIOR_ANALYTICS])

        mock_fetch.assert_called_once()
        call_kwargs = mock_fetch.call_args.kwargs
        assert (
            call_kwargs.get("enable_retries") is False
        ), "test_module_command must call _fetch_alerts with enable_retries=False to prevent timeout"


class TestGetEventsCommand:
    """Tests for get_events_command function"""

    @freeze_time("2021-01-10T00:00:00Z")
    def test_get_events_command_behavior_analytics(self, client, mocker):
        """
        Given:
            - Valid client and Behavior Analytics event type
        When:
            - Calling get_events_command
        Then:
            - Ensure events are returned
        """
        mocker.patch.object(
            client,
            "_fetch_alerts",
            return_value=[{"id": "1", "alert_time": 1609459200000, "update_time": 1609459200000}],
        )

        events, results = get_events_command(client, {"event_type": BEHAVIOR_ANALYTICS, "limit": "10"})
        assert len(events) == 1
        assert events[0]["id"] == "1"

    @freeze_time("2021-01-10T00:00:00Z")
    def test_get_events_command_with_pagination(self, client, mocker):
        """
        Given:
            - Limit of 150 events
            - API returns events in multiple pages
        When:
            - Calling get_events_command
        Then:
            - Ensure all events are fetched via pagination
        """
        page1 = [{"id": f"1-{i}", "alert_time": 1000 - i} for i in range(90)]
        page2 = [{"id": f"2-{i}", "alert_time": 910 - i} for i in range(60)]

        mocker.patch.object(
            client,
            "_fetch_alerts",
            side_effect=[page1, page2],
        )

        events, results = get_events_command(client, {"event_type": BEHAVIOR_ANALYTICS, "limit": "150"})
        assert len(events) == 150


class TestFetchEventsCommand:
    """Tests for fetch_events_command function"""

    @freeze_time("2021-01-10T00:00:00Z")
    def test_fetch_events_first_run(self, client, mocker):
        """
        Given:
            - Empty last_run (first fetch)
            - Behavior Analytics event type
        When:
            - Calling fetch_events_command
        Then:
            - Ensure events are fetched and next_run is set correctly
        """
        # Frozen time: 2021-01-10 00:00:00 UTC = 1610236800000 ms
        # First fetch lookback: 1 hour = 1610236800000 - 3600000 = 1610233200000 ms
        # Mock event must be AFTER 1610233200000 to be included
        mocker.patch.object(
            client,
            "_fetch_alerts",
            return_value=[{"id": "1", "alert_time": 1610235000000, "update_time": 1610235000000}],
        )

        next_run, events = fetch_events_command(
            client=client, last_run={}, event_types=[BEHAVIOR_ANALYTICS], max_events_per_type=10
        )

        assert len(events) == 1
        assert events[0]["_source_log_type"] == EVENT_TYPE_CONFIG[BEHAVIOR_ANALYTICS].source_log_type
        assert "last_fetch_BehaviorAnalytics" in next_run
        assert next_run["last_fetch_BehaviorAnalytics"] == 1610235000000

    @freeze_time("2021-01-10T00:00:00Z")
    def test_fetch_events_with_pagination(self, client, mocker):
        """
        Given:
            - max_events_per_type of 200
            - API returns events in multiple pages
        When:
            - Calling fetch_events_command
        Then:
            - Ensure pagination is used to fetch all events
        """
        page1 = [{"id": f"1-{i}", "alert_time": 1000 - i, "update_time": 1000 - i} for i in range(90)]
        page2 = [{"id": f"2-{i}", "alert_time": 910 - i, "update_time": 910 - i} for i in range(90)]
        page3 = [{"id": f"3-{i}", "alert_time": 820 - i, "update_time": 820 - i} for i in range(20)]

        mocker.patch.object(
            client,
            "_fetch_alerts",
            side_effect=[page1, page2, page3],
        )

        next_run, events = fetch_events_command(
            client=client, last_run={}, event_types=[BEHAVIOR_ANALYTICS], max_events_per_type=200
        )

        assert len(events) == 200

    @freeze_time("2021-01-10T00:00:00Z")
    def test_fetch_events_multiple_types(self, client, mocker):
        """
        Given:
            - Multiple event types configured
        When:
            - Calling fetch_events_command
        Then:
            - Ensure events from all types are fetched
        """
        behavior_events = [{"id": "1", "alert_time": 1609459200000, "update_time": 1609459200000}]
        addressable_events = [{"id": "2", "alert_time": 1609545600000, "update_time": 1609545600000}]

        mocker.patch.object(
            client,
            "_fetch_alerts",
            side_effect=[behavior_events, addressable_events],
        )

        next_run, events = fetch_events_command(
            client=client, last_run={}, event_types=[BEHAVIOR_ANALYTICS, ADDRESSABLE_ALERTS], max_events_per_type=10
        )

        assert len(events) == 2
        assert events[0]["_source_log_type"] == EVENT_TYPE_CONFIG[BEHAVIOR_ANALYTICS].source_log_type
        assert events[1]["_source_log_type"] == EVENT_TYPE_CONFIG[ADDRESSABLE_ALERTS].source_log_type

    @freeze_time("2021-01-10T00:00:00Z")
    def test_fetch_events_no_new_events(self, client, mocker):
        """
        Given:
            - Last run with previous fetch time
            - No new events available
        When:
            - Calling fetch_events_command
        Then:
            - Ensure empty events list and updated next_run
        """
        mocker.patch.object(
            client,
            "_fetch_alerts",
            return_value=[],
        )

        last_run = {"last_fetch_BehaviorAnalytics": 1609459200000}
        next_run, events = fetch_events_command(
            client=client, last_run=last_run, event_types=[BEHAVIOR_ANALYTICS], max_events_per_type=10
        )

        assert len(events) == 0
        assert "last_fetch_BehaviorAnalytics" in next_run

    @freeze_time("2021-01-10T00:00:00Z")
    def test_fetch_events_with_error(self, client, mocker, capfd):
        """
        Given:
            - API error during fetch
        When:
            - Calling fetch_events_command
        Then:
            - Ensure error is handled and last_run is preserved
        """
        mocker.patch.object(
            client,
            "_fetch_alerts",
            side_effect=DemistoException("Error in API call [500] - Server Error"),
        )

        last_run = {"last_fetch_BehaviorAnalytics": 1609459200000}

        with capfd.disabled():
            next_run, events = fetch_events_command(
                client=client, last_run=last_run, event_types=[BEHAVIOR_ANALYTICS], max_events_per_type=10
            )

        # Should preserve last_run on error
        assert next_run["last_fetch_BehaviorAnalytics"] == 1609459200000
        assert len(events) == 0