How to join search based on multiple values?

[KMoryson]

Hi, I am trying to build a correlation that matches traffic to threat intel to figure out if it has been blocked or not. It looks like this:

Threat intel -> provides only information that given IP is malicious and recommends blocking. Thats it.
Traffic logs -> provides info on traffic that actually happened, both incoming and outgoing.

Lets say that threat intel tells me that ip 1.2.3.4 is malicious.

I do the following search:
 `index=all_traffic sourcetype=traffic "1.2.3.4"`

It shows me the IP in half/half src and dest fields, and other one is my own ip, as well as if this traffic has been dropped or not in a field called action

I need to automate this process to show me traffic containing suspicious ip.
I tried to join the following way:

`[...getting data from threat intel...]
| table suspicious_ip

| join suspicious_ip [ search index=all_traffic sourcetype=traffic  | rename src as suspicious_traffic ]`

This works only sometimes and for half of the cases, as it only looks as src and not dest. And it gets truncated because of subsearch limit.


I also tried to connect them via parenthesis like this:

`( from datamodel:"Threat_Intel"."Threat_Activity" ) AND (index=all_traffic sourcetype=traffic )

| stats values(*) as * by src, dest`


But this didn't work because I cannot use from datamodel this way.

My ideal output would be:

suspicious_ip src dest action

Any ideas?

[maciep · Accepted solution]

I think I would use a subsearch here.  If you're unfamiliar, it allows you to search the threat index and then use that data to limit a search against your traffic index.

Something like this I think might work (untested, fat-fingers likely abound)

index=traffic [
  search index=threat
  | stats count by sus_ip
  | eval sus_ip = "\"" . sus_ip . "\""
  | table sus_ip
  | mvcombine(sus_ip)
  | eval sus_ip = "(" . mvjoin(sus_ip,",") . ")"
  | eval filter = "src_ip IN " . sus_ip . " OR dst_ip IN " . sus_ip
  | return $filter
]
| table src_ip, dst_ip, action

so $filter gets built up in the sub search and is returned the main search.  So the main search becomes something like

index=traffic src_ip IN ("1.2.3.4","2.3.4.5","3.4.5.6") OR dst_ip IN ("1.2.3.4","2.3.4.5","3.4.5.6")
| tables src_ip, dst_ip, action

But you won't know which of src or dst is actually the malicious ip.  that could probably added with a bit more manipulation with an append or something after the fact.