Cortex Core - IOC

The Cortex Core - IOCs integration uses the Cortex API for detection and response, by natively integrating network, endpoint, and cloud data to stop sophisticated attacks.

Data Enrichment & Threat Intelligence · Core

Details

ID	`Cortex Core - IOC`
Provider	Open Source
Category	Data Enrichment & Threat Intelligence
From Version	5.5.0
Docker Image	`demisto/google-cloud-storage:1.0.0.10120494`
Supported Modules	Exposure Management Cortex Cloud Cloud Runtime Security Agentix Cloud Posture Security XSIAM EDR

README

The Cortex Core - IOCs integration uses the Cortex API for detection and response, by natively integrating network, endpoint, and cloud data to stop sophisticated attacks.

Configure Indicators detection

Parameter	Description	Required
Cortex XDR Severity	Map the severity of each indicator that will be synced to Cortex.	True
Tags	Supports CSV values.	False
Sync Query	The query used to collect indicators to sync from Cortex.	True
Trust any certificate (not secure)		False
Use system proxy settings		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.

core-iocs-sync

Sync your IOC with Cortex and delete the previous version.

Base Command

core-iocs-sync

Input

Argument Name	Description	Required
firstTime	For first sync, set to true. (do NOT run this twice!). Possible values are: true, false. Default is false.	Optional

Context Output

There is no context output for this command.

core-iocs-push

Push modified IOCs to Cortex.

Base Command

core-iocs-push

Input

Argument Name	Description	Required
indicator	IOCs to push. leave empty to push all recently modified IOCs.the indicators.	Optional

Context Output

There is no context output for this command.

Command example

!core-iocs-push indicator='test.com'

Human Readable Output

push done.

core-iocs-set-sync-time

Set sync time manually (Do not use this command unless you unredstandard the consequences).

Base Command

core-iocs-set-sync-time

Input

Argument Name	Description	Required
time	The time of the file creation (use UTC time zone).	Required

Context Output

There is no context output for this command.

core-iocs-create-sync-file

Creates the sync file for the manual process. Run this command when instructed by the Cortex support team.

Base Command

core-iocs-create-sync-file

Input

There are no input arguments for this command.

Context Output

There is no context output for this command.

core-iocs-enable

Enables IOCs in the Cortex tenant.

Note: This command applies only to TIM-managed indicators (indicators that were synced/pushed to the Cortex tenant by this integration). Enabling an indicator not managed by TIM has no effect.

Base Command

core-iocs-enable

Input

Argument Name	Description	Required
indicator	The indicator to enable. Only TIM-managed indicators can be enabled; indicators from other sources are not affected.	Required

Context Output

There is no context output for this command.

Command example

!core-iocs-enable indicator=1.1.1.1

Human Readable Output

indicators 1.1.1.1 enabled.

core-iocs-disable

Disables IOCs in the Cortex tenant.

Note: This command applies only to TIM-managed indicators (indicators that were synced/pushed to the Cortex tenant by this integration). Disabling an indicator not managed by TIM has no effect.

Base Command

core-iocs-disable

Input

Argument Name	Description	Required
indicator	The indicator to disable. Only TIM-managed indicators can be disabled; indicators from other sources are not affected.	Required

Context Output

There is no context output for this command.

Command example

!core-iocs-disable indicator=22.22.22.22

Human Readable Output

indicators 22.22.22.22 disabled.

Configuration parameters

url — Server URL (e.g. https://example.net)
apikey_id — API Key ID
apikey — API Key
severity — Cortex Severity
query — Sync Query

Commands (6)

core-iocs-create-sync-file
Creates the sync file for the manual process. Run this command when instructed by the Cortex support team.
core-iocs-disable
Disables IOCs in the Cortex server. Applies only to TIM-managed indicators (indicators synced/pushed by this integration); indicators from other sources are not affected.
core-iocs-enable
Enables IOCs in the Cortex tenant. Applies only to TIM-managed indicators (indicators synced/pushed by this integration); indicators from other sources are not affected.
core-iocs-push
Push modified IOCs to Cortex.
core-iocs-set-sync-time
Set sync time manually (Do not use this command unless you unredstandard the consequences).
core-iocs-sync
Sync your IOC with Cortex and delete the previous version.

Source

CoreIOCs.py CoreIOCs_test.py CoreIOCs.yml CoreIOCs_description.md command_examples.txt

Download History Raw

import pytest
from CoreIOCs import *
from freezegun import freeze_time

Client.severity = "INFO"
client = Client({"url": "https://example.com", "apikey": "apikey", "apikey_id": "apikey_id"})


def d_sort(in_dict):
    return sorted(in_dict.items())


class TestGetHeaders:
    def test_empty_case(self):
        """
        Given:
            Empty params
        Then:
            get_headers will not raise error
        """
        get_headers({})


class TestHttpRequest:
    def test_http_request_ok(self, requests_mock):
        """
        Given:
            - a client
        When:
            - http_request returns status code 200.
        Then:
            - do not raise an error
        """
        requests_mock.post("https://example.com/public_api/v1/indicators/suffix", status_code=200, json={})
        client.http_request(url_suffix="suffix", requests_kwargs={})

    @pytest.mark.parametrize("status_code", client.error_codes.keys())
    def test_http_request_error(self, requests_mock, status_code):
        """
        Given:
            - Status code
        When:
            - http_request returns this status code.
        Then:
            - Verify error message.
            - Verify exception.res status code matches the http status code.
        """
        with pytest.raises(DemistoException) as e:
            requests_mock.post("https://example.com/public_api/v1/indicators/suffix", status_code=status_code)
            client.http_request("suffix", requests_kwargs={})
        assert e.value.message == client.error_codes[status_code]
        assert e.value.res.status_code == status_code

    def test_http_request_bad_json(self, requests_mock):
        """
        Given:
            - a client
        When:
            - http_request returns a response that is not a json.
        Then:
            - Verify error message.
            - Verify demisto exception
        """
        text = "not a json"

        with pytest.raises(DemistoException) as e:
            requests_mock.post("https://example.com/public_api/v1/indicators/suffix", status_code=200, text=text)
            client.http_request("suffix", requests_kwargs={})
        assert e.value.message == f"Could not parse json out of {text}"
        assert e.value.res.status_code == 200
        assert isinstance(e.value.exception, json.JSONDecodeError)


class TestGetRequestsKwargs:
    def test_with_json(self):
        """
        Given:
            - simple json
        Then:
            - the json ready to send
        """
        _json = {"test": "test"}
        output = get_requests_kwargs(_json=_json)
        expected_output = {"data": '{"request_data": {"test": "test"}}'}
        assert (
            output == expected_output
        ), f"get_requests_kwargs(_json={_json})\n\treturns: {output}\n\t instead: {expected_output}"  # noqa: E501


class TestPrepareCommands:
    def test_prepare_get_changes(self):
        """
        Given:
            - get changes command
        Then:
            - Verify url and json format.
        """

        ts = int(datetime.now(timezone.utc).timestamp() * 1000)
        url_suffix, _json = prepare_get_changes(ts)
        assert (
            url_suffix == "get_changes"
        ), f"prepare_get_changes\n\treturns url_suffix: {url_suffix}\n\tinstead url_suffix: get_changes"  # noqa: E501
        assert _json == {"last_update_ts": ts}

    def test_prepare_enable_iocs(self):
        """
        Given:
            - enable iocs command
        Then:
            - Verify url and json format.
        """
        url_suffix, iocs = prepare_enable_iocs("8.8.8.8,domain.com")
        assert (
            url_suffix == "enable_iocs"
        ), f"prepare_enable_iocs\n\treturns url_suffix: {url_suffix}\n\tinstead url_suffix: enable_iocs"  # noqa: E501
        assert iocs == ["8.8.8.8", "domain.com"]

    def test_prepare_disable_iocs(self):
        """
        Given:
            - disable iocs command
        Then:
            - Verify url and json format.
        """
        url_suffix, iocs = prepare_disable_iocs("8.8.8.8,domain.com")
        assert (
            url_suffix == "disable_iocs"
        ), f"prepare_disable_iocs\n\treturns url_suffix: {url_suffix}\n\tinstead url_suffix: disable_iocs"  # noqa: E501
        assert iocs == ["8.8.8.8", "domain.com"]


class TestCreateFile:
    path = "test_data/sync_file_test.json"
    data_test_create_file_sync = [
        ("Domain_iocs", "Domain_sync_file"),
        ("IP_iocs", "IP_sync_file"),
        ("File_iocs", "File_sync_file"),
    ]
    data_test_create_file_iocs_to_keep = [
        ("Domain_iocs", "Domain_iocs_to_keep_file"),
        ("IP_iocs", "IP_iocs_to_keep_file"),
        ("File_iocs", "File_iocs_to_keep_file"),
    ]

    @classmethod
    def setup_method(cls):
        # creates the file
        with open(TestCreateFile.path, "w") as _file:
            _file.write("")

    @classmethod
    def teardown_method(cls):
        # removes the file when done
        os.remove(TestCreateFile.path)

    @staticmethod
    def get_file(path):
        with open(path) as _file:
            return _file.read()

    @staticmethod
    def get_all_iocs(go_over, extension):
        iocs = []
        total = 0
        data = []
        for in_iocs, out_iocs in go_over:
            ioc = json.loads(TestCreateFile.get_file(f"test_data/{in_iocs}.json"))
            iocs.extend(ioc["iocs"])
            total += ioc["total"]
            data.append(TestCreateFile.get_file(f"test_data/{out_iocs}.{extension}"))

        all_iocs = {"iocs": iocs, "total": total}
        all_data = "".join(data)
        return all_iocs, all_data

    def test_create_file_sync_without_iocs(self, mocker):
        """
        Given:
            - Sync command
        When:
            - there is no iocs
        Then:
            - Verify sync file data.
        """
        mocker.patch.object(demisto, "searchIndicators", return_value={"total": 0})
        create_file_sync(TestCreateFile.path)
        data = self.get_file(TestCreateFile.path)
        expected_data = ""
        assert data == expected_data, f"create_file_sync with no iocs\n\tcreates: {data}\n\tinstead: {expected_data}"

    @pytest.mark.parametrize("in_iocs, out_iocs", data_test_create_file_sync)
    def test_create_file_sync(self, in_iocs, out_iocs, mocker):
        """
        Given:
            - Sync command
        When:
            - iocs type is a specific type.
        Then:
            - Verify sync file data.
        """
        mocker.patch.object(demisto, "searchIndicators", return_value=json.loads(self.get_file(f"test_data/{in_iocs}.json")))  # noqa: E501
        create_file_sync(TestCreateFile.path)
        data = self.get_file(TestCreateFile.path)
        expected_data = self.get_file(f"test_data/{out_iocs}.txt")
        assert data == expected_data, f"create_file_sync with {in_iocs} iocs\n\tcreates: {data}\n\tinstead: {expected_data}"

    def test_create_file_sync_all_types(self, mocker):
        """
        Given:
            - Sync command
        When:
            - iocs as all types
        Then:
            - Verify sync file data.
        """
        all_iocs, expected_data = self.get_all_iocs(self.data_test_create_file_sync, "txt")
        mocker.patch.object(demisto, "searchIndicators", return_value=all_iocs)
        create_file_sync(TestCreateFile.path)
        data = self.get_file(TestCreateFile.path)
        assert data == expected_data, f"create_file_sync with all iocs\n\tcreates: {data}\n\tinstead: {expected_data}"

    data_test_create_file_with_empty_indicators = [{}, {"value": "11.11.11.11"}, {"indicator_type": "IP"}]

    @pytest.mark.parametrize("defective_indicator", data_test_create_file_with_empty_indicators)
    def test_create_file_sync_with_empty_indicators(self, defective_indicator, mocker):
        """
        Given:
            - Sync command
        When:
            - a part iocs dont have all required data
        Then:
            - Verify sync file data.
        """
        all_iocs, expected_data = self.get_all_iocs(self.data_test_create_file_sync, "txt")
        all_iocs["iocs"].append(defective_indicator)
        all_iocs["total"] += 1
        mocker.patch.object(demisto, "searchIndicators", return_value=all_iocs)
        warnings = mocker.patch.object(demisto, "debug")
        create_file_sync(TestCreateFile.path)
        data = self.get_file(TestCreateFile.path)
        assert data == expected_data, f"create_file_sync with all iocs\n\tcreates: {data}\n\tinstead: {expected_data}"
        error_msg = warnings.call_args.args[0]
        assert error_msg.startswith(
            "unexpected IOC format in key: '"
        ), f"create_file_sync empty message\n\tstarts: {error_msg}\n\tinstead: unexpected IOC format in key: '"  # noqa: E501
        assert error_msg.endswith(
            f"', {defective_indicator!s}"
        ), f"create_file_sync empty message\n\tends: {error_msg}\n\tinstead: ', {defective_indicator!s}"  # noqa: E501

    def test_create_file_iocs_to_keep_without_iocs(self, mocker):
        """
        Given:
            - iocs to keep command
        When:
            - there is no iocs
        Then:
            - Verify iocs to keep file data.
        """

        mocker.patch.object(demisto, "searchIndicators", return_value={"total": 0})
        create_file_iocs_to_keep(TestCreateFile.path)
        data = self.get_file(TestCreateFile.path)
        expected_data = ""
        assert data == expected_data, f"create_file_iocs_to_keep with no iocs\n\tcreates: {data}\n\tinstead: {expected_data}"

    @pytest.mark.parametrize("in_iocs, out_iocs", data_test_create_file_iocs_to_keep)
    def test_create_file_iocs_to_keep(self, in_iocs, out_iocs, mocker):
        """
        Given:
            - iocs to keep command
        When:
            - iocs type is a specific type.
        Then:
            - Verify iocs to keep file data.
        """
        mocker.patch.object(demisto, "searchIndicators", return_value=json.loads(self.get_file(f"test_data/{in_iocs}.json")))
        create_file_iocs_to_keep(TestCreateFile.path)
        data = self.get_file(TestCreateFile.path)
        expected_data = self.get_file(f"test_data/{out_iocs}.txt")
        assert (
            data == expected_data
        ), f"create_file_iocs_to_keep with {in_iocs} iocs\n\tcreates: {data}\n\tinstead: {expected_data}"  # noqa: E501

    def test_create_file_iocs_to_keep_all_types(self, mocker):
        """
        Given:
            - iocs to keep command
        When:
            - iocs as all types
        Then:
            - Verify iocs to keep file data.
        """
        all_iocs, expected_data = self.get_all_iocs(self.data_test_create_file_iocs_to_keep, "txt")
        mocker.patch.object(demisto, "searchIndicators", return_value=all_iocs)
        create_file_iocs_to_keep(TestCreateFile.path)
        data = self.get_file(TestCreateFile.path)
        assert data == expected_data, f"create_file_iocs_to_keep with all iocs\n\tcreates: {data}\n\tinstead: {expected_data}"


class TestDemistoIOCToCore:
    data_test_demisto_expiration_to_core = [
        (None, -1),
        ("", -1),
        ("0001-01-01T00:00:00Z", -1),
        ("2020-06-03T00:00:00Z", 1591142400000),
    ]

    @pytest.mark.parametrize("demisto_expiration, core_expiration", data_test_demisto_expiration_to_core)
    def test_demisto_expiration_to_core(self, demisto_expiration, core_expiration):
        """
        Given:
            - demisto indicator expiration
        Then:
            - Verify XDR expiration.
        """

        output = demisto_expiration_to_core(demisto_expiration)
        assert (
            core_expiration == output
        ), f"demisto_expiration_to_core({demisto_expiration})\n\treturns: {output}\n\tinstead: {core_expiration}"  # noqa: E501

    data_test_demisto_reliability_to_core = [
        (None, "F"),
        ("A - Completely reliable", "A"),
        ("B - Usually reliable", "B"),
        ("C - Fairly reliable", "C"),
        ("D - Not usually reliable", "D"),
        ("E - Unreliable", "E"),
        ("F - Reliability cannot be judged", "F"),
    ]

    @pytest.mark.parametrize("demisto_reliability, core_reliability", data_test_demisto_reliability_to_core)
    def test_demisto_reliability_to_core(self, demisto_reliability, core_reliability):
        """
        Given:
            - demisto indicator reliability
        Then:
            - Verify XDR reliability.
        """

        output = demisto_reliability_to_core(demisto_reliability)
        assert (
            output == core_reliability
        ), f"demisto_reliability_to_core({demisto_reliability})\n\treturns: {output}\n\tinstead: {core_reliability}"  # noqa: E501

    data_test_demisto_types_to_core = [("File", "HASH"), ("IP", "IP"), ("Domain", "DOMAIN_NAME")]

    @pytest.mark.parametrize("demisto_type, core_type", data_test_demisto_types_to_core)
    def test_demisto_types_to_core(self, demisto_type, core_type):
        """
        Given:
            - demisto indicator type
        Then:
            - Verify XDR type.
        """

        output = demisto_types_to_core(demisto_type)
        assert output == core_type, f"demisto_reliability_to_core({demisto_type})\n\treturns: {output}\n\tinstead: {core_type}"

    data_test_demisto_vendors_to_core = [
        (
            {"moduleID": {"sourceBrand": "test", "reliability": "A - Completely reliable", "score": 2}},
            {"vendor_name": "test", "reputation": "SUSPICIOUS", "reliability": "A"},
        ),
        (
            {"moduleID": {"reliability": "A - Completely reliable", "score": 2}},
            {"vendor_name": "moduleID", "reputation": "SUSPICIOUS", "reliability": "A"},
        ),
        (
            {"moduleID": {"sourceBrand": "test", "score": 2}},
            {"vendor_name": "test", "reputation": "SUSPICIOUS", "reliability": "F"},
        ),
        (
            {"moduleID": {"reliability": "A - Completely reliable", "score": 0}},
            {"vendor_name": "moduleID", "reputation": "UNKNOWN", "reliability": "A"},
        ),
    ]

    @pytest.mark.parametrize("demisto_vendor, core_vendor", data_test_demisto_vendors_to_core)
    def test_demisto_vendors_to_core(self, demisto_vendor, core_vendor):
        """
        Given:
            - demisto indicator vendors reports.
        Then:
            - Verify XDR vendors format.
        """

        output = demisto_vendors_to_core(demisto_vendor)[0]
        assert (
            output == core_vendor
        ), f"demisto_vendors_to_core({demisto_vendor})\n\treturns: {d_sort(output)}\n\tinstead: {d_sort(core_vendor)}"  # noqa: E501

    data_test_demisto_ioc_to_core = [
        (
            {"value": "11.11.11.11", "indicator_type": "IP", "score": 2},
            {"expiration_date": -1, "indicator": "11.11.11.11", "reputation": "SUSPICIOUS", "severity": "INFO", "type": "IP"},
        ),
        (
            {"value": "11.11.11.11", "indicator_type": 100, "score": 2},
            {"expiration_date": -1, "indicator": "11.11.11.11", "reputation": "SUSPICIOUS", "severity": "INFO", "type": "100"},
        ),
        (
            {"value": "11.11.11.11", "indicator_type": "IP"},
            {"expiration_date": -1, "indicator": "11.11.11.11", "reputation": "UNKNOWN", "severity": "INFO", "type": "IP"},
        ),
        (
            {"value": "11.11.11.11", "indicator_type": "IP", "expiration": "2020-06-03T00:00:00Z"},
            {
                "expiration_date": 1591142400000,
                "indicator": "11.11.11.11",
                "reputation": "UNKNOWN",
                "severity": "INFO",
                "type": "IP",
            },  # noqa: E501
        ),
        (
            {
                "value": "11.11.11.11",
                "indicator_type": "IP",
                "comments": [{"type": "IndicatorCommentTimeLine", "content": "test"}],
            },  # noqa: E501
            {"expiration_date": -1, "indicator": "11.11.11.11", "reputation": "UNKNOWN", "severity": "INFO", "type": "IP"},
        ),
        (
            {
                "value": "11.11.11.11",
                "indicator_type": "IP",
                "comments": [{"type": "IndicatorCommentRegular", "content": "test"}],
            },  # noqa: E501
            {
                "expiration_date": -1,
                "indicator": "11.11.11.11",
                "reputation": "UNKNOWN",
                "severity": "INFO",
                "type": "IP",
                "comment": "test",
            },  # noqa: E501
        ),
        (
            {
                "value": "11.11.11.11",
                "indicator_type": "IP",
                "comments": [
                    {"type": "IndicatorCommentRegular", "content": "test"},
                    {"type": "IndicatorCommentRegular", "content": "this is the comment"},
                ],
            },  # noqa: E501
            {
                "expiration_date": -1,
                "indicator": "11.11.11.11",
                "reputation": "UNKNOWN",
                "severity": "INFO",
                "type": "IP",
                "comment": "this is the comment",
            },  # noqa: E501
        ),
        (
            {"value": "11.11.11.11", "indicator_type": "IP", "aggregatedReliability": "A - Completely reliable"},
            {
                "expiration_date": -1,
                "indicator": "11.11.11.11",
                "reputation": "UNKNOWN",
                "severity": "INFO",
                "type": "IP",
                "reliability": "A",
            },  # noqa: E501
        ),
        (
            {"value": "11.11.11.11", "indicator_type": "IP", "CustomFields": {"threattypes": {"threatcategory": "Malware"}}},  # noqa: E501
            {
                "expiration_date": -1,
                "indicator": "11.11.11.11",
                "reputation": "UNKNOWN",
                "severity": "INFO",
                "type": "IP",
                "class": "Malware",
            },  # noqa: E501
        ),
        (
            {"value": "11.11.11.11", "indicator_type": "IP", "moduleToFeedMap": {"module": {"sourceBrand": "test", "score": 2}}},  # noqa: E501
            {
                "expiration_date": -1,
                "indicator": "11.11.11.11",
                "reputation": "UNKNOWN",
                "severity": "INFO",
                "type": "IP",
                "vendors": [{"vendor_name": "test", "reputation": "SUSPICIOUS", "reliability": "F"}],
            },  # noqa: E501
        ),
    ]

    @pytest.mark.parametrize("demisto_ioc, core_ioc", data_test_demisto_ioc_to_core)
    def test_demisto_ioc_to_core(self, demisto_ioc, core_ioc):
        """
        Given:
            - demisto indicator.
        Then:
            - Verify XDR indicator format.
        """

        output = demisto_ioc_to_core(demisto_ioc)
        assert (
            output == core_ioc
        ), f"demisto_ioc_to_core({demisto_ioc})\n\treturns: {d_sort(output)}\n\tinstead: {d_sort(core_ioc)}"  # noqa: E501

    def test_empty_demisto_ioc_to_core(self, mocker):
        warnings = mocker.patch.object(demisto, "debug")
        output = demisto_ioc_to_core({})
        assert output == {}, "demisto_ioc_to_core({})\n\treturns: " + str(d_sort(output)) + "\n\tinstead: {}"
        assert warnings.call_args.args[0] == "unexpected IOC format in key: 'value', {}"


class TestCoreIOCToDemisto:
    data_test_core_expiration_to_demisto = [
        (-1, "Never"),
        (1591142400000, "2020-06-03T00:00:00Z"),
        (1592142400000, "2020-06-14T13:46:40Z"),
    ]

    @pytest.mark.parametrize("core_expiration, demisto_expiration", data_test_core_expiration_to_demisto)
    def test_core_expiration_to_demisto(self, core_expiration, demisto_expiration):
        """
        Given:
            - expiration in XDR format.
        Then:
            - expiration in demisto format.
        """
        output = core_expiration_to_demisto(core_expiration)
        assert (
            output == demisto_expiration
        ), f"core_expiration_to_demisto({core_expiration})\n\treturns: {output}\n\tinstead: {demisto_expiration}"  # noqa: E501

    data_test_core_ioc_to_demisto = [
        (
            {
                "RULE_ID": 863,
                "RULE_INSERT_TIME": 1591165763753,
                "RULE_MODIFY_TIME": 1591166095668,
                "RULE_SEVERITY": "SEV_010_INFO",
                "NUMBER_OF_HITS": 0,
                "RULE_SOURCE": "XSOAR TIM",
                "RULE_COMMENT": "",
                "RULE_STATUS": "DISABLED",
                "BS_STATUS": "DONE",
                "BS_TS": 1591165801230,
                "BS_RETRIES": 1,
                "RULE_EXPIRATION_TIME": -1,
                "IOC_TYPE": "HASH",
                "RULE_INDICATOR": "fa66f1e0e318b6d7b595b6cee580dc0d8e4ac38fbc8dbfcac6ad66dbe282832e",
                "REPUTATION": "GOOD",  # noqa: E501
                "RELIABILITY": None,
                "VENDORS": None,
                "KLASS": None,
                "IS_DEFAULT_TTL": False,
                "RULE_TTL": -1,
                "MARKED_DELETED": 0,
            },
            {
                "value": "fa66f1e0e318b6d7b595b6cee580dc0d8e4ac38fbc8dbfcac6ad66dbe282832e",
                "type": "File",
                "score": 1,
                "fields": {"expirationdate": "Never", "tags": "Cortex Core", "corestatus": "disabled"},
            },
        ),
        (
            {
                "RULE_ID": 861,
                "RULE_INSERT_TIME": 1591165763753,
                "RULE_MODIFY_TIME": 1591166095668,
                "RULE_SEVERITY": "SEV_010_INFO",
                "NUMBER_OF_HITS": 0,
                "RULE_SOURCE": "XSOAR TIM",
                "RULE_COMMENT": "",
                "RULE_STATUS": "DISABLED",
                "BS_STATUS": "DONE",
                "BS_TS": 1591165801784,
                "BS_RETRIES": 1,
                "RULE_EXPIRATION_TIME": -1,
                "IOC_TYPE": "DOMAIN_NAME",
                "RULE_INDICATOR": "test.com",
                "REPUTATION": "GOOD",  # noqa: E501
                "RELIABILITY": None,
                "VENDORS": None,
                "KLASS": None,
                "IS_DEFAULT_TTL": False,
                "RULE_TTL": -1,
                "MARKED_DELETED": 0,
            },
            {
                "value": "test.com",
                "type": "Domain",
                "score": 1,
                "fields": {"expirationdate": "Never", "tags": "Cortex Core", "corestatus": "disabled"},
            },
        ),
        (
            {
                "RULE_ID": 862,
                "RULE_INSERT_TIME": 1591165763753,
                "RULE_MODIFY_TIME": 1591166095668,
                "RULE_SEVERITY": "SEV_010_INFO",
                "NUMBER_OF_HITS": 0,
                "RULE_SOURCE": "XSOAR TIM",
                "RULE_COMMENT": "",
                "RULE_STATUS": "ENABLED",
                "BS_STATUS": "DONE",
                "BS_TS": 1591165801784,
                "BS_RETRIES": 1,
                "RULE_EXPIRATION_TIME": -1,
                "IOC_TYPE": "DOMAIN_NAME",
                "RULE_INDICATOR": "test.co.il",
                "REPUTATION": "SUSPICIOUS",
                "RELIABILITY": "A",
                "VENDORS": [{"vendor_name": "Cortex Core - IOC", "reputation": "SUSPICIOUS", "reliability": "A"}],
                "KLASS": None,
                "IS_DEFAULT_TTL": False,
                "RULE_TTL": -1,
                "MARKED_DELETED": 0,
            },
            {
                "value": "test.co.il",
                "type": "Domain",
                "score": 2,
                "fields": {"expirationdate": "Never", "tags": "Cortex Core", "corestatus": "enabled"},
            },
        ),
    ]


class TestCommands:
    # test commands full flow
    class TestIOCSCommand:
        def test_iocs_command_with_enable(self, mocker):
            """
            Given:
                - enable command
            Then:
                - Verify enable command is called.
            """
            mocker.patch.object(demisto, "command", return_value="core-iocs-enable")
            mocker.patch.object(demisto, "args", return_value={"indicator": "11.11.11.11"})
            mocker.patch("CoreIOCs.Client.http_request", return_value={})
            outputs = mocker.patch("CoreIOCs.return_outputs")
            enable_ioc = mocker.patch("CoreIOCs.prepare_enable_iocs", side_effect=prepare_enable_iocs)
            iocs_command(client)
            output = outputs.call_args.args[0]
            assert (
                output == "indicators 11.11.11.11 enabled."
            ), f"enable command\n\tprints:  {output}\n\tinstead: indicators 11.11.11.11 enabled."  # noqa: E501
            assert enable_ioc.call_count == 1, "enable command not called"

        def test_iocs_command_with_disable(self, mocker):
            """
            Given:
                - disable command
            Then:
                - Verify disable command is called.
            """

            mocker.patch.object(demisto, "command", return_value="core-iocs-disable")
            mocker.patch.object(demisto, "args", return_value={"indicator": "11.11.11.11"})
            mocker.patch("CoreIOCs.Client.http_request", return_value={})
            outputs = mocker.patch("CoreIOCs.return_outputs")
            disable_ioc = mocker.patch("CoreIOCs.prepare_disable_iocs", side_effect=prepare_disable_iocs)
            iocs_command(client)
            output = outputs.call_args.args[0]
            assert (
                output == "indicators 11.11.11.11 disabled."
            ), f"disable command\n\tprints:  {output}\n\tinstead: indicators 11.11.11.11 disabled."  # noqa: E501
            assert disable_ioc.call_count == 1, "disable command not called"

    def test_sync(self, mocker):
        http_request = mocker.patch.object(Client, "http_request", return_value={"reply": True})
        iocs, _ = TestCreateFile.get_all_iocs(TestCreateFile.data_test_create_file_sync, "txt")
        mocker.patch.object(demisto, "searchIndicators", returnvalue=iocs)
        mocker.patch("CoreIOCs.return_outputs")
        mocker.patch("CoreIOCs.upload_file_to_bucket")
        mocker.patch.object(demisto, "setIntegrationContext")
        sync(client)
        assert http_request.call_args.kwargs["url_suffix"] == "sync_tim_iocs", "sync command url changed"

    def test_sync_failure_reply(self, mocker):
        """
        Given:
            - A client and a scenario where the sync http_request returns a reply that is not True.
        When:
            - The sync function is called.
        Then:
            - A DemistoException should be raised with the appropriate error message.
        """
        error_message = "Sync failed due to an internal error."
        mocker.patch.object(Client, "http_request", return_value={"reply": error_message})
        mocker.patch.object(demisto, "searchIndicators", returnvalue={"total": 0, "iocs": []})
        mocker.patch("CoreIOCs.return_outputs")
        mocker.patch("CoreIOCs.upload_file_to_bucket")
        mocker.patch.object(demisto, "setIntegrationContext")

        with pytest.raises(DemistoException) as e:
            sync(client)
        assert str(e.value) == f"Unable to sync IOCs:\n{error_message}"

    def test_get_sync_file(self, mocker):
        iocs, _ = TestCreateFile.get_all_iocs(TestCreateFile.data_test_create_file_sync, "txt")
        mocker.patch.object(demisto, "searchIndicators", returnvalue=iocs)
        return_results_mock = mocker.patch("CoreIOCs.return_results")
        get_sync_file()
        assert return_results_mock.call_args[0][0]["File"] == "core-sync-file"

    def test_set_sync_time(self, mocker):
        mocker_reurn_results = mocker.patch("CoreIOCs.return_results")
        mocker_set_context = mocker.patch.object(demisto, "setIntegrationContext")
        set_sync_time("2021-11-25T00:00:00")
        mocker_reurn_results.assert_called_once_with("Successfully set sync time to 2021-11-25T00:00:00.")
        call_args = mocker_set_context.call_args[0][0]
        assert call_args["ts"] == 1637798400000
        assert call_args["time"] == "2021-11-25T00:00:00Z"
        assert call_args["iocs_to_keep_time"]

    def test_set_sync_time_with_invalid_time(self):
        with pytest.raises(ValueError, match="invalid time format."):
            set_sync_time("test")

    @freeze_time("2020-06-03T02:00:00Z")
    def test_iocs_to_keep(self, mocker):
        http_request = mocker.patch.object(Client, "http_request", return_value={"reply": True})
        iocs, _ = TestCreateFile.get_all_iocs(TestCreateFile.data_test_create_file_iocs_to_keep, "txt")
        mocker.patch.object(demisto, "searchIndicators", returnvalue=iocs)
        mocker.patch("CoreIOCs.upload_file_to_bucket")
        iocs_to_keep(client)
        assert http_request.call_args.kwargs["url_suffix"] == "iocs_to_keep", "iocs_to_keep command url changed"

    def test_tim_insert_jsons(self, mocker):
        http_request = mocker.patch.object(Client, "http_request")
        mocker.patch.object(demisto, "getIntegrationContext", return_value={"time": "2020-06-03T00:00:00Z"})
        iocs, _ = TestCreateFile.get_all_iocs(TestCreateFile.data_test_create_file_sync, "txt")
        mocker.patch.object(demisto, "searchIndicators", return_value=iocs)
        mocker.patch("CoreIOCs.return_outputs")
        tim_insert_jsons(client)
        assert http_request.call_args.kwargs["url_suffix"] == "tim_insert_jsons/", "tim_insert_jsons command url changed"

    @freeze_time("2020-06-03T02:00:00Z")
    def test_tim_insert_jsons_no_time_in_integration_context(self, mocker):
        http_request = mocker.patch.object(Client, "http_request")
        mocker.patch.object(demisto, "getIntegrationContext", return_value={})
        info_logger = mocker.patch.object(demisto, "info")
        iocs, _ = TestCreateFile.get_all_iocs(TestCreateFile.data_test_create_file_sync, "txt")
        mocker.patch.object(demisto, "searchIndicators", return_value=iocs)
        mocker.patch("CoreIOCs.return_outputs")
        tim_insert_jsons(client)
        info_logger.assert_called_with(
            "Could not find 'time' field in integration context, will use from_date='2020-06-02T02:00:00Z'"
        )
        assert http_request.call_args.kwargs["url_suffix"] == "tim_insert_jsons/", "tim_insert_jsons command url changed"


class TestParams:
    tags_test = [
        (
            {"value": "11.11.11.11", "indicator_type": "IP", "score": 2},
            {"expiration_date": -1, "indicator": "11.11.11.11", "reputation": "SUSPICIOUS", "severity": "INFO", "type": "IP"},
            {"tlp_color": ""},
            "Cortex Core",
            None,
        ),
        (
            {"value": "11.11.11.11", "indicator_type": "IP", "score": 2},
            {"expiration_date": -1, "indicator": "11.11.11.11", "reputation": "SUSPICIOUS", "severity": "INFO", "type": "IP"},
            {"tag": "tag1"},
            "tag1",
            None,
        ),
        (
            {"value": "11.11.11.11", "indicator_type": "IP", "score": 2},
            {"expiration_date": -1, "indicator": "11.11.11.11", "reputation": "SUSPICIOUS", "severity": "INFO", "type": "IP"},
            {"feedTags": "tag2", "tlp_color": "AMBER"},
            "tag2",
            "AMBER",
        ),
    ]


def test_file_deleted_for_create_file_sync(mocker):
    file_path = "test"
    mocker.patch("CoreIOCs.get_temp_file", return_value=file_path)
    open(file_path, "w").close()

    def raise_function(*_args, **_kwargs):
        raise DemistoException(file_path)

    mocker.patch("CoreIOCs.create_file_sync", new=raise_function)
    with pytest.raises(DemistoException):
        get_sync_file()
    assert os.path.exists(file_path) is False


data_test_test_file_deleted = [
    (sync, "create_file_sync"),
    (iocs_to_keep, "create_file_iocs_to_keep"),
]


@pytest.mark.parametrize("method_to_test,iner_method", data_test_test_file_deleted)
@freeze_time("2020-06-03T02:00:00Z")
def test_file_deleted(mocker, method_to_test, iner_method):
    file_path = "test"
    mocker.patch("CoreIOCs.get_temp_file", return_value=file_path)
    open(file_path, "w").close()

    def raise_function(*_args, **_kwargs):
        raise DemistoException(file_path)

    mocker.patch(f"CoreIOCs.{iner_method}", new=raise_function)
    with pytest.raises(DemistoException):
        method_to_test(None)
    assert os.path.exists(file_path) is False