Here is an overview of what I'm trying to accomplish. I have created a table that uses information in the threat activity index that shows shows the connections by source IPs to malicious IPs. I need to query our firewall index to determine if the connection was allowed or blocked by the destination and add it to the table.
The sub search would need to be something like
"index=firewall sourcetype=firewall:IPS dest=$destIP$ (results from root search) | table action".
Im not sure how I would run a query that would add the action field to the table.
index=threat_activity source="Threat - Source And Destination Matches - Threat Gen" dest* threat_match_field=dest | eval time=strftime(_time, "%d/%m/%y %H:%M:%S") | iplocation dest| stats count as "Count" sparkline(Count) as "Sparkline" values(dest) as "Malicious IPs" values(threat_key) as "Threat Feed" earliest(time) AS "Earliest Time", latest(time) AS "Latest Time" values(Country) as "Country" by src |table src "Sparkline" "Malicious IPs" "Country" "Threat Feed" "Earliest Time" "Latest Time" Count|rename src as "Source IP" |sort -Count
1) Please verify the format of the firewall records. What fields need to be matched to know it is the right record?
2) As a best practice, don't reformat or rename anything until as late as possible. You are reformatting the
_time of every record, but if there are 50 records, you only really need the
_time of the first and last.
3) You are rolling all threats and all dest together into one record for each src. If you care about what each dest did, then you can't do that.
The following assumes that only the
_time matter for matching the firewall up to the threat record. (ie the firewall events do not have
src). It assumes that the
_time on the firewall will always record at the same time or before the
_time on the threat.
(index=threat_activity source="Threat - Source And Destination Matches - Threat Gen" dest* threat_match_field=dest" OR (index=firewall sourcetype=firewall:IPS dest=* action=*) | rename COMMENT as "list the fields you need to keep" | fields index threat_key src dest action | rename COMMENT as "sort them so that firewall is before the threat for each moment" | sort 0 dest _time index | rename COMMENT as "copy the action off the firewall records, then drop the firewall records" | streamstats current=f last(action) as prioraction by src dest | where isnull(action) | rename prioraction as action | rename COMMENT as "mark groups if the action has gone back and forth. If not, all events will correctly get group 1" | streamstats current=f last(action) as prioraction by src dest | eval newgroup=case(isnull(prioraction),1, action!=prioraction,1, true(),0) by src dest | streamstats sum(newgroup) as groupno | rename COMMENT as "find the country and roll up the info" | iplocation dest | eventstats count as SrcDestCount by src dest | eventstats count as SrcCount by src | stats count as GroupCount sparkline(GroupCount) as Sparkline, values(SrcDestCount) as SrcDestCount, values(SrcCount) as SrcCount, values(threat_key) as ThreatKey earliest(_time) as EarliestTime, latest(_time) as LatestTime, values(action) as Action, values(Country) as Country by src dest groupno | rename COMMENT as "reformat and present" | eval EarliestTime = strftime(EarliestTime,"%Y-%m-%d %H:%M:%S") | eval LatestTime = strftime(LatestTime,"%Y-%m-%d %H:%M:%S") | table src dest Sparkline GroupCount ThreatKey EarliestTime LatestTime Action Country SrcCount | rename src as "Source IP", dest as "Malicious IP" | sort - Count + "Source IP" "Malicious IP" EarliestTime
This is air code, but it should be pretty close.
This is a huge help! Thank you! I'm still new to splunk so it helps tremendously when comments are added.
I do get a malformed error on line 19. I not so sure what the issue is?
"Error in 'eval' command: The expression is malformed"
It's tremendously inefficient, so I don't recommend it in most cases, but it sounds like you're looking for the
Be forewarned that it will kick off a new search for each event you feed into it. Also, pay close attention to the
maxsearches attribute, which defaults to 10.