from collections import namedtuple from copy import deepcopy from unittest.mock import MagicMock, patch import demistomock as demisto import pytest import SplunkPy as splunk from CommonServerPython import * from pytest_mock import MockerFixture from splunklib import client, results from splunklib.binding import AuthenticationError, HTTPError RETURN_ERROR_TARGET = "SplunkPy.return_error" DICT_RAW_RESPONSE = ( '"1528755951, url="https://test.url.com", search_name="NG_SIEM_UC25- High number of hits against ' 'unknown website from same subnet", action="allowed", dest="bb.bbb.bb.bbb , cc.ccc.ccc.cc , ' 'xx.xx.xxx.xx , yyy.yy.yyy.yy , zz.zzz.zz.zzz , aa.aa.aaa.aaa", distinct_hosts="5", ' 'first_3_octets="1.1.1", first_time="06/11/18 17:34:07 , 06/11/18 17:37:55 , 06/11/18 17:41:28 , ' '06/11/18 17:42:05 , 06/11/18 17:42:38", info_max_time="+Infinity", info_min_time="0.000", ' 'src="xx.xx.xxx.xx , yyy.yy.yyy.yy , zz.zzz.zz.zzz , aa.aa.aaa.aaa", u_category="unknown", ' 'user="xyz\\a1234 , xyz\\b5678 , xyz\\c91011 , xyz\\d121314 , unknown", website="2.2.2.2""' ) DICT_RAW_RESPONSE_WITH_MESSAGE_ID = ( '"1528755951, message-id="1", url="https://test.url.com", ' 'search_name="NG_SIEM_UC25- High number of hits against ' 'unknown website from same subnet", action="allowed", dest="bb.bbb.bb.bbb , ' 'cc.ccc.ccc.cc , xx.xx.xxx.xx , yyy.yy.yyy.yy , zz.zzz.zz.zzz , aa.aa.aaa.aaa", ' 'distinct_hosts="5", ' 'first_3_octets="1.1.1", first_time="06/11/18 17:34:07 , ' "06/11/18 17:37:55 , 06/11/18 17:41:28 , " '06/11/18 17:42:05 , 06/11/18 17:42:38", info_max_time="+Infinity", info_min_time="0.000", ' 'src="xx.xx.xxx.xx , yyy.yy.yyy.yy , zz.zzz.zz.zzz , aa.aa.aaa.aaa", u_category="unknown", ' 'user="xyz\\a1234 , xyz\\b5678 , xyz\\c91011 , xyz\\d121314 , unknown", website="2.2.2.2""' ) LIST_RAW = ( "Feb 13 09:02:55 1,2020/02/13 09:02:55,001606001116,THREAT,url," "1,2020/02/13 09:02:55,10.1.1.1,1.2.3.4,0.0.0.0,0.0.0.0,rule1,jordy,,web-browsing,vsys1,trust,untrust," "ethernet1/2,ethernet1/1,forwardAll,2020/02/13 09:02:55,59460,1,62889,80,0,0,0x208000,tcp,alert," '"ushship.com/xed/config.bin",(9999),not-resolved,informational,client-to-server,' "0,0x0,1.1.22.22-5.6.7.8,United States,0,text/html" ) RAW_WITH_MESSAGE = ( '{"@timestamp":"2019-10-15T13:30:08.578-04:00","message":"{"TimeStamp":"2019-10-15 13:30:08",' '"CATEGORY_1":"CONTACT","ASSOCIATEOID":"G2N2TJETBRAAX68V","HOST":' '"step-up-authentication-api.gslb.es.oneadp.com","SCOPE[4]":"PiSvcsProvider\\/payroll","SCOPE[19]":' '"\\/api\\/events\\/core\\/v1\\/user-status","CONTEXT":"\\/smsstepup","FLOW":"API","X-REAL-IP":' '"2.2.2.2","PRODUCT_CODE":"WFNPortal","X-FORWARDED-PROTO":"http","ERROR_ID":"4008",' '"SCOPE[23]":"\\/security\\/notification-communication-response-value.accept","REQ_URL":' '"http:\\/\\/step-up-authentication-api.gslb.es.blabla.com\\/smsstepup\\/events\\/core\\/v1\\/step-up-' 'user-authorization-request.evaluate","SCOPE[35]":"autopay\\/payroll\\/v1\\/cafeteria-plan-' 'configurations\\/{configurationItemID}","SCOPE_MATCHED":"Y","SCOPE[43]":"communication\\/n' 'otification-message-template.add","SCOPE[11]":"\\/ISIJWSUserSecurity","SCOPE[27]":"autopay\\/events' '\\/payroll\\/v1\\/earning-configuration.add","ORGOID":"G2SY6MR3ATKA232T","SCOPE[8]":"\\/' 'ISIJWSAssociatesService","SCOPE[39]":"autopay\\/payroll\\/v1\\/earning-configurations",' '"SETUP_SELF":"N","SCOPE[47]":"communication\\/notification.publish","SCOPE[15]":"' '\\/OrganizationSoftPurge","X-FORWARDED-HOST":"step-up-authentication-api.gslb.es.blabla.com",' '"ADP-MESSAGEID":"a1d57ed2-1fe6-4800-be7a-26cd89bhello","CNAME":"JRJG INC","CONTENT-LENGTH":' '"584","SCOPE[31]":"autopay\\/events\\/payroll\\/v1\\/earning-configuration.remove","CID":"BSTAR00044"' ',"ACTOR_UID":"ABinters@BSTAR00044","SECURE_API_MODE":"HTTPS_SECURE","X-REQUEST-ID":' '"2473a981bef27bc8444e510adc12234a","SCOPE[1]":"AVSSCP\\/Docstash\\/Download","SCOPE[18]":' '"\\/api\\/events\\/core\\/v1\\/product-role.assign","BLOCK_SESSION":"Y","CONSUMER_ID":' '"ab2e715e-41c4-43d6-bff7-fc2d713hello","SCOPE[34]":"autopay\\/payroll\\/v1\\/cafeteria-plan-' 'configurations","SCOPE[46]":"communication\\/notification-message-template.remove","MODULE":' '"STEPUP_API","SCOPE[9]":"\\/ISIJWSClientService","SCOPE[10]":"\\/ISIJWSJobsService","SCOPE[22]":' '"\\/api\\/person-account-registration","SCOPE[38]":"autopay\\/payroll\\/v1\\/deposit-configurations",' '"SUBJECT_ORGOID":"G2SY6MR3ATKA232T","SCOPE[5]":"\\/Associate","SCOPE[14]":"\\/Organization",' '"SCOPE[26]":"WFNSvcsProvider\\/payrollPi","EVENT_ID":"9ea87118-5679-5b0e-a67f-1abd8ccabcde",' '"SCOPE[30]":"autopay\\/events\\/payroll\\/v1\\/earning-configuration.payroll-accumulators.modify",' '"X-FORWARDED-PORT":"80","SCOPE[42]":"autopay\\/payroll\\/v1\\/worker-employment-records","JTI":' '"867b6d06-47cf-40ab-8dd7-bd0d57babcde","X-DOMAIN":"secure.api.es.abc.com","SOR_CODE":' '"WFNPortal","SCOPE[29]":"autopay\\/events\\/payroll\\/v1\\/earning-configuration.configuration' '-tags.modify","SCOPE[2]":"AVSSCP\\/Docstash\\/Get","OUTPUT_TYPE":"FAIL","ERR_MSG":"BLOCK_SESSION",' '"TRANS_ID":"3AF-D30-7CTTCQ","SCOPE[45]":"communication\\/notification-message-template.read",' '"USE_HISTORY":"Y","SCHEME":"http","SCOPE[13]":"\\/ISIJWSUsersService","SCOPE[21]":"\\/api\\/person",' '"SCOPE[33]":"autopay\\/events\\/payroll\\/v1\\/worker-insurable-payments.modify","X-FORWARDED-FOR":' '"8.8.8.8, 10.10.10.10, 1.2.3.4, 5.6.7.8","SCOPE[17]":"\\/api\\/core\\/v1\\/organization",' '"SCOPE[25]":"\\/step-up-user-authorization.initiate","SCOPE[6]":"\\/Associate\\/PIC","SCOPE[37]":' '"autopay\\/payroll\\/v1\\/cafeteria-plan-configurations\\/{configurationItemID}\\/' 'payroll-item-configurations\\/{payrollItemID}","FLOW_TYPE":"REST","SCOPE[41]":' '"autopay\\/payroll\\/v1\\/payroll-output","CONSUMERAPPOID":"WFNPortal","RESOURCE":' '"\\/events\\/core\\/v1\\/step-up-user-authorization-request.evaluate","USER-AGENT":' '"Apache-HttpClient\\/4.5.5 (Java\\/10.0.1)","SCOPE[3]":"AVSSCP\\/Docstash\\/List",' '"SUB_CATEGORY_1":"worker.businessCommunication.email.change","TIME":"9","X-SCHEME":' '"http","ADP-CONVERSATIONID":"stY46PpweABoT5JX04CZGCeBbX8=","SCOPE[12]":' '"\\/ISIJWSUserSecurityService","SCOPE[24]":"\\/step-up-user-authorization-request.evaluate",' '"SCOPE[32]":"autopay\\/events\\/payroll\\/v1\\/retro-pay-request.add","SCOPE[44]":' '"communication\\/notification-message-template.change","ACTION":"POST","SCOPE[7]":' '"\\/AssociateSoftPurge","SCOPE[16]":"\\/api\\/authentication","X-ORIGINAL-URI":' '"\\/smsstepup\\/events\\/core\\/v1\\/step-up-user-authorization-request.evaluate","SCOPE[28]":' '"autopay\\/events\\/payroll\\/v1\\/earning-configuration.change","SCOPE[36]":' '"autopay\\/payroll\\/v1\\/cafeteria-plan-configurations\\/{configurationItemID}\\/payroll-item' '-configurations","SESSION_ID":"f50be909-9e4f-408d-bf77-68499012bc35","SCOPE[20]":' '"\\/api\\/events\\/core\\/v1\\/user.provision","SUBJECT_AOID":"G370XX6XYCABCDE",' '"X-ORIGINAL-FORWARDED-FOR":"1.1.1.1, 3.3.3.3, 4.4.4.4","SCOPE[40]":' '"autopay\\/payroll\\/v1\\/employer-details"}","TXID":"3AF-D30-ABCDEF","ADP-MessageID":' '"a1d57ed2-1fe6-4800-be7a-26cd89bf686d","SESSIONID":"stY46PpweFToT5JX04CZGMeCvP8=","ORGOID":' '"G2SY6MR3ATKA232T","AOID":"G2N2TJETBRAAXAAA","MSGID":"a1d57ed2-1fe6-0000-be7a-26cd89bf686d"}' ) SAMPLE_RESPONSE = [ results.Message("INFO-TEST", "test message"), { "_bkt": "notable~668~66D21DF4-F4FD-4886-A986-82E72ADCBFE9", "_cd": "668:17198", "_indextime": "1596545116", "_raw": '1596545116, search_name="Endpoint - Recurring Malware Infection - Rule", count="17", ' 'day_count="8", dest="ACME-workstation-012", info_max_time="1596545100.000000000", ' 'info_min_time="1595939700.000000000", info_search_time="1596545113.965466000", ' 'signature="Trojan.Gen.2"', "_serial": "50", "_si": ["ip-172-31-44-193", "notable"], "_sourcetype": "stash", "_time": "2020-08-04T05:45:16.000-07:00", "dest": "ACME-workstation-012", "dest_asset_id": "028877d3c80cb9d87900eb4f9c9601ea993d9b63", "dest_asset_tag": ["cardholder", "pci", "americas"], "dest_bunit": "americas", "dest_category": ["cardholder", "pci"], "dest_city": "Pleasanton", "dest_country": "USA", "dest_ip": "192.168.3.12", "dest_is_expected": "TRUE", "dest_lat": "37.694452", "dest_long": "-121.894461", "dest_nt_host": "ACME-workstation-012", "dest_pci_domain": ["trust", "cardholder"], "dest_priority": "medium", "dest_requires_av": "TRUE", "dest_risk_object_type": "system", "dest_risk_score": "15680", "dest_should_timesync": "TRUE", "dest_should_update": "TRUE", "host": "ip-172-31-44-193", "host_risk_object_type": "system", "host_risk_score": "0", "index": "notable", "linecount": "1", "priorities": "medium", "priority": "medium", "risk_score": "15680", "rule_description": "Endpoint - Recurring Malware Infection - Rule", "rule_name": "Endpoint - Recurring Malware Infection - Rule", "rule_title": "Endpoint - Recurring Malware Infection - Rule", "security_domain": "Endpoint - Recurring Malware Infection - Rule", "severity": "unknown", "signature": "Trojan.Gen.2", "source": "Endpoint - Recurring Malware Infection - Rule", "sourcetype": "stash", "splunk_server": "ip-172-31-44-193", "urgency": "low", "owner": "unassigned", "event_id": "66D21DF4-F4FD-4886-A986-82E72ADCBFE9@@notable@@5aa44496ec8e5cf45c78ab230189a4ca", }, { "_bkt": "notable~3252~66D21DF4-F4FD-4886-A986-82E72ADCBFE9", "_cd": "3252:4913", "_eventtype_color": "none", "_indextime": "1737544322", "_raw": '1596545116, search_name="Endpoint - Recurring Malware Infection - Rule", count="17", ' 'day_count="8", dest="ACME-workstation-012", info_max_time="1596545100.000000000", ' 'info_min_time="1595939700.000000000", info_search_time="1596545113.965466000", ' 'signature="Trojan.Gen.2"', "_serial": "12", "_si": ["ip-1-1-1-1", "notable"], "_sourcetype": "stash", "_time": "2025-01-22T11:12:02.000+00:00", "comment": [ "change all fields", "changed to in progress", "changed to pending", ], "count": "1", "drilldown_earliest": "0.000", "drilldown_earliest_offset": "0.000", "drilldown_latest": "+Infinity", "drilldown_latest_offset": "+Infinity", "drilldown_name": "View infections on ACME-code-001", "drilldown_search": '| from datamodel:"Malware"."Malware_Attacks" | search dest="ACME-code-001"', "event_hash": "734b6c7bcd700ccd0449575164772230", "event_id": "test_id", "eventtype": "modnotable_results notable modnotable_results notable", "extract_assets": '["src", "dest", "dvc", "orig_host"]', "extract_identities": '["src_user", "user"]', "host": "ip-172-31-44-193", "host_risk_object_type": "system", "host_risk_score": "0", "index": "notable", "indexer_guid": "66D21DF4-F4FD-4886-A986-82E72ADCBFE9", "info_max_time": "+Infinity", "info_min_time": "0.000", "info_search_time": "1737504174.093091000", "investigation_profiles": "{}", "lastTime": "1737544072", "linecount": "2", "orig_action_name": "notable", "orig_rid": "0.6979", "owner": "test_owner", "owner_realname": "test_owner", "priorities": "critical", "priority": "critical", "review_time": ["1737547610.488234", "1737545623.764639", "1737545033.187136"], "reviewer": ["test_owner", "test_owner", "test_owner"], "risk_score": "24160", "rule_description": "A high or critical priority host (ACME-code-001) was detected with malware.", "rule_id": "test_id", "rule_name": "High Or Critical Priority Host With Malware Detected", "rule_title": "High Or Critical Priority Host With Malware Detected", "savedsearch_description": "Alerts when an infection is noted on a host with high or critical priority.", "search_name": "Endpoint - High Or Critical Priority Host With Malware - Rule", "security_domain": "endpoint", "severity": "high", "signature": "127", "source": "Endpoint - High Or Critical Priority Host With Malware - Rule", "sourcetype": "stash", "splunk_server": "ip-1-1-1-1", "status": "3", "status_default": "false", "status_description": "Closure is pending some action.", "status_end": "false", "status_group": "Open", "status_label": "Pending", "tag": ["modaction_result", "test_user"], "tag::eventtype": "modaction_result", "timestamp": "none", "urgency": "informational", "last_modified_timestamp": "1737547610.488234", }, ] SAMPLE_INCIDENT_REVIEW_RESPONSE = [ { "_key": "test_id_1737547610.49", "comment": "test comment", "last_modified_timestamp": "1737547610.488234", "owner": "test_owner", "owner_realname": "test_owner", "reviewer": "test_owner", "reviewer_realname": "test_owner", "rule_id": "test_id", "rule_name": "High Or Critical Priority Host With Malware Detected", "status": "3", "status_default": "false", "status_description": "Closure is pending some action.", "status_end": "false", "status_group": "Open", "status_label": "Pending", "urgency": "informational", } ] EXPECTED = { "action": "allowed", "dest": "bb.bbb.bb.bbb , cc.ccc.ccc.cc , xx.xx.xxx.xx , yyy.yy.yyy.yy , zz.zzz.zz.zzz , aa.aa.aaa.aaa", "distinct_hosts": "5", "first_3_octets": "1.1.1", "first_time": "06/11/18 17:34:07 , 06/11/18 17:37:55 , 06/11/18 17:41:28 , 06/11/18 17:42:05 , 06/11/18 17:42:38", "info_max_time": "+Infinity", "info_min_time": "0.000", "search_name": "NG_SIEM_UC25- High number of hits against unknown website from same subnet", "src": "xx.xx.xxx.xx , yyy.yy.yyy.yy , zz.zzz.zz.zzz , aa.aa.aaa.aaa", "u_category": "unknown", "user": "xyz\\a1234 , xyz\\b5678 , xyz\\c91011 , xyz\\d121314 , unknown", "website": "2.2.2.2", "url": "https://test.url.com", } EXPECTED_WITH_MESSAGE_ID = { "message-id": "1", "action": "allowed", "dest": "bb.bbb.bb.bbb , cc.ccc.ccc.cc , xx.xx.xxx.xx , yyy.yy.yyy.yy , zz.zzz.zz.zzz , aa.aa.aaa.aaa", "distinct_hosts": "5", "first_3_octets": "1.1.1", "first_time": "06/11/18 17:34:07 , 06/11/18 17:37:55 , 06/11/18 17:41:28 , 06/11/18 17:42:05 , 06/11/18 17:42:38", "info_max_time": "+Infinity", "info_min_time": "0.000", "search_name": "NG_SIEM_UC25- High number of hits against unknown website from same subnet", "src": "xx.xx.xxx.xx , yyy.yy.yyy.yy , zz.zzz.zz.zzz , aa.aa.aaa.aaa", "u_category": "unknown", "user": "xyz\\a1234 , xyz\\b5678 , xyz\\c91011 , xyz\\d121314 , unknown", "website": "2.2.2.2", "url": "https://test.url.com", } URL_TESTING_IN = '"url="https://test.com?key=val"' URL_TESTING_OUT = {"url": "https://test.com?key=val"} # checking a case where the last character for each value was cut RESPONSE = ( "NAS-IP-Address=2.2.2.2, NAS-Port=50222, NAS-Identifier=de-wilm-251littl-idf3b-s2, NAS-Port-Type=" "Ethernet, NAS-Port-Id=GigabitEthernet2/0/05" ) POSITIVE = { "NAS-IP-Address": "2.2.2.2", "NAS-Identifier": "de-wilm-251littl-idf3b-s2", "NAS-Port": "50222", "NAS-Port-Id": "GigabitEthernet2/0/05", "NAS-Port-Type": "Ethernet", } # testing the ValueError and json sections RAW_JSON = '{"Test": "success"}' RAW_STANDARD = '"Test="success"' RAW_JSON_AND_STANDARD_OUTPUT = {"Test": "success"} class Jobs: def __init__(self, status, service): self.oneshot = lambda x, **kwargs: x state = namedtuple("state", "content") self.state = state(content={"dispatchState": str(status)}) self.service = service def __getitem__(self, arg): return 0 def create(self, query, **kwargs): job = client.Job(sid="123456", service=self.service, **kwargs) job.resultCount = 0 job._state = self.state return job class Service: def __init__(self, status): self.jobs = Jobs(status, self) self.status = status self.disable_v2_api = False self.namespace = {"app": "test", "owner": "test", "sharing": "global"} self._abspath = lambda x, **kwargs: x def get(self, path_segment, owner=None, app=None, headers=None, sharing=None, **query): return {"status": "200", "body": "test", "headers": {"content-type": "application/json"}, "reason": "OK"} def job(self, sid): return self.jobs def test_raw_to_dict(): actual_raw = DICT_RAW_RESPONSE response = splunk.rawToDict(actual_raw) response_with_message = splunk.rawToDict(DICT_RAW_RESPONSE_WITH_MESSAGE_ID) list_response = splunk.rawToDict(LIST_RAW) raw_message = splunk.rawToDict(RAW_WITH_MESSAGE) empty = splunk.rawToDict("") url_test = splunk.rawToDict(URL_TESTING_IN) character_check = splunk.rawToDict(RESPONSE) assert response == EXPECTED assert response_with_message == EXPECTED_WITH_MESSAGE_ID assert list_response == {} assert raw_message.get("SCOPE[29]") == "autopay\\/events\\/payroll\\/v1\\/earning-configuration.configuration-tags.modify" assert isinstance(raw_message, dict) assert empty == {} assert url_test == URL_TESTING_OUT assert character_check == POSITIVE assert splunk.rawToDict(RAW_JSON) == RAW_JSON_AND_STANDARD_OUTPUT assert splunk.rawToDict(RAW_STANDARD) == RAW_JSON_AND_STANDARD_OUTPUT assert splunk.rawToDict('drilldown_search="key IN ("test1","test2")') == {"drilldown_search": "key IN (test1,test2)"} assert splunk.rawToDict( '123456, sample_account="sample1", sample_account="sample2", sample_account="sample3", distinct_count_ac="5"' ) == {"sample_account": "sample1, sample2, sample3", "distinct_count_ac": "5"} @pytest.mark.parametrize( "text, output", [ ("", [""]), ('"",', ['"",']), # a value shouldn't do anything special ("woopwoop", ["woopwoop"]), # a normal key value without quotes ("abc=123", ['abc="123"']), # add a comma at the end ("abc=123,", ['abc="123"']), # a normal key value with quotes ('cbd="123"', ['cbd="123"']), # check all wrapped with quotes removed ('"abc="123""', ['abc="123"']), # we need to remove 111 at the start. ('111, cbd="123"', ['cbd="123"']), # Testing with/without quotes and/or spaces: ("abc=123,cbd=123", ['abc="123"', 'cbd="123"']), ('abc=123,cbd="123"', ['abc="123"', 'cbd="123"']), ('abc="123",cbd=123', ['abc="123"', 'cbd="123"']), ('abc="123",cbd="123"', ['abc="123"', 'cbd="123"']), ("abc=123, cbd=123", ['abc="123"', 'cbd="123"']), ('abc=123, cbd="123"', ['abc="123"', 'cbd="123"']), ('cbd="123", abc=123', ['abc="123"', 'cbd="123"']), ('cbd="123",abc=123', ['abc="123"', 'cbd="123"']), # Continue testing quotes with more values: ("xyz=321,cbd=123,abc=123", ['xyz="321"', 'abc="123"', 'cbd="123"']), ('xyz=321,cbd="123",abc=123', ['xyz="321"', 'abc="123"', 'cbd="123"']), ('xyz="321",cbd="123",abc=123', ['xyz="321"', 'abc="123"', 'cbd="123"']), ('xyz="321",cbd="123",abc="123"', ['xyz="321"', 'abc="123"', 'cbd="123"']), # Testing nested quotes (the main reason for quote_group): # Try to remove the start 111. ('111, cbd="a="123""', ['cbd="a="123""']), ('cbd="a="123""', ['cbd="a="123""']), ('cbd="a="123", b=321"', ['cbd="a="123", b="321""']), ('cbd="a=123, b=321"', ['cbd="a="123", b="321""']), ('cbd="a=123, b="321""', ['cbd="a="123", b="321""']), ('cbd="a="123", b="321""', ['cbd="a="123", b="321""']), ('cbd="a=123, b=321"', ['cbd="a="123", b="321""']), ('xyz=123, cbd="a="123", b=321"', ['xyz="123"', 'cbd="a="123", b="321""']), ('xyz="123", cbd="a="123", b="321""', ['xyz="123"', 'cbd="a="123", b="321""']), ('xyz="123", cbd="a="123", b="321"", qqq=2', ['xyz="123"', 'cbd="a="123", b="321""', 'qqq="2"']), ('xyz="123", cbd="a="123", b="321"", qqq="2"', ['xyz="123"', 'cbd="a="123", b="321""', 'qqq="2"']), ], ) def test_quote_group(text, output): assert sorted(splunk.quote_group(text)) == sorted(output) data_test_replace_keys = [ ({}, {}), ({"test": "test"}, {"test": "test"}), ({"test.": "test."}, {"test_": "test."}), ({"te.st": "te.st"}, {"te_st": "te.st"}), ({"te[st": "te[st"}, {"te_st": "te[st"}), ({"te]st": "te]st"}, {"te_st": "te]st"}), ({"te)st": "te)st"}, {"te_st": "te)st"}), ({"te(st": "te(st"}, {"te_st": "te(st"}), ("", ""), (None, None), ] @pytest.mark.parametrize("dict_in, dict_out", data_test_replace_keys) def test_replace_keys(dict_in, dict_out): out = splunk.replace_keys(deepcopy(dict_in)) assert out == dict_out, f"replace_keys({dict_in}) got: {out} instead: {dict_out}" def test_parse_time_to_minutes_no_error(): splunk.FETCH_TIME = "3 hours" res = splunk.parse_time_to_minutes() assert res == 180 def test_parse_time_to_minutes_invalid_time_integer(mocker): return_error_mock = mocker.patch(RETURN_ERROR_TARGET) splunk.FETCH_TIME = "abc hours" splunk.parse_time_to_minutes() err_msg = return_error_mock.call_args[0][0] assert ( err_msg == "Error: Invalid fetch time, need to be a positive integer with the time unit afterwards " "e.g '2 months, 4 days'." ) def test_splunk_submit_event_hec_command(mocker): text = "a msg with a failure." class MockRes: def __init__(self, text): self.text = text mocker.patch.object(splunk, "splunk_submit_event_hec", return_value=MockRes(text)) return_error_mock = mocker.patch(RETURN_ERROR_TARGET) splunk.splunk_submit_event_hec_command(params={"hec_url": "mock_url"}, args={"entry_id": "some_entry"}, service=Service) err_msg = return_error_mock.call_args[0][0] assert err_msg == f"Could not send event to Splunk {text}" def check_request_channel(args: dict): """ Check if args contains a request_channel, return the proper text. Args: args: A dict of args. Returns: A MockResRequestChannel with the correct text value. """ if args.get("request_channel"): return MockResRequestChannel('{"text":"Success","code":0,"ackId":1}') else: return MockResRequestChannel('{"text":"Data channel is missing","code":10}') class MockResRequestChannel: def __init__(self, text): self.text = text def test_splunk_submit_event_hec_command_request_channel(mocker): """ Given - An args dict that contains a request_channel and a dummy params. When - Executing splunk_submit_event_hec_command function Then - The return result object contains the correct message. """ args = {"request_channel": "11111111-1111-1111-1111-111111111111", "entry_id": "some_entry"} mocker.patch.object(splunk, "splunk_submit_event_hec", return_value=check_request_channel(args)) moc = mocker.patch.object(demisto, "results") splunk.splunk_submit_event_hec_command(params={"hec_url": "mock_url"}, args=args, service=Service) readable_output = moc.call_args[0][0] assert readable_output == "The events were sent successfully to Splunk. AckID: 1" def test_splunk_submit_event_hec_command_without_request_channel(mocker): """ Given - An args dict that doesn't contain a request_channel and a dummy params. When - Executing splunk_submit_event_hec_command function Then - The return result object contains the correct message. """ args = {"entry_id": "some_entry"} mocker.patch.object(splunk, "splunk_submit_event_hec", return_value=check_request_channel(args)) return_error_mock = mocker.patch(RETURN_ERROR_TARGET) splunk.splunk_submit_event_hec_command(params={"hec_url": "mock_url"}, args=args, service=Service) err_msg = return_error_mock.call_args[0][0] assert err_msg == 'Could not send event to Splunk {"text":"Data channel is missing","code":10}' def test_parse_time_to_minutes_invalid_time_unit(mocker): return_error_mock = mocker.patch(RETURN_ERROR_TARGET) splunk.FETCH_TIME = "3 hoursss" splunk.parse_time_to_minutes() err_msg = return_error_mock.call_args[0][0] assert err_msg == "Error: Invalid time unit." SEARCH_RESULT = [ {"But": {"This": "is"}, "Very": "Unique"}, {"Something": "regular", "But": {"This": "is"}, "Very": "Unique"}, {"Something": "natural", "But": {"This": "is a very very"}, "Very": "Unique and awesome"}, ] REGULAR_ALL_CHOSEN_FIELDS = ["Something", "But", "Very"] REGULAR_CHOSEN_FIELDS_SUBSET = ["Something", "Very"] REGEX_CHOSEN_FIELDS_SUBSET = ["Some*", "Very"] NON_EXISTING_FIELDS = ["SDFAFSD", "ASBLFKDJK"] @pytest.mark.parametrize( "search_result, chosen_fields, expected_result", [ (SEARCH_RESULT, REGULAR_ALL_CHOSEN_FIELDS, REGULAR_ALL_CHOSEN_FIELDS), (SEARCH_RESULT, REGULAR_CHOSEN_FIELDS_SUBSET, REGULAR_CHOSEN_FIELDS_SUBSET), (SEARCH_RESULT, REGEX_CHOSEN_FIELDS_SUBSET, REGULAR_CHOSEN_FIELDS_SUBSET), (SEARCH_RESULT, NON_EXISTING_FIELDS, []), ], ) def test_commands(search_result, chosen_fields, expected_result): from SplunkPy import update_headers_from_field_names headers = update_headers_from_field_names(search_result, chosen_fields) assert expected_result == headers APPS = ["app"] STORES = ["store"] EMPTY_CASE = {} STORE_WITHOUT_APP = {"kv_store_collection_name": "test"} JUST_APP_NAME = {"app_name": "app"} # happens in splunk-kv-store-collections-list command CREATE_COMMAND = {"app_name": "app", "kv_store_name": "not_store"} CORRECT = {"app_name": "app", "kv_store_collection_name": "store"} INCORRECT_STORE_NAME = {"app_name": "app", "kv_store_collection_name": "not_store"} data_test_check_error = [ (EMPTY_CASE, "app not found"), (STORE_WITHOUT_APP, "app not found"), (JUST_APP_NAME, "empty"), (CREATE_COMMAND, "empty"), (CORRECT, "empty"), (INCORRECT_STORE_NAME, "KV Store not found"), ] @pytest.mark.parametrize("args, out_error", data_test_check_error) def test_check_error(args, out_error): class Service: def __init__(self): self.apps = APPS self.kvstore = STORES try: splunk.check_error(Service(), args) raise splunk.DemistoException("empty") except splunk.DemistoException as error: output = str(error) assert output == out_error, f"check_error(service, {args})\n\treturns: {output}\n\tinstead: {out_error}" EMPTY_CASE = {} JUST_KEY = {"key": "key"} WITH_ALL_PARAMS = {"key": "demisto", "value": "is awesome", "limit": 1, "query": "test"} STANDARD_KEY_VAL = {"key": "demisto", "value": "is awesome"} KEY_AND_LIMIT = {"key": "key", "limit": 1} KEY_AND_QUERY = {"key": "key", "query": "test_query"} QUERY = {"query": "test_query"} QUERY_AND_VALUE = {"query": "test_query", "value": "awesome"} data_test_build_kv_store_query = [ (EMPTY_CASE, str(EMPTY_CASE)), (JUST_KEY, str(EMPTY_CASE)), (STANDARD_KEY_VAL, '{"demisto": "is awesome"}'), (WITH_ALL_PARAMS, '{"demisto": "is awesome"}'), (KEY_AND_LIMIT, {"limit": 1}), (KEY_AND_QUERY, "test_query"), (QUERY, "test_query"), (QUERY_AND_VALUE, "test_query"), ] @pytest.mark.parametrize("args, expected_query", data_test_build_kv_store_query) def test_build_kv_store_query(args, expected_query, mocker): mocker.patch("SplunkPy.get_key_type", return_value=None) output = splunk.build_kv_store_query(None, args) assert output == expected_query, f"build_kv_store_query({args})\n\treturns: {output}\n\tinstead: {expected_query}" data_test_build_kv_store_query_with_key_val = [ ({"key": "demisto", "value": "is awesome"}, str, '{"demisto": "is awesome"}'), ({"key": "demisto", "value": "1"}, int, '{"demisto": 1}'), ({"key": "demisto", "value": "True"}, bool, '{"demisto": true}'), ] @pytest.mark.parametrize("args, _type, expected_query", data_test_build_kv_store_query_with_key_val) def test_build_kv_store_query_with_key_val(args, _type, expected_query, mocker): mocker.patch("SplunkPy.get_key_type", return_value=_type) output = splunk.build_kv_store_query(None, args) assert output == expected_query, f"build_kv_store_query({args})\n\treturns: {output}\n\tinstead: {expected_query}" test_test_get_key_type = [ ({"field.key": "number"}, float), ({"field.key": "string"}, str), ({"field.key": "cidr"}, str), ({"field.key": "boolean"}, bool), ({"field.key": "empty"}, None), ({"field.key": "time"}, str), ] @pytest.mark.parametrize("keys_and_types, expected_type", test_test_get_key_type) def test_get_key_type(keys_and_types, expected_type, mocker): mocker.patch("SplunkPy.get_keys_and_types", return_value=keys_and_types) output = splunk.get_key_type(None, "key") assert output == expected_type, f"get_key_type(kv_store, key)\n\treturns: {output}\n\tinstead: {expected_type}" EMPTY_CASE = {} WITHOUT_FIELD = {"empty": "number"} STRING_FIELD = {"field.test": "string"} NUMBER_FIELD = {"field.test": "number"} INDEX = {"index.test": "string"} MIXED = {"field.test": "string", "empty": "field"} data_test_get_keys_and_types = [ (EMPTY_CASE, EMPTY_CASE), (WITHOUT_FIELD, EMPTY_CASE), (STRING_FIELD, {"field.test": "string"}), (NUMBER_FIELD, {"field.test": "number"}), (INDEX, {"index.test": "string"}), (MIXED, {"field.test": "string"}), ] @pytest.mark.parametrize("raw_keys, expected_keys", data_test_get_keys_and_types) def test_get_keys_and_types(raw_keys, expected_keys): class KVMock: def __init__(self): pass def content(self): return raw_keys output = splunk.get_keys_and_types(KVMock()) assert output == expected_keys, f"get_keys_and_types(kv_store)\n\treturns: {output}\n\tinstead: {expected_keys}" START_OUTPUT = "#### configuration for name store\n| field name | type |\n| --- | --- |" EMPTY_OUTPUT = "" STANDARD_CASE = {"field.test": "number"} STANDARD_OUTPUT = "\n| field.test | number |" data_test_get_kv_store_config = [({}, EMPTY_OUTPUT), (STANDARD_CASE, STANDARD_OUTPUT)] @pytest.mark.parametrize("fields, expected_output", data_test_get_kv_store_config) def test_get_kv_store_config(fields, expected_output, mocker): class Name: def __init__(self): self.name = "name" mocker.patch("SplunkPy.get_keys_and_types", return_value=fields) output = splunk.get_kv_store_config(Name()) expected_output = f"{START_OUTPUT}{expected_output}" assert output == expected_output class TestFetchRemovingIrrelevantIncidents: notable1 = {"status": "5", "event_id": "3"} notable2 = {"status": "6", "event_id": "4"} # In order to mock the service.jobs.oneshot() call in the fetch_notables function, we need to create # the following two classes class Jobs: def __init__(self): self.oneshot = lambda x, **kwargs: TestFetchForLateIndexedEvents.notable1 class Service: def __init__(self): self.jobs = TestFetchForLateIndexedEvents.Jobs() def test_backwards_compatible(self, mocker: MockerFixture): """ Given - Incident IDs that were fetched in the last fetch round with the epoch time of their occurrence When - Fetching notables Then - Make sure that the last fetched IDs now hold the start of the fetch window, and not the epoch time """ from SplunkPy import UserMappingObject mocker.patch.object(demisto, "setLastRun") mock_last_run = { "time": "2024-02-12T10:00:00", "latest_time": "2024-02-19T10:00:00", "found_incidents_ids": {"1": 1700497516}, } mock_params = {"fetchQuery": "`notable` is cool", "fetch_limit": 2} mocker.patch("demistomock.getLastRun", return_value=mock_last_run) mocker.patch("demistomock.params", return_value=mock_params) mocker.patch("splunklib.results.JSONResultsReader", return_value=[self.notable1, self.notable2]) service = self.Service() set_last_run_mocker = mocker.patch("demistomock.setLastRun") mapper = UserMappingObject(service, False) splunk.fetch_incidents(service, mapper, "from_xsoar", "from_splunk") last_fetched_ids = set_last_run_mocker.call_args_list[0][0][0]["found_incidents_ids"] assert last_fetched_ids == { "1": {"occurred_time": "2024-02-19T10:00:00"}, "3": {"occurred_time": "2024-02-19T10:00:00"}, "4": {"occurred_time": "2024-02-19T10:00:00"}, } def test_remove_irrelevant_fetched_incident_ids(self, mocker: MockerFixture): """ Given - Incident IDs that were fetched in the last fetch round When - Fetching notables Then - Make sure that the fetched IDs that are no longer in the fetch window are removed """ from SplunkPy import UserMappingObject mocker.patch.object(demisto, "setLastRun") mock_last_run = { "time": "2024-02-12T10:00:00", "latest_time": "2024-02-19T10:00:00", "found_incidents_ids": {"1": {"occurred_time": "2024-02-12T08:59:59"}, "2": {"occurred_time": "2024-02-18T10:00:00"}}, } mock_params = {"fetchQuery": "`notable` is cool", "fetch_limit": 2} mocker.patch("demistomock.getLastRun", return_value=mock_last_run) mocker.patch("demistomock.params", return_value=mock_params) mocker.patch("splunklib.results.JSONResultsReader", return_value=[self.notable1, self.notable2]) service = self.Service() set_last_run_mocker = mocker.patch("demistomock.setLastRun") mapper = UserMappingObject(service, False) splunk.fetch_incidents(service, mapper, "from_xsoar", "from_splunk") last_fetched_ids = set_last_run_mocker.call_args_list[0][0][0]["found_incidents_ids"] assert last_fetched_ids == { "2": {"occurred_time": "2024-02-18T10:00:00"}, "3": {"occurred_time": "2024-02-19T10:00:00"}, "4": {"occurred_time": "2024-02-19T10:00:00"}, } def test_daylight_saving_time_delta_handling(self, mocker: MockerFixture): """ Given - Incident IDs that were fetched in the last fetch round - A fetch window that overlaps with a daylight saving time transition period When - Fetching notables and filtering irrelevant incident IDs Then - Make sure that incidents within the DAYLIGHT_SAVING_TIME_DELTA buffer are kept in cache - This prevents duplicate incidents during clock changes """ from SplunkPy import remove_irrelevant_incident_ids # Setup: Create a scenario where an incident is just outside the normal window # but within the DST delta buffer window_start_time = "2024-03-10T10:00:00" # DST transition date example window_end_time = "2024-03-10T12:00:00" # Incident that occurred 59 minutes before window start (within 1-hour DST buffer) incident_within_dst_buffer = "2024-03-10T09:01:00" # Incident that occurred 61 minutes before window start (outside DST buffer) incident_outside_dst_buffer = "2024-03-10T08:59:00" last_run_fetched_ids = { "incident_within_buffer": {"occurred_time": incident_within_dst_buffer}, "incident_outside_buffer": {"occurred_time": incident_outside_dst_buffer}, } # Execute the function filtered_ids = remove_irrelevant_incident_ids(last_run_fetched_ids, window_start_time, window_end_time) # Verify: Incident within DST buffer should be kept assert "incident_within_buffer" in filtered_ids, "Incident within DAYLIGHT_SAVING_TIME_DELTA should be kept in cache" # Verify: Incident outside DST buffer should be removed assert ( "incident_outside_buffer" not in filtered_ids ), "Incident outside DAYLIGHT_SAVING_TIME_DELTA should be removed from cache" # Verify the kept incident has the correct structure assert filtered_ids["incident_within_buffer"]["occurred_time"] == incident_within_dst_buffer class TestFetchForLateIndexedEvents: notable1 = {"status": "5", "event_id": "id_1"} notable2 = {"status": "6", "event_id": "id_2"} # In order to mock the service.jobs.oneshot() call in the fetch_notables function, we need to create # the following two classes class Jobs: def __init__(self): self.oneshot = lambda x, **kwargs: TestFetchForLateIndexedEvents.notable1 class Service: def __init__(self): self.jobs = TestFetchForLateIndexedEvents.Jobs() # If late_indexed_pagination is True, then we exclude the last fetched ids (check by using fetch query), # and kwargs_oneshot['offset'] == 0 def test_fetch_query_and_oneshot_args(self, mocker: MockerFixture): """ Given - Mocked incidents api response - The key "late_indexed_pagination" in the last run object is set to True - Some incident IDs that were fetched in the last fetch round When - Fetching notables Then - Make sure that last fetched incident IDs are specified to be excluded from the fetch query - Make sure that the offset of the fetch query is set to 0 """ from SplunkPy import UserMappingObject mocker.patch.object(demisto, "setLastRun") mock_last_run = { "time": "2018-10-24T14:13:20", "late_indexed_pagination": True, "found_incidents_ids": {"1234": 1700497516, "5678": 1700497516}, } mock_params = {"fetchQuery": "something"} mocker.patch("demistomock.getLastRun", return_value=mock_last_run) mocker.patch("demistomock.params", return_value=mock_params) mocker.patch("splunklib.results.JSONResultsReader", return_value=[self.notable1]) service = self.Service() oneshot_mocker = mocker.patch.object(service.jobs, "oneshot", side_effect=service.jobs.oneshot) mapper = UserMappingObject(service, False) splunk.fetch_incidents(service, mapper, "from_xsoar", "from_splunk") assert oneshot_mocker.call_args_list[0][0][0] == 'something | where not event_id in ("1234","5678")' assert oneshot_mocker.call_args_list[0][1]["offset"] == 0 # If (num_of_dropped == FETCH_LIMIT and '`notable`' in fetch_query), then late_indexed_pagination should be set to True def test_first_condition_for_late_indexed_pagination(self, mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch): """ Given - Incident IDs that were fetched in the last fetch round - Mocked incidents api response, that have IDs as the last fetched IDs (which means that num_of_dropped == FETCH_LIMIT) - `notable` is in the fetch query When - Fetching notables Then - Make sure that the key "late_indexed_pagination" in last run object is set to True """ from SplunkPy import UserMappingObject # MonkeyPatch can be used to patch global variables monkeypatch.setattr(splunk, "FETCH_LIMIT", 2) mocker.patch.object(demisto, "setLastRun") mock_last_run = {"time": "2018-10-24T14:13:20", "found_incidents_ids": {"id_1": 1700497516, "id_2": 1700497516}} mock_params = {"fetchQuery": "`notable` is cool", "fetch_limit": 2} mocker.patch("demistomock.getLastRun", return_value=mock_last_run) mocker.patch("demistomock.params", return_value=mock_params) mocker.patch("splunklib.results.JSONResultsReader", return_value=[self.notable1, self.notable2]) set_last_run_mocker = mocker.patch("demistomock.setLastRun") service = self.Service() mapper = UserMappingObject(service, False) splunk.fetch_incidents(service, mapper, "from_xsoar", "from_splunk") assert set_last_run_mocker.call_args_list[0][0][0]["late_indexed_pagination"] is True # If (len(incidents) == FETCH_LIMIT and late_indexed_pagination), then late_indexed_pagination should be set to True def test_second_condition_for_late_indexed_pagination(self, mocker: MockerFixture, monkeypatch: pytest.MonkeyPatch): """ Given - Incident IDs that were fetched in the last fetch round - Mocked incidents api response, where only new incidents are fetched (which means that len(incidents) == FETCH_LIMIT) - The key "late_indexed_pagination" in the last run object is set to True When - Fetching notables Then - Make sure that the key "late_indexed_pagination" in last run object is set to True """ from SplunkPy import UserMappingObject # MonkeyPatch can be used to patch global variables monkeypatch.setattr(splunk, "FETCH_LIMIT", 2) mocker.patch.object(demisto, "setLastRun") mock_last_run = { "time": "2018-10-24T14:13:20", "late_indexed_pagination": True, "found_incidents_ids": {"1234": 1700497516, "5678": 1700497516}, } mock_params = {"fetchQuery": "`notable` is cool", "fetch_limit": 2} mocker.patch("demistomock.getLastRun", return_value=mock_last_run) mocker.patch("demistomock.params", return_value=mock_params) mocker.patch("splunklib.results.JSONResultsReader", return_value=[self.notable1, self.notable2]) set_last_run_mocker = mocker.patch("demistomock.setLastRun") service = self.Service() mapper = UserMappingObject(service, False) splunk.fetch_incidents(service, mapper, "from_xsoar", "from_splunk") assert set_last_run_mocker.call_args_list[0][0][0]["late_indexed_pagination"] is True def test_fetch_incidents(mocker): """ Given - mocked incidents api response - a mapper which should not map the user owner into the incident response When - executing the fetch incidents flow Then - make sure the incident response is valid. - make sure that the owner is not part of the incident response """ from SplunkPy import UserMappingObject mocker.patch.object(demisto, "incidents") mocker.patch.object(demisto, "setLastRun") mock_last_run = {"time": "2018-10-24T14:13:20"} mock_params = {"fetchQuery": "something"} mocker.patch("demistomock.getLastRun", return_value=mock_last_run) mocker.patch("demistomock.params", return_value=mock_params) service = mocker.patch("splunklib.client.connect", return_value=None) mocker.patch("splunklib.results.JSONResultsReader", return_value=deepcopy(SAMPLE_RESPONSE)) mapper = UserMappingObject(service, False) splunk.fetch_incidents(service, mapper, "from_xsoar", "from_splunk") incidents = demisto.incidents.call_args[0][0] assert demisto.incidents.call_count == 1 assert len(incidents) == 2 assert incidents[0]["name"] == "Endpoint - Recurring Malware Infection - Rule : Endpoint - Recurring Malware Infection - Rule" assert not incidents[0].get("owner") SPLUNK_RESULTS = [ { "rawJSON": '{"source": "This is the alert type", "field_name1": "field_val1", "field_name2": "field_val2"}', "details": "Endpoint - High Or Critical Priority Host With Malware - Rule", "labels": [{"type": "security_domain", "value": "Endpoint - High Or Critical Priority Host With Malware - Rule"}], } ] EXPECTED_OUTPUT = { "This is the alert type": {"source": "This is the alert type", "field_name1": "field_val1", "field_name2": "field_val2"} } def test_create_mapping_dict(): mapping_dict = splunk.create_mapping_dict(SPLUNK_RESULTS, type_field="source") assert mapping_dict == EXPECTED_OUTPUT def test_fetch_notables(mocker): """ Given - mocked incidents api response - a mapper which should not map the user owner into the incident response When - executing the fetch notables flow Then - make sure the incident response is valid. - make sure that the owner is not part of the incident response """ mocker.patch.object(splunk.client.Job, "is_done", return_value=True) mocker.patch.object(splunk.client.Job, "results", return_value=None) mocker.patch.object( splunk, "ENABLED_ENRICHMENTS", [splunk.ASSET_ENRICHMENT, splunk.DRILLDOWN_ENRICHMENT, splunk.IDENTITY_ENRICHMENT] ) mocker.patch.object(demisto, "incidents") mocker.patch.object(demisto, "setLastRun") mock_last_run = {"time": "2018-10-24T14:13:20"} mock_params = {"fetchQuery": "something"} mocker.patch("demistomock.getLastRun", return_value=mock_last_run) mocker.patch("demistomock.params", return_value=mock_params) service = Service("DONE") mocker.patch("splunklib.results.JSONResultsReader", return_value=deepcopy(SAMPLE_RESPONSE)) mapper = splunk.UserMappingObject(service, False) splunk.fetch_incidents( service, mapper=mapper, comment_tag_to_splunk="comment_tag_to_splunk", comment_tag_from_splunk="comment_tag_from_splunk" ) cache_object = splunk.Cache.load_from_integration_context(get_integration_context()) assert cache_object.submitted_notables notable = cache_object.submitted_notables[0] incident_from_cache = notable.to_incident(mapper, "comment_tag_to_splunk", "comment_tag_from_splunk") incidents = demisto.incidents.call_args[0][0] assert demisto.incidents.call_count == 1 assert len(incidents) == 0 assert ( incident_from_cache["name"] == "Endpoint - Recurring Malware Infection - Rule : Endpoint - " "Recurring Malware Infection - Rule" ) assert not incident_from_cache.get("owner") # now call second time to make sure that the incident fetched splunk.fetch_incidents( service, mapper=mapper, comment_tag_to_splunk="comment_tag_to_splunk", comment_tag_from_splunk="comment_tag_from_splunk" ) incidents = demisto.incidents.call_args[0][0] assert len(incidents) == 2 assert incidents[0]["name"] == "Endpoint - Recurring Malware Infection - Rule : Endpoint - Recurring Malware Infection - Rule" assert not incidents[0].get("owner") def test_fetch_notables_with_creation_time1(mocker: MockerFixture): """ Given: A configuration using "creation time" as the notable time source in demisto parameters. When: The fetch_notables function is called. Then: The function should query Splunk using the earliest_time and latest_time fields in the search kwargs. """ mocker.patch.object( demisto, "params", return_value={"notable_time_source": "creation time", "fetchQuery": "something", "occurrence_look_behind": "0"}, ) mocker.patch.object(splunk, "parse_time_to_minutes", return_value=10) mocker.patch.object(results, "JSONResultsReader", return_value=[]) # Mock the service object mock_service = mocker.MagicMock() mock_search = mocker.MagicMock() mock_service.jobs.oneshot.return_value = mock_search # Mock the search results mock_search.results = mocker.MagicMock(return_value=[]) # Mock the mapper object mock_mapper = mocker.MagicMock() # Create a mock for the Cache mock_cache = mocker.MagicMock() # Call the function splunk.fetch_notables( service=mock_service, mapper=mock_mapper, comment_tag_to_splunk="comment_to_splunk", comment_tag_from_splunk="comment_from_splunk", cache_object=mock_cache, enrich_notables=False, ) # Verify that the service.jobs.oneshot was called with "creation time" in the kwargs call_args = mock_service.jobs.oneshot.call_args[1] # The query should include "creation time" in the search criteria assert "earliest_time" in call_args assert "latest_time" in call_args assert "index_earliest" not in call_args assert "index_latest" not in call_args def test_fetch_notables_with_index_time1(mocker: MockerFixture): """ Given: A configuration using "index time" as the notable time source in demisto parameters. When: The fetch_notables function is called. Then: The function should query Splunk using the index_earliest and index_latest fields in the search kwargs. """ mocker.patch.object( demisto, "params", return_value={"notable_time_source": "index time", "fetchQuery": "something", "occurrence_look_behind": "0"}, ) mocker.patch.object(splunk, "parse_time_to_minutes", return_value=10) mocker.patch.object(results, "JSONResultsReader", return_value=[]) # Mock the service object mock_service = mocker.MagicMock() mock_search = mocker.MagicMock() mock_service.jobs.oneshot.return_value = mock_search # Mock the search results mock_search.results = mocker.MagicMock(return_value=[]) # Mock the mapper object mock_mapper = mocker.MagicMock() # Create a mock for the Cache mock_cache = mocker.MagicMock() # Call the function splunk.fetch_notables( service=mock_service, mapper=mock_mapper, comment_tag_to_splunk="comment_to_splunk", comment_tag_from_splunk="comment_from_splunk", cache_object=mock_cache, enrich_notables=False, ) # Verify that the service.jobs.oneshot was called with "creation time" in the kwargs call_args = mock_service.jobs.oneshot.call_args[1] # The query should include "creation time" in the search criteria assert "index_earliest" in call_args assert "index_latest" in call_args assert "earliest_time" not in call_args assert "latest_time" not in call_args """ ========== Enriching Fetch Mechanism Tests ========== """ @pytest.mark.parametrize( "integration_context, output", [({splunk.INCIDENTS: ["incident"]}, ["incident"]), ({splunk.INCIDENTS: []}, []), ({}, [])] ) def test_fetch_incidents_for_mapping(integration_context, output, mocker): """ Scenario: When a user configures a mapper using Fetch from Instance when the enrichment mechanism is working, we save the ready incidents in the integration context. Given: - List of ready incidents - An empty list of incidents - An empty integration context object When: - fetch_incidents_for_mapping is called Then: - Return the expected result """ mocker.patch.object(demisto, "info") mocker.patch.object(demisto, "incidents") splunk.fetch_incidents_for_mapping(integration_context) assert demisto.incidents.call_count == 1 assert demisto.incidents.call_args[0][0] == output def test_reset_enriching_fetch_mechanism(mocker): """ Scenario: When a user is willing to reset the enriching fetch mechanism and start over. Given: - An integration context object with not empty Cache and incidents When: - reset_enriching_fetch_mechanism is called Then: - Check that the integration context does not contain this fields """ integration_context = {splunk.CACHE: "cache_string", splunk.INCIDENTS: ["i1", "i2"], "wow": "wow"} mocker.patch("SplunkPy.get_integration_context", return_value=integration_context) mocker.patch("SplunkPy.set_integration_context") splunk.reset_enriching_fetch_mechanism() assert integration_context == {"wow": "wow"} @pytest.mark.parametrize( "drilldown_creation_time, asset_creation_time, enrichment_timeout, output", [ (datetime.utcnow().isoformat(), datetime.utcnow().isoformat(), 5, False), ( (datetime.utcnow() - timedelta(minutes=6)).isoformat(), datetime.utcnow().isoformat(), 5, True, ), ], ) def test_is_enrichment_exceeding_timeout(mocker, drilldown_creation_time, asset_creation_time, enrichment_timeout, output): """ Scenario: When one of the notable's enrichments is exceeding the timeout, we want to create an incident with all the data gathered so far. Given: - Two enrichments that none of them exceeds the timeout. - An enrichment exceeding the timeout and one that does not exceeds the timeout. When: - is_enrichment_process_exceeding_timeout is called Then: - Return the expected result """ mocker.patch.object(splunk, "ENABLED_ENRICHMENTS", return_value=[splunk.DRILLDOWN_ENRICHMENT, splunk.ASSET_ENRICHMENT]) notable = splunk.Notable({splunk.EVENT_ID: "id"}) notable.enrichments.append(splunk.Enrichment(splunk.DRILLDOWN_ENRICHMENT, creation_time=drilldown_creation_time)) notable.enrichments.append(splunk.Enrichment(splunk.ASSET_ENRICHMENT, creation_time=asset_creation_time)) assert notable.is_enrichment_process_exceeding_timeout(enrichment_timeout) is output INCIDENT_1 = {"name": "incident1", "rawJSON": json.dumps({})} INCIDENT_2 = {"name": "incident2", "rawJSON": json.dumps({})} @pytest.mark.parametrize("incidents, output", [([], []), ([INCIDENT_1, INCIDENT_2], [INCIDENT_1, INCIDENT_2])]) def test_store_incidents_for_mapping(incidents, output): """ Scenario: Store ready incidents in integration context, to be retrieved by a user configuring a mapper and selecting "Fetch from instance" when the enrichment mechanism is working. Given: - An empty list of incidents - A list of two incidents When: - store_incidents_for_mapping is called Then: - Return the expected result """ splunk.set_integration_context({}) splunk.store_incidents_for_mapping(incidents) assert splunk.get_integration_context().get(splunk.INCIDENTS, []) == output @pytest.mark.parametrize( "notable_data, raw, earliest, latest", [ ({}, {}, "", ""), ( {"drilldown_earliest": f"${splunk.INFO_MIN_TIME}$", "drilldown_latest": f"${splunk.INFO_MAX_TIME}$"}, {splunk.INFO_MIN_TIME: "1", splunk.INFO_MAX_TIME: "2"}, "1", "2", ), ( { "drilldown_earliest": "1", "drilldown_latest": "2", }, {}, "1", "2", ), ], ) def test_get_drilldown_timeframe(notable_data, raw, earliest, latest, mocker): """ Scenario: Trying to get the drilldown's timeframe from the notable's data Given: - An empty notable's data - An notable's data that the info of the timeframe is in the raw field - An notable's data that the info is in the data dict When: - get_drilldown_timeframe is called Then: - Return the expected result """ mocker.patch.object(demisto, "info") earliest_offset, latest_offset = splunk.get_drilldown_timeframe(notable_data, raw) assert earliest_offset == earliest assert latest_offset == latest @pytest.mark.parametrize( "raw_field, notable_data, expected_field, expected_value", [ ("field|s", {"field": "1"}, "field", "1"), ("field", {"field": "1"}, "field", "1"), ("field|s", {"_raw": "field=1, value=2"}, "field", "1"), ("x", {"y": "2"}, "", ""), # A raw field that is a substring of another field must not collide ("src_ip", {"src": "host1", "src_ip": "1.2.3.4"}, "src_ip", "1.2.3.4"), ("src_ip|s", {"src": "host1", "src_ip": "1.2.3.4"}, "src_ip", "1.2.3.4"), ("src_ip", {"_raw": "src=host1, src_ip=1.2.3.4"}, "src_ip", "1.2.3.4"), ], ) def test_get_notable_field_and_value(raw_field, notable_data, expected_field, expected_value, mocker): """ Scenario: When building the drilldown search query, we search for the field in the raw search query and search for its real name in the notable's data or in the notable's raw data. We also ignore Splunk advanced syntax such as "|s, |h, ..." Given: - A raw field that has the same name in the notable's data - A raw field that has "|s" as a suffix in the raw search query and its value is in the notable's data - A raw field that has "|s" as a suffix in the raw search query and its value is in the notable's raw data - A raw field that is not is the notable's data or in the notable's raw data When: - get_notable_field_and_value is called Then: - Return the expected result """ mocker.patch.object(demisto, "error") field, value = splunk.get_notable_field_and_value(raw_field, notable_data) assert field == expected_field assert value == expected_value @pytest.mark.parametrize( "notable_data, search, raw, is_query_name, expected_search", [ ({"a": "1", "_raw": "c=3"}, "search a=$a|s$ c=$c$ suffix", {"c": "3"}, False, 'search a="1" c="3" suffix'), ({"a": ["1", "2"], "b": "3"}, "search a=$a|s$ b=$b|s$ suffix", {}, False, 'search (a="1" OR a="2") b="3" suffix'), ({"a": "1", "_raw": "b=3", "event_id": "123"}, "search a=$a|s$ c=$c$ suffix", {"b": "3"}, False, ""), ( {"signature": "Backdoor.test"}, "View related '$signature$' events for $dest$", {"dest": "ACME-test-005"}, True, "View related 'Backdoor.test' events for ACME-test-005", ), ( {}, 'View all wineventlogs involving user="$user$"', {"user": "test"}, True, 'View all wineventlogs involving user="test"', ), ({}, "Test query name", {}, True, "Test query name"), ( {"user": "test\\crusher"}, 'index="test" | where user = $user|s$', {}, False, 'index="test" | where user="test\\\\crusher"', ), ( {"user": "test\\crusher"}, 'index="test" | where user = "$user|s$"', {}, False, 'index="test" | where user="test\\\\crusher"', ), ( {"countryNameA": '"test\\country"', "countryNameB": '""'}, 'search countryA="$countryNameA|s$" countryB=$countryNameB|s$', {}, False, 'search countryA="test\\country" countryB=""', ), ({"test": "test_user"}, "search countryA=\\$this is a test\\$", {}, False, "search countryA=\\$this is a test\\$"), # A field whose name is a substring of the queried field must resolve to the correct field ( {"src": "host1", "src_ip": "1.2.3.4"}, "search src_ip=$src_ip$", {}, False, 'search src_ip="1.2.3.4"', ), ], ids=[ "search query fields in notables data and raw data", "search query fields in notable data more than one value", "search query fields don't exist in notable data and raw data", "query name fields in notables data and raw data", "query name fields in raw data", "query name without fields to replace", "search query with a user field that contains a backslash", "search query with a user field that is surrounded by quotation marks and contains a backslash", "search query fields in notable data more than one value, with one empty value", "search query with $ as part of the search - no need to replace", "search query with a field name that is a substring of another field", ], ) def test_build_drilldown_search(notable_data, search, raw, is_query_name, expected_search, mocker): """ Scenario: When building the drilldown search query, we replace every field in between "$" sign with its corresponding query part (key & value). Given: - A raw search query with fields both in the notable's data and in the notable's raw data - A raw search query with fields in the notable's data that has more than one value - A raw search query with fields that does not exist in the notable's data or in the notable's raw data - A raw query name with fields both in the notable's data and in the notable's raw data - A raw query name with fields in the notable's raw data - A raw query name without any fields to replace. - A raw query search with a user field that contains a backslash - A raw query search with a user field that is surrounded by quotation marks and contains a backslash When: - build_drilldown_search is called Then: - Return the expected result """ mocker.patch.object(demisto, "error") mocker.patch.object(demisto, "params", return_value={}) parsed_query = splunk.build_drilldown_search(notable_data, search, raw, is_query_name) assert parsed_query == expected_search @pytest.mark.parametrize( "notable_data, prefix, fields, query_part", [ ({"user": ["u1", "u2"]}, "identity", ["user"], '(identity="u1" OR identity="u2")'), ({"_raw": "1233, user=u1"}, "user", ["user"], 'user="u1"'), ( {"user": ["u1", "u2"], "_raw": "1321, src_user=u3"}, "user", ["user", "src_user"], '(user="u1" OR user="u2" OR user="u3")', ), ({}, "prefix", ["field"], ""), ], ) def test_get_fields_query_part(notable_data, prefix, fields, query_part): """ Scenario: When building an enrichment search query, we search for values in the notable's data / notable's raw data and fill them in the raw search query to create a searchable query. Given: - One field with multiple values, values in the data - One field, value is in the raw data - Two fields with multiple values, values in both the data and the raw data - An empty notable data, field does not exists When: - get_fields_query_part is called Then: - Return the expected result """ assert splunk.get_fields_query_part(notable_data, prefix, fields) == query_part @pytest.mark.parametrize( "enrichments, expected_result", [ ( [ splunk.Enrichment(splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1"), splunk.Enrichment(splunk.DRILLDOWN_ENRICHMENT, enrichment_id="2"), splunk.Enrichment(splunk.DRILLDOWN_ENRICHMENT, enrichment_id="3"), ], 3, ), ( [ splunk.Enrichment(splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1"), splunk.Enrichment(splunk.ASSET_ENRICHMENT, enrichment_id="2"), splunk.Enrichment(splunk.IDENTITY_ENRICHMENT, enrichment_id="3"), ], 1, ), ( [ splunk.Enrichment(splunk.ASSET_ENRICHMENT, enrichment_id="1"), splunk.Enrichment(splunk.ASSET_ENRICHMENT, enrichment_id="2"), splunk.Enrichment(splunk.IDENTITY_ENRICHMENT, enrichment_id="3"), ], 0, ), ], ids=[ "A Notable with 3 drilldown enrichments", "A Notable with 1 drilldown enrichment, 1 asset enrichment and 1 identity enrichment", "A Notable with 2 asset enrichments and 1 identity enrichment", ], ) def test_drilldown_searches_counter(enrichments, expected_result): """ Tests the drilldown searches enrichment counter. Given: - A Notable with 3 drilldown enrichments. - A Notable with 1 drilldown enrichment, 1 asset enrichment and 1 identity enrichment. - A Notable with 2 asset enrichments and 1 identity enrichment. When: - drilldown_searches_counter function is called Then: - Return the expected result - number of drilldown enrichments. """ notable = splunk.Notable({}, notable_id="id", enrichments=enrichments) assert notable.drilldown_searches_counter() == expected_result @pytest.mark.parametrize( "enrichments, expected_data", [ ( [ splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.SUCCESSFUL, query_name="query_name1", query_search="query_search1", data=[{"result1": "a"}, {"result2": "b"}], ), splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="2", status=splunk.Enrichment.SUCCESSFUL, query_name="query_name2", query_search="query_search2", data=[{"result1": "c"}, {"result2": "d"}], ), splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="3", status=splunk.Enrichment.SUCCESSFUL, query_name="query_name3", query_search="query_search3", data=[{"result1": "e"}, {"result2": "f"}], ), ], [ { "query_name": "query_name1", "query_search": "query_search1", "query_results": [{"result1": "a"}, {"result2": "b"}], "enrichment_status": splunk.Enrichment.SUCCESSFUL, }, { "query_name": "query_name2", "query_search": "query_search2", "query_results": [{"result1": "c"}, {"result2": "d"}], "enrichment_status": splunk.Enrichment.SUCCESSFUL, }, { "query_name": "query_name3", "query_search": "query_search3", "query_results": [{"result1": "e"}, {"result2": "f"}], "enrichment_status": splunk.Enrichment.SUCCESSFUL, }, ], ), ( [ splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.SUCCESSFUL, query_name="query_name1", query_search="query_search1", data=[{"result1": "a"}, {"result2": "b"}], ), splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="2", status=splunk.Enrichment.SUCCESSFUL, query_name="query_name2", query_search="query_search2", data=[{"result1": "c"}, {"result2": "d"}], ), ], [ { "query_name": "query_name1", "query_search": "query_search1", "query_results": [{"result1": "a"}, {"result2": "b"}], "enrichment_status": splunk.Enrichment.SUCCESSFUL, }, { "query_name": "query_name2", "query_search": "query_search2", "query_results": [{"result1": "c"}, {"result2": "d"}], "enrichment_status": splunk.Enrichment.SUCCESSFUL, }, ], ), ( [ splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.SUCCESSFUL, query_name="query_name1", query_search="query_search1", data=[{"result1": "a"}, {"result2": "b"}], ) ], [{"result1": "a"}, {"result2": "b"}], ), ([], None), ], ids=[ "A Notable with 3 drilldown enrichments, 1 asset enrichment and 1 identity enrichment", "A Notable with 2 drilldown enrichment, 1 asset enrichment and 1 identity enrichment", "A Notable with 1 drilldown enrichment, 1 asset enrichment and 1 identity enrichment", "A Notable without drilldown enrichments, 1 asset enrichments and 1 identity enrichment", ], ) def test_to_incident_notable_enrichments_data(enrichments, expected_data): """ Tests the logic of the Notable.to_incident() function, regarding the results data of multiple drilldown enrichments. Given: 1. A Notable with 3 drilldown enrichments, 1 asset enrichment and 1 identity enrichment. 2. A Notable with 2 drilldown enrichment, 1 asset enrichment and 1 identity enrichment. 3. A Notable with 1 drilldown enrichment, 1 asset enrichment and 1 identity enrichment. 4. A Notable without drilldown enrichments, 1 asset enrichments and 1 identity enrichment. When: - Notable.to_incident() function is called Then: - Verify that the data of the notable includes the expected enrichements result as follow: 1. A dictionary with the results of the 3 drilldown searches by query names. 2. A dictionary with the results of the 2 drilldown searches by query names. 3. A list of the drilldown searches results (backwards competability). 4. No 'Drilldown' key in the notables data. """ notable = splunk.Notable({}, notable_id="id", enrichments=enrichments) enrichments_to_add = [ splunk.Enrichment( splunk.ASSET_ENRICHMENT, enrichment_id="111", status=splunk.Enrichment.SUCCESSFUL, data=[{"result1": "a"}, {"result2": "b"}], ), splunk.Enrichment( splunk.IDENTITY_ENRICHMENT, enrichment_id="222", status=splunk.Enrichment.FAILED, data=[{"result1": "a"}, {"result2": "b"}], ), ] notable.enrichments.extend(enrichments_to_add) service = Service("DONE") mapper = splunk.UserMappingObject(service, False) notable.to_incident(mapper, "comment_tag_to_splunk", "comment_tag_from_splunk") assert notable.data.get(splunk.ASSET_ENRICHMENT) == [{"result1": "a"}, {"result2": "b"}] assert notable.data.get(splunk.IDENTITY_ENRICHMENT) == [{"result1": "a"}, {"result2": "b"}] assert notable.data.get(splunk.DRILLDOWN_ENRICHMENT) == expected_data @pytest.mark.parametrize( "enrichments, enrichment_type, expected_stauts_result", [ ( [ splunk.Enrichment( splunk.ASSET_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.SUCCESSFUL, data=[{"result1": "a"}, {"result2": "b"}], ) ], splunk.ASSET_ENRICHMENT, True, ), ( [ splunk.Enrichment( splunk.ASSET_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.FAILED, data=[{"result1": "a"}, {"result2": "b"}], ) ], splunk.ASSET_ENRICHMENT, False, ), ( [ splunk.Enrichment( splunk.IDENTITY_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.SUCCESSFUL, data=[{"result1": "a"}, {"result2": "b"}], ) ], splunk.IDENTITY_ENRICHMENT, True, ), ( [ splunk.Enrichment( splunk.IDENTITY_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.FAILED, data=[{"result1": "a"}, {"result2": "b"}], ) ], splunk.IDENTITY_ENRICHMENT, False, ), ( [ splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.SUCCESSFUL, query_name="query_name1", query_search="query_search1", data=[{"result1": "a"}, {"result2": "b"}], ) ], splunk.DRILLDOWN_ENRICHMENT, True, ), ( [ splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.FAILED, query_name="query_name1", query_search="query_search1", data=[{"result1": "a"}, {"result2": "b"}], ) ], splunk.DRILLDOWN_ENRICHMENT, False, ), ( [ splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.SUCCESSFUL, data=[{"result1": "a"}, {"result2": "b"}], ), splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.FAILED, data=[{"result1": "a"}, {"result2": "b"}], ), ], splunk.DRILLDOWN_ENRICHMENT, True, ), ( [ splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.FAILED, data=[{"result1": "a"}, {"result2": "b"}], ), splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.SUCCESSFUL, data=[{"result1": "a"}, {"result2": "b"}], ), ], splunk.DRILLDOWN_ENRICHMENT, True, ), ( [ splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.FAILED, data=[{"result1": "a"}, {"result2": "b"}], ), splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.FAILED, data=[{"result1": "a"}, {"result2": "b"}], ), ], splunk.DRILLDOWN_ENRICHMENT, False, ), ( [ splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.SUCCESSFUL, data=[{"result1": "a"}, {"result2": "b"}], ), splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.SUCCESSFUL, data=[{"result1": "a"}, {"result2": "b"}], ), ], splunk.DRILLDOWN_ENRICHMENT, True, ), ( [ splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.FAILED, data=[{"result1": "a"}, {"result2": "b"}], ), splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.SUCCESSFUL, data=[{"result1": "a"}, {"result2": "b"}], ), splunk.Enrichment( splunk.DRILLDOWN_ENRICHMENT, enrichment_id="1", status=splunk.Enrichment.FAILED, data=[{"result1": "a"}, {"result2": "b"}], ), ], splunk.DRILLDOWN_ENRICHMENT, True, ), ], ids=[ "A Notable with 1 successful Asset enrichment", "A Notable with 1 failed Asset enrichment", "A Notable with 1 successful Identity enrichment", "A Notable with 1 failed Identity enrichment", "A Notable with 1 successful Drilldown enrichment", "A Notable with 1 failed Drilldown enrichment", "A Notable with 1 successful Drilldown enrichment and 1 failed drilldown enrichment (the first is successful)", "A Notable with 1 successful Drilldown enrichment and 1 failed drilldown enrichment (the second is successful)", "A Notable with 2 Drilldown enrichments [failed, failed]", "A Notable with 2 Drilldown enrichments [successful, successful]", "A Notable with 3 Drilldown enrichments [failed, successful, failed]", ], ) def test_to_incident_notable_enrichments_status(enrichments, enrichment_type, expected_stauts_result): """ Tests the logic of the Notable.to_incident() function, regarding the statuses of enrichments. Given: 1. A Notable with 1 successful Asset enrichment. 2. A Notable with 1 failed Asset enrichment. 3. A Notable with 1 successful Identity enrichment. 4. A Notable with 1 failed Identity enrichment. 5. A Notable with 1 successful Drilldown enrichment. 6. A Notable with 1 failed Drilldown enrichment. 7. A Notable with 1 successful Drilldown enrichment and 1 failed drilldown enrichment (the first is successful). 8. A Notable with 1 successful Drilldown enrichment and 1 failed drilldown enrichment (the second is successful). 9. A Notable with 2 Drilldown enrichments [failed, failed]. 10. A Notable with 2 Drilldown enrichments [successful, successful]. 11. A Notable with 3 Drilldown enrichments [failed, successful, failed]. When: - Notable.to_incident() function is called Then: - Verify that the status of the notable enrichments is as follow: 1. Asset Enrichment status is: successful_asset_enrichment = True. 2. Asset Enrichment status is: successful_asset_enrichment = False. 3. Identity Enrichment status is: successful_identity_enrichment = True. 4. Identity Enrichment status is: successful_identity_enrichment = False. # In Drilldown enrichment - if at least one drilldown enrichment is successful the status is Success. 5. Drilldown Enrichment status is: successful_drilldown_enrichment = True. 6. Drilldown Enrichment status is: successful_drilldown_enrichment = False. 7. Drilldown Enrichment status is: successful_drilldown_enrichment = True. 8. Drilldown Enrichment status is: successful_drilldown_enrichment = True. 9. Drilldown Enrichment status is: successful_drilldown_enrichment = False. 10. Drilldown Enrichment status is: successful_drilldown_enrichment = True. 11. Drilldown Enrichment status is: successful_drilldown_enrichment = True. """ notable = splunk.Notable({}, notable_id="id", enrichments=enrichments) service = Service("DONE") mapper = splunk.UserMappingObject(service, False) notable.to_incident(mapper, "comment_tag_to_splunk", "comment_tag_from_splunk") assert notable.data[splunk.ENRICHMENT_TYPE_TO_ENRICHMENT_STATUS[enrichment_type]] == expected_stauts_result @pytest.mark.parametrize( "spl_search, expected", [ # Single backslashes inside a field="value" filter must be doubled ( 'eventcode IN (1, 2) field_a="\\foo\\bar\\baz" | head 1', 'eventcode IN (1, 2) field_a="\\\\foo\\\\bar\\\\baz" | head 1', ), # Already-doubled values are left unchanged (idempotent) ('field_a="\\\\foo\\\\bar"', 'field_a="\\\\foo\\\\bar"'), # Values without backslashes are untouched ('field_a="10.0.0.1" field_b="abc"', 'field_a="10.0.0.1" field_b="abc"'), # rex / free-text quoted strings (not preceded by '=') must NOT be modified ( 'index=x | rex field=field_a "value: (?.*)"', 'index=x | rex field=field_a "value: (?.*)"', ), # Regex literals inside SPL function calls (quote follows '(' or ',', NOT a field token) # must NOT be re-escaped. ( '| eval field_a=replace(field_a,"(\\\\)","\\\\\\\\")', '| eval field_a=replace(field_a,"(\\\\)","\\\\\\\\")', ), # Multiple genuine field="value" filters in one query are all doubled ( 'field_a="\\foo\\bar" field_b="\\\\baz\\\\qux"', 'field_a="\\\\foo\\\\bar" field_b="\\\\baz\\\\qux"', ), # A dotted field name is still treated as a field filter ( 'parent.child="\\foo\\bar"', 'parent.child="\\\\foo\\\\bar"', ), ], ids=[ "single backslashes are doubled", "already-doubled backslashes are unchanged", "no backslashes are untouched", "rex regex quoted string is not modified", "eval/replace regex literal is not over-escaped", "multiple field filters are all doubled", "dotted field name is treated as a field filter", ], ) def test_escape_backslashes_in_field_filters(spl_search, expected): """ Scenario: A drilldown search arrives as JSON; after json.loads, backslashes inside field filter values are collapsed to single backslashes, which Splunk SPL cannot match. We re-escape them. Given: - An SPL search with a field="value" filter value containing single backslashes. - An SPL search whose field filter values are already correctly escaped. - An SPL search without backslashes. - An SPL search with a rex/free-text quoted string containing a backslash. - An SPL search with a regex literal inside a function call (eval/replace). - An SPL search with multiple field filters. - An SPL search with a dotted field name. When: - escape_backslashes_in_field_filters is called. Then: - Backslashes inside genuine field="value" filters are doubled, the operation is idempotent, and rex/free-text quoted strings and regex literals inside function calls are left untouched. """ assert splunk.escape_backslashes_in_field_filters(spl_search) == expected def test_parse_drilldown_searches_preserves_backslashes(): """ Given: - A 'drilldown_searches' JSON payload where a field="value" filter value contains backslashes. When: - Running splunk.parse_drilldown_searches. Then: - The parsed 'search' keeps the backslashes escaped (doubled) for Splunk SPL. """ searches = [ '[{"name":"Show events","search":"(index=idx_a OR index=idx_b) ' 'eventcode IN (1, 2) field_a=\\"\\\\foo\\\\bar\\\\baz\\" ' '| head 1","earliest_offset":"1","latest_offset":"2","disabled":false}]' ] parsed = splunk.parse_drilldown_searches(searches) assert parsed[0]["search"] == ("(index=idx_a OR index=idx_b) eventcode IN (1, 2) " 'field_a="\\\\foo\\\\bar\\\\baz" | head 1') def test_parse_drilldown_searches(): """ Given: - A list of valid Json strings with splunk drilldown searches data. When: - Running the splunk.parse_drilldown_searches function Then: - Verify that the search data was parsed into a python dictionary as expected. """ searches = [ '{"name":"View related \'$signature$\' events for $dest$","search":"| from datamodel:\\"Malware\\".' '\\"Malware_Attacks\\" | search dest=$dest|s$ signature=$signature|s$","earliest":17145' '63300,"latest":1715168700}', '{"name":"View related \'$category$\' events for $signature$","search":"| from datamodel:\\"Malw' 'are\\".\\"Malware_Attacks\\" \\n| fields category, dest, signature | search dest=$dest|s$ signature=' '$signature|s$","earliest":1714563300,"latest":1715168700}', ] parsed_searches = splunk.parse_drilldown_searches(searches) for search in parsed_searches: assert isinstance(search, dict) assert parsed_searches == [ { "name": "View related '$signature$' events for $dest$", "search": '| from datamodel:"Malware"."Malware_Attacks" | search dest=$dest|s$ signature=$signature|s$', "earliest": 1714563300, "latest": 1715168700, }, { "name": "View related '$category$' events for $signature$", "search": '| from datamodel:"Malware"."Malware_Attacks" \n| fields category, dest, signature | search dest=$dest|s$ ' "signature=$signature|s$", "earliest": 1714563300, "latest": 1715168700, }, ] @pytest.mark.parametrize( "notable_data, expected_call_count", [ ({"event_id": "test_id", "drilldown_search": "test_search", "drilldown_searches": ["test_search1", "test_search2"]}, 0), ({"event_id": "test_id", "drilldown_search": "", "drilldown_searches": ["test_search1", "test_search2"]}, 1), ({"event_id": "test_id", "drilldown_searches": ["test_search1", "test_search2"]}, 1), ], ids=[ "A notable data with both 'drilldown_search' and 'drilldown_searches' keys with values", "A notable data with both 'drilldown_search' and 'drilldown_searches' keys but 'drilldown_search' has no value", "A notable data with 'drilldown_searches' key only", ], ) def test_drilldown_enrichment_main_condition(mocker, notable_data, expected_call_count): """ Tests the logic of the first (main) condition in the drilldown_enrichment() function. We want to make sure that in a case that the notable data include both 'drilldown_search' and 'drilldown_searches' keys (happens when there is only one drilldown search to enrich) the 'drilldown_search' value will be taken to maintain backwards cometability. In any other case the value of the 'drilldown_searches' key will be used. Given: 1. A notable data that includes both 'drilldown_search' and 'drilldown_searches' keys with values. 2. A notable data that includes both 'drilldown_search' and 'drilldown_searches' keys but 'drilldown_search' has no value. 3. A notable data that includes 'drilldown_searches' key only. When: - Running the splunk.drilldown_enrichment function Then: - Verify that: 1. The value of the 'drilldown_search' key is taken (to maintain backwards competability), and therefore we don't call the parse_drilldown_searches function. 2. The value of the 'drilldown_searches' key is taken, and therefore we call the parse_drilldown_searches function. 3. The value of the 'drilldown_searches' key is taken, and therefore we call the parse_drilldown_searches function. """ mock_parse_drilldown_searches = mocker.patch("SplunkPy.parse_drilldown_searches", return_value=[]) service = Service("DONE") splunk.drilldown_enrichment(service, notable_data, 5) assert mock_parse_drilldown_searches.call_count == expected_call_count @pytest.mark.parametrize( "notable_data, expected_call_count", [ ({"event_id": "test_id", "drilldown_search": "test_search", "drilldown_searches": [{}], "_raw": "{'test':1}"}, 1), ( { "event_id": "test_id", "drilldown_searches": [ '{"name":"View related \'$signature$\' events for $dest$","search":"| from datamodel:\\"Malware\\".\\"Malwa' 're_Attacks\\" | search dest=$dest|s$ signature=$signature|s$","earliest":1714563300,"latest":1715168700}', '{"name":"View related \'$category$\' events for $signature$","search":"| from datamodel:\\"Malware\\".\\"M' 'alware_Attacks\\" \\n| fields category, dest, signature | search dest=$dest|s$ signature=$signature|s$",' '"earliest":1714563300,"latest":1715168700}', ], }, 0, ), ], ids=["A notable data with one drilldown search", "A notable data with multiple drilldown searches"], ) def test_drilldown_enrichment_get_timeframe(mocker, notable_data, expected_call_count): """ Tests that in a case of one drildown search we extract the search timeframe from the notable data by calling the get_drilldown_timeframe() function, and in a case of multiple drilldown searches, we get the timeframe from the drilldown search data dictionary without calling the get_drilldown_timeframe() function. Given: 1. A notable data with one drilldown search. 2. A notable data with multiple drilldown searches. When: - Running the splunk.get_drilldown_timeframe function. Then: - Verify that: 1. The timeframe is determined according to fields in the notable data and raw data by using the get_drilldown_timeframe function. 2. The timeframe is determined according to fields of each drilldown search data dict. """ mock_get_drilldown_timeframe = mocker.patch("SplunkPy.get_drilldown_timeframe", return_value=("", "")) mocker.patch("SplunkPy.build_drilldown_search", return_value="") service = Service("DONE") splunk.drilldown_enrichment(service, notable_data, 5) assert mock_get_drilldown_timeframe.call_count == expected_call_count @pytest.mark.parametrize( "notable_data, expected_result", [ ( { "event_id": "test_id", "drilldown_name": "View all login attempts by system $src$", "drilldown_search": '| from datamodel:"Authentication"."Authentication" | search src=$src|s$', "drilldown_searches": '{"name":"View all login attempts by system $src$","search":"| from datamodel:\\"Authent' 'ication\\".\\"Authentication\\" | search src=$src|s$","earliest":1715040000,' '"latest":1715126400}', "_raw": "src='test_src'", "drilldown_latest": "1715126400.000000000", "drilldown_earliest": "1715040000.000000000", }, [ ( "View all login attempts by system 'test_src'", '| from datamodel:"Authentication"."Authentication" | search src="\'test_src\'"', ) ], ), ( { "event_id": "test_id2", "drilldown_searches": [ '{"name":"View all login attempts by system $src$","search":"| from datamodel:\\"Authentication\\".\\"Authe' 'ntication\\" | search src=$src|s$","earliest":1715040000,"latest":1715126400}', '{"name":"View all test involving user=\\"$user$\\"","search":"index=\\"test\\"\\n| where ' 'user = $user|s$","earliest":1716955500,"latest":1716959400}', ], "_raw": "src='test_src', user='test_user'", }, [ ( "View all login attempts by system 'test_src'", '| from datamodel:"Authentication"."Authentication" | search src="\'test_src\'"', ), ("View all test involving user=\"'test_user'\"", 'search index="test"\n| where user="\'test_user\'"'), ], ), ( { "event_id": "test_id3", "drilldown_searches": [ '{"name":"View all login attempts by system $src$","search":"| from datamodel:\\"Authentication\\".\\"Authe' 'ntication\\" | search src=$src|s$","earliest_offset":1715040000,"latest_offset":1715126400}', '{"name":"View all test involving user=\\"$user$\\"","search":"index=\\"test\\"\\n| where ' 'user = $user|s$","earliest_offset":1716955500,"latest_offset":1716959400}', ], "_raw": "src='test_src', user='test_user'", }, [ ( "View all login attempts by system 'test_src'", '| from datamodel:"Authentication"."Authentication" | search src="\'test_src\'"', ), ("View all test involving user=\"'test_user'\"", 'search index="test"\n| where user="\'test_user\'"'), ], ), ], ids=[ "A notable data with one drilldown search enrichment", "A notable data with two drilldown searches which contained the earlies in 'earliest' key ", "A notable data with two drilldown searches which contained the earlies in 'earliest_offset' key ", ], ) def test_drilldown_enrichment(notable_data, expected_result): """ Tests the logic of the drilldown_enrichment function. Given: 1. A notable data with one drilldown search enrichment. 2. A notable data with multiple (two) drilldown searches to enrich. When: - Running the splunk.drilldown_enrichment function. Then: - Verify that the returned jobs and queries are as expected. """ from splunklib import client service = Service("DONE") jobs_and_queries = splunk.drilldown_enrichment(service, notable_data, 5) for i in range(len(jobs_and_queries)): job_and_queries = jobs_and_queries[i] assert job_and_queries[0] == expected_result[i][0] assert job_and_queries[1] == expected_result[i][1] assert isinstance(job_and_queries[2], client.Job) @pytest.mark.parametrize( "notable_data, expected_result", [ ( { "event_id": "test_id", "drilldown_name": "View all login attempts by system $src$", "drilldown_search": "NULL", "drilldown_searches": [ '{"name":"View all login attempts by system $src$","search":"| from datamodel:\\"Authent' 'ication\\".\\"Authentication\\" | search src=$src|s$","earliest":1715040000,' '"latest":1715126400}' ], "_raw": "src='test_src'", "drilldown_latest": "1715126400.000000000", "drilldown_earliest": "1715040000.000000000", }, [ ( "View all login attempts by system 'test_src'", '| from datamodel:"Authentication"."Authentication" | search src="\'test_src\'"', ) ], ), ( { "event_id": "test_id", "drilldown_name": "View all login attempts by system $src$", "drilldown_search": '| from datamodel:"Authentication"."Authentication" | search src=$src|s$', "drilldown_searches": "[]", "_raw": "src='test_src'", "drilldown_latest": "1715126400.000000000", "drilldown_earliest": "1715040000.000000000", }, [ ( "View all login attempts by system 'test_src'", '| from datamodel:"Authentication"."Authentication" | search src="\'test_src\'"', ) ], ), ], ) def test_drilldown_enrichment_fillnull(mocker: MockerFixture, notable_data, expected_result): """ Tests the drilldown enrichment process when a 'splunk.FILLNULL_VALUE' is "NULL" Given: 1. A notable data with a drilldown_search=NULL 2. A notable data with a drilldown_searches is empty When: Performing drilldown enrichment to generate search jobs and queries Then: - The generated queries match the expected enriched queries - The fillnull value is correctly applied during query construction """ from splunklib import client service = Service("DONE") mock_params = {"fetchQuery": "`notable` is cool | fillnull value=NULL"} mocker.patch("demistomock.params", return_value=mock_params) jobs_and_queries = splunk.drilldown_enrichment(service, notable_data, 5) for i in range(len(jobs_and_queries)): job_and_queries = jobs_and_queries[i] assert job_and_queries[0] == expected_result[i][0] assert job_and_queries[1] == expected_result[i][1] assert isinstance(job_and_queries[2], client.Job) def test_drilldown_enrichment_query_earliest(mocker: MockerFixture): """ Tests the drilldown enrichment process when query contains earliest filter Given: A drilldown data without drilldown_earliest and drilldown_latest values, the drilldown search query contains earliest filter. When: Performing drilldown enrichment to generate search jobs and queries Then: The generated query match the expected enriched query and contains then earliest value """ from splunklib import client service = Service("DONE") mock_params = {"fetchQuery": "`notable` is cool | fillnull value=NULL"} mocker.patch("demistomock.params", return_value=mock_params) notable_data = { "event_id": "test_id", "drilldown_name": "View all login attempts by system $src$", "drilldown_search": '| from datamodel:"Authentication"."Authentication" | search src=$src|s$ | earliest=1d', "drilldown_searches": "[]", "_raw": "src='test_src'", } jobs_and_queries = splunk.drilldown_enrichment(service, notable_data, 5) for i in range(len(jobs_and_queries)): job_and_queries = jobs_and_queries[i] assert job_and_queries[0] == "View all login attempts by system 'test_src'" assert ( job_and_queries[1] == '| from datamodel:"Authentication"."Authentication" | search src="\'test_src\'" | earliest=1d' ) assert isinstance(job_and_queries[2], client.Job) @pytest.mark.parametrize( "notable_data, debug_log_message", [ ({"event_id": "test_id"}, "drill-down was not properly configured for notable test_id"), ( { "event_id": "test_id", "drilldown_name": "View all login attempts by system $src$", "drilldown_search": '| from datamodel:"Authentication"."Authentication" | search src=$src|s$', "_raw": "src='test_src'", "drilldown_latest": "", "drilldown_earliest": "", }, "Failed getting the drilldown timeframe for notable test_id", ), ( { "event_id": "test_id", "drilldown_name": "View all login attempts by system $src$", "drilldown_search": '| from datamodel:"Authentication"."Authentication" | search src=$src|s$', "_raw": "", "drilldown_latest": "00101", "drilldown_earliest": "00001", }, "Couldn't build search query for notable test_id with the following drilldown search ", ), ( { "event_id": "test_id", "drilldown_searches": [ '{"name":"View all login attempts by system $src$","search":"| from datamodel:\\"Authentica' 'tion\\".\\"Authentication\\" | search src=$src|s$","earliest":"","latest":""}', '{"name":"View all test involving user=\\"$user$\\"","search":"index=\\"test\\"\\n| where user =' '$user|s$","earliest":"","latest":""}', ], "_raw": "src='test_src', user='test_user'", }, "Failed getting the drilldown timeframe for notable test_id", ), ( { "event_id": "test_id", "drilldown_searches": [ '{"name":"View all login attempts by system $src$","search":"| from datamodel:\\"Authentic' 'ation\\".\\"Authentication\\" | search src=$src|s$","earliest":"","latest":""}', '{"name":"View all test involving user=\\"$user$\\"","search":"index=\\"test\\"\\n| where user =' '$user|s$","earliest":"","latest":""}', ], "_raw": "", }, "Couldn't build search query for notable test_id with the following drilldown search", ), ], ids=[ "A notable data without drilldown enrichment data", "A notable data with a single drilldown enrichment without search timeframe data", "A notable data with a single drilldown enrichment with an invalid search query", "A notable data with multiple drilldown enrichments without search timeframe data", "A notable data with multiple drilldown enrichments with invalid search queries", ], ) def test_drilldown_enrichment_no_enrichement_cases(mocker, notable_data, debug_log_message): """ Tests the logic of the drilldown_enrichment function when for some reason the enrichments raw data is invalid. Given: 1. A notable data without drilldown enrichment data. 2. A notable data with a single drilldown enrichment without search timeframe data. 3. A notable data with a single drilldown enrichment with an invalid search query. 4. A notable data with multiple drilldown enrichments without search timeframe data. 5. A notable data with multiple drilldown enrichments with invalid search queries. When: - Running the splunk.drilldown_enrichment function. Then: - Verify that the returned value is a tuple of None values as expected. """ debug_log = mocker.patch.object(demisto, "debug") mocker.patch.object(demisto, "error") service = Service("DONE") jobs_and_queries = splunk.drilldown_enrichment(service, notable_data, 5) for i in range(len(jobs_and_queries)): assert jobs_and_queries[i] == (None, None, None) assert debug_log_message in debug_log.call_args.args[0] """ ========== Mirroring Mechanism Tests ========== """ @pytest.mark.parametrize( "last_update, demisto_params, splunk_time_timestamp", [ ("2021-02-22T18:39:47.753+00:00", {"timezone": "0"}, 1614019187.753), ("2021-02-22T18:39:47.753+02:00", {"timezone": "+120"}, 1614019187.753), ("2021-02-22T20:39:47.753+02:00", {"timezone": "0"}, 1614019187.753), ("2021-02-09T16:41:30.589575+02:00", {}, ""), ], ) def test_get_last_update_in_splunk_time(last_update, demisto_params, splunk_time_timestamp, mocker): """Tests the conversion of the Demisto server time into timestamp in Splunk Server time Given: - The last update time in the Demisto server - The timezone in the Splunk Server When: Converting the time in the Demisto server into timestamp in Splunk Server time Then: - Conversion is correct - An Exception is raised in case that Splunk Server timezone is not specified in Demisto params """ mocker.patch.object(demisto, "params", return_value=demisto_params) if demisto_params: assert splunk.get_last_update_in_splunk_time(last_update) == splunk_time_timestamp else: error_msg = "Cannot mirror incidents when timezone is not configured. Please enter the " "timezone of the Splunk server being used in the integration configuration." with pytest.raises(Exception, match=error_msg): splunk.get_last_update_in_splunk_time(last_update) @pytest.mark.parametrize( "notable_data, func_call_kwargs, expected_closure_data", [ # A Notable with a "Closed" status label ( [ results.Message("INFO-TEST", "test message"), { "status_label": "Closed", "event_id": "id", "rule_id": "id", "status_end": "true", "last_modified_timestamp": "1737547610.49", }, ], { "close_incident": True, "close_end_statuses": False, "close_extra_labels": [], }, { "EntryContext": {"mirrorRemoteId": "id"}, "Type": EntryType.NOTE, "Contents": { "dbotIncidentClose": True, "closeReason": 'Notable event was closed on Splunk with status "Closed".', }, "ContentsFormat": EntryFormat.JSON, }, ), # A Notable with a "New" status label (shouldn't close) ( [ results.Message("INFO-TEST", "test message"), { "status_label": "New", "event_id": "id", "rule_id": "id", "status_end": "false", "last_modified_timestamp": "1737547610.50", }, ], { "close_incident": True, "close_end_statuses": False, "close_extra_labels": [], }, None, ), # A Notable with a custom status label that is on close_extra_labels (should close) ( [ results.Message("INFO-TEST", "test message"), { "status_label": "Custom", "event_id": "id", "rule_id": "id", "status_end": "false", "last_modified_timestamp": "1737547610.51", }, ], { "close_incident": True, "close_end_statuses": False, "close_extra_labels": ["Custom"], }, { "EntryContext": {"mirrorRemoteId": "id"}, "Type": EntryType.NOTE, "Contents": { "dbotIncidentClose": True, "closeReason": 'Notable event was closed on Splunk with status "Custom".', }, "ContentsFormat": EntryFormat.JSON, }, ), # A Notable with close_extra_labels that don't include status_label (shouldn't close) ( [ results.Message("INFO-TEST", "test message"), { "status_label": "Custom", "event_id": "id", "rule_id": "id", "status_end": "false", "last_modified_timestamp": "1737547610.52", }, ], { "close_incident": True, "close_end_statuses": False, "close_extra_labels": ["A", "B"], }, None, ), # A Notable that has status_end as true with close_end_statuses as true (should close) ( [ results.Message("INFO-TEST", "test message"), { "status_label": "Custom", "event_id": "id", "rule_id": "id", "status_end": "true", "last_modified_timestamp": "1737547610.53", }, ], { "close_incident": True, "close_end_statuses": True, "close_extra_labels": [], }, { "EntryContext": {"mirrorRemoteId": "id"}, "Type": EntryType.NOTE, "Contents": { "dbotIncidentClose": True, "closeReason": 'Notable event was closed on Splunk with status "Custom".', }, "ContentsFormat": EntryFormat.JSON, }, ), # A Notable that has status_end as true with close_end_statuses as false (shouldn't close) ( [ results.Message("INFO-TEST", "test message"), { "status_label": "Custom", "event_id": "id", "rule_id": "id", "status_end": "true", "last_modified_timestamp": "1737547610.54", }, ], { "close_incident": True, "close_end_statuses": False, "close_extra_labels": [], }, None, ), # A Notable that is both on close_extra_labels, # and has status_end as true with close_end_statuses as true (should close) ( [ results.Message("INFO-TEST", "test message"), { "status_label": "Custom", "event_id": "id", "rule_id": "id", "status_end": "true", "last_modified_timestamp": "1737547610.55", }, ], { "close_incident": True, "close_end_statuses": True, "close_extra_labels": ["Custom"], }, { "EntryContext": {"mirrorRemoteId": "id"}, "Type": EntryType.NOTE, "Contents": { "dbotIncidentClose": True, "closeReason": 'Notable event was closed on Splunk with status "Custom".', }, "ContentsFormat": EntryFormat.JSON, }, ), ], ) def test_get_modified_remote_data_command_close_incident( mocker, notable_data: list[results.Message | dict], func_call_kwargs: dict, expected_closure_data: dict ): class Jobs: def oneshot(self, **kwargs): assert kwargs["output_mode"] == splunk.OUTPUT_MODE_JSON return notable_data class Service: def __init__(self): self.jobs = Jobs() expected_entries = {"EntryContext": {"mirrorRemoteId": "id"}, "Type": EntryType.NOTE, "ContentsFormat": EntryFormat.JSON} args = {"lastUpdate": "2021-02-09T16:41:30.589575+02:00", "id": "id"} mocker.patch.object(demisto, "params", return_value={"timezone": "0"}) mocker.patch.object(demisto, "debug") mocker.patch.object(demisto, "info") mocker.patch("SplunkPy.results.JSONResultsReader", return_value=notable_data) mocker.patch.object(demisto, "results") service = Service() splunk.get_modified_remote_data_command( service, args, mapper=splunk.UserMappingObject(service, False), comment_tag_from_splunk="comment_tag_from_splunk", **func_call_kwargs, ) results = demisto.results.call_args[0][0] expected_entries["Contents"] = notable_data[1] expected_results = [expected_entries] if expected_closure_data: expected_results.append(expected_closure_data) assert demisto.results.call_count == 1 assert results == expected_results def test_get_remote_data_command_with_message(mocker): """ Test for the get_remote_data_command function with a message. This test verifies that when the splunk-sdk returns a message, the function correctly logs the message using demisto.info(). Args: mocker: The mocker object for patching and mocking. Returns: None """ service = mocker.patch.object(client, "Service") mocker.patch.object(demisto, "info") mocker.patch.object(demisto, "params", return_value={"timezone": "0"}) func_call_kwargs = { "args": {"lastUpdate": "2021-02-09T16:41:30.589575+02:00", "id": "id"}, "close_incident": True, "close_end_statuses": True, "close_extra_labels": ["Custom"], "mapper": splunk.UserMappingObject(service, False), "comment_tag_from_splunk": "from_splunk", } mocker.patch("SplunkPy.results.JSONResultsReader", return_value=[results.Message("INFO-test", "test message")]) splunk.get_modified_remote_data_command(service, **func_call_kwargs) assert demisto.info.call_args[0][0] == "Splunk-SDK message: test message" def test_fetch_with_error_in_message(mocker): """ Given - fetch result from Splunk return Error message When - fetch incidents Then - assert DemistoException is raised """ mock_params = {"fetchQuery": "something", "parseNotableEventsRaw": True} mocker.patch("demistomock.getLastRun", return_value={"time": "2018-10-24T14:13:20"}) mocker.patch("demistomock.params", return_value=mock_params) mocker.patch("splunklib.results.JSONResultsReader", return_value=[results.Message("FATAL", "Error")]) # run service = mocker.patch("splunklib.client.connect") with pytest.raises(DemistoException) as e: splunk.fetch_incidents(service, None, None, None) assert "Failed to fetch incidents, check the provided query in Splunk web search" in e.value.message def test_get_modified_remote_data_command_add_comment(mocker): """ Test case for get_modified_remote_data_command with comment addition. Given: - notable data with new comment When: new comment added in splunk Then: - ensure the comment added as a new note - ensure the event was updated """ test_id = "test_event_id" notable_data = { "status_label": "New", "rule_id": test_id, "event_id": test_id, "status_end": "false", "comment": "new comment from splunk", "reviewer": "admin", "last_modified_timestamp": "1737547610.56", "review_time": "1612881691.589575", } entry_tempale = {"EntryContext": {"mirrorRemoteId": test_id}, "Type": 1} expected_comment_entry = entry_tempale | { "Contents": "new comment from splunk", "ContentsFormat": "text", "Tags": ["from_splunk"], "Note": True, } expected_notable_entry = entry_tempale | {"Contents": notable_data, "ContentsFormat": "json"} mocker.patch.object(demisto, "params", return_value={"timezone": "0"}) mocker.patch("SplunkPy.results.JSONResultsReader", return_value=[notable_data]) mocker.patch.object(demisto, "results") service = mocker.patch.object(client, "Service") func_call_kwargs = { "args": {"lastUpdate": "2021-02-09T16:41:30.589575+02:00", "id": "id"}, "close_incident": True, "close_end_statuses": True, "close_extra_labels": ["Custom"], "mapper": splunk.UserMappingObject(service, False), "comment_tag_from_splunk": "from_splunk", } splunk.get_modified_remote_data_command(service, **func_call_kwargs) results = demisto.results.call_args[0][0][0] notable_data.update({"SplunkComments": [{"Comment": "new comment from splunk"}]}) note_results = demisto.results.call_args[0][0][1] assert demisto.results.call_count == 1 assert results == expected_notable_entry assert note_results == expected_comment_entry def test_get_modified_remote_data_command(mocker): updated_incidet_review = { "rule_id": "id", "event_id": "id", "last_modified_timestamp": "1737547610.56", } service = mocker.patch.object(client, "Service") func_call_kwargs = { "args": {"lastUpdate": "2021-02-09T16:41:30.589575+02:00", "id": "id"}, "close_incident": True, "close_end_statuses": True, "close_extra_labels": ["Custom"], "mapper": splunk.UserMappingObject(service, False), "comment_tag_from_splunk": "from_splunk", } mocker.patch.object(demisto, "params", return_value={"timezone": "0"}) mocker.patch("SplunkPy.results.JSONResultsReader", return_value=[updated_incidet_review]) mocker.patch.object(demisto, "results") splunk.get_modified_remote_data_command(service, **func_call_kwargs) results = demisto.results.call_args[0][0][0]["Contents"] assert demisto.results.call_count == 1 assert results == updated_incidet_review def test_edit_notable_event__failed_to_update(mocker, requests_mock): """ Given - notable event with id ID100 When - updating the event with invalid owner 'dbot' - the service should return error string message 'ValueError: Invalid owner value.' Then - ensure the error message parsed correctly and returned to the user """ test_base_url = "https://test.url.com:8089/" test_token = "token12345" test_args = {"eventIDs": "ID100", "owner": "dbot"} mocker.patch.object(splunk, "return_error") requests_mock.post(f"{test_base_url}services/notable_update", json="ValueError: Invalid owner value.") splunk.splunk_edit_notable_event_command(base_url=test_base_url, token=test_token, auth_token=None, args=test_args) assert splunk.return_error.call_count == 1 error_message = splunk.return_error.call_args[0][0] assert error_message == "Could not update notable events: ID100: ValueError: Invalid owner value." @pytest.mark.parametrize( "args, params, call_count, success", [ ( {"delta": {"status": "2"}, "remoteId": "12345", "status": 2, "incidentChanged": True}, {"host": "ec.com", "port": "8089", "authentication": {"identifier": "i", "password": "p"}}, 4, True, ), ( {"delta": {"status": "2"}, "remoteId": "12345", "status": 2, "incidentChanged": True}, {"host": "ec.com", "port": "8089", "authentication": {"identifier": "i", "password": "p"}}, 3, False, ), ( {"delta": {"status": "2"}, "remoteId": "12345", "status": 2, "incidentChanged": True}, {"host": "ec.com", "port": "8089", "authentication": {"identifier": "i", "password": "p"}, "close_notable": True}, 5, True, ), ], ) def test_update_remote_system(args, params, call_count, success, mocker, requests_mock): class Service: def __init__(self): self.token = "fake_token" self.basic = True self._auth_headers = [("Authentication", self.token)] mocker.patch.object(demisto, "info") mocker.patch.object(demisto, "debug") base_url = "https://" + params["host"] + ":" + params["port"] + "/" requests_mock.post(f"{base_url}services/auth/login", json={"sessionKey": "session_key"}) requests_mock.post( f"{base_url}services/notable_update", json={"success": success, "message": "wow"}, ) if not success: mocker.patch.object(demisto, "error") service = Service() mapper = splunk.UserMappingObject(service, False) assert ( splunk.update_remote_system_command( args, params, service, None, mapper=mapper, comment_tag_to_splunk="comment_tag_to_splunk" ) == args["remoteId"] ) assert demisto.debug.call_count == call_count if not success: assert demisto.error.call_count == 1 NOTABLE = { "rule_name": "string", "rule_title": "string", "security_domain": "string", "index": "string", "rule_description": "string", "risk_score": "string", "host": "string", "host_risk_object_type": "string", "dest_risk_object_type": "string", "dest_risk_score": "string", "splunk_server": "string", "_sourcetype": "string", "_indextime": "string", "_time": "string", "src_risk_object_type": "string", "src_risk_score": "string", "_raw": "string", "urgency": "string", "owner": "string", "info_min_time": "string", "info_max_time": "string", "comment": "string", "reviewer": "string", "rule_id": "string", "action": "string", "app": "string", "authentication_method": "string", "authentication_service": "string", "bugtraq": "string", "bytes": "string", "bytes_in": "string", "bytes_out": "string", "category": "string", "cert": "string", "change": "string", "change_type": "string", "command": "string", "comments": "string", "cookie": "string", "creation_time": "string", "cve": "string", "cvss": "string", "date": "string", "description": "string", "dest": "string", "dest_bunit": "string", "dest_category": "string", "dest_dns": "string", "dest_interface": "string", "dest_ip": "string", "dest_ip_range": "string", "dest_mac": "string", "dest_nt_domain": "string", "dest_nt_host": "string", "dest_port": "string", "dest_priority": "string", "dest_translated_ip": "string", "dest_translated_port": "string", "dest_type": "string", "dest_zone": "string", "direction": "string", "dlp_type": "string", "dns": "string", "duration": "string", "dvc": "string", "dvc_bunit": "string", "dvc_category": "string", "dvc_ip": "string", "dvc_mac": "string", "dvc_priority": "string", "dvc_zone": "string", "file_hash": "string", "file_name": "string", "file_path": "string", "file_size": "string", "http_content_type": "string", "http_method": "string", "http_referrer": "string", "http_referrer_domain": "string", "http_user_agent": "string", "icmp_code": "string", "icmp_type": "string", "id": "string", "ids_type": "string", "incident": "string", "ip": "string", "mac": "string", "message_id": "string", "message_info": "string", "message_priority": "string", "message_type": "string", "mitre_technique_id": "string", "msft": "string", "mskb": "string", "name": "string", "orig_dest": "string", "orig_recipient": "string", "orig_src": "string", "os": "string", "packets": "string", "packets_in": "string", "packets_out": "string", "parent_process": "string", "parent_process_id": "string", "parent_process_name": "string", "parent_process_path": "string", "password": "string", "payload": "string", "payload_type": "string", "priority": "string", "problem": "string", "process": "string", "process_hash": "string", "process_id": "string", "process_name": "string", "process_path": "string", "product_version": "string", "protocol": "string", "protocol_version": "string", "query": "string", "query_count": "string", "query_type": "string", "reason": "string", "recipient": "string", "recipient_count": "string", "recipient_domain": "string", "recipient_status": "string", "record_type": "string", "registry_hive": "string", "registry_key_name": "string", "registry_path": "string", "registry_value_data": "string", "registry_value_name": "string", "registry_value_text": "string", "registry_value_type": "string", "request_sent_time": "string", "request_payload": "string", "request_payload_type": "string", "response_code": "string", "response_payload_type": "string", "response_received_time": "string", "response_time": "string", "result": "string", "return_addr": "string", "rule": "string", "rule_action": "string", "sender": "string", "service": "string", "service_hash": "string", "service_id": "string", "service_name": "string", "service_path": "string", "session_id": "string", "sessions": "string", "severity": "string", "severity_id": "string", "sid": "string", "signature": "string", "signature_id": "string", "signature_version": "string", "site": "string", "size": "string", "source": "string", "sourcetype": "string", "src": "string", "src_bunit": "string", "src_category": "string", "src_dns": "string", "src_interface": "string", "src_ip": "string", "src_ip_range": "string", "src_mac": "string", "src_nt_domain": "string", "src_nt_host": "string", "src_port": "string", "src_priority": "string", "src_translated_ip": "string", "src_translated_port": "string", "src_type": "string", "src_user": "string", "src_user_bunit": "string", "src_user_category": "string", "src_user_domain": "string", "src_user_id": "string", "src_user_priority": "string", "src_user_role": "string", "src_user_type": "string", "src_zone": "string", "state": "string", "status": "string", "status_code": "string", "status_description": "string", "subject": "string", "tag": "string", "ticket_id": "string", "time": "string", "time_submitted": "string", "transport": "string", "transport_dest_port": "string", "type": "string", "uri": "string", "uri_path": "string", "uri_query": "string", "url": "string", "url_domain": "string", "url_length": "string", "user": "string", "user_agent": "string", "user_bunit": "string", "user_category": "string", "user_id": "string", "user_priority": "string", "user_role": "string", "user_type": "string", "vendor_account": "string", "vendor_product": "string", "vlan": "string", "xdelay": "string", "xref": "string", } DRILLDOWN = { "Drilldown": { "action": "string", "app": "string", "authentication_method": "string", "authentication_service": "string", "bugtraq": "string", "bytes": "string", "bytes_in": "string", "bytes_out": "string", "category": "string", "cert": "string", "change": "string", "change_type": "string", "command": "string", "comments": "string", "cookie": "string", "creation_time": "string", "cve": "string", "cvss": "string", "date": "string", "description": "string", "dest": "string", "dest_bunit": "string", "dest_category": "string", "dest_dns": "string", "dest_interface": "string", "dest_ip": "string", "dest_ip_range": "string", "dest_mac": "string", "dest_nt_domain": "string", "dest_nt_host": "string", "dest_port": "string", "dest_priority": "string", "dest_translated_ip": "string", "dest_translated_port": "string", "dest_type": "string", "dest_zone": "string", "direction": "string", "dlp_type": "string", "dns": "string", "duration": "string", "dvc": "string", "dvc_bunit": "string", "dvc_category": "string", "dvc_ip": "string", "dvc_mac": "string", "dvc_priority": "string", "dvc_zone": "string", "file_hash": "string", "file_name": "string", "file_path": "string", "file_size": "string", "http_content_type": "string", "http_method": "string", "http_referrer": "string", "http_referrer_domain": "string", "http_user_agent": "string", "icmp_code": "string", "icmp_type": "string", "id": "string", "ids_type": "string", "incident": "string", "ip": "string", "mac": "string", "message_id": "string", "message_info": "string", "message_priority": "string", "message_type": "string", "mitre_technique_id": "string", "msft": "string", "mskb": "string", "name": "string", "orig_dest": "string", "orig_recipient": "string", "orig_src": "string", "os": "string", "packets": "string", "packets_in": "string", "packets_out": "string", "parent_process": "string", "parent_process_id": "string", "parent_process_name": "string", "parent_process_path": "string", "password": "string", "payload": "string", "payload_type": "string", "priority": "string", "problem": "string", "process": "string", "process_hash": "string", "process_id": "string", "process_name": "string", "process_path": "string", "product_version": "string", "protocol": "string", "protocol_version": "string", "query": "string", "query_count": "string", "query_type": "string", "reason": "string", "recipient": "string", "recipient_count": "string", "recipient_domain": "string", "recipient_status": "string", "record_type": "string", "registry_hive": "string", "registry_key_name": "string", "registry_path": "string", "registry_value_data": "string", "registry_value_name": "string", "registry_value_text": "string", "registry_value_type": "string", "request_payload": "string", "request_payload_type": "string", "request_sent_time": "string", "response_code": "string", "response_payload_type": "string", "response_received_time": "string", "response_time": "string", "result": "string", "return_addr": "string", "rule": "string", "rule_action": "string", "sender": "string", "service": "string", "service_hash": "string", "service_id": "string", "service_name": "string", "service_path": "string", "session_id": "string", "sessions": "string", "severity": "string", "severity_id": "string", "sid": "string", "signature": "string", "signature_id": "string", "signature_version": "string", "site": "string", "size": "string", "source": "string", "sourcetype": "string", "src": "string", "src_bunit": "string", "src_category": "string", "src_dns": "string", "src_interface": "string", "src_ip": "string", "src_ip_range": "string", "src_mac": "string", "src_nt_domain": "string", "src_nt_host": "string", "src_port": "string", "src_priority": "string", "src_translated_ip": "string", "src_translated_port": "string", "src_type": "string", "src_user": "string", "src_user_bunit": "string", "src_user_category": "string", "src_user_domain": "string", "src_user_id": "string", "src_user_priority": "string", "src_user_role": "string", "src_user_type": "string", "src_zone": "string", "state": "string", "status": "string", "status_code": "string", "subject": "string", "tag": "string", "ticket_id": "string", "time": "string", "time_submitted": "string", "transport": "string", "transport_dest_port": "string", "type": "string", "uri": "string", "uri_path": "string", "uri_query": "string", "url": "string", "url_domain": "string", "url_length": "string", "user": "string", "user_agent": "string", "user_bunit": "string", "user_category": "string", "user_id": "string", "user_priority": "string", "user_role": "string", "user_type": "string", "vendor_account": "string", "vendor_product": "string", "vlan": "string", "xdelay": "string", "xref": "string", } } ASSET = { "Asset": { "asset": "string", "asset_id": "string", "asset_tag": "string", "bunit": "string", "category": "string", "city": "string", "country": "string", "dns": "string", "ip": "string", "is_expected": "string", "lat": "string", "long": "string", "mac": "string", "nt_host": "string", "owner": "string", "pci_domain": "string", "priority": "string", "requires_av": "string", } } IDENTITY = { "Identity": { "bunit": "string", "category": "string", "email": "string", "endDate": "string", "first": "string", "identity": "string", "identity_tag": "string", "last": "string", "managedBy": "string", "nick": "string", "phone": "string", "prefix": "string", "priority": "string", "startDate": "string", "suffix": "string", "watchlist": "string", "work_city": "string", "work_lat": "string", "work_long": "string", } } def test_get_cim_mapping_field_command(mocker): """Scenario: When the mapping is based on Splunk CIM.""" fields = splunk.get_cim_mapping_field_command() assert fields == {"Notable Data": NOTABLE, "Drilldown Data": DRILLDOWN, "Asset Data": ASSET, "Identity Data": IDENTITY} def test_build_search_human_readable(mocker): """ Given: table headers in query When: building a human readable table as part of splunk-search Then: Test headers are calculated correctly: * comma-separated, space-separated * support commas and spaces inside header values (if surrounded with parenthesis) * rename headers """ func_patch = mocker.patch("SplunkPy.update_headers_from_field_names") results = [ { "ID": 1, "Header with space": "h1", "header3": 1, "header_without_space": "1234", "old_header_1": "1", "old_header_2": "2", }, { "ID": 2, "Header with space": "h2", "header3": 2, "header_without_space": "1234", "old_header_1": "1", "old_header_2": "2", }, ] args = { "query": 'something | table ID "Header with space" header3 header_without_space ' 'comma,separated "Single,Header,with,Commas" old_header_1 old_header_2 | something else' " | rename old_header_1 AS new_header_1 old_header_2 AS new_header_2" } expected_headers = [ "ID", "Header with space", "header3", "header_without_space", "comma", "separated", "Single,Header,with,Commas", "new_header_1", "new_header_2", ] splunk.build_search_human_readable(args, results, sid="123456") headers = func_patch.call_args[0][1] assert headers == expected_headers def test_build_search_human_readable_multi_table_in_query(mocker): """ Given: multiple table headers in query When: building a human readable table as part of splunk-search Then: Test headers are calculated correctly: * all expected header exist without duplications """ args = {"query": " table header_1, header_2 | stats state_1, state_2 | table header_1, header_2, header_3, header_4"} results = [ {"header_1": "val_1", "header_2": "val_2", "header_3": "val_3", "header_4": "val_4"}, ] expected_headers_hr = "|header_1|header_2|header_3|header_4|\n|---|---|---|---|" hr = splunk.build_search_human_readable(args, results, sid="123456") assert expected_headers_hr in hr @pytest.mark.parametrize("polling, fast_mode", [(False, True), (True, True)]) def test_build_search_kwargs(polling, fast_mode): """ Given: The splunk-search command args. When: Running the build_search_kwargs to build the search query kwargs. Then: Ensure the query kwargs as expected. """ args = { "earliest_time": "2021-11-23T10:10:10", "latest_time": "2021-11-23T10:10:20", "app": "test_app", "fast_mode": fast_mode, "polling": polling, } kwargs_normalsearch = splunk.build_search_kwargs(args, polling) for field in args: if field == "polling": assert "exec_mode" in kwargs_normalsearch if polling: assert kwargs_normalsearch["exec_mode"] == "normal" else: assert kwargs_normalsearch["exec_mode"] == "blocking" elif field == "fast_mode" and fast_mode: assert kwargs_normalsearch["adhoc_search_level"] == "fast" else: assert field in kwargs_normalsearch @pytest.mark.parametrize("polling,status", [(False, "DONE"), (True, "DONE"), (True, "RUNNING")]) def test_splunk_search_command(mocker, polling, status): """ Given: A search query with args. When: Running the splunk_search_command with and without polling. Then: Ensure the result as expected in polling and in regular search. """ mock_args = { "query": "query", "earliest_time": "2021-11-23T10:10:10", "latest_time": "2020-10-20T10:10:20", "app": "test_app", "fast_mode": "false", "polling": polling, } mocker.patch.object(ScheduledCommand, "raise_error_if_not_supported") search_result = splunk.splunk_search_command(Service(status), mock_args) search_result = search_result if isinstance(search_result, CommandResults) else search_result[0] if search_result.scheduled_command: assert search_result.outputs["Status"] == status assert search_result.scheduled_command._args["sid"] == "123456" else: assert search_result.outputs["Splunk.Result"] == [] assert search_result.readable_output == "### Splunk Search results for query:\nsid: 123456\n**No entries.**\n" @pytest.mark.parametrize( "messages,expected_msg", [({"fatal": ["fatal msg"]}, "fatal msg"), ({"error": ["error msg"]}, "error msg")] ) def test_err_in_splunk_search(mocker, messages, expected_msg): """ Given: A wrong search query. When: Running the splunk_search_command. Then: Ensure the result as expected in polling and in regular search. """ mock_args = { "query": "wrong search query", "earliest_time": "2021-11-23T10:10:10", "latest_time": "2020-10-20T10:10:20", "fast_mode": "false", } service = Service(status="FAILED") service.jobs.state.content["messages"] = messages with pytest.raises(DemistoException) as e: splunk.splunk_search_command(service, mock_args) assert f"Failed to run the search in Splunk: {expected_msg}" in str(e) @pytest.mark.parametrize( argnames="credentials", argvalues=[{"username": "test", "password": "test"}, {"splunkToken": "token", "password": "test"}] ) def test_module_test(mocker, credentials): """ Given: - Credentials for connecting Splunk When: - Run test-module command Then: - Validate the info method was called """ # prepare mocker.patch.object(client.Service, "info") mocker.patch.object(client.Service, "login") service = client.Service(**credentials) # run splunk.test_module(service, {}) # validate assert service.info.call_count == 1 @pytest.mark.parametrize( argnames="credentials", argvalues=[{"username": "test", "password": "test"}, {"splunkToken": "token", "password": "test"}] ) def test_module__exception_raised(mocker, credentials): """ Given: - AuthenticationError was occurred When: - Run test-module command Then: - Validate the expected message was returned """ # prepare def exception_raiser(): raise AuthenticationError mocker.patch.object(AuthenticationError, "__init__", return_value=None) mocker.patch.object(client.Service, "info", side_effect=exception_raiser) mocker.patch.object(client.Service, "login") return_error_mock = mocker.patch(RETURN_ERROR_TARGET) service = client.Service(**credentials) # run splunk.test_module(service, {}) # validate assert return_error_mock.call_args[0][0] == "Authentication error, please validate your credentials." def test_module_hec_url(mocker): """ Given: - hec_url was is in params When: - Run test-module command Then: - Validate that the request.get was called with the expected args """ # prepare mocker.patch.object(client.Service, "info") mocker.patch.object(client.Service, "login") mocker.patch.object(requests, "get") service = client.Service(username="test", password="test") # run splunk.test_module(service, {"hec_url": "test_hec_url"}) # validate assert requests.get.call_args[0][0] == "test_hec_url/services/collector/health" def test_module_message_object(mocker): """ Given: - query results with one message item. When: - Run test-module command. Then: - Validate the test_module run successfully and the info method was called once. """ # prepare message = results.Message("DEBUG", "There's something in that variable...") mocker.patch("splunklib.results.JSONResultsReader", return_value=[message]) service = mocker.patch("splunklib.client.connect", return_value=None) # run splunk.test_module(service, {"isFetch": True, "fetchQuery": "something"}) # validate assert service.info.call_count == 1 def test_labels_with_non_str_values(mocker): """ Given: - Raw response with values in _raw that stored as dict or list When: - Fetch incidents Then: - Validate the Labels created in the incident are well formatted to avoid server errors on json.Unmarshal """ from SplunkPy import UserMappingObject # prepare raw = { "message": "Authentication of user via Radius", "actor_obj": {"id": "test", "type": "User", "alternateId": "test", "displayName": "test"}, "actor_list": [{"id": "test", "type": "User", "alternateId": "test", "displayName": "test"}], "actor_tuple": ("id", "test"), "num_val": 100, "bool_val": False, "float_val": 100.0, } mocked_response: list[results.Message | dict] = deepcopy(SAMPLE_RESPONSE) mocked_response[1]["_raw"] = json.dumps(raw) mock_last_run = {"time": "2018-10-24T14:13:20"} mock_params = {"fetchQuery": "something", "parseNotableEventsRaw": True} mocker.patch.object(demisto, "incidents") mocker.patch.object(demisto, "setLastRun") mocker.patch("demistomock.getLastRun", return_value=mock_last_run) mocker.patch("demistomock.params", return_value=mock_params) mocker.patch("splunklib.results.JSONResultsReader", return_value=mocked_response) # run service = mocker.patch("splunklib.client.connect", return_value=None) mapper = UserMappingObject(service, False) splunk.fetch_incidents( service, mapper, comment_tag_to_splunk="comment_tag_to_splunk", comment_tag_from_splunk="comment_tag_from_splunk" ) incidents = demisto.incidents.call_args[0][0] # validate assert demisto.incidents.call_count == 1 assert len(incidents) == 2 labels = incidents[0]["labels"] assert len(labels) >= 7 assert all(isinstance(label["value"], str) for label in labels) def test_empty_string_as_app_param_value(mocker): """ Given: - A mock to demisto.params that contains an 'app' key with an empty string as its value When: - Run splunk.get_connection_args() function Then: - Validate that the value of the 'app' key in connection_args is '-' """ # prepare mock_params = {"app": "", "host": "111", "port": "111"} # run connection_args = splunk.get_connection_args(mock_params) # validate assert connection_args.get("app") == "-" @pytest.mark.parametrize( argnames="host, expected_host", argvalues=[ ("8.8.8.8", "8.8.8.8"), ("8.8.8.8/", "8.8.8.8"), ("https://www.test.com", "www.test.com"), ("https://www.test.com/", "www.test.com"), ], ) def test_host_param(host, expected_host): """ Given: - Different host values When: - Run get_connection_args() function Then: - Ensure the host is as expected """ params = {"host": host, "port": "111"} actuall_host = splunk.get_connection_args(params)["host"] assert actuall_host == expected_host OWNER_MAPPING = [ {"xsoar_user": "test_xsoar", "splunk_user": "test_splunk", "wait": True}, {"xsoar_user": "test_not_full", "splunk_user": "", "wait": True}, {"xsoar_user": "", "splunk_user": "test_not_full", "wait": True}, ] MAPPER_CASES_XSOAR_TO_SPLUNK = [ ( "", "unassigned", "UserMapping: Could not find splunk user matching xsoar's . Consider adding it to the splunk_xsoar_users lookup.", ), ( "not_in_table", "unassigned", "UserMapping: Could not find splunk user matching xsoar's not_in_table. " "Consider adding it to the splunk_xsoar_users lookup.", ), ] @pytest.mark.parametrize("xsoar_name, expected_splunk, expected_msg", MAPPER_CASES_XSOAR_TO_SPLUNK) def test_owner_mapping_mechanism_xsoar_to_splunk(mocker, xsoar_name, expected_splunk, expected_msg): """ Given: - different xsoar values When: - fetching, or mirroring Then: - validates the splunk user is correct """ def mocked_get_record(col, value_to_search): return filter(lambda x: x[col] == value_to_search, OWNER_MAPPING[:-1]) service = mocker.patch("splunklib.client.connect", return_value=None) mapper = splunk.UserMappingObject( service, True, table_name="splunk_xsoar_users", xsoar_user_column_name="xsoar_user", splunk_user_column_name="splunk_user" ) mocker.patch.object(mapper, "_get_record", side_effect=mocked_get_record) error_mock = mocker.patch.object(demisto, "error") s_user = mapper.get_splunk_user_by_xsoar(xsoar_name) assert s_user == expected_splunk if error_mock.called: assert error_mock.call_args[0][0] == expected_msg MAPPER_CASES_SPLUNK_TO_XSOAR = [ ("test_splunk", "test_xsoar", None), ( "test_not_full", "", "UserMapping: Xsoar user matching splunk's test_not_full is empty. Fix the record in splunk_xsoar_users lookup.", ), ( "unassigned", "", "UserMapping: Could not find xsoar user matching splunk's unassigned. Consider adding it to the" " splunk_xsoar_users lookup.", ), ( "not_in_table", "", "UserMapping: Could not find xsoar user matching splunk's not_in_table. " "Consider adding it to the splunk_xsoar_users lookup.", ), ] @pytest.mark.parametrize("splunk_name, expected_xsoar, expected_msg", MAPPER_CASES_SPLUNK_TO_XSOAR) def test_owner_mapping_mechanism_splunk_to_xsoar(mocker, splunk_name, expected_xsoar, expected_msg): """ Given: - different xsoar values When: - fetching, or mirroring Then: - validates the splunk user is correct """ def mocked_get_record(col, value_to_search): return filter(lambda x: x[col] == value_to_search, OWNER_MAPPING) service = mocker.patch("splunklib.client.connect", return_value=None) mapper = splunk.UserMappingObject( service, True, table_name="splunk_xsoar_users", xsoar_user_column_name="xsoar_user", splunk_user_column_name="splunk_user" ) mocker.patch.object(mapper, "_get_record", side_effect=mocked_get_record) error_mock = mocker.patch.object(demisto, "error") s_user = mapper.get_xsoar_user_by_splunk(splunk_name) assert s_user == expected_xsoar if error_mock.called: assert error_mock.call_args[0][0] == expected_msg COMMAND_CASES = [ ( {"xsoar_username": "test_xsoar"}, # case normal single username was provided [{"SplunkUser": "test_splunk", "XsoarUser": "test_xsoar"}], ), ( {"xsoar_username": "test_xsoar, Non existing"}, # case normal multiple usernames were provided [{"SplunkUser": "test_splunk", "XsoarUser": "test_xsoar"}, {"SplunkUser": "unassigned", "XsoarUser": "Non existing"}], ), ( {"xsoar_username": "Non Existing,"}, # case normal&empty multiple usernames were provided [ {"SplunkUser": "unassigned", "XsoarUser": "Non Existing"}, {"SplunkUser": "Could not map splunk user, Check logs for more info.", "XsoarUser": ""}, ], ), ( {"xsoar_username": ["test_xsoar", "Non existing"]}, # case normal&missing multiple usernames were provided [{"SplunkUser": "test_splunk", "XsoarUser": "test_xsoar"}, {"SplunkUser": "unassigned", "XsoarUser": "Non existing"}], ), ( {"xsoar_username": ["test_xsoar", "Non existing"], "map_missing": False}, # case normal & missing multiple usernames were provided without missing's mapping activated [ {"SplunkUser": "test_splunk", "XsoarUser": "test_xsoar"}, {"SplunkUser": "Could not map splunk user, Check logs for more info.", "XsoarUser": "Non existing"}, ], ), ( {"xsoar_username": "Non Existing,", "map_missing": False}, # case missing&empty multiple usernames were provided [ {"SplunkUser": "Could not map splunk user, Check logs for more info.", "XsoarUser": "Non Existing"}, {"SplunkUser": "Could not map splunk user, Check logs for more info.", "XsoarUser": ""}, ], ), ] @pytest.mark.parametrize("xsoar_names, expected_outputs", COMMAND_CASES) def test_get_splunk_user_by_xsoar_command(mocker, xsoar_names, expected_outputs): """ Given: a list of xsoar users When: trying to get splunk matching users Then: validates correctness of list """ def mocked_get_record(col, value_to_search): return filter(lambda x: x[col] == value_to_search, OWNER_MAPPING[:-1]) service = mocker.patch("splunklib.client.connect", return_value=None) mapper = splunk.UserMappingObject( service, True, table_name="splunk_xsoar_users", xsoar_user_column_name="xsoar_user", splunk_user_column_name="splunk_user" ) # Ignoring logging pytest error mocker.patch.object(demisto, "error") mocker.patch.object(mapper, "_get_record", side_effect=mocked_get_record) res = mapper.get_splunk_user_by_xsoar_command(xsoar_names) assert res.outputs == expected_outputs @pytest.mark.parametrize( argnames="username, expected_username, basic_auth", argvalues=[("test_user", "test_user", False), ("test@_basic", "test", True)], ) def test_basic_authentication_param(mocker, username, expected_username, basic_auth): """ Given: - the username contain '@_basic' suffix When: - connecting to Splunk server Then: - validate the connection args was sent as expected """ mocked_params = { "host": "test_host", "port": "8089", "proxy": "false", "authentication": {"identifier": username, "password": "test_password"}, } mocker.patch.object(client, "connect") mocker.patch.object(demisto, "params", return_value=mocked_params) mocker.patch.object(demisto, "command", return_value="not_impl_command") with pytest.raises(NotImplementedError): splunk.main() assert client.connect.call_args[1]["username"] == expected_username assert ("basic" in client.connect.call_args[1]) == basic_auth @pytest.mark.parametrize( argnames="host, expected_base_url", argvalues=[ ("8.8.8.8", "https://8.8.8.8:8089/"), ("https://www.test.com", "https://www.test.com:8089/"), ("http://www.test.com", "https://http://www.test.com:8089/"), # we don't want to silently replace http with https ], ) def test_base_url(mocker, host, expected_base_url): """ Given: - Different host values When: - Running the splunk-notable-event-edit command Then: - Ensure the base URL is built as expected """ mocked_params = { "host": host, "port": "8089", "proxy": "false", "authentication": {"identifier": "username", "password": "test_password"}, } mocker.patch.object(demisto, "command", return_value="splunk-notable-event-edit") mocker.patch.object(demisto, "params", return_value=mocked_params) mocker.patch.object(client, "connect") cmd = mocker.patch.object(splunk, "splunk_edit_notable_event_command") splunk.main() assert cmd.call_args[0][0] == expected_base_url @pytest.mark.parametrize( "item, expected", [({"message": "Test message"}, False), (results.Message("INFO", "Test message"), True)] ) def test_handle_message(item: dict | results.Message, expected: bool): """ Tests that passing a results.Message object returns True """ assert splunk.handle_message(item) is expected def test_single_drilldown_searches(mocker): """ Given: - notable with single string represent dict, in the drilldown_searches key. When: - call to drilldown_enrichment. Then: - validate there is no errors in the process. """ drilldown_searches = json.dumps( {"name": "test drilldown", "search": "| from datamodel: test", "earliest": 1719218100, "latest": 1719823500} ) mocker.patch.object(demisto, "error") mocker.patch.object(splunk, "build_drilldown_search", return_value=None) splunk.drilldown_enrichment( service=None, notable_data={"drilldown_searches": drilldown_searches, "event_id": "test_id"}, num_enrichment_events=1 ) assert demisto.error.call_count == 0, "Something was wrong in the drilldown_enrichment process" @pytest.mark.parametrize( "drilldown_data, expected", [ ({"drilldown_search": "test"}, ["test"]), ({"drilldown_searches": '{"search_1":"test_1"}'}, [{"search_1": "test_1"}]), ( {"drilldown_searches": ['{"search_1":"test_1"}', '{"search_2":"test_2"}']}, [{"search_1": "test_1"}, {"search_2": "test_2"}], ), ({"drilldown_searches": '[{"search_1":"test_1"}]'}, [{"search_1": "test_1"}]), ( {"drilldown_searches": '[{"search_1":"test_1"}, {"search_2":"test_2"}]'}, [{"search_1": "test_1"}, {"search_2": "test_2"}], ), ], ) def test_get_drilldown_searches(drilldown_data, expected): """ Given: - 1. A notable data with a single 'old' (string value in the 'drilldown_search' key) drilldown enrichment data . 2. A notable data with a single drilldown enrichments as json string in the 'new' key (drilldown_searches). 3. A notable data with multiple drilldown enrichments as json string in the 'new' key (drilldown_searches). 4. A notable data with a single drilldown enrichments as json list string in the 'new' key (drilldown_searches). 5. A notable data with a multiple drilldown enrichments as json list string in the 'new' key (drilldown_searches). When: - call to get_drilldown_searches. Then: - validate the result are as expected. """ assert splunk.get_drilldown_searches(drilldown_data) == expected @pytest.mark.parametrize( "drilldown_search, expected_res", [ ('{"name":"test", "query":"|key="the value""}', 'key="the value"'), ('{"name":"test", "query":"|key in (line_1\nline_2)"}', "key in (line_1,line_2)"), ('{"name":"test", "query":"search a=$a|s$ c=$c$ suffix"}', "search a=$a|s$ c=$c$ suffix"), ], ) def test_escape_invalid_chars_in_drilldown_json(drilldown_search, expected_res): """ Scenario: When extracting the drilldown search query which are a json string, we should escape unescaped JSON special characters. Given: - A raw search query with text like 'key="a value"'. - A raw search query with text like where 'key in (a\nb)' which it should be 'key in (a,b)'. - A raw search query with normal json string, should not be changed by this function. When: - escape_invalid_chars_in_drilldown_json is called Then: - Return the expected result """ import json res = splunk.escape_invalid_chars_in_drilldown_json(drilldown_search) assert expected_res in json.loads(res)["query"] # Define minimal classes to simulate the service and index behavior class Index: def __init__(self, name): self.name = name class ServiceIndex: def __init__(self, indexes): self.indexes = [Index(name) for name in indexes] @pytest.mark.parametrize( "fields, expected", [ # Valid JSON input ('{"key": "value"}', {"key": "value"}), # Valid JSON with multiple key-value pairs ('{"key1": "value1", "key2": 2}', {"key1": "value1", "key2": 2}), # Invalid JSON input (non-JSON string) ("not a json string", {"fields": "not a json string"}), # Another invalid JSON input (partially structured JSON) ("{'key': 'value'}", {"fields": "{'key': 'value'}"}), ], ) def test_parse_fields(fields, expected): """ Given: A string representing fields, which may be a valid JSON string or a regular string. When: The parse_fields function is called with the given string. Then: If the string is valid JSON, the function returns a dictionary of the parsed fields. If the string is not valid JSON, the function returns a dictionary with a single key-value pair, where the entire input string is the key. """ from SplunkPy import parse_fields result = parse_fields(fields) assert result == expected @pytest.mark.parametrize( "event, batch_event_data, entry_id, expected_data", [ ("Somthing happened", None, None, '{"event": "Somthing happened", "fields": {"field1": "value1"}, "index": "main"}'), ( None, "{'event': 'some event', 'index': 'some index'} {'event': 'some event', 'index': 'some index'}", None, "{'event': 'some event', 'index': 'some index'} {'event': 'some event', 'index': 'some index'}", ), # Batch event data ( None, None, "some entry_id", "{'event': 'some event', 'index': 'some index'} {'event': 'some event', 'index': 'some index'}", ), ( None, """{'event': "some event's", 'index': 'some index'} {'event': 'some event', 'index': 'some index'}""", None, """{'event': "some event's", 'index': 'some index'} {'event': 'some event', 'index': 'some index'}""", ), # with ' ( None, None, "some entry_id", "{'event': 'some event', 'index': 'some index'} {'event': 'some event', 'index': 'some index'}", ), ], ) @patch("SplunkPy.VERIFY_CERTIFICATE", True) @patch("requests.post") @patch("SplunkPy.get_events_from_file") @patch("SplunkPy.parse_fields") def test_splunk_submit_event_hec( mock_parse_fields, mock_get_events_from_file, mock_post, event, batch_event_data, entry_id, expected_data, ): """ Given: Different types of event submission (single event, batch event, entry_id). When: Calling splunk_submit_event_hec. Then: Ensure a POST request is sent with the correct data and headers. """ from SplunkPy import splunk_submit_event_hec # Arrange hec_token = "valid_token" baseurl = "https://splunk.example.com" fields = '{"field1": "value1"}' parsed_fields = {"field1": "value1"} # Mocks mock_parse_fields.return_value = parsed_fields if entry_id: # Entry ID mock_get_events_from_file.return_value = ( "{'event': 'some event', 'index': 'some index'} {'event': 'some event', 'index': 'some index'}" ) # Act splunk_submit_event_hec( hec_token=hec_token, baseurl=baseurl, event=event, fields=fields, host=None, index="main", source_type=None, source=None, time_=None, request_channel="test_channel", batch_event_data=batch_event_data, entry_id=entry_id, service=MagicMock(), ) mock_post.assert_called_once_with( f"{baseurl}/services/collector/event", data=expected_data, headers={ "Authorization": f"Splunk {hec_token}", "Content-Type": "application/json", "X-Splunk-Request-Channel": "test_channel", }, verify=True, ) def test_splunk_submit_event_hec_command_no_required_arguments(): """Given: none of these arguments: 'entry_id', 'event', 'batch_event_data' When: Runing splunk-submit-event-hec command Then: An exception is thrown """ from SplunkPy import splunk_submit_event_hec_command with pytest.raises( DemistoException, match=r"Invalid input: Please specify one of the following arguments: `event`, `batch_event_data`, or `entry_id`.", ): splunk_submit_event_hec_command({"hec_url": "hec_url"}, None, {}) @pytest.mark.parametrize(argnames="should_map_user", argvalues=[True, False]) def test_get_modified_remote_data_command_with_user_mapping(mocker, should_map_user): """Given: - Different values for the splunk.UserMappingObject.should_map arguments and `notable` query response without 'owner' key When: - Runing test_get_modified_remote_data_command Then: - Verify the correct owner are returned. """ notable_without_owner = deepcopy(SAMPLE_RESPONSE[2]) del notable_without_owner["owner"] mapped_user = "mapped_splunk_user" mocker.patch.object(demisto, "results") mocker.patch.object(demisto, "params", return_value={"timezone": "0"}) mocker.patch.object(splunk.UserMappingObject, "get_xsoar_user_by_splunk", return_value=mapped_user) mocker.patch("SplunkPy.results.JSONResultsReader", side_effect=lambda res: res) mocked_service = mocker.patch("SplunkPy.client.Service") mocker.patch("SplunkPy.get_integration_context") mocked_service.jobs.oneshot = ( lambda query, **kwargs: [SAMPLE_INCIDENT_REVIEW_RESPONSE[0]] if "`incident_review`" in query else [notable_without_owner] ) splunk.get_modified_remote_data_command( mocked_service, args={"lastUpdate": "2021-02-09T16:41:30.589575+02:00"}, mapper=splunk.UserMappingObject(mocked_service, should_map_user), comment_tag_from_splunk="comment_tag_from_splunk", close_incident=True, close_end_statuses=False, close_extra_labels=[], ) contents = demisto.results.call_args[0][0][0]["Contents"] expected_owner = mapped_user if should_map_user else SAMPLE_INCIDENT_REVIEW_RESPONSE[0]["owner"] assert len(contents["SplunkComments"]) == 3 assert contents["owner"] == expected_owner def test_mirror_in_with_enrichment_enabled(mocker): """ Given: - Drilldown Enrichmnet enabled in the instance configuration When: - Mirror in run (get-modified-remote-data) Then: - Validate the integration context stored the "delta" for the incident which sent to enrichment but not yet created in order to create the incident with the updated fields. """ # create an integration context in order to simulate the context in a normal run. integration_context = { splunk.CACHE: json.dumps( {splunk.SUBMITTED_NOTABLES: [splunk.Notable(SAMPLE_RESPONSE[2])]}, default=lambda obj: obj.__dict__ ), } mocker.patch("SplunkPy.set_integration_context") mocker.patch("SplunkPy.get_integration_context", return_value=integration_context) mocker.patch.object(demisto, "params", return_value={"timezone": "0"}) mocker.patch.object(splunk, "ENABLED_ENRICHMENTS", new=[splunk.DRILLDOWN_ENRICHMENT]) mocker.patch("SplunkPy.results.JSONResultsReader", side_effect=lambda res: res) mocker.patch.object(splunk.UserMappingObject, "get_xsoar_user_by_splunk", return_value="after_mirror_owner") mocked_service = mocker.patch("SplunkPy.client.Service") # the get_modified_remote_data send two queries to Splunk, # the first one used the `incident_review` macro and the second used the `notable` notable_delta = {"status_label": "after_mirror_status", "urgency": "after_mirror_urgency"} updated_notable = SAMPLE_RESPONSE[2] | notable_delta mocked_service.jobs.oneshot = ( lambda query, **kwargs: [SAMPLE_INCIDENT_REVIEW_RESPONSE[0]] if "`incident_review`" in query else [updated_notable] ) splunk.get_modified_remote_data_command( mocked_service, args={"lastUpdate": "2021-02-09T16:41:30.589575+02:00"}, mapper=splunk.UserMappingObject(mocked_service, True), comment_tag_from_splunk="comment_tag_from_splunk", close_incident=True, close_end_statuses=False, close_extra_labels=[], ) notable_id = SAMPLE_RESPONSE[2]["event_id"] mirrored_enriching_natables = splunk.set_integration_context.call_args[0][0][splunk.MIRRORED_ENRICHING_NOTABLES] actual_mirrored_notable_delta = mirrored_enriching_natables[notable_id] # 'SplunkComments' is calculated and added in the mirror process and not returned from Splunk assert "SplunkComments" not in updated_notable assert len(actual_mirrored_notable_delta["SplunkComments"]) == 3 assert actual_mirrored_notable_delta["owner"] == "after_mirror_owner" assert all(actual_mirrored_notable_delta[k] == v for k, v in notable_delta.items()) def test_format_splunk_note_for_xsoar_basic(): """ Given: - A Splunk note with URL-encoded title and content. When: - Formatting the note for XSOAR. Then: - The title and content are URL-decoded and separated by a blank line, ending with a newline. """ note = {"title": "My%20Title", "content": "Line%201%0ALine%202"} expected = "My Title\n\nLine 1\nLine 2\n" assert splunk.format_splunk_note_for_xsoar(note) == expected def test_format_splunk_note_for_xsoar_empty_content(): """ Given: - A Splunk note with a URL-encoded title and empty content. When: - Formatting the note for XSOAR. Then: - The title is URL-decoded and followed by two newlines (current implementation behavior). """ note = {"title": "Only%20Title", "content": ""} expected = "Only Title\n\n" assert splunk.format_splunk_note_for_xsoar(note) == expected def test_format_splunk_note_for_xsoar_missing_fields(): """ Given: - A Splunk note missing both title and content. When: - Formatting the note for XSOAR. Then: - Returns two newlines as per current implementation. """ note = {} expected = "\n\n" assert splunk.format_splunk_note_for_xsoar(note) == expected def test_user_mapping_used_cache(mocker): """ Given: - A KVStore table exist in SPlunk to map the Splunk user to the XSOAR user. When: - Call to the function to map the user. Then: - Validate that the function use cache to store the mapped values and called only once. """ mocker.patch.object(demisto, "error") mocked_service = mocker.patch("SplunkPy.client.Service") mapper = splunk.UserMappingObject(mocked_service, True) for _ in range(5): mapper.get_xsoar_user_by_splunk("test_splunk_user") assert mocked_service.kvstore.__getitem__().data.query.call_count == 1 @pytest.mark.parametrize( "query, expected_query", [ ("search index=_internal", "search index=_internal"), ("| inputlookup some_lookup", "| inputlookup some_lookup"), ("index=_internal", "search index=_internal"), ], ) def test_splunk_job_create_command(mocker, query, expected_query): mocked_service = mocker.patch("SplunkPy.client.Service") mocked_create_job = MagicMock() mocked_service.jobs.create = mocked_create_job mocker.patch("SplunkPy.return_results") args = {"query": query} splunk.splunk_job_create_command(mocked_service, args) mocked_create_job.assert_called_once_with(expected_query, exec_mode="normal", app="") def mock_service_job(sid): class MockJob: def __init__(self, state): self.state = MagicMock() self.state.content = {"dispatchState": state} class MockResponse: def __init__(self, status, reason, body): self.status = status self.reason = reason self.body = body self.headers = {} class MockBody: def __init__(self, message): self.message = message def read(self): return self.message if sid == "valid_sid": return MockJob("DONE") elif sid == "running_sid": return MockJob("RUNNING") elif sid == "error_sid": raise HTTPError(MockResponse("418", "I'm a teapot", MockBody("I won't brew coffee."))) else: raise HTTPError(MockResponse("404", "Not Found", MockBody("Unknown sid."))) @patch("SplunkPy.client.Service") def test_splunk_job_status_valid(mock_service): mock_service.job.side_effect = mock_service_job service = mock_service args = {"sid": "valid_sid"} result = splunk.splunk_job_status(service, args) assert len(result) == 1 assert result[0].outputs == {"SID": "valid_sid", "Status": "DONE"} assert "Splunk Job Status" in result[0].readable_output @patch("SplunkPy.client.Service") def test_splunk_job_status_running(mock_service): mock_service.job.side_effect = mock_service_job service = mock_service args = {"sid": "running_sid"} result = splunk.splunk_job_status(service, args) assert len(result) == 1 assert result[0].outputs == {"SID": "running_sid", "Status": "RUNNING"} assert "Splunk Job Status" in result[0].readable_output @patch("SplunkPy.client.Service") def test_splunk_job_status_not_found(mock_service): mock_service.job.side_effect = mock_service_job service = mock_service args = {"sid": "invalid_sid"} result = splunk.splunk_job_status(service, args) assert len(result) == 1 assert result[0].readable_output == "Not found job for SID: invalid_sid" @patch("SplunkPy.client.Service") def test_splunk_job_status_418_error(mock_service): mock_service.job.side_effect = mock_service_job service = mock_service args = {"sid": "error_sid"} result = splunk.splunk_job_status(service, args) assert len(result) == 1 assert ( "Querying splunk for SID: error_sid resulted in the following error HTTP 418 I'm a teapot -- I won't brew coffee" in result[0].readable_output ) @patch("SplunkPy.client.Service") def test_splunk_job_status_multiple_sids(mock_service): mock_service.job.side_effect = mock_service_job service = mock_service args = {"sid": "valid_sid,running_sid,invalid_sid"} result = splunk.splunk_job_status(service, args) assert len(result) == 3 assert result[0].outputs == {"SID": "valid_sid", "Status": "DONE"} assert result[1].outputs == {"SID": "running_sid", "Status": "RUNNING"} assert result[2].readable_output == "Not found job for SID: invalid_sid" def test_splunk_search_parse_bad_chars(): """ Given: The splunk search output contains a json string with invalid chars. (e.g. 0xa0, 0xd1 etc.) When: Attempting to parse the results from splunk search. Then: The parsing removes the bad chars and proceeds successfully. """ import io bad_search_output = b'{"preview": false, "init_offset": 0, "messages": [], "fields": [{"name": "Message"}, {"name": "_bkt"}, \ {"name": "_cd"}, {"name": "_indextime"}, {"name": "_pre_msg"}, {"name": "_raw"}, {"name": "_serial"}, {"name": "_si"}, \ {"name": "_sourcetype"}, {"name": "_time"}, {"name": "host"}, {"name": "index"}, {"name": "linecount"}, \ {"name": "source"}, {"name": "sourcetype"}, {"name": "splunk_server"}], \ "results": [{"Message": "Service \xd1started\xa0 successfully.", "_bkt": "main~1111~00000000-0000-0000-0000-000000000000", \ "_cd": "1111:0000000", "_indextime": "5555555555", "_pre_msg": "04/23/2025 08:04:41 AM\\nLogName=Test log\\n\ SourceName=Server\\nEventCode=0\\nEventType=4\\nType=Information\xa0\\nComputerName=#COMPUTERNAME#\\nTaskCategory=\ Test log Server\\nOpCode=Info\\nRecordNumber=3\\nKeywords=Classic", "_raw": "04/23/2025 08:04:41 AM\\nLogName=Test log\\n\ SourceName=Server\\nEventCode=0\\nEventType=4\\nType=Information\xa0\\nComputerName=#COMPUTERNAME#\\nTaskCategory=Test log \ Server\\nOpCode=Info\\nRecordNumber=3\\nKeywords=Classic\\nMessage=Service started successfully.\\n", "_serial": "1", \ "_si": ["ip-000-00-00-000", "main"], "_sourcetype": "WinEventLog", "_time": "2025-04-23T05:04:41.000-03:00", \ "host": "127.0.0.1", "index": "main", "linecount": "13", "source": "WinEventLog:Server", "sourcetype": "WinEventLog", \ "splunk_server": "ip-000-00-00-000"}], "highlighted": {}}' expected_res = ( [ { "Message": "Service started successfully.", "_bkt": "main~1111~00000000-0000-0000-0000-000000000000", "_cd": "1111:0000000", "_indextime": "5555555555", "_pre_msg": ( "04/23/2025 08:04:41 AM\nLogName=Test log\nSourceName=Server\nEventCode=0\nEventType=4\nType=Information\n" "ComputerName=#COMPUTERNAME#\nTaskCategory=Test log Server\nOpCode=Info\nRecordNumber=3\nKeywords=Classic" ), "_raw": ( "04/23/2025 08:04:41 AM\nLogName=Test log\nSourceName=Server\nEventCode=0\nEventType=4\nType=Information\n" "ComputerName=#COMPUTERNAME#\nTaskCategory=Test log Server\nOpCode=Info\nRecordNumber=3\nKeywords=Classic\n" "Message=Service started successfully.\n" ), "_serial": "1", "_si": ["ip-000-00-00-000", "main"], "_sourcetype": "WinEventLog", "_time": "2025-04-23T05:04:41.000-03:00", "host": "127.0.0.1", "index": "main", "linecount": "13", "source": "WinEventLog:Server", "sourcetype": "WinEventLog", "splunk_server": "ip-000-00-00-000", } ], [{"Indicator": "127.0.0.1", "Type": "hostname", "Vendor": "Splunk", "Score": 0, "isTypedIndicator": True}], ) mock_result_batch = io.BytesIO(bad_search_output) res = splunk.parse_batch_of_results(mock_result_batch, 10, "") assert res == expected_res @pytest.mark.parametrize( "args", [ {"entry_id": "entry_id"}, {"index": "invalid_index", "event": {"event_data": "event_data"}}, ], ) def test_splunk_submit_event_hec_command_invalid_index(mocker, requests_mock, args): """ Given: - An event to submit to Splunk via HEC with an invalid index. When: - Calling splunk_submit_event_hec_command. Then: - The function should not raise an exception - as we don't check for invalid indexes. """ from SplunkPy import splunk_submit_event_hec_command from splunklib.client import Service mocker.patch("SplunkPy.get_events_from_file", return_value=[{"event": "test", "index": "invalid_index"}]) mocker.patch(RETURN_ERROR_TARGET) requests_mock.post("https://splunk.test.com/services/collector/event") service_mock = MagicMock(spec=Service) index_mock = MagicMock() index_mock.name = "valid_index" service_mock.indexes = [index_mock] splunk_submit_event_hec_command( params={"hec_token": "token", "hec_url": "https://splunk.test.com"}, service=service_mock, args=args ) def test_get_modified_remote_data_skips_cached_events(mocker): """ Given: - An initial run of get_modified_remote_data_command processes an event. When: - get_modified_remote_data_command is called a second time, and the same event is mirrored again. Then: - Ensure the event from the second run is skipped because its key (event_id:timestamp) is already in the cache. """ from SplunkPy import get_modified_remote_data_command test_id = "event_123" timestamp = "1737547610.49" event_key = f"{test_id}:{timestamp}" func_call_kwargs = { "args": {"lastUpdate": "2021-02-09T16:41:30.589575+02:00", "id": "id"}, "close_incident": False, "close_end_statuses": False, "close_extra_labels": ["Custom"], "mapper": splunk.UserMappingObject(MagicMock(), False), "comment_tag_from_splunk": "from_splunk", } # Mock Splunk API responses incident_review_response = [ { "rule_id": test_id, "last_modified_timestamp": timestamp, "event_id": test_id, } ] # === First Run: Process and cache the event === mocker.patch("splunklib.results.JSONResultsReader", return_value=incident_review_response) mocker.patch("SplunkPy.get_integration_context", return_value={}) mocker.patch("SplunkPy.demisto.results") mocker.patch("SplunkPy.demisto.params", return_value={"timezone": "0"}) set_context_mock = mocker.patch("SplunkPy.set_integration_context") extensive_log_mock = mocker.patch("SplunkPy.extensive_log") get_modified_remote_data_command(MagicMock(), **func_call_kwargs) results = demisto.results.call_args[0][0][0]["Contents"] assert results["event_id"] == test_id # Assert the event was cached set_context_mock.assert_called_once() cached_context = set_context_mock.call_args[0][0] assert cached_context.get("processed_mirror_in_events_cache") == [event_key] # === Second Run: Should skip the cached event === mocker.patch("SplunkPy.get_integration_context", return_value=cached_context) set_context_mock.reset_mock() demisto.results.reset_mock() extensive_log_mock.reset_mock() get_modified_remote_data_command(MagicMock(), **func_call_kwargs) # Assert no new events were processed results = demisto.results.call_args[0][0] assert len(results) == 0 assert extensive_log_mock.call_args_list[0].contains("mirror-in: no notables was changed since") # ============================================================================================ # COMMENT HANDLING TESTS FOR get_modified_remote_data_command for supported Splunk ES versions # ============================================================================================ def test_get_modified_remote_data_legacy_comment_mechanism(mocker): """ Test the legacy comment mechanism for Splunk ES versions below 8.0.0. Given: - Splunk ES version is below 8.0.0 - A notable with comments in the 'comment' field from incident_review When: - get_modified_remote_data_command is called Then: - Comments are extracted from the 'comment' field in the notable data - Comment entries are created with proper tags - The notable search is performed to get full event details """ from SplunkPy import get_modified_remote_data_command test_id = "legacy_comment_test" timestamp = "1737547610.49" comment_text = "This is a legacy comment from Splunk" func_call_kwargs = { "args": {"lastUpdate": "2021-02-09T16:41:30.589575+02:00", "id": "id"}, "close_incident": False, "close_end_statuses": False, "close_extra_labels": ["Custom"], "mapper": splunk.UserMappingObject(MagicMock(), False), "comment_tag_from_splunk": "from_splunk", } # Mock incident_review response incident_review_response = [ { "rule_id": test_id, "last_modified_timestamp": timestamp, "event_id": test_id, "comment": comment_text, } ] # Mock notable search response with full event details notable_search_response = [ { "rule_id": test_id, "event_id": test_id, "comment": [comment_text, "Additional comment"], "status": "new", } ] # Mock Splunk ES version check to return False (legacy version) mocker.patch("SplunkPy.is_splunk_es_version_or_higher", return_value=False) mocker.patch("splunklib.results.JSONResultsReader", side_effect=[incident_review_response, notable_search_response]) mocker.patch("SplunkPy.get_integration_context", return_value={}) mocker.patch("SplunkPy.set_integration_context") mocker.patch("SplunkPy.demisto.params", return_value={"timezone": "0"}) results_mock = mocker.patch("SplunkPy.demisto.results") get_modified_remote_data_command(MagicMock(), **func_call_kwargs) # Verify results structure results = results_mock.call_args[0][0] assert len(results) == 2 # 1 notable entry + 1 comment entry # Verify notable entry notable_entry = results[0] assert notable_entry["EntryContext"]["mirrorRemoteId"] == test_id assert notable_entry["Contents"]["rule_id"] == test_id # Verify comment entry comment_entry = results[1] assert comment_entry["Type"] == 1 # EntryType.NOTE assert comment_entry["Tags"] == ["from_splunk"] assert comment_entry["Note"] is True assert comment_entry["EntryContext"]["mirrorRemoteId"] == test_id # Verify that all comments are stored under SplunkComments in the notable data notable_data = notable_entry["Contents"] assert "SplunkComments" in notable_data assert isinstance(notable_data["SplunkComments"], list) assert len(notable_data["SplunkComments"]) == 2 # Both comments from notable search # Verify comment structure and content splunk_comments = notable_data["SplunkComments"] comment_texts = [comment["Comment"] for comment in splunk_comments] assert comment_text in comment_texts assert "Additional comment" in comment_texts # Verify each comment has the correct structure for comment in splunk_comments: assert isinstance(comment, dict) assert "Comment" in comment assert isinstance(comment["Comment"], str) def test_get_modified_remote_data_new_comment_mechanism_es8(mocker): """ Test the new comment mechanism for Splunk ES version 8.0.0 and above. Given: - Splunk ES version is 8.0.0 or higher - A notable that was modified - Comments are stored in mc_notes KV store When: - get_modified_remote_data_command is called Then: - Comments are fetched from mc_notes KV store via get_comments_data_new - Comment entries are created with proper structure - No additional notable search is performed """ from SplunkPy import get_modified_remote_data_command test_id = "new_comment_test" timestamp = "1737547610.49" func_call_kwargs = { "args": {"lastUpdate": "2021-02-09T16:41:30.589575+02:00", "id": "id"}, "close_incident": False, "close_end_statuses": False, "close_extra_labels": ["Custom"], "mapper": splunk.UserMappingObject(MagicMock(), False), "comment_tag_from_splunk": "from_splunk", } # Mock incident_review response incident_review_response = [ { "rule_id": test_id, "last_modified_timestamp": timestamp, "event_id": test_id, } ] # Mock KV store response with comments data # Note: create_time needs to be greater than lastUpdate (1612874490.589575) to create entry mock_comments_data = [ { "notable_id": test_id, "comment": "New comment from mc_notes", "update_time": 1757409703.589575, # Greater than lastUpdate "create_time": 1757409703, # Greater than lastUpdate } ] # Mock Splunk ES version check to return True (ES 8.0+) mocker.patch("SplunkPy.is_splunk_es_version_or_higher", return_value=True) mocker.patch("splunklib.results.JSONResultsReader", return_value=incident_review_response) # Mock KV store operations instead of get_comments_data_new mock_service = MagicMock() mock_kv_store = MagicMock() mock_kv_store.data.query.return_value = mock_comments_data mock_service.kvstore.__getitem__.return_value = mock_kv_store # Mock the helper functions used by get_comments_data_new mocker.patch("SplunkPy.get_comments_data_old", return_value=[]) mocker.patch("SplunkPy.format_splunk_note_for_xsoar", return_value="New comment from mc_notes") mocker.patch("SplunkPy.get_integration_context", return_value={}) mocker.patch("SplunkPy.set_integration_context") mocker.patch("SplunkPy.demisto.params", return_value={"timezone": "0"}) results_mock = mocker.patch("SplunkPy.demisto.results") get_modified_remote_data_command(mock_service, **func_call_kwargs) # Verify results structure results = results_mock.call_args[0][0] assert len(results) == 2 # 1 notable entry + 1 comment entry # Verify notable entry notable_entry = results[0] assert notable_entry["EntryContext"]["mirrorRemoteId"] == test_id # Verify comment entry from new mechanism comment_entry = results[1] assert comment_entry["Type"] == 1 # EntryType.NOTE assert comment_entry["Tags"] == ["from_splunk"] assert comment_entry["Contents"] == "New comment from mc_notes" assert comment_entry["ContentsFormat"] == "markdown" # Verify MD format def test_get_modified_remote_data_new_mechanism_with_multiple_comments(mocker): """ Test the new comment mechanism handles multiple comments correctly. Given: - Splunk ES version is 8.0.0 or higher - Multiple comments exist for a notable in mc_notes When: - get_modified_remote_data_command is called Then: - All comments are fetched and processed - Multiple comment entries are created - Comments are properly tagged and formatted """ from SplunkPy import get_modified_remote_data_command test_id = "multi_comment_test" timestamp = "1737547610.49" func_call_kwargs = { "args": {"lastUpdate": "2021-02-09T16:41:30.589575+02:00", "id": "id"}, "close_incident": False, "close_end_statuses": False, "close_extra_labels": ["Custom"], "mapper": splunk.UserMappingObject(MagicMock(), False), "comment_tag_from_splunk": "from_splunk", } # Mock incident_review response incident_review_response = [ { "rule_id": test_id, "last_modified_timestamp": timestamp, "event_id": test_id, } ] # Mock KV store response with multiple comments data mock_comments_data = [ { "notable_id": test_id, "comment": "First comment from mc_notes", "update_time": 1757409703.589575, "create_time": 1757409703, }, { "notable_id": test_id, "comment": "Second comment from mc_notes", "update_time": 1757409704.589575, "create_time": 1757409704, }, ] # Mock Splunk ES version check to return True (ES 8.0+) mocker.patch("SplunkPy.is_splunk_es_version_or_higher", return_value=True) mocker.patch("splunklib.results.JSONResultsReader", return_value=incident_review_response) # Mock KV store operations mock_service = MagicMock() mock_kv_store = MagicMock() mock_kv_store.data.query.return_value = mock_comments_data mock_service.kvstore.__getitem__.return_value = mock_kv_store # Mock the helper functions used by get_comments_data_new mocker.patch("SplunkPy.get_comments_data_old", return_value=[]) mocker.patch("SplunkPy.format_splunk_note_for_xsoar", side_effect=lambda note: note["comment"]) mocker.patch("SplunkPy.get_integration_context", return_value={}) mocker.patch("SplunkPy.set_integration_context") mocker.patch("SplunkPy.demisto.params", return_value={"timezone": "0"}) results_mock = mocker.patch("SplunkPy.demisto.results") get_modified_remote_data_command(mock_service, **func_call_kwargs) # Verify results structure results = results_mock.call_args[0][0] assert len(results) == 3 # 1 notable entry + 2 comment entries # Verify notable entry notable_entry = results[0] assert notable_entry["EntryContext"]["mirrorRemoteId"] == test_id # Verify both comment entries comment_entries = results[1:] assert len(comment_entries) == 2 for comment_entry in comment_entries: assert comment_entry["Type"] == 1 assert comment_entry["Tags"] == ["from_splunk"] assert comment_entry["ContentsFormat"] == "markdown" assert "comment from mc_notes" in comment_entry["Contents"] def test_get_modified_remote_data_legacy_vs_new_mechanism_behavior(mocker): """ Test that the correct comment mechanism is used based on Splunk ES version. Given: - Same notable data and test conditions When: - get_modified_remote_data_command is called with different ES versions Then: - Legacy mechanism (< 8.0.0) uses get_comments_data_old and notable search - New mechanism (>= 8.0.0) uses get_comments_data_new and mc_notes """ from SplunkPy import get_modified_remote_data_command test_id = "version_comparison_test" timestamp = "1737547610.49" func_call_kwargs = { "args": {"lastUpdate": "2021-02-09T16:41:30.589575+02:00", "id": "id"}, "close_incident": False, "close_end_statuses": False, "close_extra_labels": ["Custom"], "mapper": splunk.UserMappingObject(MagicMock(), False), "comment_tag_from_splunk": "from_splunk", } incident_review_response = [ { "rule_id": test_id, "last_modified_timestamp": timestamp, "event_id": test_id, "comment": "Legacy comment", } ] notable_search_response = [ { "rule_id": test_id, "event_id": test_id, "comment": "Legacy comment", } ] # Test Legacy Mechanism (ES < 8.0.0) mocker.patch("splunklib.results.JSONResultsReader", side_effect=[incident_review_response, notable_search_response]) mocker.patch("SplunkPy.get_integration_context", side_effect=[{}, {}]) mocker.patch("SplunkPy.set_integration_context") mocker.patch("SplunkPy.demisto.params", return_value={"timezone": "0"}) mocker.patch("SplunkPy.is_splunk_es_version_or_higher", return_value=False) get_comments_data_old_mock = mocker.patch("SplunkPy.get_comments_data_old", return_value=[]) get_comments_data_new_mock = mocker.patch("SplunkPy.get_comments_data_new", return_value=[]) results_mock = mocker.patch("SplunkPy.demisto.results") mock_service_legacy = MagicMock() get_modified_remote_data_command(mock_service_legacy, **func_call_kwargs) # Verify legacy mechanism was used get_comments_data_old_mock.assert_called_once() get_comments_data_new_mock.assert_not_called() # Reset mocks for new mechanism test get_comments_data_old_mock.reset_mock() get_comments_data_new_mock.reset_mock() results_mock.reset_mock() # Test New Mechanism (ES >= 8.0.0) mocker.patch("SplunkPy.is_splunk_es_version_or_higher", return_value=True) mocker.patch("splunklib.results.JSONResultsReader", return_value=incident_review_response) mock_service_new = MagicMock() get_modified_remote_data_command(mock_service_new, **func_call_kwargs) # Verify new mechanism was used get_comments_data_new_mock.assert_called_once() get_comments_data_old_mock.assert_not_called() def test_fetch_notables_with_comments_es8_mechanism(mocker): """ Test that fetch brings comments even when they're not part of the main fetch query results in ES 8.0+. Given: - Splunk ES version is 8.0.0 or higher - Fetch operation is running - Comments exist in mc_notes but not in the main notable search results When: - fetch_notables is called Then: - Comments are fetched via get_comments_data_new with is_fetch=True - Comments are stored in the notable under 'comment' key for incident creation - The fetch process includes comments even if they weren't in the original search """ from SplunkPy import fetch_notables test_id = "fetch_comment_test" # Mock notable data without comments in the main search notable_data = { "rule_id": test_id, "event_id": test_id, "_time": "2021-02-09T16:41:30.589575+02:00", "rule_name": "Test Rule", "status": "new", } # Mock search results (no comments in main search) search_results = [notable_data] # Mock that ES version check returns True mocker.patch("SplunkPy.is_splunk_es_version_or_higher", return_value=True) # Mock the search execution mocker.patch("splunklib.results.JSONResultsReader", return_value=search_results) # Mock KV store operations for get_comments_data_new mock_comments_data = [ { "notable_id": test_id, "comment": "Comment from fetch", "update_time": 1757409703.589575, "create_time": 1757409703, } ] mock_kv_store = MagicMock() mock_kv_store.data.query.return_value = mock_comments_data # Mock helper functions mocker.patch("SplunkPy.get_comments_data_old", return_value=[]) mocker.patch("SplunkPy.get_fetch_start_times", return_value=("", "")) mocker.patch("SplunkPy.remove_irrelevant_incident_ids") mocker.patch("SplunkPy.format_splunk_note_for_xsoar", return_value="Comment from fetch") # Mock other dependencies mocker.patch("SplunkPy.demisto.getLastRun", return_value={}) mocker.patch( "SplunkPy.demisto.params", return_value={ "timezone": "0", "fetchQuery": "|`notable`", "earliest_fetch_time_fieldname": "_time", "latest_fetch_time_fieldname": "_time", }, ) mocker.patch("SplunkPy.demisto.setLastRun") mocker.patch("SplunkPy.demisto.incidents") mocker.patch("SplunkPy.get_current_splunk_time", return_value="2021-02-09T17:41:30.589575+02:00") # Create mock service and mapper mock_service = MagicMock() mock_service.jobs.oneshot.return_value = MagicMock() mock_service.kvstore.__getitem__.return_value = mock_kv_store mock_mapper = splunk.UserMappingObject(MagicMock(), False) # Call fetch_notables fetch_notables( service=mock_service, mapper=mock_mapper, comment_tag_to_splunk="to_splunk", comment_tag_from_splunk="from_splunk", ) # Verify that KV store was accessed (indicating get_comments_data_new was called) mock_service.kvstore.__getitem__.assert_called_with("mc_notes") def test_fetch_vs_mirror_comment_storage_difference(mocker): """ Test that comments are stored differently in fetch vs mirror-in operations. Given: - Same notable and comment data - ES version 8.0.0 or higher When: - get_comments_data_new is called with is_fetch=True vs is_fetch=False Then: - In fetch (is_fetch=True): comments stored under 'comment' key as list of strings - In mirror-in (is_fetch=False): comments stored under 'SplunkComments' key as list of dicts """ from SplunkPy import get_comments_data_new test_id = "comment_storage_test" # Mock notable map id_to_notable_map = { test_id: { "rule_id": test_id, "event_id": test_id, } } # Mock KV store response with comments mock_comments_data = [ { "notable_id": test_id, "comment": "First comment", "update_time": 1612874490.589575, "create_time": 1612874490, }, { "notable_id": test_id, "comment": "Second comment", "update_time": 1612874500.589575, "create_time": 1612874500, }, ] # Mock service and KV store mock_service = MagicMock() mock_kv_store = MagicMock() mock_kv_store.data.query.return_value = mock_comments_data mock_service.kvstore.__getitem__.return_value = mock_kv_store # Mock helper functions mocker.patch("SplunkPy.get_comments_data_old", return_value=[]) mocker.patch("SplunkPy.format_splunk_note_for_xsoar", side_effect=lambda note: note["comment"]) # Test fetch mode (is_fetch=True) fetch_notable_map = id_to_notable_map.copy() get_comments_data_new( service=mock_service, id_to_notable_map=fetch_notable_map, comment_tag_from_splunk="from_splunk", last_update_splunk_timestamp=1612874430.589575, is_fetch=True, ) # Verify fetch storage format assert "comment" in fetch_notable_map[test_id] assert isinstance(fetch_notable_map[test_id]["comment"], list) assert all(isinstance(comment, str) for comment in fetch_notable_map[test_id]["comment"]) assert "Second comment" in fetch_notable_map[test_id]["comment"] # Most recent first assert "First comment" in fetch_notable_map[test_id]["comment"] # Test mirror-in mode (is_fetch=False) mirror_notable_map = { test_id: { "rule_id": test_id, "event_id": test_id, } } get_comments_data_new( service=mock_service, id_to_notable_map=mirror_notable_map, comment_tag_from_splunk="from_splunk", last_update_splunk_timestamp=1612874430.589575, is_fetch=False, ) # Verify mirror-in storage format assert "SplunkComments" in mirror_notable_map[test_id] assert isinstance(mirror_notable_map[test_id]["SplunkComments"], list) assert all(isinstance(comment, dict) and "Comment" in comment for comment in mirror_notable_map[test_id]["SplunkComments"]) assert mirror_notable_map[test_id]["SplunkComments"][0]["Comment"] == "Second comment" # Most recent first assert mirror_notable_map[test_id]["SplunkComments"][1]["Comment"] == "First comment" @pytest.mark.parametrize( "query, expected", [ # Positive cases ("index=your_index earliest=-5m", True), ("index=your_index earliest= @d", True), ("index=your_index earliest =@mon", True), ('index=your_index earliest = "01/01/2025:00:00:00"', True), ("no keyword here", False), ("earliest", False), ], ) def test_earliest_time_exists_in_query(query, expected): """ Given: - A query string that may or may not contain an 'earliest=' pattern. When: - earliest_time_exists_in_query is invoked with the query. Then: - It returns True only when 'earliest' is followed by an equals sign with optional whitespace (e.g., 'earliest=...', 'earliest = ...'). - Otherwise, it returns False. """ from SplunkPy import earliest_time_exists_in_query assert earliest_time_exists_in_query(query) == expected def test_splunk_get_indexes_command_success(mocker): """ Given: - A Splunk service object. - The REST API query for indexes succeeds. When: - calling splunk_get_indexes_command. Then: - Ensure the command returns the expected indexes from the REST API query. - Ensure the fallback mechanism (service.indexes) is NOT used. """ from SplunkPy import splunk_get_indexes_command # Mock the service and the oneshot job results service = mocker.MagicMock() mock_result = {"name": "main", "count": "100"} mocker.patch("splunklib.results.JSONResultsReader", return_value=[mock_result]) # Mock return_results to capture the output return_results_mock = mocker.patch("SplunkPy.return_results") # Call the function splunk_get_indexes_command(service, "search") # Verify results assert return_results_mock.call_count == 1 results = return_results_mock.call_args[0][0] assert results.raw_response == json.dumps([mock_result]) # Verify oneshot was called service.jobs.oneshot.assert_called_once() def test_splunk_get_indexes_command_fallback(mocker): """ Given: - A Splunk service object. - The REST API query for indexes fails. - The direct API (service.indexes) succeeds. When: - calling splunk_get_indexes_command. Then: - Ensure the command returns the expected indexes from the direct API. - Ensure the error is logged and fallback is attempted. """ from SplunkPy import splunk_get_indexes_command # Mock the service service = mocker.MagicMock() # Mock oneshot to raise an exception service.jobs.oneshot.side_effect = Exception("REST API Failed") # Mock service.indexes to return a list of indexes mock_index = mocker.MagicMock() mock_index.name = "history" # Mocking dictionary access for the index object since the code uses index["totalEventCount"] mock_index.__getitem__.return_value = "50" service.indexes = [mock_index] # Mock logging and return_results error_mock = mocker.patch("demistomock.error") debug_mock = mocker.patch("demistomock.debug") return_results_mock = mocker.patch("SplunkPy.return_results") # Call the function splunk_get_indexes_command(service, "search") # Verify error was logged assert error_mock.call_count == 1 assert "Failed to get indexes using REST API query approach" in error_mock.call_args[0][0] # Verify fallback was attempted (debug log) fallback_log_found = False for call in debug_mock.call_args_list: if "Falling back to direct API approach" in call[0][0]: fallback_log_found = True break assert fallback_log_found # Verify results expected_result = [{"name": "history", "count": "50"}] assert return_results_mock.call_count == 1 results = return_results_mock.call_args[0][0] assert results.raw_response == json.dumps(expected_result) def test_splunk_get_indexes_command_failure(mocker): """ Given: - A Splunk service object. - Both the REST API query and the direct API fail. When: - calling splunk_get_indexes_command. Then: - Ensure a DemistoException is raised with details from both errors. """ from SplunkPy import splunk_get_indexes_command # Mock the service service = mocker.MagicMock() # Mock oneshot to raise an exception service.jobs.oneshot.side_effect = Exception("REST API Failed") # Mock service.indexes to raise an exception (property access raises exception) type(service).indexes = mocker.PropertyMock(side_effect=Exception("Direct API Failed")) # Mock logging error_mock = mocker.patch("demistomock.error") # Call the function and expect DemistoException with pytest.raises(DemistoException) as e: splunk_get_indexes_command(service, "search") assert "Failed to retrieve indexes using both methods" in str(e.value) assert "REST API error: REST API Failed" in str(e.value) assert "Direct API error: Direct API Failed" in str(e.value) # Verify errors were logged assert error_mock.call_count >= 2 # One for REST failure, one for Direct failure # ========== ResponseSizeValidator Tests ========== def test_response_size_validator_no_warning_below_threshold(mocker): """ Given: - Data size below the 20 MB threshold When: - ResponseSizeValidator validates the data Then: - return_results is not called - validated flag is set to True """ validator = splunk.ResponseSizeValidator() mock_return_results = mocker.patch("SplunkPy.return_results") # Create data below threshold (1 MB) - as list of dicts small_data = [{"data": "x" * (1 * 1024 * 1024)}] validator.validate_and_report(small_data) mock_return_results.assert_not_called() assert validator.validated is True def test_response_size_validator_warning_above_threshold(mocker): """ Given: - Data size above the 20 MB threshold (e.g., 25 MB) When: - ResponseSizeValidator validates the data Then: - return_results is called with warning message - validated flag is set to True - Warning message contains "WARNING" prefix """ validator = splunk.ResponseSizeValidator() mock_return_results = mocker.patch("SplunkPy.return_results") # Create data above threshold (25 MB) - as list of dicts large_data = [{"data": "x" * (25 * 1024 * 1024)}] validator.validate_and_report(large_data) mock_return_results.assert_called_once() warning = mock_return_results.call_args[0][0] assert "WARNING" in warning assert "25." in warning # Size will be around 25 MB assert "20" in warning assert "normal usage size" in warning assert validator.validated is True def test_response_size_validator_warning_shown_only_once(mocker): """ Given: - Multiple batches of data, all above threshold When: - ResponseSizeValidator validates each batch Then: - return_results is called only on the first validation - Subsequent validations don't call return_results - validated flag remains True after first validation """ validator = splunk.ResponseSizeValidator() mock_return_results = mocker.patch("SplunkPy.return_results") # Create data above threshold (25 MB) - as list of dicts large_data = [{"data": "x" * (25 * 1024 * 1024)}] # First validation should call return_results validator.validate_and_report(large_data) assert mock_return_results.call_count == 1 assert validator.validated is True # Second validation should not call return_results (already validated) validator.validate_and_report(large_data) assert mock_return_results.call_count == 1 assert validator.validated is True # Third validation should also not call return_results validator.validate_and_report(large_data) assert mock_return_results.call_count == 1 assert validator.validated is True def test_response_size_validator_just_above_threshold(mocker): """ Given: - Data size just above the 20 MB threshold (20 MB + 1 byte) When: - ResponseSizeValidator validates the data Then: - return_results is called with warning message - validated flag is set to True """ validator = splunk.ResponseSizeValidator() mock_return_results = mocker.patch("SplunkPy.return_results") # Create data just above threshold (20 MB + 1 byte) - as list of dicts just_above_data = [{"data": "x" * (20 * 1024 * 1024 + 1)}] validator.validate_and_report(just_above_data) mock_return_results.assert_called_once() warning = mock_return_results.call_args[0][0] assert "WARNING" in warning assert validator.validated is True def test_response_size_validator_message_format(mocker): """ Given: - Data size of 30 MB (above threshold) When: - ResponseSizeValidator validates the data Then: - Warning message contains all required elements: * "WARNING" prefix * Actual size in MB * Threshold size in MB * "normal usage size" phrase """ validator = splunk.ResponseSizeValidator() mock_return_results = mocker.patch("SplunkPy.return_results") # Create data of 30 MB - as list of dicts data_30mb = [{"data": "x" * (30 * 1024 * 1024)}] validator.validate_and_report(data_30mb) mock_return_results.assert_called_once() warning = mock_return_results.call_args[0][0] assert warning.startswith("WARNING:") assert "30." in warning # Size will be around 30 MB assert "20" in warning assert "normal usage size" in warning assert "exceeds" in warning.lower()