Is there anyway to verify how many repeated alerts are being suppressed by Throttling.?

[yourfriend]

Hi Team, 

                   We are reviewing the use cases in our Splunk Enterprise security, We have given Throttling as 1 day for a use case, but want to check how many alerts are being suppressed by Throttling action. Is there any search query or any way how to check that.?

                    Is there anyway that we can show the proof that throttling is working fine.?

                    Thanks in advance for your help.

[Azeemering]

I assume you are running correlation searches that create notables since you are running Splunk ES.

You can check the notable index for the suppressions.

Use the macro in the SA-ThreatIntelligence app to see all suppressions:

`suppressed_notables("*")`

Otherwise here is the macro expanded:

(index=notable source="*") 
| eval indexer_guid=replace('_bkt',".*~(.+)","\\1"), event_hash=md5(('_time' . '_raw')), event_id=((((indexer_guid . "@@") . index) . "@@") . event_hash), rule_id=event_id 
| search event_id="*" 
| fields - "host_*" 
| tags outputfield=tag 
| eval tag=mvdedup(mvappend(tag,NULL,orig_tag)) 
| dedup rule_id 
| lookup update=true notable_xref_lookup event_id OUTPUTNEW xref_name as notable_xref_name,xref_id as notable_xref_id 
| eval notable_xref=mvzip(notable_xref_name,notable_xref_id,":") 
| lookup update=true correlationsearches_lookup _key as source OUTPUTNEW annotations, security_domain, severity, rule_name, description as savedsearch_description, rule_title, rule_description, drilldown_name, drilldown_search, drilldown_earliest_offset, drilldown_latest_offset, default_status, default_owner, default_disposition, next_steps, investigation_profiles, extract_artifacts, recommended_actions 
| eval rule_name=if(isnull(rule_name),source,rule_name), rule_title=if(isnull(rule_title),rule_name,rule_title), drilldown_earliest=case(isint(drilldown_earliest_offset),('_time' - drilldown_earliest_offset),(drilldown_earliest_offset == "$info_min_time$"),info_min_time,true(),null()), drilldown_latest=case(isint(drilldown_latest_offset),('_time' + drilldown_latest_offset),(drilldown_latest_offset == "$info_max_time$"),info_max_time,true(),null()), security_domain=if(isnull(security_domain),"threat",lower(security_domain)), rule_description=case(isnotnull(rule_description),rule_description,isnotnull(savedsearch_description),savedsearch_description,true(),"unknown"), governance_lookup_type="default" 
| lookup update=true governance_lookup savedsearch as source, lookup_type as governance_lookup_type OUTPUT governance, control 
| eval governance_lookup_type="tag" 
| lookup update=true governance_lookup savedsearch as source, tag, lookup_type as governance_lookup_type OUTPUT governance as governance_tag, control as control_tag 
| eval governance=mvappend(governance,NULL,governance_tag), control=mvappend(control,NULL,control_tag) 
| fields - governance_lookup_type, governance_tag, control_tag 
| eval temp_time=(time() + 86400) 
| lookup update=true event_time_field=temp_time incident_review_lookup rule_id OUTPUT owner as new_owner, urgency as new_urgency, status as new_status, disposition as new_disposition 
| lookup update=true event_time_field=temp_time incident_review_comment_lookup rule_id OUTPUT time as review_time,user as reviewer,comment 
| eval owner=if(isnotnull(new_owner),new_owner,owner), status=case(isnotnull(new_status),new_status,isnotnull(status),status,true(),default_status), urgency=if(isnotnull(new_urgency),new_urgency,urgency), disposition=if(isnotnull(new_disposition),new_disposition,default_disposition) 
| fields - temp_time, new_owner, new_status, new_urgency, new_disposition 
| eval temp_status=if(isnull(status),-1,status) 
| lookup update=true reviewstatuses_lookup _key as temp_status OUTPUT status,label as status_label,description as status_description,default as status_default,end as status_end 
| eval status=if(isnull(status_label),0,status), status_label=if(isnull(status_label),"Unassigned",status_label), status_description=if(isnull(status_description),"unknown",status_description), status_default=case(match(status_default,"1|[Tt]|[Tt][Rr][Uu][Ee]"),"true",match(status_default,"0|[Ff]|[Ff][Aa][Ll][Ss][Ee]"),"false",true(),status_default), status_end=case(match(status_end,"1|[Tt]|[Tt][Rr][Uu][Ee]"),"true",match(status_end,"0|[Ff]|[Ff][Aa][Ll][Ss][Ee]"),"false",true(),status_end), status_group=case((status_default == "true"),"New",(status_end == "true"),"Closed",(status == 0),"New",true(),"Open") 
| fields - temp_status 
| eval temp_disposition=if(isnull(disposition),-3,disposition) 
| lookup update=true disposition_lookup _key as temp_disposition OUTPUT status as disposition,label as disposition_label,description as disposition_description,default as disposition_default 
| eval disposition=if(isnull(disposition),"disposition:0",disposition), disposition_label=if(isnull(disposition_label),"Unassigned",disposition_label), disposition_description=if(isnull(disposition_description),"An error is preventing the event from having a valid disposition.",disposition_description), disposition_default=case(match(disposition_default,"1|[Tt]|[Tt][Rr][Uu][Ee]"),"true",match(disposition_default,"0|[Ff]|[Ff][Aa][Ll][Ss][Ee]"),"false",true(),disposition_default) 
| fields - temp_disposition 
| eval owner=case(isnotnull(owner),owner,isnotnull(default_owner),default_owner,true(),"unassigned") 
| lookup update=true user_realnames_lookup user as "owner" OUTPUTNEW realname as "owner_realname" 
| eval owner_realname=if(isnull(owner_realname),owner,owner_realname) 
| lookup update=true user_realnames_lookup user as "creator" OUTPUTNEW realname as "creator_realname" 
| eval creator_realname=if(isnull(creator_realname),creator,creator_realname), severities=if(in(lower(severity),"critical","high","medium","low","informational"),lower(severity),"unknown"), severity=case((severities == "critical"),"critical",(severities == "high"),"high",(severities == "medium"),"medium",(severities == "low"),"low",(severities == "informational"),"informational",true(),"unknown"), priorities=lower(mvappend(priority,host_priority,orig_host_priority,src_priority,dest_priority,dvc_priority,src_user_priority,user_priority,host_owner_priority,orig_host_owner_priority,src_owner_priority,dest_owner_priority,dvc_owner_priority)), priority=case((priorities == "critical"),"critical",(priorities == "high"),"high",(priorities == "medium"),"medium",(priorities == "low"),"low",(priorities == "informational"),"informational",true(),"unknown") 
| lookup local=true urgency_lookup priority,severity OUTPUTNEW urgency 
| eval urgency=if(in(lower(urgency),"critical","high","medium","low","informational"),lower(urgency),"unknown") 
| typer 
| tags outputfield=tag 
| eval tag=mvdedup(mvappend(tag,NULL,orig_tag)) 
| rex field=eventtype "notable_suppression-(?<suppression>.+)" 
| lookup update=true risk_correlation_by_system_lookup risk_object as "orig_host" OUTPUT risk_score as "_orig_host_risk_score" 
| eval orig_host_risk_score=if(isnotnull(orig_host_risk_score),orig_host_risk_score,'_orig_host_risk_score') 
| fields - _orig_host_risk_score 
| lookup update=true risk_correlation_by_system_lookup risk_object as "dvc" OUTPUT risk_score as "_dvc_risk_score" 
| eval dvc_risk_score=if(isnotnull(dvc_risk_score),dvc_risk_score,'_dvc_risk_score') 
| fields - _dvc_risk_score 
| lookup update=true risk_correlation_by_system_lookup risk_object as "src" OUTPUT risk_score as "_src_risk_score" 
| eval src_risk_score=if(isnotnull(src_risk_score),src_risk_score,'_src_risk_score') 
| fields - _src_risk_score 
| lookup update=true risk_correlation_by_system_lookup risk_object as "dest" OUTPUT risk_score as "_dest_risk_score" 
| eval dest_risk_score=if(isnotnull(dest_risk_score),dest_risk_score,'_dest_risk_score') 
| fields - _dest_risk_score 
| lookup update=true risk_correlation_by_user_lookup risk_object as "src_user" OUTPUT risk_score as "_src_user_risk_score" 
| eval src_user_risk_score=if(isnotnull(src_user_risk_score),src_user_risk_score,'_src_user_risk_score') 
| fields - _src_user_risk_score 
| lookup update=true risk_correlation_by_user_lookup risk_object as "user" OUTPUT risk_score as "_user_risk_score" 
| search (eventtype=notable_suppression-* OR suppression=*) 
| eval user_risk_score=if(isnotnull(user_risk_score),user_risk_score,'_user_risk_score') 
| fields - _user_risk_score 
| eval host_risk_object_type=if(isnotnull(host_risk_score),"system",null()), orig_host_risk_object_type=if(isnotnull(orig_host_risk_score),"system",null()), dvc_risk_object_type=if(isnotnull(dvc_risk_score),"system",null()), src_risk_object_type=if(isnotnull(src_risk_score),"system",null()), dest_risk_object_type=if(isnotnull(dest_risk_score),"system",null()), src_user_risk_object_type=if(isnotnull(src_user_risk_score),"user",null()), user_risk_object_type=if(isnotnull(user_risk_score),"user",null()), risk_score=if(isnotnull(risk_score),risk_score,max(orig_host_risk_score,dvc_risk_score,src_risk_score,dest_risk_score,src_user_risk_score,user_risk_score)), notable_type=if((isnotnull(risk_object) AND isnotnull(risk_object_type)),"risk_event","notable")

[yourfriend]

@Azeemering , I have tried the search query which you have given but its not showing any events. Can you please help me in any other way so that I can get to know those suppressed alerts. I want to show the proof of throttling is working fine and the alerts are being suppressed in the given time range.

[Azeemering]

I understand what you are trying to prove..I have had to do this myself too hence my solution that I used and worked.
Do you have that macro? Go to Advanced Search and check if you have it.

`suppressed_notables`

is a macro that comes with SA-ThreatIntelligence...and you should have that if you are running Splunk Enterprise Security.

Do you have events in the notable index at all? 

index=notable

When you search for your specific rule that you are suppressing by going to:

index=notable source="your rule name - Rule"

Do you find it at all?

[yourfriend]

I have checked what you have said we have that two macros (I am sharing it pics too), but when I tried to open it, It is showing I don't have permission for that. So what to do now, If I contact my admin teammate he would have access for that right.? He can help me right.? And what to do after this actually.? How to check the suppressed alerts after opening that macros.? You already helped me alot, just give me last step so that I can finish my task.

Thank you so much @Azeemering You are a real Gem of a person. I would appreciate heartfully for your help buddy.

[Azeemering]

That is correct. An admin is able to run these macro's or they need to give you access to run them and or access to the notable index too.

Then you can run:

`suppressed_notables` | stats c by rule_name

This will give you a count of the amount of suppressions per rule name / colleration search.

[yourfriend]

@Azeemering , with admin access, In splunk search we have to run this search query right.?

 which you have given yesterday.

`suppressed_notables` | stats c by rule_name

index="notable" `suppressed_notables` | stats c by rule_name , (I will place correlation name in place of rule_name)

But still this query doesn't seems correct to me, am I missing anything here like  how to eval macro in this search query.

Can pls let me know the exact search query or else exact steps how to get the suppressed events.? one by one. So that I will give these details to my admin colleague and get those results.

[Azeemering]

Eval a macro in search query does not make sense....

When you press CTRL-SHIFT-E in the spl search bar the macro will expand to this:

(index=notable source="*") 
| eval indexer_guid=replace('_bkt',".*~(.+)","\\1"), event_hash=md5(('_time' . '_raw')), event_id=((((indexer_guid . "@@") . index) . "@@") . event_hash), rule_id=event_id 
| search event_id="*" 
| fields - "host_*" 
| tags outputfield=tag 
| eval tag=mvdedup(mvappend(tag,NULL,orig_tag)) 
| dedup rule_id 
| lookup update=true notable_xref_lookup event_id OUTPUTNEW xref_name as notable_xref_name,xref_id as notable_xref_id 
| eval notable_xref=mvzip(notable_xref_name,notable_xref_id,":") 
| lookup update=true correlationsearches_lookup _key as source OUTPUTNEW annotations, security_domain, severity, rule_name, description as savedsearch_description, rule_title, rule_description, drilldown_name, drilldown_search, drilldown_earliest_offset, drilldown_latest_offset, default_status, default_owner, default_disposition, next_steps, investigation_profiles, extract_artifacts, recommended_actions 
| eval rule_name=if(isnull(rule_name),source,rule_name), rule_title=if(isnull(rule_title),rule_name,rule_title), drilldown_earliest=case(isint(drilldown_earliest_offset),('_time' - drilldown_earliest_offset),(drilldown_earliest_offset == "$info_min_time$"),info_min_time,true(),null()), drilldown_latest=case(isint(drilldown_latest_offset),('_time' + drilldown_latest_offset),(drilldown_latest_offset == "$info_max_time$"),info_max_time,true(),null()), security_domain=if(isnull(security_domain),"threat",lower(security_domain)), rule_description=case(isnotnull(rule_description),rule_description,isnotnull(savedsearch_description),savedsearch_description,true(),"unknown"), governance_lookup_type="default" 
| lookup update=true governance_lookup savedsearch as source, lookup_type as governance_lookup_type OUTPUT governance, control 
| eval governance_lookup_type="tag" 
| lookup update=true governance_lookup savedsearch as source, tag, lookup_type as governance_lookup_type OUTPUT governance as governance_tag, control as control_tag 
| eval governance=mvappend(governance,NULL,governance_tag), control=mvappend(control,NULL,control_tag) 
| fields - governance_lookup_type, governance_tag, control_tag 
| eval temp_time=(time() + 86400) 
| lookup update=true event_time_field=temp_time incident_review_lookup rule_id OUTPUT owner as new_owner, urgency as new_urgency, status as new_status, disposition as new_disposition 
| lookup update=true event_time_field=temp_time incident_review_comment_lookup rule_id OUTPUT time as review_time,user as reviewer,comment 
| eval owner=if(isnotnull(new_owner),new_owner,owner), status=case(isnotnull(new_status),new_status,isnotnull(status),status,true(),default_status), urgency=if(isnotnull(new_urgency),new_urgency,urgency), disposition=if(isnotnull(new_disposition),new_disposition,default_disposition) 
| fields - temp_time, new_owner, new_status, new_urgency, new_disposition 
| eval temp_status=if(isnull(status),-1,status) 
| lookup update=true reviewstatuses_lookup _key as temp_status OUTPUT status,label as status_label,description as status_description,default as status_default,end as status_end 
| eval status=if(isnull(status_label),0,status), status_label=if(isnull(status_label),"Unassigned",status_label), status_description=if(isnull(status_description),"unknown",status_description), status_default=case(match(status_default,"1|[Tt]|[Tt][Rr][Uu][Ee]"),"true",match(status_default,"0|[Ff]|[Ff][Aa][Ll][Ss][Ee]"),"false",true(),status_default), status_end=case(match(status_end,"1|[Tt]|[Tt][Rr][Uu][Ee]"),"true",match(status_end,"0|[Ff]|[Ff][Aa][Ll][Ss][Ee]"),"false",true(),status_end), status_group=case((status_default == "true"),"New",(status_end == "true"),"Closed",(status == 0),"New",true(),"Open") 
| fields - temp_status 
| eval temp_disposition=if(isnull(disposition),-3,disposition) 
| lookup update=true disposition_lookup _key as temp_disposition OUTPUT status as disposition,label as disposition_label,description as disposition_description,default as disposition_default 
| eval disposition=if(isnull(disposition),"disposition:0",disposition), disposition_label=if(isnull(disposition_label),"Unassigned",disposition_label), disposition_description=if(isnull(disposition_description),"An error is preventing the event from having a valid disposition.",disposition_description), disposition_default=case(match(disposition_default,"1|[Tt]|[Tt][Rr][Uu][Ee]"),"true",match(disposition_default,"0|[Ff]|[Ff][Aa][Ll][Ss][Ee]"),"false",true(),disposition_default) 
| fields - temp_disposition 
| eval owner=case(isnotnull(owner),owner,isnotnull(default_owner),default_owner,true(),"unassigned") 
| lookup update=true user_realnames_lookup user as "owner" OUTPUTNEW realname as "owner_realname" 
| eval owner_realname=if(isnull(owner_realname),owner,owner_realname) 
| lookup update=true user_realnames_lookup user as "creator" OUTPUTNEW realname as "creator_realname" 
| eval creator_realname=if(isnull(creator_realname),creator,creator_realname), severities=if(in(lower(severity),"critical","high","medium","low","informational"),lower(severity),"unknown"), severity=case((severities == "critical"),"critical",(severities == "high"),"high",(severities == "medium"),"medium",(severities == "low"),"low",(severities == "informational"),"informational",true(),"unknown"), priorities=lower(mvappend(priority,host_priority,orig_host_priority,src_priority,dest_priority,dvc_priority,src_user_priority,user_priority,host_owner_priority,orig_host_owner_priority,src_owner_priority,dest_owner_priority,dvc_owner_priority)), priority=case((priorities == "critical"),"critical",(priorities == "high"),"high",(priorities == "medium"),"medium",(priorities == "low"),"low",(priorities == "informational"),"informational",true(),"unknown") 
| lookup local=true urgency_lookup priority,severity OUTPUTNEW urgency 
| eval urgency=if(in(lower(urgency),"critical","high","medium","low","informational"),lower(urgency),"unknown") 
| typer 
| tags outputfield=tag 
| eval tag=mvdedup(mvappend(tag,NULL,orig_tag)) 
| rex field=eventtype "notable_suppression-(?<suppression>.+)" 
| lookup update=true risk_correlation_by_system_lookup risk_object as "orig_host" OUTPUT risk_score as "_orig_host_risk_score" 
| eval orig_host_risk_score=if(isnotnull(orig_host_risk_score),orig_host_risk_score,'_orig_host_risk_score') 
| fields - _orig_host_risk_score 
| lookup update=true risk_correlation_by_system_lookup risk_object as "dvc" OUTPUT risk_score as "_dvc_risk_score" 
| eval dvc_risk_score=if(isnotnull(dvc_risk_score),dvc_risk_score,'_dvc_risk_score') 
| fields - _dvc_risk_score 
| lookup update=true risk_correlation_by_system_lookup risk_object as "src" OUTPUT risk_score as "_src_risk_score" 
| eval src_risk_score=if(isnotnull(src_risk_score),src_risk_score,'_src_risk_score') 
| fields - _src_risk_score 
| lookup update=true risk_correlation_by_system_lookup risk_object as "dest" OUTPUT risk_score as "_dest_risk_score" 
| eval dest_risk_score=if(isnotnull(dest_risk_score),dest_risk_score,'_dest_risk_score') 
| fields - _dest_risk_score 
| lookup update=true risk_correlation_by_user_lookup risk_object as "src_user" OUTPUT risk_score as "_src_user_risk_score" 
| eval src_user_risk_score=if(isnotnull(src_user_risk_score),src_user_risk_score,'_src_user_risk_score') 
| fields - _src_user_risk_score 
| lookup update=true risk_correlation_by_user_lookup risk_object as "user" OUTPUT risk_score as "_user_risk_score" 
| search (eventtype=notable_suppression-* OR suppression=*) 
| eval user_risk_score=if(isnotnull(user_risk_score),user_risk_score,'_user_risk_score') 
| fields - _user_risk_score 
| eval host_risk_object_type=if(isnotnull(host_risk_score),"system",null()), orig_host_risk_object_type=if(isnotnull(orig_host_risk_score),"system",null()), dvc_risk_object_type=if(isnotnull(dvc_risk_score),"system",null()), src_risk_object_type=if(isnotnull(src_risk_score),"system",null()), dest_risk_object_type=if(isnotnull(dest_risk_score),"system",null()), src_user_risk_object_type=if(isnotnull(src_user_risk_score),"user",null()), user_risk_object_type=if(isnotnull(user_risk_score),"user",null()), risk_score=if(isnotnull(risk_score),risk_score,max(orig_host_risk_score,dvc_risk_score,src_risk_score,dest_risk_score,src_user_risk_score,user_risk_score)), notable_type=if((isnotnull(risk_object) AND isnotnull(risk_object_type)),"risk_event","notable") 
| stats c by rule_name