XQL Query: Looking at Multiple Events

[SARowe_NZ]

Hey there,

I need some help with a query please.

I want a query that returns instances of two events, but ONLY when the events do NOT occur within 2 minutes of each other.

dataset = xdr_data
| filter (event_type = ENUM.PROCESS and event_sub_type = ENUM.PROCESS_STOP and actor_process_image_name = "PanGPS.exe") or (action_evtlog_event_id = 1074 and action_evtlog_message contains "has initiated the restart of computer")

The above returns all the stuff I want to look at. What I want to do is further filter it where the two events are independent of each other and don't occur together within a couple of minutes.

In plain English I want to know when the PanGPS.exe process is stopped, but only when it is not due to a system reboot.

Thanks!

[zarnous]

Hi 

If i understood your request correctly, this can be done using the timestamp_diff function in XQL - 
https://docs-cortex.paloaltonetworks.com/r/Cortex-XDR/Cortex-XDR-XQL-Language-Reference/timestamp_di... 

Based on that , Try adding those lines to your query and let me know if that gives you the result you are looking for :

| alter CurrentTime = current_time()
| alter Timediff_M = timestamp_diff(CurrentTime , _time, "MINUTE")
| filter Timediff_M > 2

[SARowe_NZ]

Heya,

Thanks for that! - I did look at the timestamp_diff reference page but could not understand it in the context of what I'm trying to do.

Your lines definitely return a valid query but don't actually seem to filter anything. I have the same number of results and the instance of what I wish to filter out is still there. For example in the screenshot below the first line is the PanGPS.exe process being stopped, and the second line is the OS being restarted. They appear within 2 minutes of each other so I wish to filter them out.

Here is the complete query I am using:

dataset = xdr_data
| filter (event_type = ENUM.PROCESS and event_sub_type = ENUM.PROCESS_STOP and actor_process_image_name = "PanGPS.exe") or (action_evtlog_event_id = 1074 and action_evtlog_message contains "has initiated the restart of computer")
| alter CurrentTime = current_time()
| alter Timediff_M = timestamp_diff(CurrentTime , _time, "MINUTE")
| filter Timediff_M > 2

As current_time() is relative the number will always be greater than 2 so I thought about extracting the time based on day+hour+minute+second then converting that string to a number. This all works but I don't know where to go from there to filter it so that it looks at the value of ROW 1, compares it to the value of ROW 2 and filters both of them out if they are within x

dataset = xdr_data
| filter (event_type = ENUM.PROCESS and event_sub_type = ENUM.PROCESS_STOP and actor_process_image_name = "PanGPS.exe") or (action_evtlog_event_id = 1074 and action_evtlog_message contains "has initiated the restart of computer")
| alter time_string = concat(to_string(extract_time(_time, "DAY")) + to_string(extract_time(_time, "HOUR")) + to_string(extract_time(_time, "MINUTE")) + to_string(extract_time(_time, "SECOND")))
| fields time_string
| alter time_num = to_number(time_string)
| fields time_num

One other thought I had was to use dedupe with the above (e.g. concat day+hour+minute and dedupe everything) this would create duplicate entries where they share the same minute (close enough to what I want) and remove all duplicates, but would still leave one entry which is not what we want. Maybe there is a way to filter out any records that have been de-duped? or use something with dedupe to remove all entries and not just all-1?

It also feels like I'm missing the logic to do this sensibly. XQL queries are a bit above my pay grade 😉

[SARowe_NZ]

Interestingly this does partially work:

dataset = xdr_data
| filter (event_type = ENUM.PROCESS and event_sub_type = ENUM.PROCESS_STOP and actor_process_image_name = "PanGPS.exe") or (action_evtlog_event_id = 1074 and action_evtlog_message contains "has initiated the restart of computer")
| alter time_string = concat(to_string(extract_time(_time, "DAY")) + to_string(extract_time(_time, "HOUR")) + to_string(extract_time(_time, "MINUTE"))) //+ to_string(extract_time(_time, "SECOND")))
//| fields time_string
| alter time_num = to_number(time_string)
//| fields time_num
| comp var(time_num) as testx by time_num
| filter testx != 0

It seems to give testx a value of 0 for all records where time_num is a duplicate so its easy to filter out.

Unfortunately the "comp" function also seems to strip out all other fields/columns except those called in the function itself so the result is accurate, but only shows the "testx" and "time_num" fields instead of all the other info like hostname/ip/user etc.

Maybe it is easier to work with this and somehow get the fields back instead of working on a different logic?

[bbucao · Accepted solution]

Hi SARowe_NZ,

If I understand your use-case correctly I think using a left join could work. A left join will include all results from the parent query plus any results from the join result where it intersects with the parent. In the example below I am using the _time field with various functions and operators to only join the results if the shutdown event occurs within one minute (in seconds) of the process stop event, as well as requiring the hostnames to match. The results have some process stop events joined with related shutdown events. Next, I just filter for any log rows where my field "related_shutdown_event" equals null, which leaves me with the process stop events that do not have a related shutdown event. Let me know if this works for you!

dataset = xdr_data

| filter event_type = ENUM.PROCESS and event_sub_type = ENUM.PROCESS_STOP and actor_process_image_name = "PanGPS.exe"

|join conflict_strategy = both type = left (dataset = xdr_data
| filter action_evtlog_event_id = 1074 and action_evtlog_message contains "has initiated the restart of computer"|fields agent_hostname, action_evtlog_description as related_shutdown_event) as resulting_records agent_hostname = resulting_records.agent_hostname and add(to_epoch(_time, "SECONDS"), 60) >= to_epoch(resulting_records._time) and subtract(to_epoch(_time, "SECONDS"), 60) <= to_epoch(resulting_records._time)

|filter related_shutdown_event = null

|dedup _time by desc agent_hostname

Regards,
Ben

[SARowe_NZ]

This is amazing. Looks like it works perfectly. Thanks so much!

[SARowe_NZ]

Hey Ben,

Not sure if you're still around but, if so, I seem to have an issue with the dedupe on the above. In testing I killed off the PanGPS.exe at the times 14:01, 14:06, 14:32. With the dedupe off I see all entries (but many instances of them), but if I turn on the dedupe I only get the entry for 14:01 which I can't understand as the dedupe is by _time and so I would expect this to only dedupe entries where the time is the same (eg 14:32 and 14:32).

Any ideas?