About dtburrows3

dtburrows3 · ‎12-21-2023

In these kinds of situations in Splunk I generally do something like this to replace empty strings with actual null values. | foreach err_field* [ | eval <<FIELD>>=if( '<<FIELD>>'=="" OR match('<<FIELD>>', "^\s*$"), null(), '<<FIELD>>' ) ] | eval err_final=coalesce(err_field1, err_field2, err_field3, err_field4) You can see the coalesce works as expected after replacing nullifying the empty strings. Note: this is also replacing any values in the err_field* fields that is only whitespace in addition to empty strings.

dtburrows3 · ‎12-20-2023

Oh okay I just assumed it was a Splunk lookup. So if you are indexing the data from a CSV then you can probably do something like this (assuming field extractions are in place) index=<index> sourcetype=<sourcetype> | table [ | makeresults | fields - _time | eval ID=[ | search index=<index> sourcetype=<sourcetype> | stats latest(ID) as ID | return $ID ], field_list_id_zero="NAME,STATUS,DATE,ACTION", field_list_id_positive="DATE-Changed,ID,NAME,DATE_DOWN,ACTION", final_field_list=if( 'ID'==0, 'field_list_id_zero', 'field_list_id_positive' ) | fields + final_field_list | return $final_field_list ] where <index> and <sourcetype> is where your CSV is being indexed.

dtburrows3 · ‎12-20-2023

Not sure exactly how your ID value is being derived in this situation but you may be able to utilize a subsearch holding you list of fields for each scenario and then set up an eval if() function to determine which to use based on the value in the ID field. Then with a return command you can return that conditional field list back into the parent search after a fields command. Something like this. | inputlookup <lookup> | fields [ | makeresults | fields - _time | eval ``` Not sure how the ID is being derived but there should be a variety of ways to get it here ``` ``` From lookup method ``` ``` ID=[ | inputlookup <lookup> | stats max(ID) as ID | return $ID ] ``` ``` From token method ``` ``` ID=$ID_token$ ``` ``` This is hardcoded for a POC ``` ID=1, field_list_id_zero="NAME,STATUS,DATE,ACTION", field_list_id_positive="DATE-Changed,ID,NAME,DATE_DOWN,ACTION", final_field_list=if( 'ID'==0, 'field_list_id_zero', 'field_list_id_positive' ) | fields + final_field_list | return $final_field_list ] Sample output when ID=0 Sample output when ID>0

dtburrows3 · ‎12-20-2023

Awesome! Glad it's working out so far. Feel free to leave reply if you run into any issues and I we can try to resolve.

dtburrows3 · ‎12-20-2023

You may be able to use streamstats assuming that there is some degree off distribution of _time between each event. <base_search> | rex field=_raw "Processing\s+(?<process>[^\-]+)\-" | rex field=_raw "Person\s+Name\:\s+(?<person_name>[^\,]+)\," | sort 0 +_time | streamstats reset_before="("isnotnull(process)")" values(process) as current_process | streamstats window=2 first(_raw) as previous_log | eval checked_person_name=if( match(previous_log, "\-Check\s+for\s+Person\-"), 'person_name', null() ) | stats min(_time) as _time by current_process, checked_person_name | fields + _time, current_process, checked_person_name The final output should look something like this The table before the final stats aggregation looked like this and show more context around what the streamstats are doing here. Note: For this method to work properly _timestamps of each process event shouldn't be exactly the same, there would need to be some sort of step up in time to the next event (event if it is milliseconds). This is because we need the events in the correct sequence for the streamstats to work as expected.

dtburrows3 · ‎12-20-2023

Maybe something like this? | multisearch [ | search index=messages* MSG_src="AAAAA" MSG_DOMAIN="BBBBBB" MSG_TYPE="CC *" | rename MSGID as MSGID1 ] [ | search index=messages* MSG_src="CCCCCC", MSG_DOMAIN="DDDDDDD", MSG_TYPE="Workflow Start" | rex field=_raw "<pmt>(?<pmt>.*)<\/pmt>" | rex field=_raw "<EventId>(?<MSGID1>.*)<\/EventId>" | search pmt="EEEEEEE" ] | stats ``` first occurrence timestamp of msg_id in search_1 ``` earliest(eval(case(match(MSG_TYPE, "^C{2}\s+"), _time))) as first_event_epoch, ``` first occurrence timestamp of msg_id in search_2 ``` earliest(eval(case('MSG_TYPE'=="Workflow Start", _time))) as second_event_epoch by MSGID1 ``` calculate the time difference between the msg_id showing up in each search ``` | eval diff_seconds=if( ``` if the msg_id didn't show up in the second search but did show up in the first ``` isnull(second_event_epoch) AND isnotnull(first_event_epoch), ``` calculate how long ago from now the msg_id was seen in search_1 ``` now()-'first_event_epoch', ``` msg_id exists in both searches, calculate the time difference between them in seconds ``` 'second_event_epoch'-'first_event_epoch' ), ``` convert time difference to hours``` diff_hours='diff_seconds'/(60*60), ``` human readable format ``` duration_seconds=tostring(diff_seconds, "duration") ``` filter off everything the has less than a 1 hour difference ``` | where 'diff_hours'>1

dtburrows3 · ‎12-20-2023

It sounds like you timestamps "created" and "last_login" have the format "%Y-%m-%d" in the events. Trying to convert them to epoch using a different format will not work. For example If you have a situations where your events have these field in a mixture of both formats, maybe you could adjust your eval to be something more like this? | eval dormancy=if( last_login="(never)", round((now()-case(match(created, "^\d{4}\-\d{2}\-\d{2}"), strptime(created,"%Y-%m-%d"), match(created, "^\d{4}\/\d{2}\/\d{2}"), strptime(created,"%Y/%m/%d")))/86400), round((now()-case(match(last_login, "^\d{4}\-\d{2}\-\d{2}"), strptime(last_login,"%Y-%m-%d"), match(last_login, "^\d{4}\/\d{2}\/\d{2}"), strptime(last_login,"%Y/%m/%d")))/86400) ) This seem to extract both formats properly

dtburrows3 · ‎12-19-2023

You may need to specify the "total_time" field as the field to return descriptive statistic on instead of using it as a by-field in this search. Something like this, index=idx_prd_analysis sourcetype="type:prd_analysis:result" corp="AUS" | eval total_time='End_time'-'Start_time' | stats median(total_time) as median, min(total_time) as min, max(total_time) as max, p25(total_time) as lowerquartile, p75(total_time) as upperquartile | eval iqr='upperquartile'-'lowerquartile', scalar=1.5, lowerwhisker='median'-('scalar'*'iqr'), upperwhisker='median'+('scalar'*'iqr')

dtburrows3 · ‎12-19-2023

I was able to find a provenance="UI:Report" inside of index=_introspection sourcetype=search_telemetry that I think will have the data you are after. Example SPL: index=_introspection sourcetype=search_telemetry desc.provenance="UI:Report" earliest=-90d@d latest=now | stats values(host) as hosts, latest(timestamp) as last_run_epoch by "desc.app", "desc.savedsearch_name" | eval days_since_last_run=((now()-'last_run_epoch')/(60*60*24)), duration_since_last_run=tostring((now()-'last_run_epoch'), "duration") | convert ctime(last_run_epoch) as last_run_timestamp

dtburrows3 · ‎12-19-2023

I think this regex will capture just the value for template. \s+template:\s+([^,]+) The character after the carrot inside the square brackets means match on a character not in this list. And adding a "+" after is a quantifier for 1 or more times. So doing "template: (?<TemplateID>[^-]+)" is matching on all characters after 'template: ' up until a "-" (which I dont see one in the example. So replacing the "-" with a "," I think will extract the value as intended.

dtburrows3 · ‎12-19-2023

This seemed to do what you are asking about if I understood you question correctly. | sendemail to="<email>" message="this is a test message: testing token $$token$$"

dtburrows3 · ‎12-19-2023

Should be able to check index=_audit for search run times for each individual search running on dashboards. Something like this. index="_audit" action=search provenance="UI:Dashboard:*" | table _time, user, app, provenance, savedsearch_name, search_id, total_run_time, event_count, result_count, search_et, search_lt | eval dashboard_id=mvindex(split(provenance, ":"), 2, -1)

dtburrows3 · ‎12-19-2023

Doing some SPL like this may lead you in the right direction if I am understanding you question correctly. Note: The top portion of this code is just generating sampe data, the meat of the solution is where the comments start ``` <comment> ``` | makeresults | eval input_value="83, 9123, 272529, 1234" | append [ | makeresults | eval input_value="851056, 714062, 6234, 91258,272476" ] | append [ | makeresults | eval input_value="28, 10001, 18, 99923,1027385" ] ``` Generating field with the comma delimited list of Keys ``` | eval Keys="272476, 272529, 274669, 714062, 714273, 845143, 851056, 853957, 855183" ``` Splitting both Keys and simulated user input fields into multivalued fields ``` | eval mv_Keys=trim(split(Keys, ","), " "), mv_input_value=trim(split(input_value, ","), " ") ``` looping through each entry in a multivalue field 'mv_input_value' and checking if it exists in the list of Keys ``` | eval intersecting_keys=case( isnull(mv_input_value), null(), mvcount(mv_input_value)==1, if('mv_input_value'=='mv_Keys', 'mv_input_value', null()), mvcount(mv_input_value)>1, mvmap(mv_input_value, if('mv_input_value'=='mv_Keys', 'mv_input_value', null())) ) Results show in the screenshot You can split the comma delimited lists into MV fields and then loop through one of them to individually check if that number exists in another multivalued field. In this example I did this and created a new field 'intersecting_keys' to return the number that exist in both fields.

dtburrows3 · ‎12-19-2023

Not sure if this is exactly what you are looking for but I think it is pretty close. I got this output by stringing together a couple of streamstats with window=<int> and reset_before=<criteria> parameters | sort 0 +Machine, +time | streamstats count as row | eval TimeStamp=strftime(time, "%m/%d/%Y %H:%M:%S") | fields - _time | fields + row, Machine, TimeStamp, time | streamstats window=3 count as running_count, min(time) as min_time, max(time) as max_time by Machine | eval seconds_diff='time'-'min_time', duration_diff=tostring(seconds_diff, "duration") | streamstats window=3 reset_before="("seconds_diff>300")" count as running_count by Machine | eval Occurrence=if( 'seconds_diff'<=300 AND 'running_count'==3, "TRUE", "FALSE" ) | fields + row, Machine, TimeStamp, Occurrence Here is the full SPL I used to generate the screenshot (results may vary because of the use of relative_time()) | makeresults | eval Machine="machine 1", time=relative_time(now(), "-2h@s") | append [ | makeresults | eval Machine="machine 1", time=relative_time(now(), "-2h+18s@s") ] | append [ | makeresults | eval Machine="machine 1", time=relative_time(now(), "-2h+34s@s") ] | append [ | makeresults | eval Machine="machine 2", time=relative_time(now(), "+4d@d+20h@h+31m@m+48s@s") ] | append [ | makeresults | eval Machine="machine 1", time=relative_time(now(), "-2h+52s@s") ] | append [ | makeresults | eval Machine="machine 2", time=relative_time(now(), "+4d+5h+5m+2s") ] | append [ | makeresults | eval Machine="machine 1", time=relative_time(now(), "-2h+302s@s") ] | append [ | makeresults | eval Machine="machine 1", time=relative_time(now(), "+2d-5h+18s@s") ] | append [ | makeresults | eval Machine="machine 2", time=relative_time(now(), "+4d+5h+18s@s") ] | append [ | makeresults | eval Machine="machine 2", time=relative_time(now(), "+4d+5h+2m+1s") ] | append [ | makeresults | eval Machine="machine 2", time=relative_time(now(), "+4d+5h+2m+34s") ] | append [ | makeresults | eval Machine="machine 2", time=relative_time(now(), "+4d+5h+4m-12s") ] | append [ | makeresults | eval Machine="machine 2", time=relative_time(now(), "+4d@d+20h@h+43m@m+5s@s") ] | sort 0 +Machine, +time | streamstats count as row | eval TimeStamp=strftime(time, "%m/%d/%Y %H:%M:%S") | fields - _time | fields + row, Machine, TimeStamp, time | streamstats window=3 count as running_count, min(time) as min_time, max(time) as max_time by Machine | eval seconds_diff='time'-'min_time', duration_diff=tostring(seconds_diff, "duration") | streamstats window=3 reset_before="("seconds_diff>300")" count as running_count by Machine | eval Occurrence=if( 'seconds_diff'<=300 AND 'running_count'==3, "TRUE", "FALSE" ) | fields + row, Machine, TimeStamp, Occurrence

dtburrows3 · ‎12-18-2023

yea unfortunately mvexpand can be memory intensive. I would say limit your fieldset as much as possible before using it and see if that helps. It actually may work to just do a, <base_search> | stats count by "records{}.properties.flows{}.flows{}.flowTuples{}" | eval time=mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 0), src_ip=mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 1), dst_ip=mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 2), src_port=mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 3), dst_port=mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 4), protocol=mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 5), traffic_flow=mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 6), traffic_result=mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 7) | stats sum(count) as total by src_ip, dst_ip this should tally up all the individual flow_tuples from events and then we can eval to split it out and then sum it all up by src, dest IP. I think this get around the need for an MVexpand. Let me know if that works!

dtburrows3 · ‎12-18-2023

First glance it looks like this the regex you provided is looking for double quote to help find the pattern and the sample event don't seem to have any. I think you may have better luck using a regex closer to REGEX = \s+([^:]+?):\s+([^,]+) I havent tested locally but on regex 101 it looks like it matches pretty well

dtburrows3 · ‎12-18-2023

To retain the associations for any sort of analysis you may need to mvexpand the "records{}.properties.flows{}.flows{}.flowTuples{}" field itself. stats aggregation using 2 multivalued fields as by-fields can be misleading for the final output. Below is a table of the event you shared on the initial post after using the mvexpand and then extracting out the individual fields after. SPL to do this | mvexpand "records{}.properties.flows{}.flows{}.flowTuples{}" | eval time=mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 0), src_ip=mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 1), dest_ip=mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 2), src_port=mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 3), dest_port=mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 4), protocol=mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 5), traffic_flow=mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 6), traffic_result=mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 7) Doing a stats count by src_ip and dst_ip should make more sense using the data formatted in this way.

dtburrows3 · ‎12-18-2023

You can give these evals a go. I would check and make sure you are getting everything properly as expected. I don't have access to any sourcetype="mscs:nsg:flow" data at the moment so I just am using simulated data based off of your screenshots. If you are happy with the output then you could add them as calculated fields in local/props.conf (I would make sure that they don't step on any existing knowledge object in the app though) | eval time=if(isnotnull('records{}.properties.flows{}.flows{}.flowTuples{}'), case(mvcount('records{}.properties.flows{}.flows{}.flowTuples{}')==1, mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 0), mvcount('records{}.properties.flows{}.flows{}.flowTuples{}')>1, mvmap('records{}.properties.flows{}.flows{}.flowTuples{}', mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 0))), 'time') | eval src_ip=if(isnotnull('records{}.properties.flows{}.flows{}.flowTuples{}'), case(mvcount('records{}.properties.flows{}.flows{}.flowTuples{}')==1, mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 1), mvcount('records{}.properties.flows{}.flows{}.flowTuples{}')>1, mvmap('records{}.properties.flows{}.flows{}.flowTuples{}', mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 1))), 'src_ip') | eval dst_ip=if(isnotnull('records{}.properties.flows{}.flows{}.flowTuples{}'), case(mvcount('records{}.properties.flows{}.flows{}.flowTuples{}')==1, mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 2), mvcount('records{}.properties.flows{}.flows{}.flowTuples{}')>1, mvmap('records{}.properties.flows{}.flows{}.flowTuples{}', mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 2))), 'dst_ip') | eval src_port=if(isnotnull('records{}.properties.flows{}.flows{}.flowTuples{}'), case(mvcount('records{}.properties.flows{}.flows{}.flowTuples{}')==1, mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 3), mvcount('records{}.properties.flows{}.flows{}.flowTuples{}')>1, mvmap('records{}.properties.flows{}.flows{}.flowTuples{}', mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 3))), 'src_port') | eval dst_port=if(isnotnull('records{}.properties.flows{}.flows{}.flowTuples{}'), case(mvcount('records{}.properties.flows{}.flows{}.flowTuples{}')==1, mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 4), mvcount('records{}.properties.flows{}.flows{}.flowTuples{}')>1, mvmap('records{}.properties.flows{}.flows{}.flowTuples{}', mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 4))), 'dst_port') | eval protocol=if(isnotnull('records{}.properties.flows{}.flows{}.flowTuples{}'), case(mvcount('records{}.properties.flows{}.flows{}.flowTuples{}')==1, mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 5), mvcount('records{}.properties.flows{}.flows{}.flowTuples{}')>1, mvmap('records{}.properties.flows{}.flows{}.flowTuples{}', mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 5))), 'protocol') | eval traffic_flow=if(isnotnull('records{}.properties.flows{}.flows{}.flowTuples{}'), case(mvcount('records{}.properties.flows{}.flows{}.flowTuples{}')==1, mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 6), mvcount('records{}.properties.flows{}.flows{}.flowTuples{}')>1, mvmap('records{}.properties.flows{}.flows{}.flowTuples{}', mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 6))), 'traffic_flow') | eval traffic_result=if(isnotnull('records{}.properties.flows{}.flows{}.flowTuples{}'), case(mvcount('records{}.properties.flows{}.flows{}.flowTuples{}')==1, mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 7), mvcount('records{}.properties.flows{}.flows{}.flowTuples{}')>1, mvmap('records{}.properties.flows{}.flows{}.flowTuples{}', mvindex(split('records{}.properties.flows{}.flows{}.flowTuples{}', ","), 7))), 'traffic_result') Also, not sure if there are ever events formatted slightly differently because only a single flow occurred and it would no longer be an array in the json event, therefore changing the overall extracted field name to something like "records{}.properties.flows{}.flows.flowTuples{}". From the look at the microsoft_azure app configs, it looks like its only every referencing "records{}.properties.flows{}.flows{}.flowTuples{}" for it's extractions so I just made the assumption that events will be formatted this way.

dtburrows3 · ‎12-18-2023

Tried this out and came back with this. Format might be a little different than what you asked for but I think tells the same story. | bucket span=1m _time | stats count as TPS by _time | eventstats min(TPS) as min_TPS, max(TPS) as max_TPS | foreach *_TPS [ | eval <<MATCHSTR>>_TPS_epoch=if( 'TPS'=='<<MATCHSTR>>_TPS', mvappend( '<<MATCHSTR>>_TPS_epoch', '_time' ), '<<MATCHSTR>>_TPS_epoch' ) ] | stats avg(TPS) as avg_TPS, first(*_TPS) as *_TPS, first(*_TPS_epoch) as *_TPS_epoch | eval avg_TPS=round('avg_TPS', 2) | foreach *_TPS_epoch [ | eval <<MATCHSTR>>_TPS_timestamps=case( mvcount('<<FIELD>>')==1, strftime('<<FIELD>>', "%x %X"), mvcount('<<FIELD>>')>1, mvmap('<<FIELD>>', strftime('<<FIELD>>', "%x %X")) ), <<MATCHSTR>>_TPS_json=json_object( "type", "<<MATCHSTR>>", "TPS", '<<MATCHSTR>>_TPS', "Timestamps", '<<MATCHSTR>>_TPS_timestamps' ), combined_TPS_json=mvappend( 'combined_TPS_json', '<<MATCHSTR>>_TPS_json' ) ] | fields + combined_TPS_json, avg_TPS | addinfo | eval search_time_window_end=strftime(info_max_time, "%x %X"), search_time_window_start=strftime(info_min_time, "%x %X"), avg_TPS_time_window='search_time_window_start'." --> ".'search_time_window_end' | eval combined_TPS_json=mvappend( 'combined_TPS_json', json_object( "type", "avg", "TPS", 'avg_TPS', "Timestamps", 'avg_TPS_time_window' ) ) | mvexpand combined_TPS_json | fromjson combined_TPS_json | fields - combined_TPS_json | fields + type, TPS, Timestamps Output should look something like this. You should also be able to change the time bucket span form 1m back to 1s since that is how it was setup in your initial query.

dtburrows3 · ‎12-18-2023

You should be able to utilize built in Splunk JSON commands and/or JSON functions to build out valid STIX 2.1 objects from Splunk events. Here is some sample SPL to give you an example of how to build out the json from individual fields in Splunk. | makeresults | fields - _time ``` gen properties data ``` | eval enum=split("attack-pattern|campaign", "|"), description="The type of this object, which MUST be the literal `attack-pattern`.", type="string" | tojson str(enum) str(type) str(description) output_field=properties | fields - enum, type, description ``` gen ID data ``` | eval title="id", pattern="^attack-pattern--" | tojson str(title) str(pattern) output_field=id | fields - title, pattern ``` gen Name data ``` | eval type="string", description="The name used to identify the Attack Pattern." | tojson str(type) str(description) output_field=name | fields - type, description ``` gen description data ``` | eval type="string", description="A description that provides more details and context about the Attack Pattern, potentially including its purpose and its key characteristics." | tojson str(type) str(description) output_field=description | fields - type, description ``` gen kill_chain_phases data ``` | eval "$ref"="../common/kill-chain-phase.json" | tojson str($ref) output_field=items | fields - "$ref" | eval type="array", description="The list of kill chain phases for which this attack pattern is used.", minItems=1 | tojson str(type) str(description) json(items) num(minItems) output_field=kill_chain_phases | fields - minItems, items, description, type | tojson json(properties) json(type) json(id) json(name) json(description) json(kill_chain_phases) output_field=allOf | fields - properties, type, id, name, description, kill_chain_phases | eval ref="../common/core.json" | tojson str(ref) output_field=allOf_2 | eval allOf=mvappend( 'allOf_2', 'allOf' ) | fields - allOf_2 | eval required="name", type="object", description="Attack Patterns are a type of TTP that describe ways that adversaries attempt to compromise targets. ", title="attack-pattern", "$schema"="http://json-schema.org/draft-04/schema#" | tojson str($schema) str(title) str(description) str(type) json(allOf) str(required) output_field=stix_2_payload | fields + stix_2_payload For this example I used a generative command to put together sample data first, but if you are building from a Splunk event then the fields should all be derived from _raw or already extracted. The example is more of a demonstration on how to build a valid STIX 2.1 json object using Splunk. Below is the json object build out from the SPL above. { "$schema": "http://json-schema.org/draft-04/schema#", "allOf": [ { "ref": "../common/core.json" }, { "id": { "pattern": "^attack-pattern--", "title": "id" }, "kill_chain_phases": { "description": "The list of kill chain phases for which this attack pattern is used.", "items": { "$ref": "../common/kill-chain-phase.json" }, "minItems": 1, "type": "array" }, "name": { "description": "The name used to identify the Attack Pattern.", "type": "string" }, "properties": { "description": "The type of this object, which MUST be the literal `attack-pattern`.", "enum": [ "attack-pattern", "campaign" ], "type": "string" } } ], "description": "Attack Patterns are a type of TTP that describe ways that adversaries attempt to compromise targets. ", "required": "name", "title": "attack-pattern", "type": "object" }

dtburrows3 · ‎12-17-2023

I was able to achieve this on my local instance by a stats aggregation by "severity" field and then doing a transpose of results so that the splunk chary visualization will display it this way. Example of SPL: <base_search> | stats count as count by severity | transpose header_field=severity column_name=severity | fields + severity, critical, high, medium, low, informational In the dashboard XML you should be able to add this option tag to your bar chart visualization to assign colors for each unique severity value. <option name="charting.fieldColors">{"critical":0xFF0000,"high":0xFF7F50,"medium":0xFFBF00,"low":0xDFFF00,"informational":0x40E0D0}</option> Screenshot of results. Full SPL used to replicate on my local instance: | makeresults count=377 | eval severity="high" | append [ | makeresults count=1118 | eval severity="medium" ] | append [ | makeresults count=119 | eval severity="critical" ] | append [ | makeresults count=1001 | eval severity="low" ] | append [ | makeresults count=41 | eval severity="informational" ] | stats count as count by severity | transpose header_field=severity column_name=severity | fields + severity, critical, high, medium, low, informational

dtburrows3 · ‎12-17-2023

Looks like as of Splunk 8.1 you should be able to pipe in tokens into the sendemail command. https://docs.splunk.com/Documentation/Splunk/9.1.2/SearchReference/Sendemail If you have a scenario where you may need to email multiple groups across different applications from one search than you may be able to utilize the "map" command piped directly to the "sendemail" command. (May need to some testing on this, but I'm pretty sure I have seen this done before) Note: map command will attempt to dispatch a search for each row from the parent search returned. There is a default limit of the max searches it will attempt to send (maxsearches=10) You can find more here. https://docs.splunk.com/Documentation/Splunk/9.1.2/SearchReference/Map Example: | search index IN ("app_index_1", "app_index_2") CASE(ERROR) | bucket span=1h _time ``` mapping notification user by eval or lookup or possible derived from the _raw data (This example is just setting a hardcoded list of users to each application) ``` | eval notification_users_email=case( 'index'=="app_index_1", mvappend("user_1@acme.com", "user_3@acme.com", "user_5@acme.com"), 'index'=="app_index_2", "user_2@acme.com" ) | stats count as error_count values(notification_users_email) as notification_users_email by _time, app ``` Trigger criteria for more than 50 errors for specific application in a 1 hour time window ``` | where 'error_count'>50 ``` prepare notification user field to be formatted for the sendemail command (convert multivalue field of unique values to a comma delimeted list of users) ``` | eval notification_users_email=mvjoin(notification_users_email, ",") ``` prepare message to send to email list for the application ``` | eval message='app'." had ".'error_count'." errors in a one hour time window. Please investigate..." ``` each row returned from the parent search will be piped into the map command and send out its own email to the list of users associated with the applications meeting the alert criteria ``` | map search="sendemail to=\"$notification_users_email$\" message=\"$message$\"" maxsearches=10

dtburrows3 · ‎12-17-2023

I think this SPL will do what you are looking for. index=o365 UserloginFailed* | iplocation ClientIP | search Country!=Australia | stats values(Country) as Country by user | where mvcount(Country)>1

dtburrows3 · ‎12-17-2023

Nevermind @ITWhisperer beat me too it! There is probably a couple ways of doing this but this seemed to work for me on my local <base_search> | eval time=strptime(TimeStamp, "%Y-%m-%d %H:%M:%S.%Q") | appendpipe [ | bucket span=1m time | stats sum(SumTotalErrors) as sumErrors by time | eval bucket_type="1 minute" ] | appendpipe [ | bucket span=2m time | stats sum(SumTotalErrors) as sumErrors by time | eval bucket_type="2 minutes" ] | appendpipe [ | bucket span=3m time | stats sum(SumTotalErrors) as sumErrors by time | eval bucket_type="3 minutes" ] | appendpipe [ | bucket span=5m time | stats sum(SumTotalErrors) as sumErrors by time | eval bucket_type="5 minutes" ] | appendpipe [ | bucket span=1h time | stats sum(SumTotalErrors) as sumErrors by time | eval bucket_type="1 hour" ] | stats count as sample_size, avg(sumErrors) as avg_sumErrors by bucket_type | eval "Average Error Rate (Human Readable)"=round(avg_sumErrors, 0)." Errors per ".'bucket_type' | addinfo | eval search_time_window_sampled=tostring('info_max_time'-'info_min_time', "duration") | fields - info_*_time, info_sid | sort 0 +sample_size Not quite a loop but I am curious about this so I will keep trying out different things. Output should look something like this As for the dashboard, you can set up an input token (dropdown) to allow the user to select a span and use that token on the | bucket span=$span$ time then do your stats command.

dtburrows3 · ‎12-17-2023

I agree with @richgalloway response here for this particular example. However if you want a more generalize approach to parsing out a table being ingested in Splunk with this format in _raw (capable of accounting for different headers and not having to build out a regex for each time) you may want to try the SPL below. <base_search> ``` Doing this before an mvexpand occurring later in the search pipeline can help attribute the expanded data back to it's orginal event (if needed) ``` | streamstats count as event_count ``` Extract Info from table title (this is the one line that would need to be adjusted for each unique table depending on title format and data you would want extracted from it ``` | rex field=_raw "Disk\s+Utilization\s+for\s+(?<Server>[^\s]+)\s+in\s+(?<Region>[^\s]+)\s+(?<Environment>[^\s]+)\s+\-(?<Server_IP>[^\s]+)" ``` Parse individual rows from the table as a multivalued field ``` | spath input=_raw path=table.tr output=data | fields - _raw ``` mvexpand each row to it's own event ``` | mvexpand data ``` replace tags with standardized breakers for parsing ``` | eval item_data=replace(replace(replace(replace('data', "\<\/t(?:h|d)\>\<t(?:h|d)[^\>]*\>", "<BREAK>"), "^\<t(?:h|d)[^\>]*\>", "<START>"), "\<\/t(?:h|d)\>", "<END>"), "(\<START\>|\<BREAK\>)(\<BREAK\>|\<END\>)", "\1#NULL#\2") ``` set a row count for each table ``` | streamstats count as row_number by event_count ``` assign row type depending on the row number (assumes that the table has a header) ``` | eval row_number='row_number'-1, row_type=if( 'row_number'==0, "header", "data" ), ``` remove start and end tags ``` data_mv=split(replace(replace(item_data, "\<START\>", ""), "\<END\>", ""), "<BREAK>"), ``` trim of any leading or tailing spaces from each entry in the table ``` data_mv=case( mvcount(data_mv)==1, trim(data_mv, " "), mvcount(data_mv)>1, mvmap(data_mv, trim(data_mv, " ")) ) | fields - data, item_data ``` pull header fieldnames into each row of the table to stitch it all together ``` | eventstats list(eval(case('row_type'=="header", 'data_mv'))) as header by event_count ``` mvzip to attribute each table entry to the header field (we will loop through this field to build the table further down the search pipeline) ``` | eval data_mv_zip=mvzip(header, data_mv, "<DELIM>") | fields - header, data_mv ``` we no longer need the row that only holds the header info now that it has been pulled into each data entry row of the table ``` | where NOT 'row_type'=="header" ``` initialize json object that will hold KV pairs of header_fieldname/row_entry ``` | eval table_json=json_object() ``` loop through the MV field and build the json_object containing all KV assignments ``` | foreach mode=multivalue data_mv_zip [ | eval key=mvindex(split('<<ITEM>>', "<DELIM>"), 0), value=mvindex(split('<<ITEM>>', "<DELIM>"), 1), table_json=json_set(table_json, 'key', 'value') ] | fields - key, value, row_type, data_mv_zip ``` spath the json_object just built ``` | spath input=table_json ``` remove the json_object now that we dont need it ``` | fields - table_json ``` list out data in the order that you want it displayed ``` | table _time, event_count, Server, Region, Environment, Server_IP, "Filesystem", "Type", "Blocks", "Used %", "Available %", "Usage %", "Mounted on" ``` remove any placeholder #NULL# values with actual null() values since table has been generated ``` | foreach * [ | eval <<FIELD>>=if( '<<FIELD>>'=="#NULL#", null(), '<<FIELD>>' ) ] You can see the output I got locally using this method below. This method should also account for null values in a the html table as well, whereas the regex shared from your original post makes the assumption that table entries always have a leading and trailing space wrapping a non-null value. Which is true for this table you shared, but for generalizing parsing a html table you may run into issues doing it that way. As a proof on concept here is some output of some simulation data I put together with 3 separate events, each with their own html table and sets of data with their own headers and piping them through this method. Below is the SPL used to generate this screenshot. | makeresults | eval _raw="<p align='center'><font size='4' color=blue>Example: Table 1</font></p> <table align=center border=2> <tr style=background-color:#2711F0 ><th>sample_field_1</th><th>sample_field_2</th><th>sample_field_3</th><th>sample_field_4</th></tr> <tr><td>table1_row1_column1</td><td></td><td>table1_row1_column3</td><td></td></tr> <tr><td>table1_row2_column1</td><td>table1_row2_column2</td><td>table1_row2_column3</td><td>table1_row2_column4</td></tr> <tr><td>table1_row3_column1</td><td></td><td>table1_row3_column3</td><td>table1_row3_column4</td></tr> <tr><td>table1_row4_column1</td><td>table1_row4_column2</td><td>table1_row4_column3</td><td>table1_row4_column4</td></tr> <tr><td></td><td></td><td></td><td>table1_row5_column4</td></tr> </table></body></html>" | append [ | makeresults | eval _raw="<p align='center'><font size='4' color=blue>Example: Table 2</font></p> <table align=center border=2> <tr style=background-color:#2711F0 ><th>sample_field_5</th><th>sample_field_6</th><th>sample_field_7</th></tr> <tr><td></td><td>table2_row1_column2</td><td>table2_row1_column3</td></tr> <tr><td>table2_row2_column1</td><td>table2_row2_column2</td><td>table2_row2_column3</td></tr> <tr><td>table2_row3_column1</td><td></td><td>table2_row3_column3</td></tr> <tr><td></td><td>table2_row4_column2</td><td></td></tr> <tr><td>table2_row5_column1</td><td>table2_row5_column2</td><td>table2_row5_column3</td></tr> </table></body></html>" ] | append [ | makeresults | eval _raw="<p align='center'><font size='4' color=blue>Example: Table 3</font></p> <table align=center border=2> <tr style=background-color:#2711F0 ><th>sample_field_1</th><th>sample_field_2</th><th>sample_field_3</th><th>sample_field_4</th><th>sample_field_5</th><th>sample_field_6</th></tr> <tr><td>table3_row1_column1</td><td></td><td>table1_row1_column3</td><td></td><td>table1_row1_column5</td><td>table1_row1_column6</td></tr> <tr><td>table3_row2_column1</td><td>table3_row2_column2</td><td>table1_row2_column3</td><td>table3_row2_column4</td><td>table1_row2_column5</td><td>table1_row2_column6</td></tr> </table></body></html>" ] ``` Extract Table Titles ``` | spath input=_raw path=p.font output=table_title | streamstats count as event_count | spath input=_raw path=table.tr output=data | fields - _raw | mvexpand data | eval item_data=replace(replace(replace(replace('data', "\<\/t(?:h|d)\>\<t(?:h|d)[^\>]*\>", "<BREAK>"), "^\<t(?:h|d)[^\>]*\>", "<START>"), "\<\/t(?:h|d)\>", "<END>"), "(\<START\>|\<BREAK\>)(\<BREAK\>|\<END\>)", "\1#NULL#\2") | streamstats count as row_number by event_count | eval row_number='row_number'-1, row_type=if( 'row_number'==0, "header", "data" ), data_mv=split(replace(replace(item_data, "\<START\>", ""), "\<END\>", ""), "<BREAK>"), data_mv=case( mvcount(data_mv)==1, trim(data_mv, " "), mvcount(data_mv)>1, mvmap(data_mv, trim(data_mv, " ")) ) | fields - data, item_data | eventstats list(eval(case('row_type'=="header", 'data_mv'))) as header by event_count | eval data_mv_zip=mvzip(header, data_mv, "<DELIM>") | fields - header, data_mv | where NOT 'row_type'=="header" | eval table_json=json_object() | foreach mode=multivalue data_mv_zip [ | eval key=mvindex(split('<<ITEM>>', "<DELIM>"), 0), value=mvindex(split('<<ITEM>>', "<DELIM>"), 1), table_json=json_set(table_json, 'key', 'value') ] | fields - key, value, row_type, data_mv_zip | spath input=table_json | fields - table_json | fields + _time, table_title, event_count, sample_field_* | foreach * [ | eval <<FIELD>>=if( '<<FIELD>>'=="#NULL#", null(), '<<FIELD>>' ) ]

Posts	153
Solutions	52
Karma Given	12
Karma Received	95
Member Since	‎12-11-2023

Online Status	Online
Date Last Visited	4 hours ago

Re: Splunk coalesce

Re: Change table columns based on field value

Re: Change table columns based on field value

Re: How can I account for delay in response

Re: How to get the result of next event by searchi...

Re: How can I account for delay in response

Re: strptime calculation not working correctly wi...

Re: Box plot error in 'stats' command

Re: When was a Report last run?

Re: How can I extract just a section of a field?

Re: Sending $name$ in email notification

Re: how to check dashboard load time ?

Re: Searching a block of text for multiple values ...

Re: Delta time between each row - calculate freque...

Re: How do we extract multi-value fields from nest...

Re: Field Extraction with transforms configuration

Re: How do we extract multi-value fields from nest...

Re: How do we extract multi-value fields from nest...

Re: Max/Min/Avg TPS

Re: How to convert splunk event to stix 2.1 json

Re: Arranging column chart X-axis labels in static...

Re: How to send an email to users present the resu...

Re: Search data by result count

Re: How to create a sensitive table?

Re: Need to expand and create a table from one row...