import json from unittest.mock import Mock import pytest import demistomock as demisto from CommonServerPython import DemistoException, FeedIndicatorType, ThreatIntel from FeedMISP import ( Client, build_indicators, build_indicators_from_galaxies, build_indicators_iterator, build_params_dict, fetch_attributes_command, get_galaxy_indicator_type, get_ip_type, handle_file_type_fields, parsing_user_query, update_indicator_fields, ) def test_build_indicators_iterator_success(): """ Given - A list of attributes returned from MISP When - Attributes are well formed Then - Create an iterator of indicators """ attributes = { "response": { "Attribute": [ { "id": "123", "type": "sha256", "value": "123456789", }, ] } } attributes_iterator = build_indicators_iterator(attributes, "some_url") assert len(attributes_iterator) == 1 assert attributes_iterator[0]["value"] == attributes["response"]["Attribute"][0] def test_build_indicators_iterator_fail(): """ Given - A list of attributes returned from MISP When - Attributes are not well formed Then - Raise a KeyError """ with pytest.raises(KeyError): attributes = {"wrong_key": "some_value"} build_indicators_iterator(attributes, "some_url") def test_handle_file_type_fields_simple_hash(): """ Given - Indicator created from an attribute of a hash When - Indicator contains only hash value Then - Set the hash value as a field of a matching hash name """ raw_type = "sha256" indicator_obj = {"value": "somehashvalue", "type": "File", "service": "MISP", "fields": {}} handle_file_type_fields(raw_type, indicator_obj) assert indicator_obj["fields"]["SHA256"] == "somehashvalue" def test_handle_file_type_fields_hash_and_filename(): """ Given - Indicator created from an attribute of a hash and filename When - Indicator contains both hash value and a filename Then - Set the hash value as a field of a matching hash name, update the value of the indicator to the hash value and add 'Associated File Names' field with the filename """ raw_type = "filename|sha256" indicator_obj = {"value": "file.exe|somehashvalue", "type": "File", "service": "MISP", "fields": {}} handle_file_type_fields(raw_type, indicator_obj) assert indicator_obj["fields"]["SHA256"] == "somehashvalue" assert indicator_obj["fields"]["Associated File Names"] == "file.exe" assert indicator_obj["value"] == "somehashvalue" def test_parsing_user_query_success(): """ Given - A json string query When - query is good Then - create a dict from json string """ querystr = '{"returnFormat": "json","limit": "3", "type": {"OR": ["ip-src"]}, "tags": {"OR": ["tlp:%"]}}' params = parsing_user_query(querystr, limit=40000) assert len(params) == 6 assert params["includeEventTags"] is True def test_parsing_user_query_bad_query(): """ Given - A json string query When - json syntax is incorrect Then - raise a DemistoException """ with pytest.raises(DemistoException): querystr = '{"returnFormat": "json", "type": {"OR": ["md5"]}, "tags": {"OR": ["tlp:%"]' parsing_user_query(querystr, limit=4) def test_parsing_user_query_change_format(): """ Given - A json parsed result from qualys When - query has a unsupported return format Then - change return format to json """ querystr = '{"returnFormat": "xml", "type": {"OR": ["md5"]}, "tags": {"OR": ["tlp:%"]}}' params = parsing_user_query(querystr, limit=4) assert params["returnFormat"] == "json" def test_parsing_user_query_remove_timestamp(): """ Given - A json parsed result from qualys When - query has timestamp parameter Then - Return query without the timestamp parameter """ good_query = ( '{"returnFormat": "json", "type": {"OR": ["md5"]}, "tags": {"OR": ["tlp:%"]}, "page": 1, "limit": 2,' ' "attribute_timestamp": "1617875568", "includeEventTags": true}' ) querystr = '{"returnFormat": "json", "timestamp": "1617875568", "type": {"OR": ["md5"]}, "tags": {"OR": ["tlp:%"]}}' params = parsing_user_query(querystr, limit=2) assert good_query == json.dumps(params) def test_get_galaxy_indicator_type_success(): """ Given - Galaxy name When - Galaxy name is of a supported type Then - Return the matching indicator type """ galaxy_name = 'misp-galaxy:mitre-attack-pattern="Testing - R123"' assert get_galaxy_indicator_type(galaxy_name) == ThreatIntel.ObjectsNames.ATTACK_PATTERN def test_get_galaxy_indicator_type_doesnt_exist(): """ Given - Galaxy name When - Galaxy name is not supported Then - Return None """ galaxy_name = 'misp-galaxy:doesnt-exist="Testing - R123"' assert get_galaxy_indicator_type(galaxy_name) is None def test_build_indicators_from_galaxies(): """ Given - Indicator with galaxy tags When - Only one of the two galaxies is supported Then - Return List with an indicator created from the supported galaxy """ indicator_obj = { "value": "some_value", "type": "IP", "service": "MISP", "fields": {}, "rawJSON": { "value": { "Tag": [ { "name": 'misp-galaxy:mitre-attack-pattern="Some Value - R1234"', }, {"name": 'misp-galaxy:amitt-misinformation-pattern="fake galaxy"'}, ] } }, } galaxy_indicators = build_indicators_from_galaxies(indicator_obj, "Suspicious") assert len(galaxy_indicators) == 1 assert galaxy_indicators[0]["value"] == "Some Value" assert galaxy_indicators[0]["type"] == ThreatIntel.ObjectsNames.ATTACK_PATTERN @pytest.mark.parametrize( "indicator, feed_tags, expected_calls", [ ( { "value": "some_value", "type": "IP", "service": "MISP", "fields": {}, "rawJSON": { "value": { "Tag": [ { "name": 'misp-galaxy:mitre-attack-pattern="Some Value - R1234"', } ] } }, }, None, 1, ), ( { "value": "some_value", "type": "IP", "service": "MISP", "fields": {}, "rawJSON": {"value": {}}, }, ["test", "test2"], 1, ), ( { "value": "some_value", "type": "IP", "service": "MISP", "fields": {}, "rawJSON": {"value": {}}, }, None, 0, ), ], ) def test_update_indicator_fields(mocker, indicator: dict, feed_tags: list | None, expected_calls: int): """ Given: - indicator and feed_tags argument When: - the update_indicator_fields function runs Then: - Ensure the update_indicator_fields function is called if the feed_tags argument is passed even though the indicator has no tag. - Ensure the update_indicator_fields function is called when the indicator has tag. - Ensure the update_indicator_fields function is not called when the indicator has no tag and no feed_tags argument is sent. """ handle_tags_fields_mock = mocker.patch("FeedMISP.handle_tags_fields") update_indicator_fields(indicator, None, "test", feed_tags) assert handle_tags_fields_mock.call_count == expected_calls @pytest.mark.parametrize( "indicator, indicator_type", [ ({"value": "1.1.1.1"}, FeedIndicatorType.IP), ({"value": "1.1.1.1/24"}, FeedIndicatorType.CIDR), ({"value": "2001:0db8:85a3:0000:0000:8a2e:0370:7334"}, FeedIndicatorType.IPv6), ({"value": "2001:0db8:85a3:0000:0000:8a2e:0370:7334/64"}, FeedIndicatorType.IPv6CIDR), ], ) def test_get_ip_type(indicator, indicator_type): assert get_ip_type(indicator) == indicator_type def test_search_query_indicators_pagination(mocker): """ Given: - All relevant arguments for the command When: - the fetch_attributes_command function runs Then: - Ensure the pagination mechanism return the expected result (good http response is returned) """ client = Client( base_url="example", authorization="auth", verify=False, proxy=False, timeout=60, performance=False, max_indicator_to_fetch=2, ) returned_result_1 = { "response": { "Attribute": [ { "id": "1", "event_id": "1", "object_id": "0", "object_relation": None, "category": "Payload delivery", "type": "sha256", "to_ids": True, "uuid": "5fd0c620", "timestamp": "1607517728", "distribution": "5", "sharing_group_id": "0", "comment": "malspam", "deleted": False, "disable_correlation": False, "first_seen": None, "last_seen": None, "value": "val1", "Event": {}, }, { "id": "2", "event_id": "2", "object_id": "0", "object_relation": None, "category": "Payload delivery", "type": "sha256", "to_ids": True, "uuid": "5fd0c620", "timestamp": "1607517728", "distribution": "5", "sharing_group_id": "0", "comment": "malspam", "deleted": False, "disable_correlation": False, "first_seen": None, "last_seen": None, "value": "val2", "Event": {}, }, ] } } returned_result_2 = {"response": {"Attribute": []}} mocker.patch.object(Client, "_http_request", side_effect=[returned_result_1, returned_result_2]) params_dict = { "type": "attribute", "filters": {"category": ["Payload delivery"]}, } mocker.patch("FeedMISP.LIMIT", new=2000) mocker.patch.object(demisto, "getLastRun", return_value={}) mocker.patch.object(demisto, "setLastRun") mocker.patch.object(demisto, "createIndicators") fetch_attributes_command(client, params_dict) indicators = demisto.createIndicators.call_args[0][0] assert len(indicators) == 2 def test_search_query_indicators_pagination_bad_case(mocker): """ Given: - All relevant arguments for the command When: - the fetch_attributes_command function runs Then: - Ensure the pagination mechanism raises an error (bad http response is returned) """ from CommonServerPython import DemistoException client = Client( base_url="example", authorization="auth", verify=False, proxy=False, timeout=60, performance=False, max_indicator_to_fetch=2000, ) returned_result = {"Error": "failed api call"} expected_result = "Error in API call - check the input parameters and the API Key. Error: failed api call" mocker.patch.object(Client, "_http_request", return_value=returned_result) params_dict = {"type": "attribute", "filters": {"category": ["Payload delivery"]}} with pytest.raises(DemistoException) as e: fetch_attributes_command(client, params_dict) assert str(e.value) == expected_result def test_parsing_user_query_timestamp_deprecated(): """ Given: - No input When: - The parsing_user_query function runs Then: - Ensure the parsing_user_query function correctly parses the user query JSON string, replacing the 'timestamp' key with 'attribute_timestamp' since timestamp deprecated. """ good_query = ( '{"returnFormat": "json", "type": {"OR": ["md5"]}, "tags": {"OR": ["tlp:%"]}, "page": 1,' ' "limit": 2, "attribute_timestamp": "1617875568", "includeEventTags": true}' ) query_str = '{"returnFormat": "json", "timestamp": "1617875568", "type": {"OR": ["md5"]}, "tags": {"OR": ["tlp:%"]}}' params = parsing_user_query(query_str, limit=2) assert good_query == json.dumps(params) def test_ignore_last_fetched_indicator(mocker): """ Given: - The fetch_attributes_command function is called with a client object and a params_dict. When: - The last fetched indicator is returned when already fetched. Then: - The fetch_attributes_command function should ignore the last fetched indicator and continue fetching new indicators. """ client = Client( base_url="example", authorization="auth", verify=False, proxy=False, timeout=60, performance=False, max_indicator_to_fetch=2000, ) mocked_result = { "response": { "Attribute": [ { "id": "1", "event_id": "1", "object_id": "0", "object_relation": None, "category": "Payload delivery", "type": "sha256", "to_ids": True, "uuid": "5fd0c620", "timestamp": "1607517728", "distribution": "5", "sharing_group_id": "0", "comment": "malspam", "deleted": False, "disable_correlation": False, "first_seen": None, "last_seen": None, "value": "test", "Event": {}, } ] } } mocker.patch.object(Client, "_http_request", side_effect=[mocked_result]) params_dict = { "type": "attribute", "filters": {"category": ["Payload delivery"]}, } mocked_last_run = {"last_indicator_timestamp": "1607517728", "last_indicator_value": "test"} mocker.patch.object(demisto, "getLastRun", return_value=mocked_last_run) mocker.patch.object(demisto, "setLastRun") mocker.patch.object(demisto, "createIndicators") fetch_attributes_command(client, params_dict) indicators = demisto.createIndicators.call_args assert not indicators # No indicators should be created since the latest indicator was already fetched def test_fetch_new_indicator_after_last_indicator_been_ignored(mocker): """ Given: - The fetch_attributes_command function is called with a client object and a params_dict. When: - The latest retrieved indicators been ignored and new indicator is fetched. Then: - The fetch_attributes_command function should fetch the next indicator and set the new last run. """ client = Client( base_url="example", authorization="auth", verify=False, proxy=False, timeout=60, performance=False, max_indicator_to_fetch=2000, ) mocked_result_1 = { "response": { "Attribute": [ { "id": "1", "event_id": "1", "object_id": "0", "object_relation": None, "category": "Payload delivery", "type": "sha256", "to_ids": True, "uuid": "5fd0c620", "timestamp": "1607517728", "distribution": "5", "sharing_group_id": "0", "comment": "malspam", "deleted": False, "disable_correlation": False, "first_seen": None, "last_seen": None, "value": "test1", "Event": {}, }, { "id": "2", "event_id": "2", "object_id": "0", "object_relation": None, "category": "Payload delivery", "type": "sha256", "to_ids": True, "uuid": "5fd0c620", "timestamp": "1607517729", "distribution": "5", "sharing_group_id": "0", "comment": "malspam", "deleted": False, "disable_correlation": False, "first_seen": None, "last_seen": None, "value": "test2", "Event": {}, }, ] } } mocked_result_2 = {"response": {"Attribute": []}} mocker.patch.object(Client, "_http_request", side_effect=[mocked_result_1, mocked_result_2]) params_dict = { "type": "attribute", "filters": {"category": ["Payload delivery"]}, } mocked_last_run = {"last_indicator_timestamp": "1607517728", "last_indicator_value": "test1"} mocker.patch.object(demisto, "getLastRun", return_value=mocked_last_run) setLastRun_mocked = mocker.patch.object(demisto, "setLastRun") mocker.patch.object(demisto, "createIndicators") fetch_attributes_command(client, params_dict) indicators = demisto.createIndicators.call_args[0][0] # The last ignored indicator will be re-fetched as we query his timestamp, # but the new last run will be updated with the new indicator. assert len(indicators) == 2 assert setLastRun_mocked.called def test_set_last_run_pagination(mocker): """ Given: - The set_last_run_pagination function is called with a list of indicators, a next_page value, and a last_run dictionary. When: - The function is called to set the last run with the appropriate values. Then: - Ensure the last run is set correctly with the appropriate values """ from FeedMISP import update_candidate # Sample indicators indicators = [{"value": "test1", "timestamp": "1607517728"}, {"value": "test2", "timestamp": "1607517729"}] # Test parameters last_run = {"last_indicator_timestamp": "1607517727", "last_indicator_value": "test0"} last_run_timestamp = last_run["last_indicator_timestamp"] last_run_value = last_run["last_indicator_value"] latest_indicator_timestamp = indicators[-1]["timestamp"] latest_indicator_value = indicators[-1]["value"] # Call the function update_candidate(last_run, last_run_timestamp, latest_indicator_timestamp, latest_indicator_value) # Assert that setLastRun was called with the correct arguments expected_last_run = { "last_indicator_timestamp": last_run_timestamp, "candidate_timestamp": latest_indicator_timestamp, "last_indicator_value": last_run_value, "candidate_value": latest_indicator_value, } assert last_run == expected_last_run def test_build_indicators_from_galaxies_tool_type(): """ Given: - An indicator object containing a MISP tag for a MITRE tool. When: - The build_indicators_from_galaxies function is called with the indicator object and a high reputation level. Then: - The extracted indicator should have the correct 'value' corresponding to the tool name. - The 'type' should be 'Tool'. - The 'service' should be 'MISP'. - The 'Reputation' should be 'High'. """ from FeedMISP import build_indicators_from_galaxies indicator_obj = {"rawJSON": {"value": {"Tag": [{"name": 'misp-galaxy:mitre-tool="aaa aaa"'}]}}} galaxy_indicators = build_indicators_from_galaxies(indicator_obj, "High")[0] assert galaxy_indicators["value"] == "aaa aaa" assert galaxy_indicators["type"] == "Tool" assert galaxy_indicators["service"] == "MISP" assert galaxy_indicators["Reputation"] == "High" def test_build_indicators_from_galaxies_attack_type(): """ Given: - An indicator object containing a MISP tag for a MITRE attack pattern. When: - The build_indicators_from_galaxies function is called with the indicator object and a high reputation level. Then: - The extracted indicator should have the correct 'value' corresponding to the attack pattern name. - The 'type' should be 'Attack Pattern'. - The 'service' should be 'MISP'. - The 'Reputation' should be 'High'. """ from FeedMISP import build_indicators_from_galaxies indicator_obj = {"rawJSON": {"value": {"Tag": [{"name": 'misp-galaxy:mitre-attack-pattern="aaa aaa - 1111"'}]}}} galaxy_indicators = build_indicators_from_galaxies(indicator_obj, "High")[0] assert galaxy_indicators["value"] == "aaa aaa" assert galaxy_indicators["type"] == "Attack Pattern" assert galaxy_indicators["service"] == "MISP" assert galaxy_indicators["Reputation"] == "High" def test_build_indicators_with_hostname(): """ Given: - A response from the API which contains an indicator of type hostname. When: - build_indicators executed. Then: - Successfully creates indicator and relationship. """ from FeedMISP import build_indicators client = Mock() response = { "response": { "Attribute": [ { "type": "hostname", "value": {"value": "www.test.com"}, "Tag": [{"name": 'misp-galaxy:mitre-attack-pattern="T1111"'}], } ] } } result = build_indicators(client, response, ["hostname"], "color", "url", "reputation", ["feed_tags"]) assert result[0].get("Relationships", [])[0].get("entityAType") == "Domain" assert result[0].get("Relationships", [])[0].get("entityBType") == "Attack Pattern" assert result[0].get("Relationships", [])[0].get("entityB") == "T1111" def test_parsing_user_query_with_limit(): """ Given - A json parsed result from MISP When - function has limit argument Then - Return query with the right limit value """ good_query = ( '{"returnFormat": "json", "type": {"OR": ["md5"]}, "tags": {"OR": ["tlp:%"]}, "page": 1, "limit": 3,' ' "attribute_timestamp": "1617875568", "includeEventTags": true}' ) querystr = '{"returnFormat": "json", "timestamp": "1617875568", "type": {"OR": ["md5"]}, "tags": {"OR": ["tlp:%"]}}' params = parsing_user_query(querystr, limit=3) assert good_query == json.dumps(params) def test_build_params_dict_includes_event_tags(): """ Given - No tags, no attribute types, a limit and a page When - Building the params dict sent to MISP /attributes/restSearch Then - The "includeEventTags" flag is present and set to True so MISP merges event-inherited tags (e.g. TLP) into Attribute.Tag. """ params = build_params_dict(tags=[], attribute_type=[], limit=100, page=1) assert params["includeEventTags"] is True def test_build_indicators_propagates_inherited_event_tag(): """ Given - A MISP attribute whose Tag array contains an attribute-level tag and an event-inherited tag (e.g. 'tlp:white' with inherited=1), as MISP returns when includeEventTags=True. When - build_indicators parses the attribute into an indicator. Then - The inherited 'tlp:white' tag ends up in indicator["fields"]["Tags"] alongside the attribute-level tag, satisfying the DoD that event-level TLP is surfaced to the indicator mapper. """ client = Client( base_url="example", authorization="auth", verify=False, proxy=False, timeout=60, performance=False, max_indicator_to_fetch=2000, ) response = { "response": { "Attribute": [ { "id": "1", "event_id": "1", "type": "ip-src", "value": "1.2.3.4", "Tag": [ { "id": "1", "name": "attribute-tag", "colour": "#000000", "exportable": True, "is_galaxy": False, "is_custom_galaxy": False, }, { "id": "2", "name": "tlp:white", "colour": "#ffffff", "exportable": True, "numerical_value": None, "is_galaxy": False, "is_custom_galaxy": False, "inherited": 1, }, ], }, ] } } indicators = build_indicators( client=client, response=response, attribute_type=[], tlp_color=None, url=None, reputation=None, feed_tags=[], ) assert len(indicators) == 1 tags = indicators[0]["fields"]["Tags"] assert "tlp:white" in tags assert "attribute-tag" in tags def test_fetch_empty_response_preserves_last_run(mocker): """ Given: - A previously stored watermark (last_indicator_timestamp/value) from an earlier fetch. When: - The next fetch returns an empty response (no new attributes), so no candidate is produced. Then: - setLastRun must NOT be called (the existing watermark is left untouched), instead of being overwritten with nulls which would reset the incremental cursor and cause a full re-fetch. """ client = Client( base_url="example", authorization="auth", verify=False, proxy=False, timeout=60, performance=False, max_indicator_to_fetch=2000, ) # Empty MISP response -> the while loop never runs, no candidate is produced. mocker.patch.object(Client, "_http_request", side_effect=[{"response": {"Attribute": []}}]) mocked_last_run = {"last_indicator_timestamp": "1607517728", "last_indicator_value": "test"} mocker.patch.object(demisto, "getLastRun", return_value=mocked_last_run) set_last_run = mocker.patch.object(demisto, "setLastRun") create_indicators = mocker.patch.object(demisto, "createIndicators") fetch_attributes_command(client, {}) # No new indicators should be created, and the watermark must be preserved (setLastRun not called). assert not create_indicators.called assert not set_last_run.called