from dataclasses import dataclass from unittest.mock import MagicMock import pytest from CommonServerPython import * @dataclass class MockClient: time_field = "calculatedTime" class MockHit: def __init__(self, hit_val): self._hit_val = dict(hit_val) def to_dict(self): return self._hit_val """MOCKED RESPONSES""" CUSTOM_VAL_KEY = "indicatorValue" CUSTOM_TYPE_KEY = "indicatorType" CUSTOM_HIT = {"id": "abc", CUSTOM_VAL_KEY: "5.5.5.5", CUSTOM_TYPE_KEY: "IP"} CUSTOM_HIT_ELASTIC_V7 = {"_source": {"id": "abc", CUSTOM_VAL_KEY: "5.5.5.5", CUSTOM_TYPE_KEY: "IP"}} INSIGHT_HIT_ELASTIC_V7 = {"_id": "123", "_source": {"name": "5.5.5.5", "type": "IP"}} PARSED_INSIGHT_HIT_ELASTIC_V7 = { "id": "123", "name": "5.5.5.5", "type": "IP", "value": "5.5.5.5", "rawJSON": {"id": "123", "name": "5.5.5.5", "type": "IP", "value": "5.5.5.5"}, "fields": {}, } PARSED_CUSTOM_HIT = { "id": "abc", "indicatorValue": "5.5.5.5", "indicatorType": "IP", "value": "5.5.5.5", "rawJSON": {"id": "abc", "indicatorValue": "5.5.5.5", "indicatorType": "IP", "value": "5.5.5.5"}, "type": "IP", "fields": {"tags": ["tag1", "tag2"], "trafficlightprotocol": "AMBER"}, } PARSED_INSIGHT_HIT = { "id": "1d5920f4b44b27a802bd77c4f0536f5a", "version": 3, "modified": "2020-01-26T14:16:44.641927Z", "sortValues": None, "account": "acc1", "type": "Domain", "value": "google.com", "rawName": "google.com", "createdTime": "2020-01-26T16:16:18.801688+02:00", "investigationIDs": ["57ec1eb4-454e-4561-8059-a9beb3f830c0"], "investigationsCount": 1, "sourceInstances": ["VirusTotal"], "sourceBrands": ["VirusTotal"], "isIoc": True, "lastSeen": "2020-01-26T16:16:18.801508+02:00", "firstSeen": "2020-01-26T16:16:18.801509+02:00", "lastSeenEntryID": "4@57ec1eb4-454e-4561-8059-a9beb3f830c0", "firstSeenEntryID": "4@57ec1eb4-454e-4561-8059-a9beb3f830c0", "lastReputationRun": "2020-01-26T16:16:13.219824+02:00", "isShared": True, "calculatedTime": "2020-01-26T16:16:18.801508+02:00", "score": 1, "manualSetTime": "0001-01-01T00:00:00Z", "context": [], "comment": "", "CustomFields": None, "manuallyEditedFields": None, "modifiedTime": "2020-01-26T16:16:09.855733+02:00", "moduleToFeedMap": { "VirusTotal.VirusTotal": { "reliability": "A+ - 3rd party enrichment", "rawJSON": None, "fetchTime": "2020-01-26T16:16:09.855733+02:00", "sourceBrand": "VirusTotal", "sourceInstance": "VirusTotal", "expirationPolicy": "indicatorType", "expirationInterval": 0, "expiration": "0001-01-01T00:00:00Z", "ExpirationSource": None, "bypassExclusionList": False, "type": "domain", "value": "google.com", "score": 1, "timestamp": "0001-01-01T00:00:00Z", "lastSeen": "0001-01-01T00:00:00Z", "firstSeen": "0001-01-01T00:00:00Z", "CustomFields": None, "modifiedTime": "0001-01-01T00:00:00Z", "isEnrichment": True, }, "Whois.Whois": { "reliability": "A+ - 3rd party enrichment", "rawJSON": None, "fetchTime": "2020-01-26T16:16:09.855733+02:00", "sourceBrand": "VirusTotal", "sourceInstance": "VirusTotal", "expirationPolicy": "indicatorType", "expirationInterval": 0, "expiration": "0001-01-01T00:00:00Z", "ExpirationSource": None, "bypassExclusionList": False, "type": "domain", "value": "google.com", "score": 1, "timestamp": "0001-01-01T00:00:00Z", "lastSeen": "0001-01-01T00:00:00Z", "firstSeen": "0001-01-01T00:00:00Z", "CustomFields": None, "modifiedTime": "0001-01-01T00:00:00Z", "isEnrichment": True, }, "Demisto.Demisto": { "reliability": "A+ - 3rd party enrichment", "rawJSON": None, "fetchTime": "2020-01-26T16:16:09.855733+02:00", "sourceBrand": "VirusTotal", "sourceInstance": "VirusTotal", "expirationPolicy": "indicatorType", "expirationInterval": 0, "expiration": "0001-01-01T00:00:00Z", "ExpirationSource": None, "bypassExclusionList": False, "type": "domain", "value": "google.com", "score": 1, "timestamp": "0001-01-01T00:00:00Z", "lastSeen": "0001-01-01T00:00:00Z", "firstSeen": "0001-01-01T00:00:00Z", "CustomFields": None, "modifiedTime": "0001-01-01T00:00:00Z", "isEnrichment": False, }, }, "expiration": "0001-01-01T00:00:00Z", "expirationStatus": "active", "expirationSource": None, "fields": {"tags": ["tag1", "tag2"], "trafficlightprotocol": "AMBER"}, } FEED_IOC_KEYS = ( "rawJSON", "fetchTime", "sourceBrand", "sourceInstance", "expirationPolicy", "expirationInterval", "expiration", "ExpirationSource", "bypassExclusionList", "type", "value", "score", "timestamp", "lastSeen", "firstSeen", "CustomFields", "modifiedTime", "isEnrichment", "fields", ) def test_hit_to_indicator(): import FeedElasticsearch as esf ioc = esf.hit_to_indicator(MockHit(CUSTOM_HIT), CUSTOM_VAL_KEY, CUSTOM_TYPE_KEY, None, ["tag1", "tag2"], "AMBER") assert ioc == PARSED_CUSTOM_HIT no_type_hit = dict(CUSTOM_HIT) no_type_hit[CUSTOM_TYPE_KEY] = "" ioc = esf.hit_to_indicator(MockHit(no_type_hit), CUSTOM_VAL_KEY, CUSTOM_TYPE_KEY, "IP", ["tag1", "tag2"], "AMBER") assert ioc["type"] == "IP" assert ioc[CUSTOM_TYPE_KEY] == "" ioc = esf.hit_to_indicator(MockHit(CUSTOM_HIT), CUSTOM_VAL_KEY, "", "URL", ["tag1", "tag2"], "AMBER") assert ioc["type"] == "URL" def test_hit_to_indicator_enrichment_excluded(): """ Given: - The `hit_to_indicator` function in the `FeedElasticsearch` module is used to convert a hit to an indicator. When: - Enrichment excluded is True Then: - 'enrichmentExcluded' = True should be added to the indicator. """ import FeedElasticsearch as esf ioc = esf.hit_to_indicator( MockHit(CUSTOM_HIT), CUSTOM_VAL_KEY, CUSTOM_TYPE_KEY, None, ["tag1", "tag2"], "AMBER", enrichment_excluded=True ) assert ioc.pop("enrichmentExcluded") assert ioc == PARSED_CUSTOM_HIT def test_hit_to_indicator_custom_elastic_v7(mocker): """ Background: To maintain backwards compatibility for elastic server v7 and below, we have changed several places in the code where instead of using the elasticsearch_dsl library (which is compatible to versions >= 8 we use the elasticsearch library directly). In some cases those code changes caused slight differences in API responses structure. In this test we check that the 'hit_to_indicator' function handles the generic (custom) hit object structure we get when searching for indicators hits in Elasticsearch server v7. Given: - Elasticsearch client from type 'Elasticsearch' (v7 client), and a mock response of a generic (custom) hit object we get by querying elasticsearch v7 server. When: - Running the 'hit_to_indicator' function. Then: - Make sure that the parsed hit is as expected. """ params: dict = {"client_type": "ElasticSearch"} mocker.patch.object(demisto, "params", return_value=params) import FeedElasticsearch as esf ioc = esf.hit_to_indicator(CUSTOM_HIT_ELASTIC_V7, CUSTOM_VAL_KEY, CUSTOM_TYPE_KEY, None, ["tag1", "tag2"], "AMBER") assert ioc == PARSED_CUSTOM_HIT def test_hit_to_indicator_insight_hit_v7(): """ Background: To maintain backwards compatibility for elastic server v7 and below, we have changed several places in the code where instead of using the elasticsearch_dsl library (which is compatible to versions >= 8 we use the elasticsearch library directly). In some cases those code changes caused slight differences in API responses structure. In this test we check that the 'hit_to_indicator' function handles the demisto (insight) hit object structure we get when searching for indicators hits in Elasticsearch server v7. Given: - Elasticsearch client from type 'Elasticsearch' (v7 client), and a mock response of a demisto (insight) hit object we get by querying elasticsearch v7 server. When: - Running the 'hit_to_indicator' function. Then: - Make sure that the parsed hit is as expected. """ import FeedElasticsearch as esf ioc = esf.hit_to_indicator(INSIGHT_HIT_ELASTIC_V7) assert ioc == PARSED_INSIGHT_HIT_ELASTIC_V7 def test_extract_indicators_from_insight_hit2(mocker): params: dict = {"client_type": "OpenSearch"} mocker.patch.object(demisto, "params", return_value=params) import FeedElasticsearch as esf mocker.patch.object(esf, "hit_to_indicator", return_value=dict(PARSED_INSIGHT_HIT)) ioc_lst, ioc_enrch_lst = esf.extract_indicators_from_insight_hit(PARSED_INSIGHT_HIT, ["tag1", "tag2"], "AMBER") # moduleToFeedMap with isEnrichment: False should not be added to ioc_lst assert len(ioc_lst) == 1 assert len(ioc_enrch_lst[0]) == 2 assert ioc_lst[0].get("value") # moduleToFeedMap with isEnrichment: False should be added to ioc_lst assert ioc_lst[0].get("moduleToFeedMap").get("Demisto.Demisto") assert ioc_lst[0].get("moduleToFeedMap").get("VirusTotal.VirusTotal") is None set(FEED_IOC_KEYS).issubset(ioc_enrch_lst[0][0]) set(FEED_IOC_KEYS).issubset(ioc_enrch_lst[0][1]) def test_extract_indicators_from_generic_hit(mocker): import FeedElasticsearch as esf mocker.patch.object(esf, "hit_to_indicator", return_value=PARSED_CUSTOM_HIT) ioc_lst = esf.extract_indicators_from_generic_hit( CUSTOM_HIT, CUSTOM_VAL_KEY, CUSTOM_TYPE_KEY, None, ["tag1", "tag2"], "AMBER" ) assert ioc_lst == [PARSED_CUSTOM_HIT] def test_create_enrichment_batches_one_indicator(mocker): import FeedElasticsearch as esf mocker.patch.object(esf, "hit_to_indicator", return_value=PARSED_INSIGHT_HIT) _, ioc_enrch_lst = esf.extract_indicators_from_insight_hit(PARSED_INSIGHT_HIT, ["tag1", "tag2"], "AMBER") ioc_enrch_lst_of_lsts = esf.create_enrichment_batches(ioc_enrch_lst) assert len(ioc_enrch_lst_of_lsts) == 2 assert ioc_enrch_lst_of_lsts[0][0] == ioc_enrch_lst[0][0] assert ioc_enrch_lst_of_lsts[1][0] == ioc_enrch_lst[0][1] def test_create_enrichment_batches_mult_indicators(): import FeedElasticsearch as esf ioc_enrch_lst = [[1, 2, 3], [4, 5], [6, 7, 8, 9]] ioc_enrch_lst_of_lsts = esf.create_enrichment_batches(ioc_enrch_lst) assert len(ioc_enrch_lst_of_lsts) == 4 assert ioc_enrch_lst_of_lsts[0] == [1, 4, 6] assert ioc_enrch_lst_of_lsts[1] == [2, 5, 7] assert ioc_enrch_lst_of_lsts[2] == [3, 8] assert ioc_enrch_lst_of_lsts[3] == [9] def test_elasticsearch_builder_called_with_username_password(mocker): """ Given: - basic authentication parameters are provided (username and password) - Client type is Elasticsearch (elastic v7) When: - creating an Elasticsearch client Then: - ensure the client is created with the correct parameters """ import FeedElasticsearch as esf mocker.patch("FeedElasticsearch.ELASTIC_SEARCH_CLIENT", new="Elasticsearch") es_mock = mocker.patch.object(esf.Elasticsearch, "__init__", return_value=None) username = "demisto" password = "mock" esf.ElasticsearchClient(username=username, password=password) assert es_mock.call_args[1].get("http_auth") == ("demisto", "mock") assert es_mock.call_args[1].get("api_key") is None def test_elasticsearch_builder_called_with_username_password_elastic_v8(mocker): """ In elastic version >= 8, basic auth params are transferred through the 'basic auth' key instead of the 'http_auth' key. This test check that the parameters are transferred correctly. Given: - basic authentication parameters are provided (username and password) - Client type is Elasticsearch_v8 When: - creating an Elasticsearch client Then: - ensure the client is created with the correct parameters """ import FeedElasticsearch as esf mocker.patch("FeedElasticsearch.ELASTIC_SEARCH_CLIENT", new="Elasticsearch_v8") es_mock = mocker.patch.object(esf.Elasticsearch, "__init__", return_value=None) username = "demisto" password = "mock" esf.ElasticsearchClient(username=username, password=password) assert es_mock.call_args[1].get("basic_auth") == ("demisto", "mock") assert es_mock.call_args[1].get("http_auth") is None assert es_mock.call_args[1].get("api_key") is None def test_elasticsearch_builder_called_with_api_key(mocker): """ Given: - api key authentication parameters are provided (api key id and api key) When: - creating an Elasticsearch client Then: - ensure the client is created with the correct parameters """ import FeedElasticsearch as esf es_mock = mocker.patch.object(esf.Elasticsearch, "__init__", return_value=None) api_id = "demisto" api_key = "mock" esf.ElasticsearchClient(api_key=api_key, api_id=api_id) assert es_mock.call_args[1].get("http_auth") is None assert es_mock.call_args[1].get("api_key") == (api_id, api_key) def test_elasticsearch_builder_called_with_no_creds(mocker): """ Given: - no authentication parameter are provided When: - creating an Elasticsearch client Then: - ensure the client is created with the correct parameters (edge this, this use-case should not happen as '401 Unauthorized - Incorrect or invalid username or password' message will be returned """ import FeedElasticsearch as esf es_mock = mocker.patch.object(esf.Elasticsearch, "__init__", return_value=None) esf.ElasticsearchClient() assert es_mock.call_args[1].get("http_auth") is None assert es_mock.call_args[1].get("api_key") is None def test_extract_api_from_username_password_empty(): import FeedElasticsearch as esf assert esf.extract_api_from_username_password(None, None) == (None, None) def test_extract_api_from_username_password_username_username(): import FeedElasticsearch as esf assert esf.extract_api_from_username_password("username", "password") == (None, None) def test_extract_api_from_username_password_username_api_key(): import FeedElasticsearch as esf username = esf.API_KEY_PREFIX + "api_id" assert esf.extract_api_from_username_password(username, "api_key") == ("api_id", "api_key") def test_last_run(): from FeedElasticsearch import update_last_fetch ioc_lst = [ {"id": "1", "calculatedTime": "2023-01-17T14:30:00.000Z"}, {"id": "2", "calculatedTime": "2023-01-17T14:32:00.000Z"}, {"id": "3", "calculatedTime": "2023-01-17T14:33:00.000Z"}, {"id": "4", "calculatedTime": "2023-01-17T14:33:00.000Z"}, ] last_update, last_ids = update_last_fetch(MockClient(), ioc_lst) assert set(last_ids) == {"4", "3"} assert datetime.fromtimestamp(last_update // 1000).isoformat() == "2023-01-17T14:33:00" @pytest.mark.parametrize("client_version", [("Elasticsearch"), ("OpenSearch"), ("Elasticsearch_v8")]) def test_get_indicators_by_elastic_version_generic_feed(mocker, client_version): """ This test makes sure that the right get generic indicators functions are called in accordance to the elasticsearch client version. Given: - Elasticsearch client of type: 1. Elasticsearch (elastic version <= 7) 2. OpenSearch 3. Elasticsearch_v8 When: - Running the 'get_indicators_command'. Then: - Verify that the right get indicators inner function is being called: 1. The 'get_generic_indicators_elastic_v7' is called. 2. The 'get_generic_indicators' is called. 3. The 'get_generic_indicators' is called. """ import FeedElasticsearch as esf mocker.patch("FeedElasticsearch.ELASTIC_SEARCH_CLIENT", new=client_version) mocker.patch.object(esf.Elasticsearch, "__init__", return_value=None) mocker.patch.object(esf, "get_scan_generic_format", return_value=None) mock_get_generic_indicators = mocker.patch.object(esf, "get_generic_indicators", return_value=None) mock_get_generic_indicators_elastic_v7 = mocker.patch.object(esf, "get_generic_indicators_elastic_v7", return_value=None) client = esf.ElasticsearchClient() esf.get_indicators_command( client, feed_type="Generic Feed", src_val="indicatorValue", src_type="indicatorType", default_type="IP" ) if client_version in ["Elasticsearch_v8", "OpenSearch"]: assert mock_get_generic_indicators.call_count == 1 assert mock_get_generic_indicators_elastic_v7.call_count == 0 else: # Elasticsearch v7 and below assert mock_get_generic_indicators.call_count == 0 assert mock_get_generic_indicators_elastic_v7.call_count == 1 @pytest.mark.parametrize("client_version", [("Elasticsearch"), ("OpenSearch"), ("Elasticsearch_v8")]) def test_get_indicators_by_elastic_version_demisto_feed(mocker, client_version): """ This test makes sure that the right get demisto (insight) indicators functions are called in accordance to the elasticsearch client version. Given: - Elasticsearch client of type: 1. Elasticsearch (elastic version <= 7) 2. OpenSearch 3. Elasticsearch_v8 When: - Running the 'get_indicators_command'. Then: - Verify that the right get indicators inner function is being called: 1. The 'get_demisto_indicators_elastic_v7' is called. 2. The 'get_demisto_indicators' is called. 3. The 'get_demisto_indicators' is called. """ import FeedElasticsearch as esf mocker.patch("FeedElasticsearch.ELASTIC_SEARCH_CLIENT", new=client_version) mocker.patch.object(esf.Elasticsearch, "__init__", return_value=None) mocker.patch.object(esf, "get_scan_insight_format", return_value=None) mock_get_demisto_indicators = mocker.patch.object(esf, "get_demisto_indicators", return_value=([], [])) mock_get_demisto_indicators_elastic_v7 = mocker.patch.object(esf, "get_demisto_indicators_elastic_v7", return_value=([], [])) client = esf.ElasticsearchClient() esf.get_indicators_command(client, feed_type="Cortex XSOAR MT Shared Feed", src_val="", src_type="", default_type="") if client_version in ["Elasticsearch_v8", "OpenSearch"]: assert mock_get_demisto_indicators.call_count == 1 assert mock_get_demisto_indicators_elastic_v7.call_count == 0 else: # Elasticsearch v7 and below assert mock_get_demisto_indicators.call_count == 0 assert mock_get_demisto_indicators_elastic_v7.call_count == 1 @pytest.mark.parametrize("client_version", [("Elasticsearch"), ("OpenSearch"), ("Elasticsearch_v8")]) def test_fetch_indicators_by_elastic_version(mocker, client_version): """ This test makes sure that the right fetch indicators functions are called in accordance to the elasticsearch client version. Given: - Elasticsearch client of type: 1. Elasticsearch (elastic version <= 7) 2. OpenSearch 3. Elasticsearch_v8 When: - Running the 'fetch_indicators_command'. Then: - Verify that the right get indicators inner function is being called: 1. The 'fetch_indicators_elastic_v7' is called. 2. The 'fetch_indicators' is called. 3. The 'fetch_indicators' is called. """ import FeedElasticsearch as esf mocker.patch("FeedElasticsearch.ELASTIC_SEARCH_CLIENT", new=client_version) mocker.patch.object(esf.Elasticsearch, "__init__", return_value=None) mocker.patch.object(esf, "get_last_fetch_timestamp", return_value=None) mock_fetch_indicators = mocker.patch.object(esf, "fetch_indicators", return_value=([], [])) mock_fetch_indicators_elastic_v7 = mocker.patch.object(esf, "fetch_indicators_elastic_v7", return_value=([], [])) client = esf.ElasticsearchClient() esf.fetch_indicators_command( client, feed_type="Generic Feed", src_val="indicatorValue", src_type="indicatorType", default_type="IP", last_fetch={}, fetch_limit=1000, ) if client_version in ["Elasticsearch_v8", "OpenSearch"]: assert mock_fetch_indicators.call_count == 1 assert mock_fetch_indicators_elastic_v7.call_count == 0 else: # Elasticsearch v7 and below assert mock_fetch_indicators.call_count == 0 assert mock_fetch_indicators_elastic_v7.call_count == 1 def test_get_demisto_indicators_elastic_v7(mocker): """ Tests the 'get_demisto_indicators_elastic_v7' function's logic. Given: - Elasticsearch client of type Elasticsearch (v7 and below), a mocked response of the FeedElasticsearch.scan api call (generator). When: - Running the 'get_demisto_indicators_elastic_v7'. Then: - Verify that the indicators' list is as expected. """ import FeedElasticsearch as esf mocked_scan_indicators_res_e7 = [ { "_index": "index_name", "_type": "_doc", "_id": "1", "_score": None, "_source": { "name": "7.7.7.7", "calculatedTime": "2020-01-12T15:29:01.270228+02:00", "id": "1", "type": "IP", }, "sort": [3], } ] mocked_generator = MagicMock() mocked_generator.return_value = (x for x in mocked_scan_indicators_res_e7) mocker.patch("FeedElasticsearch.scan", return_value=mocked_generator.return_value) mocker.patch.object(esf.Elasticsearch, "__init__", return_value=None) client = esf.ElasticsearchClient() client.time_field = "calculatedTime" search = esf.get_scan_insight_format(client, feed_type="Cortex XSOAR MT Shared Feed") ioc_lst, _ = esf.get_demisto_indicators_elastic_v7(client, search, None, None, None) assert len(ioc_lst) == 1 assert ioc_lst[0]["name"] == "7.7.7.7" assert ioc_lst[0]["id"] == "1" assert ioc_lst[0]["calculatedTime"] == "2020-01-12T15:29:01.270228+02:00" assert ioc_lst[0]["type"] == "IP" assert ioc_lst[0]["value"] == "7.7.7.7" def test_get_generic_indicators_elastic_v7(mocker): """ Tests the 'get_generic_indicators_elastic_v7' function's logic. Given: - Elasticsearch client of type Elasticsearch (v7 and below), a mocked response of the FeedElasticsearch.scan api call (generator). When: - Running the 'get_generic_indicators_elastic_v7'. Then: - Verify that the indicators' list is as expected. """ import FeedElasticsearch as esf mocked_scan_indicators_res_e7 = [ { "_index": "index_name", "_type": "_doc", "_id": "1", "_score": None, "_source": { "indicatorValue": "1.1.1.1", "date": "2020-01-12T15:29:01.270228+02:00", "id": "1", "indicatorType": "IP", }, "sort": [3], } ] mocked_generator = MagicMock() mocked_generator.return_value = (x for x in mocked_scan_indicators_res_e7) mocker.patch("FeedElasticsearch.scan", return_value=mocked_generator.return_value) mocker.patch.object(esf.Elasticsearch, "__init__", return_value=None) client = esf.ElasticsearchClient() client.time_field = "date" search = esf.get_scan_generic_format(client) ioc_lst = esf.get_generic_indicators_elastic_v7(client, search, "indicatorValue", "indicatorType", "IP", None, None, None) assert len(ioc_lst) == 1 assert ioc_lst[0]["indicatorValue"] == "1.1.1.1" assert ioc_lst[0]["id"] == "1" assert ioc_lst[0]["date"] == "2020-01-12T15:29:01.270228+02:00" assert ioc_lst[0]["indicatorType"] == "IP" assert ioc_lst[0]["value"] == "1.1.1.1" def test_fetch_demisto_indicators_elastic_v7(mocker): """ Tests the 'fetch_indicators_elastic_v7' function's logic for feed type - Demisto Feed. Given: - Elasticsearch client of type Elasticsearch (v7 and below), a mocked response of the FeedElasticsearch.search api call. When: - Running the 'fetch_indicators_elastic_v7'. Then: - Verify that the indicators' list is as expected. """ import FeedElasticsearch as esf mocked_search_indicators_res_e7 = { "took": 1, "timed_out": False, "_shards": {"total": 1, "successful": 1, "skipped": 0, "failed": 0}, "hits": { "total": {"value": 1, "relation": "eq"}, "max_score": None, "hits": [ { "_index": "index_name", "_type": "_doc", "_id": "1", "_score": None, "_source": { "id": "1", "version": 1, "modified": "2020-01-12T13:27:02.270302Z", "sortValues": None, "comments": [], "account": "", "type": "IP", "name": "1.1.1.1", "value": "1.1.1.1", "rawName": "1.1.1.1", "createdTime": "2020-01-12T15:27:02.270303+02:00", "investigationIDs": [], "investigationsCount": 0, "isIoc": True, "lastSeen": "2020-01-12T15:27:02.270228+02:00", "firstSeen": "2020-01-12T15:27:02.270228+02:00", "lastSeenEntryID": "API", "firstSeenEntryID": "API", "lastReputationRun": "0001-01-01T00:00:00Z", "isShared": False, "calculatedTime": "2020-01-12T15:27:02.270228+02:00", "manualSetTime": "0001-01-01T00:00:00Z", "context": None, "comment": "", "CustomFields": {"internal": False}, "ManuallyEditedFields": None, "modifiedTime": "0001-01-01T00:00:00Z", "expiration": "0001-01-01T00:00:00Z", "expirationStatus": "active", "expirationSource": { "setTime": "2020-01-12T15:27:02.27023+02:00", "source": "indicatorType", "user": "", "feedId": "", "expirationPolicy": "never", "expirationInterval": 0, }, }, "sort": [1578835622270], }, ], }, } mocker.patch.object(esf.Elasticsearch, "__init__", return_value=None) mocker.patch.object(esf.Elasticsearch, "search", return_value=mocked_search_indicators_res_e7) client = esf.ElasticsearchClient() client.time_field = "calculatedTime" client.fetch_index = "test" ioc_lst, _ = esf.fetch_indicators_elastic_v7( client, last_fetch_timestamp="", feed_type="Cortex XSOAR MT Shared Feed", fetch_limit=10000, src_type=None, src_val=None, default_type=None, ) assert len(ioc_lst) == 1 assert ioc_lst[0]["name"] == "1.1.1.1" assert ioc_lst[0]["id"] == "1" assert ioc_lst[0]["calculatedTime"] == "2020-01-12T15:27:02.270228+02:00" assert ioc_lst[0]["type"] == "IP" assert ioc_lst[0]["value"] == "1.1.1.1" def test_fetch_generic_indicators_elastic_v7_with_time_field(mocker): """ Tests the 'fetch_indicators_elastic_v7' function's logic for feed type - Generic Feed, when the user set a custom time field. When a time field is set, we fetch by searching for indicators in relation to the last fetch time which is determined by the time field values of the indicators that were fetched in the last cycle. Given: - Elasticsearch client of type Elasticsearch (v7 and below), a mocked response of the FeedElasticsearch.search api call. client.time_field is set and equals "date". When: - Running the 'fetch_indicators_elastic_v7'. Then: - Verify that the indicators' list is as expected. """ import FeedElasticsearch as esf mocked_search_indicators_res_e7 = { "took": 0, "timed_out": False, "_shards": {"total": 1, "successful": 1, "skipped": 0, "failed": 0}, "hits": { "total": {"value": 2, "relation": "eq"}, "max_score": None, "hits": [ { "_index": "test", "_type": "_doc", "_id": "1111", "_score": None, "_source": { "indicatorValue": "https://www.test.com/", "indicatorType": "URL", "date": "2024-10-14T16:03:45.735577", "id": "1111", }, "sort": [1728921825735], }, { "_index": "test", "_type": "_doc", "_id": "2222", "_score": None, "_source": { "indicatorValue": "1.1.1.1", "indicatorType": "IP", "date": "2024-10-14T16:05:55.735588", "id": "2222", }, "sort": [1728921955735], }, ], }, } mocker.patch.object(esf.Elasticsearch, "__init__", return_value=None) mocker.patch.object(esf.Elasticsearch, "search", return_value=mocked_search_indicators_res_e7) client = esf.ElasticsearchClient() client.time_field = "date" client.fetch_index = "test" ioc_lst, _ = esf.fetch_indicators_elastic_v7( client, last_fetch_timestamp="", feed_type="Generic Feed", fetch_limit=10000, src_type="indicatorType", src_val="indicatorValue", default_type="IP", ) assert len(ioc_lst) == 2 assert ioc_lst[0]["indicatorValue"] == "https://www.test.com/" assert ioc_lst[0]["id"] == "1111" assert ioc_lst[0]["date"] == "2024-10-14T16:03:45.735577" assert ioc_lst[0]["indicatorType"] == "URL" assert ioc_lst[0]["value"] == "https://www.test.com/" assert ioc_lst[1]["indicatorValue"] == "1.1.1.1" assert ioc_lst[1]["id"] == "2222" assert ioc_lst[1]["date"] == "2024-10-14T16:05:55.735588" assert ioc_lst[1]["indicatorType"] == "IP" assert ioc_lst[1]["value"] == "1.1.1.1" def test_fetch_generic_indicators_elastic_v7_without_time_field(mocker): """ Tests the 'fetch_indicators_elastic_v7' function's logic for feed type - Generic Feed, without a custom time field. When a time field isn't set, we fetch by scanning for all indicators regardless last fetch time (in every fetch cycle). Given: - Elasticsearch client of type Elasticsearch (v7 and below), a mocked response of the FeedElasticsearch.scan api call. client.time_field isn't set. When: - Running the 'fetch_indicators_elastic_v7'. Then: - Verify that the indicators' list is as expected. """ import FeedElasticsearch as esf mocked_scan_indicators_res_e7 = [ { "_index": "test", "_type": "_doc", "_id": "1", "_score": None, "_source": { "indicatorValue": "1.1.1.1", "date": "2020-01-12T15:29:01.270228+02:00", "id": "1", "indicatorType": "IP", }, "sort": [3], }, { "_index": "test", "_type": "_doc", "_id": "2", "_score": None, "_source": { "indicatorValue": "https://www.test.com/", "date": "2020-01-12T15:29:01.270228+02:00", "id": "2", "indicatorType": "URL", }, "sort": [3], }, ] mocker.patch.object(esf.Elasticsearch, "__init__", return_value=None) mocked_generator = MagicMock() mocked_generator.return_value = (x for x in mocked_scan_indicators_res_e7) mocker.patch("FeedElasticsearch.scan", return_value=mocked_generator.return_value) client = esf.ElasticsearchClient() client.fetch_index = "test" ioc_lst, _ = esf.fetch_indicators_elastic_v7( client, last_fetch_timestamp="", feed_type="Generic Feed", fetch_limit=10000, src_type="indicatorType", src_val="indicatorValue", default_type="IP", ) assert len(ioc_lst) == 2 assert ioc_lst[0]["indicatorValue"] == "1.1.1.1" assert ioc_lst[0]["id"] == "1" assert ioc_lst[0]["date"] == "2020-01-12T15:29:01.270228+02:00" assert ioc_lst[0]["indicatorType"] == "IP" assert ioc_lst[0]["value"] == "1.1.1.1" assert ioc_lst[1]["indicatorValue"] == "https://www.test.com/" assert ioc_lst[1]["id"] == "2" assert ioc_lst[1]["date"] == "2020-01-12T15:29:01.270228+02:00" assert ioc_lst[1]["indicatorType"] == "URL" assert ioc_lst[1]["value"] == "https://www.test.com/" @pytest.mark.parametrize( "server_details, server_version, client_version", [ ( { "name": "test1", "cluster_name": "elasticsearch", "cluster_uuid": "test_id", "version": { "number": "7.3.0", }, }, "7.3.0", "Elasticsearch_v8", ), ( { "name": "test2", "cluster_name": "elasticsearch", "cluster_uuid": "test_id", "version": { "number": "8.4.1", }, }, "8.4.1", "Elasticsearch", ), ], ids=[ "Test miss configuration error - server version is 7 while client version is 8", "Test miss configuration error - server version is 8 while client version is 7", ], ) def test_verify_es_server_version_errors(mocker, server_details, server_version, client_version): """ Tests the 'verify_es_server_version' function's logic. Given 1. Elastic search server details (response json of the requests.get) - server version is 7.3.0. Integration parameter - client type - is set to 'Elasticsearch_v8. 2. Elastic search server details (response json of the requests.get) - server version is 8.4.1. Integration parameter - client type - is set to 'Elasticsearch. (v7 and below) When - Running the verify_es_server_version function. Then - Make sure that the expected error message is raised. """ import FeedElasticsearch as esf mocker.patch("FeedElasticsearch.ELASTIC_SEARCH_CLIENT", new=client_version) with pytest.raises(ValueError) as e: esf.verify_es_server_version(server_details) assert server_version in str(e.value) def test_feed_main_enrichment_excluded(mocker): """ Given: params with tlp_color set to RED and enrichmentExcluded set to False When: Calling feed_main Then: validate enrichment_excluded is set to True """ from FeedElasticsearch import main params = {"tlp_color": "RED", "enrichmentExcluded": False, "time_field": "test", "feed_type": ["test"]} client_mocker = mocker.patch("FeedElasticsearch.ElasticsearchClient") mocker.patch("FeedElasticsearch.extract_api_from_username_password", return_value=("test", "test")) mocker.patch("FeedElasticsearch.is_xsiam_or_xsoar_saas", return_value=True) mocker.patch.object(demisto, "params", return_value=params) # Call the function under test main() # Assertion - verify that enrichment_excluded is set to True assert client_mocker.call_args_list[0].args[-1] is True