import json import demistomock as demisto import pytest def util_load_json(path): with open(path, encoding="utf-8") as f: return json.loads(f.read()) def test_crowdstrike_indicators_list_command(requests_mock): """Tests crowdstrike_indicators_list_command function Given - The following indicator type: 'domain' that was chosen by the user. - include_deleted: False When - Calling `crowdstrike_indicators_list_command` Then - convert the result to indicators list - validate the length of the indicators list - validate the Feed tags is as expected """ from CrowdStrikeIndicatorFeed import Client, crowdstrike_indicators_list_command mock_response = util_load_json("test_data/crowdstrike_indicators_list_command.json") requests_mock.post("https://api.crowdstrike.com/oauth2/token", json={"access_token": "12345"}) requests_mock.get(url="https://api.crowdstrike.com/intel/combined/indicators/v1", json=mock_response) requests_mock.get( url="https://api.crowdstrike.com/intel/entities/actors/v1?ids=GOBLINPANDA&fields=name", json={"resources": [{"name": "GOBLIN PANDA"}]}, ) feed_tags = ["Tag1", "Tag2"] client = Client( base_url="https://api.crowdstrike.com/", credentials={"identifier": "123", "password": "123"}, type="Domain", include_deleted="false", limit=2, feed_tags=feed_tags, ) args = {"limit": "2"} response = crowdstrike_indicators_list_command(client, args) assert len(response.outputs) == 3 assert len(response.raw_response) == 3 assert "Indicators from CrowdStrike Falcon Intel" in response.readable_output assert "domain_abc" in response.readable_output assert feed_tags[0] assert feed_tags[1] in response.raw_response[0]["fields"]["tags"] @pytest.mark.parametrize( "types_list, expected", [ ( ["ALL"], "type:'username',type:'domain',type:'email_address',type:'hash_md5',type:'hash_sha1'," "type:'hash_sha256',type:'registry',type:'url',type:'ip_address',type:'reports',type:'actors'," "type:'malware_families',type:'vulnerabilities'", ), (["Domain", "Email", "Registry Key"], "type:'domain',type:'email_address',type:'registry'"), ], ) def test_build_type_fql(types_list, expected): """Tests build_type_fql function Given - Indicator types that were chosen by the user. When - Calling `build_type_fql` in order to build filter for `get_indicators` Then - validate result as expected """ from CrowdStrikeIndicatorFeed import Client res = Client.build_type_fql(types_list=types_list) assert res == expected def test_create_indicators_from_response(requests_mock, mocker): """Tests build_type_fql function Given - Indicator types that were chosen by the user. When - Calling `create_indicators_from_response` in order to build indicators from response Then - validate result as expected """ from CrowdStrikeIndicatorFeed import Client mocker.patch("CrowdStrikeIndicatorFeed.get_integration_context", return_value={"GOBLINPANDA": "GOBLIN PANDA"}) raw_response = util_load_json("test_data/crowdstrike_indicators_list_command.json") expected_result = util_load_json("test_data/create_indicators_from_response.json") res = Client.create_indicators_from_response(raw_response, Client.get_actors_names_request) assert res == expected_result def test_empty_first_fetch(mocker, requests_mock): mocker.patch.object(demisto, "params", return_value={"first_fetch": ""}) mocker.patch.object(demisto, "command", return_value="") requests_mock.post("https://api.crowdstrike.com/oauth2/token", json={"access_token": "12345"}) from CrowdStrikeIndicatorFeed import main main() assert True @pytest.mark.parametrize( "field, indicator, resource, expected_results", [ ( "relations", {"type": "hash_md5", "value": "1234567890"}, {"relations": [{"type": "password"}, {"type": "username", "indicator": "abc"}]}, [ { "name": "related-to", "reverseName": "related-to", "type": "IndicatorToIndicator", "entityA": "1234567890", "entityAFamily": "Indicator", "entityAType": "hash_md5", "entityB": "abc", "entityBFamily": "Indicator", "entityBType": "Account", "fields": {}, } ], ), ( "malware_families", {"type": "type non support", "value": "1234567890"}, {"malware_families": {"relations": "Test indicator"}}, [ { "name": "type non support", "reverseName": "related-to", "type": "IndicatorToIndicator", "entityA": "1234567890", "entityAFamily": "Indicator", "entityAType": "type non support", "entityB": "relations", "entityBFamily": "Indicator", "entityBType": "Malware", "fields": {}, } ], ), ], ) def test_create_relationships_unknown_key(field, indicator, resource, expected_results): """ Given - Field type, indicator and a resource with an unknown relation type. When - Calling `create_relationships` command. Then - validate that no Key Error exception was thrown, and that only 1 relationship was created. """ from CrowdStrikeIndicatorFeed import Client, create_relationships rs_ls = create_relationships(field, indicator, resource, Client.get_actors_names_request) assert rs_ls == expected_results assert len(rs_ls) == 1 def test_reset_last_run(mocker): """ Given - No inputs. When - Calling `reset_last_run` command. Then - Ensure that the integration context dict was cleared. """ from CrowdStrikeIndicatorFeed import reset_last_run demisto_set_context_mocker = mocker.patch.object(demisto, "setIntegrationContext") reset_last_run() assert demisto_set_context_mocker.call_args.args == ({},) def test_fetch_no_indicators(mocker, requests_mock): """ Given - no indicators api response When - fetching indicators Then - Ensure empty list is returned and no exception is raised. """ from CrowdStrikeIndicatorFeed import Client mock_response = util_load_json("test_data/crowdstrike_indicators_list_command.json") requests_mock.post("https://api.crowdstrike.com/oauth2/token", json={"access_token": "12345"}) requests_mock.get(url="https://api.crowdstrike.com/intel/combined/indicators/v1", json=mock_response) feed_tags = ["Tag1", "Tag2"] client = Client( base_url="https://api.crowdstrike.com/", credentials={"identifier": "123", "password": "123"}, type="Domain", include_deleted="false", limit=2, feed_tags=feed_tags, ) mocker.patch.object(client, "get_indicators", return_value={"resources": []}) assert client.fetch_indicators(limit=10, offset=5, fetch_command=True) == [] def test_crowdstrike_to_xsoar_types(): from CrowdStrikeIndicatorFeed import CROWDSTRIKE_TO_XSOAR_TYPES assert None not in CROWDSTRIKE_TO_XSOAR_TYPES @pytest.mark.parametrize( "first_fetch, filter, integration_context, get_indicators_response, filter_arg_call, expected_results", [ ("1662650320", "", {"last_updated": "1662650343"}, {}, "(last_updated:>=1662650343)", ("", 0)), ( "1662650320", "", {"last_updated": "1662650343"}, { "resources": [ { "id": "dummy", "indicator": "dummy", "type": "hash_md5", "deleted": "False", "published_date": 1622198010, "last_updated": 1662650343, "reports": [], "actors": ["DOPPELSPIDER"], "malware_families": ["DoppelDridex"], "kill_chains": [], "ip_address_types": [], "domain_types": [], "malicious_confidence": "high", "_marker": "test_marker_test", "labels": [], } ] }, "(last_updated:>=1662650343)", ("(_marker:>'test_marker_test')", 1), ), ( "1662650320", "", {}, { "resources": [ { "id": "dummy", "indicator": "dummy", "type": "hash_md5", "deleted": "False", "published_date": 1622198010, "last_updated": 1662650343, "reports": [], "actors": ["DOPPELSPIDER"], "malware_families": ["DoppelDridex"], "kill_chains": [], "ip_address_types": [], "domain_types": [], "malicious_confidence": "high", "_marker": "test_marker_test", "labels": [], } ] }, "(last_updated:>=1662650320)", ("(_marker:>'test_marker_test')", 1), ), ( "", "", {}, { "resources": [ { "id": "dummy", "indicator": "dummy", "type": "hash_md5", "deleted": "False", "published_date": 1622198010, "last_updated": 1662650343, "reports": [], "actors": ["DOPPELSPIDER"], "malware_families": ["DoppelDridex"], "kill_chains": [], "ip_address_types": [], "domain_types": [], "malicious_confidence": "high", "_marker": "test_marker_test", "labels": [], } ] }, None, ("(_marker:>'test_marker_test')", 1), ), ], ) def test_handling_first_fetch_and_old_integration_context( mocker, requests_mock, first_fetch, filter, integration_context, get_indicators_response, filter_arg_call, expected_results ): from CrowdStrikeIndicatorFeed import Client requests_mock.post("https://api.crowdstrike.com/oauth2/token", json={"access_token": "12345"}) client = Client( base_url="https://api.crowdstrike.com/", credentials={"identifier": "123", "password": "123"}, type="ALL", include_deleted="false", limit=2, first_fetch=first_fetch, ) mocker.patch("CrowdStrikeIndicatorFeed.demisto.getIntegrationContext", return_value=integration_context) requests_mock.get( url="https://api.crowdstrike.com/intel/entities/actors/v1?ids=DOPPELSPIDER&fields=name", json={"resources": [{"name": "DOPPEL SPIDER"}]}, ) get_indicator_call = mocker.patch.object(client, "get_indicators", return_value=get_indicators_response) results = client.handle_first_fetch_context_or_pre_2_1_0(filter) assert get_indicator_call.call_args.kwargs["params"].get("filter") == filter_arg_call assert results[0] == expected_results[0] assert len(results[1]) == expected_results[1] @pytest.mark.parametrize( "indicator, expected_results", [ ({"indicator": "1.1.1.1", "type": "ip_address"}, "IP"), ({"indicator": "fe80:0000:0000:0000:91ba:7558:26d3:acde", "type": "ip_address"}, "IPv6"), ({"indicator": "test_test", "type": "username"}, "Account"), ({"indicator": "test_test", "type": "password"}, None), ], ) def test_auto_detect_indicator_type_from_cs(indicator: dict, expected_results: str | None): from CrowdStrikeIndicatorFeed import auto_detect_indicator_type_from_cs assert auto_detect_indicator_type_from_cs(indicator["indicator"], indicator["type"]) == expected_results def test_get_actors_names_request_check_output(mocker, requests_mock): """ Given - params for get_actors_names_request http request When - calling get_actors_names_request after fetching/listing indicators Then - Ensure the result is correct """ from CrowdStrikeIndicatorFeed import Client mock_response = util_load_json("test_data/crowdstrike_indicators_list_command.json") requests_mock.post("https://api.crowdstrike.com/oauth2/token", json={"access_token": "12345"}) requests_mock.get(url="https://api.crowdstrike.com/intel/combined/indicators/v1", json=mock_response) crowdstrike_client = Client( base_url="https://api.crowdstrike.com/", credentials={"identifier": "123", "password": "123"}, type="Domain", include_deleted="false", limit=2, ) requests_mock.get(url="https://api.crowdstrike.com/intel/entities/actors/v1?", json={"resources": ""}) requests_mock.get( url="https://api.crowdstrike.com/intel/entities/actors/v1?ids=123&fields=name", json={"resources": {"name": "TEST TEST"}} ) res = crowdstrike_client.get_actors_names_request(params_string="ids=123&fields=name") assert res == {"name": "TEST TEST"} def test_get_actors_names_request_called_with(mocker, requests_mock): """ Given - params for get_actors_names_request http request When - calling get_actors_names_request after fetching/listing indicators Then - Ensure the request is called with the right args """ from CrowdStrikeIndicatorFeed import Client mock_response = util_load_json("test_data/crowdstrike_indicators_list_command.json") requests_mock.post("https://api.crowdstrike.com/oauth2/token", json={"access_token": "12345"}) requests_mock.get(url="https://api.crowdstrike.com/intel/combined/indicators/v1", json=mock_response) crowdstrike_client = Client( base_url="https://api.crowdstrike.com/", credentials={"identifier": "123", "password": "123"}, type="Domain", include_deleted="false", limit=2, ) requests_mock.get(url="https://api.crowdstrike.com/intel/entities/actors/v1?", json={"resources": ""}) http_request_mock = mocker.patch.object( crowdstrike_client, "_http_request", return_value={"resources": {"name": "TEST TEST"}} ) crowdstrike_client.get_actors_names_request(params_string="ids=123&fields=name") http_request_mock.assert_called_once_with( method="GET", url_suffix="intel/entities/actors/v1?ids=123&fields=name", timeout=30, ok_codes=(200, 401) ) def test_crowdstrike_indicators_list_command_check_actors_convert(mocker, requests_mock): """ Given - params for crowdstrike_indicators_list_command http request When - calling get_actors_names_request after fetching/listing indicators Then - Ensure the response is correct """ from CrowdStrikeIndicatorFeed import Client, crowdstrike_indicators_list_command mock_response = util_load_json("test_data/crowdstrike_test_actors_convert.json") requests_mock.post("https://api.crowdstrike.com/oauth2/token", json={"access_token": "12345"}) requests_mock.get(url="https://api.crowdstrike.com/intel/combined/indicators/v1", json=mock_response) requests_mock.get( url="https://api.crowdstrike.com/intel/entities/actors/v1?ids=TESTTEST&fields=name", json={"resources": [{"name": "TEST TEST"}]}, ) mocker.patch("CrowdStrikeIndicatorFeed.get_integration_context", return_value={}) mocker.patch("CrowdStrikeIndicatorFeed.update_integration_context") feed_tags = ["Tag1", "Tag2"] crowdstrike_client = Client( base_url="https://api.crowdstrike.com/", credentials={"identifier": "123", "password": "123"}, type="Domain", include_deleted="false", limit=2, feed_tags=feed_tags, ) args = {"limit": "2"} response = crowdstrike_indicators_list_command(crowdstrike_client, args) assert len(response.outputs) == 1 assert len(response.raw_response) == 1 assert len(response.raw_response[0].get("relationships", None)) == 2 assert response.raw_response[0].get("relationships", None)[1].get("entityB", None) == "TEST TEST" def test_change_actors_from_id_to_name(mocker, requests_mock): """ Given - params for change_actors_from_id_to_name http request When - calling get_actors_names_request after fetching/listing indicators Then - Ensure the response is correct """ from CrowdStrikeIndicatorFeed import Client, change_actors_from_id_to_name requests_mock.post("https://api.crowdstrike.com/oauth2/token", json={"access_token": "12345"}) crowdstrike_client = Client( base_url="https://api.crowdstrike.com/", credentials={"identifier": "123", "password": "123"}, type="Domain", include_deleted="false", limit=2, ) actors_unparsed_array = ["TEST", "TEST1", "TEST2"] mocker.patch("CrowdStrikeIndicatorFeed.get_integration_context", return_value={"TEST": "WAS IN CONTEXT"}) requests_mock.get( url="https://api.crowdstrike.com/intel/entities/actors/v1?ids=TEST1&ids=TEST2&fields=name", json={"resources": [{"name": "Changedtest1"}, {"name": "Changedtest2"}]}, ) result = change_actors_from_id_to_name(actors_unparsed_array, crowdstrike_client.get_actors_names_request) assert result[0] == "WAS IN CONTEXT" assert result[1] == "Changedtest1" assert result[2] == "Changedtest2" # This function will alternate the response each time it's called def mock_alternate_response(request, context): # Track the number of times the post request has been called if not hasattr(mock_alternate_response, "call_count"): mock_alternate_response.call_count = 0 mock_alternate_response.call_count += 1 if mock_alternate_response.call_count == 1: # First call: Return 401 Unauthorized context.status_code = 401 return {"errors": [{"code": 401, "message": "access denied, authorization failed"}]} else: # Second call: Return 200 OK with the access token context.status_code = 200 return {"resources": []} def test_fetch_no_indicators_regenerate_token(requests_mock): """ Given - no indicators api response When - fetching indicators when API token expired. Then - Ensure empty list is returned and no exception is raised. """ from CrowdStrikeIndicatorFeed import Client requests_mock.post("https://api.crowdstrike.com/oauth2/token", json={"access_token": "12345"}) requests_mock.get(url="https://api.crowdstrike.com/intel/combined/indicators/v1", json=mock_alternate_response) feed_tags = ["Tag1", "Tag2"] client = Client( base_url="https://api.crowdstrike.com/", credentials={"identifier": "123", "password": "123"}, type="Domain", include_deleted="false", limit=2, feed_tags=feed_tags, ) assert client.fetch_indicators(limit=10, offset=5, fetch_command=True) == [] def test_get_actors_names_request_regenerate_token(requests_mock): """ Given - no actors api response When - calling get_actors_names_request after fetching/listing indicators Then - Ensure empty list is returned and no exception is raised. """ from CrowdStrikeIndicatorFeed import Client mock_response = util_load_json("test_data/crowdstrike_indicators_list_command.json") requests_mock.post("https://api.crowdstrike.com/oauth2/token", json={"access_token": "12345"}) requests_mock.get(url="https://api.crowdstrike.com/intel/combined/indicators/v1", json=mock_response) crowdstrike_client = Client( base_url="https://api.crowdstrike.com/", credentials={"identifier": "123", "password": "123"}, type="Domain", include_deleted="false", limit=2, ) requests_mock.get(url="https://api.crowdstrike.com/intel/entities/actors/v1?", json={"resources": ""}) requests_mock.get( url="https://api.crowdstrike.com/intel/entities/actors/v1?ids=123&fields=name", json=mock_alternate_response ) res = crowdstrike_client.get_actors_names_request(params_string="ids=123&fields=name") assert res == []