"""Imports""" import io import json import os import tempfile import zipfile from tempfile import mkdtemp import demistomock as demisto import pytest import EDL as edl from EDL import ( DONT_COLLAPSE, check_platform_and_version, datetime, get_indicators_to_format, initialize_edl_context, timezone, RequestArguments, FORMAT_TEXT, FORMAT_CSV, FORMAT_JSON, FORMAT_MWG, FORMAT_PROXYSG, ) from freezegun import freeze_time IOC_RES_LEN = 38 """Tests""" class TestRequestArguments: from EDL import RequestArguments context_json = { RequestArguments.CTX_QUERY_KEY: "query", RequestArguments.CTX_LIMIT_KEY: 10, RequestArguments.CTX_OFFSET_KEY: 1, RequestArguments.CTX_INVALIDS_KEY: True, RequestArguments.CTX_PORT_STRIP_KEY: True, RequestArguments.CTX_COLLAPSE_IPS_KEY: "collapse", RequestArguments.CTX_EMPTY_EDL_COMMENT_KEY: True, RequestArguments.CTX_OUT_FORMAT: "text", RequestArguments.CTX_MWG_TYPE: "string", RequestArguments.CTX_CATEGORY_DEFAULT: "bc_category", RequestArguments.CTX_CATEGORY_ATTRIBUTE: [], RequestArguments.CTX_FIELDS_TO_PRESENT: "name,type", RequestArguments.CTX_CSV_TEXT: False, RequestArguments.CTX_PROTOCOL_STRIP_KEY: False, RequestArguments.CTX_URL_TRUNCATE_KEY: False, RequestArguments.CTX_NO_TLD: True, RequestArguments.CTX_MAXIMUM_CIDR: 8, } request_args = RequestArguments( query=context_json[RequestArguments.CTX_QUERY_KEY], limit=context_json[RequestArguments.CTX_LIMIT_KEY], offset=context_json[RequestArguments.CTX_OFFSET_KEY], url_port_stripping=context_json[RequestArguments.CTX_PORT_STRIP_KEY], drop_invalids=context_json[RequestArguments.CTX_PORT_STRIP_KEY], collapse_ips=context_json[RequestArguments.CTX_COLLAPSE_IPS_KEY], add_comment_if_empty=context_json[RequestArguments.CTX_EMPTY_EDL_COMMENT_KEY], out_format=context_json[RequestArguments.CTX_OUT_FORMAT], mwg_type=context_json[RequestArguments.CTX_MWG_TYPE], category_default=context_json[RequestArguments.CTX_CATEGORY_DEFAULT], category_attribute="", fields_to_present=context_json[RequestArguments.CTX_FIELDS_TO_PRESENT], csv_text=context_json[RequestArguments.CTX_CSV_TEXT], url_protocol_stripping=context_json[RequestArguments.CTX_PROTOCOL_STRIP_KEY], url_truncate=context_json[RequestArguments.CTX_URL_TRUNCATE_KEY], no_wildcard_tld=context_json[RequestArguments.CTX_NO_TLD], maximum_cidr_size=context_json[RequestArguments.CTX_MAXIMUM_CIDR], ) def test_to_context_json(self): """ Test to_context_json transforms the class to the expected context_json Given: - request_args When: - calling to_context_json() Then: - creates a dict in the expected format """ assert self.request_args.to_context_json() == self.context_json def test_from_context_json(self): """ Test from_context_json creates an instance of the class with all the values Given: - context json with data When: - calling from_context_json() Then: - creates an instance of RequestArguments with the proper values """ actual_request_args_dict = self.RequestArguments.from_context_json(self.context_json).__dict__ expected_request_args_dict = self.request_args.__dict__ for key, val in actual_request_args_dict.items(): assert expected_request_args_dict[key] == val class TestHelperFunctions: def test_get_edl_on_demand__with_cache(self, mocker): """ Test get_edl_on_demand fetches indicators from cache Given: - No refresh signal in context - Cache has a valid value When: - calling get_edl_on_demand Then: - return the edl from the system file """ import EDL as edl edl.EDL_ON_DEMAND_CACHE_PATH = "test_data/iocs_cache_values_text.txt" edl.EDL_ON_DEMAND_CACHE_ORIGINAL_SIZE = 40 mocker.patch.object(edl, "get_integration_context", return_value={}) actual_edl, original_indicators_count = edl.get_edl_on_demand() with open(edl.EDL_ON_DEMAND_CACHE_PATH) as f: expected_edl = f.read() assert actual_edl == expected_edl assert original_indicators_count == edl.EDL_ON_DEMAND_CACHE_ORIGINAL_SIZE @freeze_time("2024-04-04 03:21:34") def test_get_edl_on_demand__with_refresh_signal(self, mocker): """ Test get_edl_on_demand fetches new indicators and stores them to cache when key is passed Given: - refresh signal in context When: - calling get_edl_on_demand Then: - save the edl to the system file - assert the edl log is as expected """ import EDL as edl expected_edl = "8.8.8.8" edl_log_line = "\nAdded | 8.8.8.8 | 8.8.8.8 | Found new Domain." ctx = { edl.EDL_ON_DEMAND_KEY: True, edl.RequestArguments.CTX_QUERY_KEY: "*", edl.RequestArguments.CTX_FIELDS_TO_PRESENT: "name,type", } tmp_dir = mkdtemp() edl.EDL_ON_DEMAND_CACHE_PATH = os.path.join(tmp_dir, "cache") mocker.patch.object(edl, "get_integration_context", return_value=ctx) mocker.patch.object(edl, "create_new_edl", return_value=(expected_edl, 1, {"Added": 1})) created_time = datetime.now(timezone.utc) with ( tempfile.NamedTemporaryFile() as wip_log_file, tempfile.NamedTemporaryFile() as log_file, tempfile.NamedTemporaryFile() as cached_edl_file, ): edl.EDL_FULL_LOG_PATH_WIP = wip_log_file.name edl.EDL_FULL_LOG_PATH = log_file.name edl.EDL_ON_DEMAND_CACHE_PATH = cached_edl_file.name wip_log_file.write(edl_log_line.encode()) wip_log_file.seek(0) actual_edl, _ = edl.get_edl_on_demand() edl_log = log_file.read() edl_log = edl_log.decode("utf-8") cached_edl = cached_edl_file.read() assert actual_edl == expected_edl == cached_edl.decode("utf-8") expected_edl_log = ( f"# Created new EDL at {created_time.isoformat()}\n\n" f"## Configuration Arguments: {{'last_query': '*', 'out_format': 'PAN-OS (text)', " f"'last_limit': 10000, 'last_offset': 0, 'drop_invalids': False, 'url_port_stripping': False, " f"'collapse_ips': \"Don't Collapse\", 'add_comment_if_empty': True, 'mwg_type': 'string', " f"'bc_category': 'bc_category', 'category_attribute': [], 'fields_to_present': 'name,type', " f"'csv_text': False, 'url_protocol_stripping': False, 'url_truncate': False, " f"'maximum_cidr_size': 8, 'no_wildcard_tld': False}}\n\n" f"## EDL stats: 1 indicators in total, 0 modified, 0 dropped, 1 added.\n\n" f"Action | Indicator | Raw Indicator | Reason{edl_log_line}" ) assert edl_log == expected_edl_log def test_iterable_to_str_1(self): """Test invalid""" from EDL import DemistoException, iterable_to_str with pytest.raises(DemistoException): invalid_list_value = 2 iterable_to_str(invalid_list_value) def test_iterable_to_str_2(self): """Test empty""" from EDL import iterable_to_str assert iterable_to_str(None) == "" assert iterable_to_str([]) == "" assert iterable_to_str({}) == "" def test_get_params_port_1(self): """Test invalid""" from CommonServerPython import DemistoException from EDL import get_params_port params = {"longRunningPort": "invalid"} with pytest.raises(DemistoException): get_params_port(params) def test_get_params_port_2(self): """Test empty""" from EDL import get_params_port params = {"longRunningPort": ""} with pytest.raises(ValueError): get_params_port(params) def test_get_params_port_3(self): """Test valid""" from EDL import get_params_port params = {"longRunningPort": "80"} assert get_params_port(params) == 80 def test_create_new_edl(self, mocker): """Sanity""" import EDL as edl f = tempfile.TemporaryFile(mode="w+t") f.write( '{"value": "https://google.com", "indicator_type": "URL"}\n' '{"value": "demisto.com:7000", "indicator_type": "URL"}\n' '{"value": "demisto.com/qwertqwertyuioplkjhgfdsazxqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyu' "iopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopyuioplkjhgfdsa" "zxqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwert" 'yuiopqwertyuiopqwertyuiopqwertyuiop", "indicator_type": "URL"}\n' '{"value": "demisto.com", "indicator_type": "URL"}' ) mocker.patch.object(edl, "get_indicators_to_format", return_value=(f, 4)) request_args = edl.RequestArguments( query="", limit=3, url_port_stripping=True, url_protocol_stripping=True, url_truncate=True ) with tempfile.NamedTemporaryFile() as wip_log_file: edl.EDL_FULL_LOG_PATH_WIP = wip_log_file.name edl_vals, original_indicators_count, edl_log_stats = edl.create_new_edl(request_args) edl_log = wip_log_file.read() edl_log = edl_log.decode("utf-8") assert ( edl_vals == "google.com\ndemisto.com\ndemisto.com/qwertqwertyuioplkjhgfdsazxqwertyuiopqwertyuiopq" "wertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwert" "yuiopqwertyuiopqwertyuiopyuioplkjhgfdsazxqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwe" "rtyuiopqwertyuiopqwertyuiop\n" ) expected_log = ( "\nAdded | google.com | https://google.com | Found new URL." "\nAdded | demisto.com | demisto.com:7000 | Found new URL." "\nAdded | demisto.com/qwertqwertyuioplkjhgfdsazxqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwe" "rtyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopyui" "oplkjhgfdsazxqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiop | demi" "sto.com/qwertqwertyuioplkjhgfdsazxqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwerty" "uiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopyuioplkjhgfdsazx" "qwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiop" "qwertyuiopqwertyuiopqwertyuiopqwertyuiop | Found new URL." "\nAdded | demisto.com | demisto.com | Found new URL." ) assert edl_log == expected_log assert edl_log_stats == {"Added": 4} f = tempfile.TemporaryFile(mode="w+t") f.write( '{"value": "https://google.com", "indicator_type": "URL"}\n' '{"value": "demisto.com:7000", "indicator_type": "URL"}\n' '{"value": "demisto.com/qwertqwertyuioplkjhgfdsazxqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyu' "iopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopyuioplkjhgfdsa" "zxqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwert" 'yuiopqwertyuiopqwertyuiopqwertyuiop", "indicator_type": "URL"}\n' '{"value": "demisto.com", "indicator_type": "URL"}' ) mocker.patch.object(edl, "get_indicators_to_format", return_value=(f, 4)) request_args = edl.RequestArguments( out_format="CSV", query="", limit=3, url_port_stripping=True, url_protocol_stripping=True, url_truncate=True, fields_to_present="name,value", ) with tempfile.NamedTemporaryFile() as wip_log_file: edl.EDL_FULL_LOG_PATH_WIP = wip_log_file.name edl_v, original_indicators_count, edl_log_stats = edl.create_new_edl(request_args) edl_log = wip_log_file.read() edl_log = edl_log.decode("utf-8") assert ( edl_v == '{"value": "https://google.com", "indicator_type": "URL"}\n' '{"value": "demisto.com:7000", "indicator_type": "URL"}\n' '{"value": "demisto.com/qwertqwertyuioplkjhgfdsazxqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyu' "iopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopyuioplkjhgfdsa" "zxqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwertyuiopqwert" 'yuiopqwertyuiopqwertyuiopqwertyuiop", "indicator_type": "URL"}\n' '{"value": "demisto.com", "indicator_type": "URL"}' ) assert edl_log == "" # No log when exporting to csv assert edl_log_stats == {} # No log when exporting to csv def test_create_new_edl_edge_cases(self, mocker, requests_mock): """ Test create_new_edl with wildcards, cidrs with prefix and non-ascii chars Given: - A list of indicators of "edge cases" When: - calling create_new_edl Then: - Ensure that the list is the same as is should. - Ensure the log is as expected. """ import EDL as edl tlds = "com\nco.uk" requests_mock.get("https://publicsuffix.org/list/public_suffix_list.dat", text=tlds) requests_mock.get("https://raw.githubusercontent.com/publicsuffix/list/master/public_suffix_list.dat", text=tlds) indicators = [ {"value": "1.1.1.1/7", "indicator_type": "CIDR"}, # prefix=7 {"value": "1.1.1.1/12", "indicator_type": "CIDR"}, # prefix=12 {"value": "*.com", "indicator_type": "Domain"}, # tld {"value": "*.co.uk", "indicator_type": "Domain"}, # tld {"value": "*.google.com", "indicator_type": "Domain"}, # no tld {"value": "aא.com", "indicator_type": "URL"}, ] # no ascii f = "\n".join(json.dumps(indicator) for indicator in indicators) request_args = edl.RequestArguments(collapse_ips=DONT_COLLAPSE, maximum_cidr_size=2) mocker.patch.object(edl, "get_indicators_to_format", return_value=(io.StringIO(f), 6)) with tempfile.NamedTemporaryFile() as wip_log_file: edl.EDL_FULL_LOG_PATH_WIP = wip_log_file.name edl_v, _, edl_log_stats = edl.create_new_edl(request_args) edl_log = wip_log_file.read() edl_log = edl_log.decode("utf-8") expected_values = set() for indicator in indicators: value = indicator.get("value") if value.startswith("*."): expected_values.add(value.lstrip("*.")) expected_values.add(value) assert set(edl_v.split("\n")) == expected_values expected_log = ( "\nAdded | 1.1.1.1/7 | 1.1.1.1/7 | Found new CIDR." "\nAdded | 1.1.1.1/12 | 1.1.1.1/12 | Found new CIDR." "\nAdded | *.com | *.com | Found new Domain." "\nAdded | *.co.uk | *.co.uk | Found new Domain." "\nAdded | *.google.com | *.google.com | Found new Domain." "\nAdded | aא.com | aא.com | Found new URL." ) assert edl_log == expected_log assert edl_log_stats == {"Added": 6} request_args = edl.RequestArguments(collapse_ips=DONT_COLLAPSE, maximum_cidr_size=8) mocker.patch.object(edl, "get_indicators_to_format", return_value=((io.StringIO(f)), 6)) with tempfile.NamedTemporaryFile() as wip_log_file: edl.EDL_FULL_LOG_PATH_WIP = wip_log_file.name edl_v, _, edl_log_stats = edl.create_new_edl(request_args) edl_log = wip_log_file.read() edl_log = edl_log.decode("utf-8") assert set(edl_v.split("\n")) == { "1.1.1.1/12", "*.com", "com", "*.co.uk", "co.uk", "*.google.com", "google.com", "aא.com", } expected_log = ( "\nDropped | 1.1.1.1/7 | 1.1.1.1/7 | CIDR exceeds max length 8." "\nAdded | 1.1.1.1/12 | 1.1.1.1/12 | Found new CIDR." "\nAdded | *.com | *.com | Found new Domain." "\nAdded | *.co.uk | *.co.uk | Found new Domain." "\nAdded | *.google.com | *.google.com | Found new Domain." "\nAdded | aא.com | aא.com | Found new URL." ) assert edl_log == expected_log assert edl_log_stats == {"Added": 5, "Dropped": 1} request_args = edl.RequestArguments(collapse_ips=DONT_COLLAPSE, no_wildcard_tld=True, maximum_cidr_size=13) mocker.patch.object(edl, "get_indicators_to_format", return_value=(io.StringIO(f), 6)) with tempfile.NamedTemporaryFile() as wip_log_file: edl.EDL_FULL_LOG_PATH_WIP = wip_log_file.name edl_v, _, edl_log_stats = edl.create_new_edl(request_args) edl_log = wip_log_file.read() edl_log = edl_log.decode("utf-8") assert set(edl_v.split("\n")) == {"*.google.com", "google.com", "aא.com"} expected_log = ( "\nDropped | 1.1.1.1/7 | 1.1.1.1/7 | CIDR exceeds max length 13." "\nDropped | 1.1.1.1/12 | 1.1.1.1/12 | CIDR exceeds max length 13." "\nDropped | com | *.com | Domain is a TLD." "\nDropped | co.uk | *.co.uk | Domain is a TLD." "\nAdded | *.google.com | *.google.com | Found new Domain." "\nAdded | aא.com | aא.com | Found new URL." ) assert edl_log == expected_log assert edl_log_stats == {"Added": 2, "Dropped": 4} def test_create_new_edl_with_offset(self, mocker, requests_mock): """ Test create_new_edl with and without offset Given: - A list of indicators When: - calling create_new_edl Then: - Ensure that the list is the same as is should with no offset and with offset=2 - Ensure the log is as expected. """ import EDL as edl tlds = "com\nco.uk" requests_mock.get("https://publicsuffix.org/list/public_suffix_list.dat", text=tlds) requests_mock.get("https://raw.githubusercontent.com/publicsuffix/list/master/public_suffix_list.dat", text=tlds) indicators = [ {"value": "1.1.1.1/7", "indicator_type": "CIDR"}, # prefix=7 {"value": "1.1.1.1/12", "indicator_type": "CIDR"}, # prefix=12 {"value": "*.com", "indicator_type": "Domain"}, # tld {"value": "*.co.uk", "indicator_type": "Domain"}, # tld {"value": "*.google.com", "indicator_type": "Domain"}, # no tld {"value": "aא.com", "indicator_type": "URL"}, ] # no ascii f = "\n".join(json.dumps(indicator) for indicator in indicators) # create_new_edl with no offset request_args = edl.RequestArguments(collapse_ips=DONT_COLLAPSE, maximum_cidr_size=8) mocker.patch.object(edl, "get_indicators_to_format", return_value=((io.StringIO(f)), 6)) with tempfile.NamedTemporaryFile() as wip_log_file: edl.EDL_FULL_LOG_PATH_WIP = wip_log_file.name edl_v, _, edl_log_stats = edl.create_new_edl(request_args) edl_log = wip_log_file.read() edl_log = edl_log.decode("utf-8") assert set(edl_v.split("\n")) == { "1.1.1.1/12", "*.com", "com", "*.co.uk", "co.uk", "*.google.com", "google.com", "aא.com", } expected_log = ( "\nDropped | 1.1.1.1/7 | 1.1.1.1/7 | CIDR exceeds max length 8." "\nAdded | 1.1.1.1/12 | 1.1.1.1/12 | Found new CIDR." "\nAdded | *.com | *.com | Found new Domain." "\nAdded | *.co.uk | *.co.uk | Found new Domain." "\nAdded | *.google.com | *.google.com | Found new Domain." "\nAdded | aא.com | aא.com | Found new URL." ) assert edl_log == expected_log assert edl_log_stats == {"Added": 5, "Dropped": 1} # create_new_edl with offset=2 request_args = edl.RequestArguments(collapse_ips=DONT_COLLAPSE, maximum_cidr_size=8, offset=2) mocker.patch.object(edl, "get_indicators_to_format", return_value=((io.StringIO(f)), 6)) with tempfile.NamedTemporaryFile() as wip_log_file: edl.EDL_FULL_LOG_PATH_WIP = wip_log_file.name edl_v, _, edl_log_stats = edl.create_new_edl(request_args) edl_log = wip_log_file.read() edl_log = edl_log.decode("utf-8") assert set(edl_v.split("\n")) == {"google.com", "co.uk", "*.co.uk", "*.google.com", "*.com", "aא.com"} assert edl_log == expected_log assert edl_log_stats == {"Added": 5, "Dropped": 1} def test_create_json_out_format(self): """ Given: - RequestArguments - Indicator info When: - request json outbound format Then: - assert the result """ from EDL import RequestArguments, create_json_out_format returned_output = [] with open("test_data/demisto_url_iocs.json") as iocs_json_f: iocs_json = json.loads(iocs_json_f.read()) # strips port numbers request_args = RequestArguments(query="", drop_invalids=True, url_port_stripping=True) for ioc in iocs_json: returned_output.append(create_json_out_format(["value", "indicator_type"], ioc, request_args, True)) assert returned_output == [ ', {"value": "1.2.3.4/wget", "indicator_type": "URL"}', ', {"value": "https://www.demisto.com/cool", "indicator_type": "URL"}', ', {"value": "https://www.demisto.com/*cool", "indicator_type": "URL"}', ] returned_output = "" not_first_call = False for ioc in iocs_json: returned_output += create_json_out_format([], ioc, request_args, not_first_call) not_first_call = True returned_output += "]" assert json.loads(returned_output) == iocs_json def test_create_csv_out_format(self): """ Given: - RequestArguments - Indicator info With CustomFields When: - request csv outbound format Then: - assert the result """ from EDL import RequestArguments, create_csv_out_format with open("test_data/demisto_url_iocs.json") as iocs_json_f: iocs_json = json.loads(iocs_json_f.read()) request_args = RequestArguments( query="", drop_invalids=True, url_port_stripping=True, url_protocol_stripping=True, fields_to_present="name,type" ) returned_output = "" not_first_call = False for ioc in iocs_json: returned_output += create_csv_out_format(not_first_call, ["value", "indicator_type", "test"], ioc, request_args) not_first_call = True assert ( returned_output == 'name,type\n"1.2.3.4/wget","URL","test"\n"www.demisto.com/cool","URL","test"\n' '"www.demisto.com/*cool","URL","None"' ) def test_create_mwg_out_format(self): """ Given: - RequestArguments - Indicator info When: - request mwg outbound format Then: - assert the result """ from EDL import RequestArguments, create_mwg_out_format with open("test_data/demisto_url_iocs.json") as iocs_json_f: iocs_json = json.loads(iocs_json_f.read()) request_args = RequestArguments(query="", drop_invalids=True, url_port_stripping=True, url_protocol_stripping=True) returned_output = "" not_first_call = False for ioc in iocs_json: returned_output += create_mwg_out_format(ioc, request_args, not_first_call) not_first_call = True assert ( returned_output == 'type=string\n"1.2.3.4/wget" "AutoFocus Feed"\n"www.demisto.com/cool" ' '"AutoFocus V2,VirusTotal,Alien Vault OTX TAXII Feed"\n"www.demisto.com/*cool" ' '"AutoFocus V2,VirusTotal,Alien Vault OTX TAXII Feed"' ) def test_create_proxysg_out_format(self): """ Given: - RequestArguments - Indicator info When: - request proxysg outbound format Then: - assert files_by_category as 3 keys - assert the result """ from EDL import RequestArguments, create_proxysg_all_category_out_format, create_proxysg_out_format files_by_category = {} with open("test_data/demisto_url_iocs.json") as iocs_json_f: iocs_json = json.loads(iocs_json_f.read()) request_args = RequestArguments(query="", drop_invalids=True, url_port_stripping=True, url_protocol_stripping=True) for ioc in iocs_json: files_by_category = create_proxysg_out_format(ioc, files_by_category, request_args) assert len(files_by_category) == 3 result_file = tempfile.TemporaryFile(mode="w+t") result_file = create_proxysg_all_category_out_format(result_file, files_by_category) result_file.seek(0) assert ( result_file.read() == "define category category1\n1.2.3.4/wget\nend\ndefine category category2\n" "www.demisto.com/cool\nend\ndefine category bc_category\nwww.demisto.com/*cool\n" "end" ) def test_validate_basic_authentication(self): """Test Authentication""" from EDL import validate_basic_authentication data = { "empty_auth": {}, "basic_missing_auth": {"Authorization": "missing basic"}, "colon_missing_auth": {"Authorization": "Basic bWlzc2luZ19jb2xvbg=="}, "wrong_length_auth": {"Authorization": "Basic YTpiOmM="}, "wrong_credentials_auth": {"Authorization": "Basic YTpi"}, "right_credentials_auth": {"Authorization": "Basic dXNlcjpwd2Q="}, } username, password = "user", "pwd" assert not validate_basic_authentication(data.get("empty_auth"), username, password) assert not validate_basic_authentication(data.get("basic_missing_auth"), username, password) assert not validate_basic_authentication(data.get("colon_missing_auth"), username, password) assert not validate_basic_authentication(data.get("wrong_length_auth"), username, password) assert not validate_basic_authentication(data.get("wrong_credentials_auth"), username, password) assert validate_basic_authentication(data.get("right_credentials_auth"), username, password) def test_ips_to_ranges_range(self): from EDL import COLLAPSE_TO_RANGES, ips_to_ranges ip_list = [ "1.1.1.1", "25.24.23.22", "22.21.20.19", "1.1.1.2", "1.2.3.4", "1.1.1.3", "2.2.2.2", "1.2.3.5", "3.3.3.0/30", "3.3.3.1", ] ip_range_list = ips_to_ranges(ip_list, COLLAPSE_TO_RANGES) assert "1.1.1.1-1.1.1.3" in ip_range_list assert "1.2.3.4-1.2.3.5" in ip_range_list assert "1.1.1.2" not in ip_range_list assert "2.2.2.2" in ip_range_list assert "25.24.23.22" in ip_range_list assert "3.3.3.0-3.3.3.3" in ip_range_list def test_ips_to_ranges_cidr(self): from EDL import COLLAPSE_TO_CIDR, ips_to_ranges ip_list = [ "1.1.1.1", "25.24.23.22", "22.21.20.19", "1.1.1.2", "1.2.3.4", "1.1.1.3", "2.2.2.2", "1.2.3.5", "3.3.3.0/30", "3.3.3.1", ] ip_range_list = ips_to_ranges(ip_list, COLLAPSE_TO_CIDR) assert "1.1.1.1" in ip_range_list assert "1.1.1.2/31" in ip_range_list assert "1.2.3.4/31" in ip_range_list assert "1.2.3.5" not in ip_range_list assert "1.1.1.3" not in ip_range_list assert "2.2.2.2" in ip_range_list assert "25.24.23.22" in ip_range_list assert "3.3.3.0/30" in ip_range_list def test_ips_to_cidrs_bad_ip(self): from EDL import COLLAPSE_TO_CIDR, ips_to_ranges ip_list = ["1.1.1.1", "1.1.1.2", "1.1.1.3", "1.2.3.4", "1.2.3.5", "doesntwork/oh"] ip_range_list = ips_to_ranges(ip_list, COLLAPSE_TO_CIDR) assert "1.1.1.1" in ip_range_list assert "1.1.1.2/31" in ip_range_list assert "1.2.3.4/31" in ip_range_list assert "doesntwork/oh" in ip_range_list assert "1.2.3.5" not in ip_range_list assert "1.1.1.3" not in ip_range_list def test_ips_to_ranges_bad_ip(self): from EDL import COLLAPSE_TO_RANGES, ips_to_ranges ip_list = ["1.1.1.1", "doesntwork/oh"] ip_range_list = ips_to_ranges(ip_list, COLLAPSE_TO_RANGES) assert "1.1.1.1" in ip_range_list assert "doesntwork/oh" in ip_range_list def test_is_valid_ip_ipv4(self): from EDL import is_valid_ip ip = "1.1.1.1" assert is_valid_ip(ip) def test_is_valid_ip_ipv6(self): from EDL import is_valid_ip ip = "2001:0db8:85a3:0000:0000:8a2e:0370:7334" assert is_valid_ip(ip) def test_is_valid_ip_cidr(self): from EDL import is_valid_ip cidr = "1.2.3.5/10" assert not is_valid_ip(cidr) def test_is_valid_ip_invalid_ip(self): from EDL import is_valid_ip ip = "This is not an IP, this is just a String" assert not is_valid_ip(ip) def test_is_valid_cidr_ipv4_network(self): from EDL import is_valid_cidr cidr = "1.2.3.5/10" assert is_valid_cidr(cidr) def test_is_valid_cidr_ipv6_network(self): from EDL import is_valid_cidr cidr = "2001:0db8:85a3:0000:0000:8a2e:0370:7334/10" assert is_valid_cidr(cidr) def test_is_valid_cidr_ip_address(self): from EDL import is_valid_cidr ip = "1.1.1.1" assert not is_valid_cidr(ip) def test_is_valid_cidr_not_a_cidr(self): from EDL import is_valid_cidr cidr = "This is not a CIDR / this is just a String" assert not is_valid_cidr(cidr) def test_is_large_cidr(self): """ Given: large CIDR (large cidr is with prefix smaller than 8 When: - Calling `is_valid_cidr` with `auto_block_large_cidrs=False` - Calling `is_valid_cidr` with `auto_block_large_cidrs=True` Then: - Should be valid if `auto_block_large_cidrs=False` - Should be invalid if `auto_block_large_cidrs=False` """ from EDL import is_large_cidr ipv4_large_cidr = "1.2.3.5/7" ipv6_large_cide = "2001:0db8:85a3:0000:0000:8a2e:0370:7334/8" assert is_large_cidr(ipv4_large_cidr, 9) assert is_large_cidr(ipv6_large_cide, 9) ipv4_small_cidr = "1.2.3.5/12" ipv6_small_cidr = "2001:0db8:85a3:0000:0000:8a2e:0370:7334/23" assert not is_large_cidr(ipv6_small_cidr, 11) assert not is_large_cidr(ipv4_small_cidr, 11) def test_is_large_cidr_not_valid(self): from EDL import is_large_cidr not_valid = "doesntwork/oh" assert not is_large_cidr(not_valid, 7) def test_get_bool_arg_or_param(self): """ Given: - bool arg that is not in args - bool arg that is in args and is in params - bool arg that is not in args but is in params When: - calling get_bool_arg_or_param Then: - return False - return arg bool equivelant value - return param bool value """ from EDL import get_bool_arg_or_param existing_key = "exists" missing_key = "missing" args = {existing_key: "true"} params = {existing_key: False, missing_key: True} # missing from both assert get_bool_arg_or_param(args, params, "") is False # exists in args and params assert get_bool_arg_or_param(args, params, existing_key) is True # exists in params only assert get_bool_arg_or_param(args, params, missing_key) is True def test_get_request_args(self): """ Test get_request_args sets RequestArgs with priority to request_args and params as fallback """ from EDL import COLLAPSE_TO_CIDR, COLLAPSE_TO_RANGES, get_request_args limit = 100 offset = "1" query = "q" strip_port = True drop_invalids = True add_comment_if_empty = True request_args = { "n": limit, "s": offset, "q": query, "sp": strip_port, "di": drop_invalids, "tr": 1, "ce": add_comment_if_empty, "v": "CSV", } params = { "edl_size": limit + 1, "indicators_query": query + "42", "url_port_stripping": not strip_port, "drop_invalids": not drop_invalids, "collapse_ips": 2, "add_comment_if_empty": not add_comment_if_empty, "format": "CSV", "no_wildcard_tld": False, "maximum_cidr_size": "8", } # request with no request_args res = get_request_args({}, params) assert res.limit == params["edl_size"] assert res.query == params["indicators_query"] assert res.url_port_stripping == params["url_port_stripping"] assert res.drop_invalids == params["drop_invalids"] assert res.collapse_ips == COLLAPSE_TO_CIDR assert res.add_comment_if_empty == params["add_comment_if_empty"] # request with full request_args res = get_request_args(request_args, params) assert res.limit == request_args["n"] assert res.query == request_args["q"] assert res.url_port_stripping == request_args["sp"] assert res.drop_invalids == request_args["di"] assert res.collapse_ips == COLLAPSE_TO_RANGES assert res.add_comment_if_empty == request_args["ce"] @pytest.mark.parametrize( "ip_networks, expected_output", [ pytest.param(["192.168.1.1/32"], {"192.168.1.1"}, id="single_ip_without_suffix"), pytest.param(["192.168.1.0/24"], {"192.168.1.0/24"}, id="small_cidr_block"), pytest.param(["10.0.0.0/24", "172.16.0.0/16"], {"10.0.0.0/24", "172.16.0.0/16"}, id="multiple_cidr_blocks"), pytest.param(["192.168.1.1/32", "10.0.0.0/8"], {"192.168.1.1", "10.0.0.0/8"}, id="mix_single_ip_and_cidr"), pytest.param(["0.0.0.0/0"], {"0.0.0.0/0"}, id="very_large_cidr_entire_ipv4_space"), ], ) def test_ip_groups_to_cidrs(self, ip_networks, expected_output): """ Given: - An iterable of IPNetwork objects representing IP addresses or CIDR blocks When: - Calling ip_groups_to_cidrs to collapse IP groups to CIDR notation Then: - Returns a set of CIDR strings with single IPs without /32 suffix - Handles very large CIDR blocks without raising IndexError """ from netaddr import IPNetwork from EDL import ip_groups_to_cidrs # Convert string CIDRs to IPNetwork objects ip_network_objects = [IPNetwork(cidr) for cidr in ip_networks] result = ip_groups_to_cidrs(ip_network_objects) assert result == expected_output @pytest.mark.parametrize( "ip_ranges, expected_output", [ pytest.param(["192.168.1.1"], {"192.168.1.1"}, id="single_ip"), pytest.param( ["192.168.1.1", "192.168.1.2", "192.168.1.3"], {"192.168.1.1-192.168.1.3"}, id="consecutive_ips_to_range" ), pytest.param(["192.168.1.1", "192.168.1.5"], {"192.168.1.1", "192.168.1.5"}, id="non_consecutive_ips"), pytest.param( ["10.0.0.1", "10.0.0.2", "172.16.0.1"], {"10.0.0.1-10.0.0.2", "172.16.0.1"}, id="mix_range_and_single_ip" ), pytest.param(["0.0.0.0/0"], {"0.0.0.0-255.255.255.255"}, id="very_large_range_entire_ipv4_space"), ], ) def test_ip_groups_to_ranges(self, ip_ranges, expected_output): """ Given: - An iterable of IP addresses or CIDR blocks When: - Calling ip_groups_to_ranges to collapse IP groups to range notation Then: - Returns a set of IP range strings (e.g., '192.168.1.1-192.168.1.3') - Single IPs are returned as-is without range notation - Handles very large IP ranges without raising IndexError """ from netaddr import IPSet from EDL import ip_groups_to_ranges # Convert IPs to IPRange objects using IPSet ip_range_objects = IPSet(ip_ranges).iter_ipranges() result = ip_groups_to_ranges(ip_range_objects) assert result == expected_output def test_initialize_edl_context(): """ Given: - the params When: - config the instance Then: - assert the integrationContext is saved properly """ params = { "edl_size": "200", "indicators_query": "*", "collapse_ips": DONT_COLLAPSE, "url_port_stripping": False, "url_protocol_stripping": True, "drop_invalids": True, "add_comment_if_empty": False, "mwg_type": "string", "category_default": "bc_category", "category_attribute": "test1,test2", "fields_filter": "value,type", "format": "CSV", "csv_text": True, "url_truncate": False, "maximum_cidr_size": "8", "no_wildcard_tld": True, } initialize_edl_context(params) assert demisto.integrationContext == { "last_query": "*", "out_format": "CSV", "last_limit": 200, "last_offset": 0, "drop_invalids": True, "url_port_stripping": False, "collapse_ips": "Don't Collapse", "add_comment_if_empty": False, "mwg_type": "string", "bc_category": "bc_category", "category_attribute": ["test1", "test2"], "fields_to_present": "name,type", "csv_text": True, "url_protocol_stripping": True, "url_truncate": False, "UpdateEDL": True, "maximum_cidr_size": 8, "no_wildcard_tld": True, } class IndicatorsSearcher: # an IndicatorsSearcher class for testing def __init__(self, limit=None): self._limit = limit self._corrent = 0 self.ioc = [ {"iocs": [{"value": "https://google.com", "indicator_type": "URL"}]}, {"iocs": [{"value": "demisto.com:7000", "indicator_type": "URL"}]}, {"iocs": [{"value": "demisto.com/qwertqwer", "indicator_type": "URL"}]}, {"iocs": [{"value": "demisto.com", "indicator_type": "URL"}]}, {"iocs": [{"value": "non ascii valuè", "indicator_type": "URL"}]}, ] def __iter__(self): return self def __next__(self): if self._corrent >= self._limit: raise StopIteration res = self.ioc[self._corrent] self._corrent += 1 return res @property def limit(self): return self._limit def test_get_indicators_to_format_csv(): """ Given: - IndicatorsSearcher with indicators - request_args of the coll When: - request indicators on csv format Then: - assert the indicators are returned properly for the requested format """ import EDL as edl indicator_searcher = IndicatorsSearcher(4) request_args = edl.RequestArguments( out_format="CSV", query="", limit=3, url_port_stripping=True, url_protocol_stripping=True, url_truncate=True, fields_to_present="name,type", ) indicators_data, _ = get_indicators_to_format(indicator_searcher, request_args) indicators_data.seek(0) indicators = indicators_data.read() assert indicators == 'name,type\n"google.com","URL"\n"demisto.com","URL"\n"demisto.com/qwertqwer","URL"\n"demisto.com","URL"' def test_get_indicators_to_format_json(): """ Given: - IndicatorsSearcher with indicators - request_args of the coll When: - request indicators on json format Then: - assert the indicators are returned properly for the requested format """ import EDL as edl indicator_searcher = IndicatorsSearcher(4) request_args = edl.RequestArguments( out_format="JSON", query="", limit=3, url_port_stripping=True, url_protocol_stripping=True, url_truncate=True, fields_to_present="name,type", ) f, _ = get_indicators_to_format(indicator_searcher, request_args) f.seek(0) indicators = f.read() assert ( indicators == '[{"value": "google.com", "indicator_type": "URL"},' ' {"value": "demisto.com", "indicator_type": "URL"},' ' {"value": "demisto.com/qwertqwer", "indicator_type": "URL"},' ' {"value": "demisto.com", "indicator_type": "URL"}]' ) def test_get_indicators_to_format_mwg(): """ Given: - IndicatorsSearcher with indicators - request_args of the coll When: - request indicators on mwg format Then: - assert the indicators are returned properly for the requested format """ import EDL as edl indicator_searcher = IndicatorsSearcher(4) request_args = edl.RequestArguments( out_format="McAfee Web Gateway", query="", limit=3, url_port_stripping=True, url_protocol_stripping=True, url_truncate=True, ) f, _ = get_indicators_to_format(indicator_searcher, request_args) f.seek(0) indicators = f.read() assert ( indicators == 'type=string\n"google.com" "from CORTEX XSOAR"\n"demisto.com" "from CORTEX XSOAR"\n' '"demisto.com/qwertqwer" "from CORTEX XSOAR"\n"demisto.com" "from CORTEX XSOAR"' ) def test_get_indicators_to_format_symantec(): """ Given: - IndicatorsSearcher with indicators - request_args of the coll When: - request indicators on symantec format Then: - assert the indicators are returned properly for the requested format """ import EDL as edl indicator_searcher = IndicatorsSearcher(4) request_args = edl.RequestArguments( out_format="Symantec ProxySG", query="", limit=3, url_port_stripping=True, url_protocol_stripping=True, url_truncate=True ) f, _ = get_indicators_to_format(indicator_searcher, request_args) f.seek(0) indicators = f.read() assert indicators == "define category bc_category\ngoogle.com\ndemisto.com\ndemisto.com/qwertqwer\ndemisto.com\nend" @pytest.mark.parametrize( "port, platform, expected_res", [ ("", {"platform": "xsiam", "version": "8.1.1"}, False), ("", {"platform": "xsiam", "version": "7.0.0"}, False), ("", {"platform": "xsoar", "version": "7.0.0"}, True), ("8888", {"platform": "xsoar", "version": "7.0.0"}, False), ("0000", {"platform": "xsoar", "version": "8.4.0"}, False), ], ) def test_no_port_param_lower_than_xsoar_8(port, platform, expected_res, mocker): """ Given: - longRuningPort param Empry with xsiam platform - longRuningPort param Empry with xsiam platform - longRuningPort param Empry with xsoar platform - longRuningPort param value with xsoar platform - longRuningPort param value with xsoar platform When: - Checking if the platform is xsoar and the version is less than 8.0.0 Then: Valdiate correct expected result - False - False - True - False - False """ mocker.patch.object(demisto, "demistoVersion", return_value=platform) assert check_platform_and_version({"longRunningPort": port}) == expected_res def test_get_indicators_to_format_text(): """ Given: - IndicatorsSearcher with indicators - request_args of the coll When: - request indicators on text format Then: - assert the indicators are returned properly for the requested format - assert the log is as expected """ import EDL as edl indicator_searcher = IndicatorsSearcher(4) request_args = edl.RequestArguments( out_format="PAN-OS (text)", query="", limit=3, url_port_stripping=True, url_protocol_stripping=True, url_truncate=True ) with tempfile.NamedTemporaryFile() as wip_log_file: edl.EDL_FULL_LOG_PATH_WIP = wip_log_file.name indicators_data, _ = get_indicators_to_format(indicator_searcher, request_args) indicators_data, indicators_stats = edl.create_text_out_format(indicators_data, request_args) indicators_log = wip_log_file.read() indicators_log = indicators_log.decode("utf-8") indicators_data.seek(0) indicators = indicators_data.read() assert indicators == "google.com\ndemisto.com\ndemisto.com/qwertqwer\ndemisto.com" expected_indicators_log = ( "\nAdded | google.com | https://google.com | Found new URL." "\nAdded | demisto.com | demisto.com:7000 | Found new URL." "\nAdded | demisto.com/qwertqwer | demisto.com/qwertqwer | Found new URL." "\nAdded | demisto.com | demisto.com | Found new URL." ) assert indicators_log == expected_indicators_log assert indicators_stats == {"Added": 4} def test_get_indicators_to_format_text_enforce_ascii(mocker): """ Given: - IndicatorsSearcher with indicators - request_args of the coll - enforce_ascii = True When: - request indicators on text format Then: - assert the indicators are returned properly for the requested format - assert the log is as expected """ import EDL as edl mocker.patch.object(demisto, "params", return_value={"enforce_ascii": True}) indicator_searcher = IndicatorsSearcher(5) request_args = edl.RequestArguments( out_format="PAN-OS (text)", query="", limit=3, url_port_stripping=True, url_protocol_stripping=True, url_truncate=True ) with tempfile.NamedTemporaryFile() as wip_log_file: edl.EDL_FULL_LOG_PATH_WIP = wip_log_file.name indicators_data, _ = get_indicators_to_format(indicator_searcher, request_args) indicators_data, indicators_stats = edl.create_text_out_format(indicators_data, request_args) indicators_log = wip_log_file.read() indicators_log = indicators_log.decode("utf-8") indicators_data.seek(0) indicators = indicators_data.read() assert indicators == "google.com\ndemisto.com\ndemisto.com/qwertqwer\ndemisto.com" expected_indicators_log = ( "\nAdded | google.com | https://google.com | Found new URL." "\nAdded | demisto.com | demisto.com:7000 | Found new URL." "\nAdded | demisto.com/qwertqwer | demisto.com/qwertqwer | Found new URL." "\nAdded | demisto.com | demisto.com | Found new URL." ) assert indicators_log == expected_indicators_log assert indicators_stats == {"Added": 4} @pytest.mark.parametrize( "raw_indicators, expected_indicators, expected_log, expected_stats", [ ( [{"value": "1.3.3.7", "indicator_type": "Domain"}], {"1.3.3.7"}, "Added | 1.3.3.7 | 1.3.3.7 | Found new Domain.", {"Added": 1}, ), ( [{"value": "1.1.1.1/7", "indicator_type": "CIDR"}], {""}, "Dropped | 1.1.1.1/7 | 1.1.1.1/7 | CIDR exceeds max length 8.", {"Dropped": 1}, ), ( [{"value": "www.Inv@l*id_token.com", "indicator_type": "Domain"}], {""}, "Dropped | www.Inv@l*id_token.com | www.Inv@l*id_token.com | Invalid tokens or port.", {"Dropped": 1}, ), ([{"value": "*.com", "indicator_type": "Domain"}], {""}, "Dropped | com | *.com | Domain is a TLD.", {"Dropped": 1}), ( [{"value": "http://www.google.com", "indicator_type": "URL"}], {"www.google.com"}, "Added | www.google.com | http://www.google.com | Found new URL.", {"Added": 1}, ), ( [{"value": "http://www.very_long_url_example_very_long.com", "indicator_type": "URL"}], {""}, "Dropped | www.very_long_url_example_very_long.com | http://www.very_long_url_example_very_long.com" " | URL exceeds max length 20.", {"Dropped": 1}, ), ( [ {"value": "1.3.3.7", "indicator_type": "IP"}, {"value": "1.3.3.8", "indicator_type": "IP"}, {"value": "1.3.3.10", "indicator_type": "IP"}, ], {"1.3.3.10", "1.3.3.7-1.3.3.8"}, "Modified | 1.3.3.7 | 1.3.3.7 | Collapsed IPv4 To Ranges.\n" "Modified | 1.3.3.8 | 1.3.3.8 | Collapsed IPv4 To Ranges.\n" "Added | 1.3.3.10 | 1.3.3.10 | Found new IPv4.", {"Added": 1, "Modified": 2}, ), ( [ {"value": "1.3.3.7", "indicator_type": "IPv6"}, {"value": "1.3.3.8", "indicator_type": "IPv6"}, {"value": "1.3.3.10", "indicator_type": "IPv6"}, ], {"1.3.3.10", "1.3.3.7-1.3.3.8"}, "Modified | 1.3.3.7 | 1.3.3.7 | Collapsed IPv6 To Ranges.\n" "Modified | 1.3.3.8 | 1.3.3.8 | Collapsed IPv6 To Ranges.\n" "Added | 1.3.3.10 | 1.3.3.10 | Found new IPv6.", {"Added": 1, "Modified": 2}, ), ], ) def test_create_log_str_from_indicators(raw_indicators, expected_indicators, expected_log, expected_stats): """ Given: - Indicator list. - Strict request arguments. Cases: Case 1: Valid new Domain. Case 2: CIDR exceeds max length. Case 3: Invalid tokens or port. Case 4: Domain is a TLD. Case 5: Valid new URL. Case 6: URL exceeds max length Case 7: Collapsed some IPv4 To Ranges. Case 8: Collapsed some IPv6 To Ranges When: - Running create_text_out_format with out_format PAN-OS (text). Then: - Ensure the log lines are as expected. - Ensure the indicator list is as expected. """ import EDL as edl edl_request_args = edl.RequestArguments( out_format="PAN-OS (text)", query="", limit=3, url_port_stripping=True, url_protocol_stripping=True, url_truncate=False, drop_invalids=True, collapse_ips="To Ranges", no_wildcard_tld=True, maximum_cidr_size=8, ) indicators_file = tempfile.TemporaryFile(mode="w+t") for raw_indicator in raw_indicators: indicators_file.write(json.dumps(raw_indicator)) indicators_file.write("\n") indicators_file.seek(0) edl.PAN_OS_MAX_URL_LEN = 20 with tempfile.NamedTemporaryFile() as log_file, tempfile.NamedTemporaryFile() as wip_log_file: edl.EDL_FULL_LOG_PATH = log_file.name edl.EDL_FULL_LOG_PATH_WIP = wip_log_file.name indicators_data, indicators_stats = edl.create_text_out_format(indicators_file, edl_request_args) indicators_data.seek(0) indicators = indicators_data.read() expected_log_entries = expected_log.split("\n") log_lines = log_file.readlines() for log_line in log_lines: assert log_line in expected_log_entries indicator_lines = indicators.split("\n") assert set(indicator_lines) == expected_indicators assert indicators_stats == expected_stats def test_route_edl_log(mocker): """ Given: - Append and prepend strings in demisto params. - A stored log in the relevant file. When: - A request to the '/log' endpoint is sent. Then: - Ensure the contents of the returned log are as the one stored in the file with append and prepend strings. """ import EDL as edl mocker.patch.object( demisto, "params", return_value={ "append_string": "+append_string+", "prepend_string": "+prepend_string+", "cache_refresh_rate": "30 minutes", }, ) request_args = edl.RequestArguments() mocker.patch.object(edl, "authenticate_app", return_value=None) mocker.patch.object(edl, "get_request_args", return_value=request_args) log_content = "test log" with tempfile.NamedTemporaryFile() as log_file: edl.EDL_FULL_LOG_PATH = log_file.name with open(edl.EDL_FULL_LOG_PATH, "w+") as f: f.write(log_content) test_app = edl.APP.test_client() response = test_app.get("/log") assert response.status_code == 200 assert response.data.decode() == f"+prepend_string+\n{log_content}+append_string+" def test_route_edl_log_empty(mocker): """ Given: - Append and prepend strings in demisto params. - An empty log in the stored log file. When: - A request to the '/log' endpoint is sent. Then: - Ensure the comment '# Empty' is returned. """ import EDL as edl mocker.patch.object( demisto, "params", return_value={ "append_string": "+append_string+", "prepend_string": "+prepend_string+", "cache_refresh_rate": "30 minutes", }, ) request_args = edl.RequestArguments() mocker.patch.object(edl, "authenticate_app", return_value=None) mocker.patch.object(edl, "get_request_args", return_value=request_args) with tempfile.NamedTemporaryFile() as log_file: edl.EDL_FULL_LOG_PATH = log_file.name with open(edl.EDL_FULL_LOG_PATH, "w+") as f: f.write("") test_app = edl.APP.test_client() response = test_app.get("/log") assert response.status_code == 200 assert response.data.decode() == "# Empty" @freeze_time("2024-04-04 03:21:34") def test_route_edl_log_too_big(mocker): """ Given: - A stored log in the relevant file that is too big to display. When: - A request to the '/log' endpoint is sent. Then: - Ensure the first response indicates that a file will be returned. - Ensure the contents of the returned log are as the one stored in the file. - Ensure the file is returned and saved as zip. """ import EDL as edl mocker.patch.object(demisto, "params", return_value={"cache_refresh_rate": "30 minutes"}) request_args = edl.RequestArguments() edl.MAX_DISPLAY_LOG_FILE_SIZE = 2 mocker.patch.object(edl, "authenticate_app", return_value=None) mocker.patch.object(edl, "get_request_args", return_value=request_args) log_content = "test log" with tempfile.NamedTemporaryFile() as log_file: edl.EDL_FULL_LOG_PATH = log_file.name with open(edl.EDL_FULL_LOG_PATH, "w+") as f: f.write(log_content) test_app = edl.APP.test_client() first_response = test_app.get("/log") second_response = test_app.get("/log") assert first_response.status_code == 200 assert first_response.data.decode() == "# Log exceeds max size. Refresh to download as file." assert second_response.status_code == 200 with zipfile.ZipFile(io.BytesIO(second_response.data), "r") as zip_log_file: zip_file_list = zip_log_file.namelist() with zip_log_file.open(zip_file_list[0]) as extracted_file: decoded_log_data = extracted_file.read().decode("utf-8") assert decoded_log_data == log_content downloaded_expected_path = f"{edl.LOGS_ZIP_FILE_PREFIX}_{datetime.now().strftime('%Y%m%d-%H%M%S')}.zip" assert os.path.exists(downloaded_expected_path) os.remove(downloaded_expected_path) @pytest.mark.parametrize(argnames="wip_exist", argvalues=[True, False]) def test_store_log_data(mocker, wip_exist): """ Given: - previous log file exist/missing. When: - call to store_log_data. Then: - ensure full_log will create only if previous log exist """ from datetime import datetime from pathlib import Path import EDL as edl tmp_dir = mkdtemp() wip_log_file = Path(tmp_dir) / "wip_log_file" full_log_file = Path(tmp_dir) / "full_log_file" if wip_exist: wip_log_file.write_text("") mocker.patch.object(edl, "EDL_FULL_LOG_PATH_WIP", new=wip_log_file.absolute()) mocker.patch.object(edl, "EDL_FULL_LOG_PATH", new=full_log_file.absolute()) request_args = edl.RequestArguments() edl.store_log_data(request_args, datetime.now(), {}) assert Path(edl.EDL_FULL_LOG_PATH).exists() == wip_exist @pytest.mark.parametrize( "out_format, fields_to_present, expected", [ # Case 1: use_legacy_query returns "" pytest.param(FORMAT_TEXT, "use_legacy_query", "", id="legacy_query_mode"), # Case 2: FORMAT_CSV with 'all' returns "" pytest.param(FORMAT_CSV, "all", "", id="csv_all_fields"), # Case 3: FORMAT_JSON with 'value' replaced to 'name' pytest.param(FORMAT_JSON, "value,type", "name,type", id="json_value_to_name"), # Case 4: FORMAT_MWG with no fields_to_present pytest.param(FORMAT_MWG, "", RequestArguments.FILTER_FIELDS_ON_FORMAT_MWG, id="mwg_default_fields"), # Case 5: FORMAT_PROXYSG with no fields_to_present pytest.param(FORMAT_PROXYSG, "", RequestArguments.FILTER_FIELDS_ON_FORMAT_PROXYSG, id="proxysg_default_fields"), # Case 6: Unknown format fallback to FILTER_FIELDS_ON_FORMAT_TEXT pytest.param("unknown_format", "", RequestArguments.FILTER_FIELDS_ON_FORMAT_TEXT, id="unknown_format_fallback"), # Case 7: Empty string returns "" (fix for XSUP-67083 - legacy mode preservation) pytest.param(FORMAT_TEXT, "", "", id="empty_string_returns_empty"), ], ) def test_get_fields_to_present(out_format, fields_to_present, expected): """ Given: - Various output formats and fields_to_present values When: - Creating RequestArguments with different field configurations Then: - Ensure get_fields_to_present returns the correct field list - Ensure empty string is treated as legacy query mode (returns "") """ args = RequestArguments(out_format=out_format, fields_to_present=fields_to_present) assert args.fields_to_present == expected @pytest.mark.parametrize( "indicator_value, indicator_type, expected_output", [ pytest.param("*.example.org", "DomainGlob", ["example.org", "*.example.org"], id="domainglob_with_wildcard_prefix"), pytest.param("*.example.org", "URL", ["example.org", "*.example.org"], id="url_with_wildcard_prefix"), pytest.param("example.com", "Domain", ["example.com"], id="domain_without_wildcard"), ], ) def test_domain_glob_wildcard_expansion(indicator_value: str, indicator_type: str, expected_output: list): """ Given: - Indicators with different types (DomainGlob, URL, Domain) and values When: - Processing indicators through create_text_out_format for PAN-OS text output Then: - Ensure DomainGlob indicators produce both wildcard and bare domain forms - Ensure URL wildcards are handled correctly - Ensure regular domains produce only the domain itself - Ensure DomainGlob type triggers expansion even without "*." prefix """ import json from io import StringIO from EDL import create_text_out_format # Create indicator data in memory (no file writing) indicator_json = json.dumps({"value": indicator_value, "indicator_type": indicator_type}) indicators_data = StringIO(indicator_json + "\n") # Create request arguments for PAN-OS text format request_args = RequestArguments(out_format=FORMAT_TEXT) # Process the indicator output, _ = create_text_out_format(indicators_data, request_args) # Parse the output - seek to beginning and read output.seek(0) output_content = output.read().strip() output_lines = output_content.split("\n") if output_content else [] # Verify the output matches expected assert sorted(output_lines) == sorted(expected_output) def test_get_request_id_without_request_context(mocker): """ Given: - get_request_id is called outside of an active Flask request context. When: - Accessing request.headers raises a RuntimeError. Then: - A fresh 12-char hex id is generated and a debug log is emitted. """ mock_request = mocker.MagicMock() mock_request.headers.get.side_effect = RuntimeError("Working outside of request context.") mocker.patch.object(edl, "request", mock_request) debug_mock = mocker.patch.object(demisto, "debug") request_id = edl.get_request_id() assert len(request_id) == 12 assert all(c in "0123456789abcdef" for c in request_id) debug_mock.assert_called_once() def test_get_request_id_reuses_forwarded_header(mocker): """ Given: - get_request_id is called within a request context that carries X-Request-ID. When: - The forwarded header is present. Then: - The forwarded id is reused as-is (so it can be grepped across NGINX -> WSGI -> EDL). """ mock_request = mocker.MagicMock() mock_request.headers.get.return_value = "forwarded-id-123" mocker.patch.object(edl, "request", mock_request) assert edl.get_request_id() == "forwarded-id-123" mock_request.headers.get.assert_called_once_with("X-Request-ID") def test_get_indicators_to_format_restores_stdout_lock_timeout(mocker): """ Given: - A server runtime that exposes demisto._stdout_lock_timeout. When: - get_indicators_to_format iterates over indicators (temporarily raising the timeout). Then: - The original _stdout_lock_timeout value is restored after the iteration completes. """ demisto._stdout_lock_timeout = 60 try: indicator_searcher = IndicatorsSearcher(4) request_args = edl.RequestArguments( out_format="TEXT", query="", limit=3, url_port_stripping=True, url_protocol_stripping=True, url_truncate=True, ) get_indicators_to_format(indicator_searcher, request_args) assert demisto._stdout_lock_timeout == 60 finally: delattr(demisto, "_stdout_lock_timeout") def test_get_indicators_to_format_noop_when_stdout_lock_timeout_absent(mocker): """ Given: - A server runtime that does NOT expose demisto._stdout_lock_timeout. When: - get_indicators_to_format iterates over indicators. Then: - The attribute is not created (the timeout tweak is a true no-op). """ if hasattr(demisto, "_stdout_lock_timeout"): delattr(demisto, "_stdout_lock_timeout") indicator_searcher = IndicatorsSearcher(4) request_args = edl.RequestArguments( out_format="TEXT", query="", limit=3, url_port_stripping=True, url_protocol_stripping=True, url_truncate=True, ) get_indicators_to_format(indicator_searcher, request_args) assert not hasattr(demisto, "_stdout_lock_timeout")