PantherFlow Examples: Threat Hunting Scenarios

Pivoting from an alert to a log search

Let's say we've received a Wiz alert about an EC2 instance being potentially misconfigured. From the alert, we can pull the associated AWS instance IDs. Then, we can search across all of our AWS logs for activity on those instances.

let alert_data = panther_signals.public.signal_alerts
| where p_event_time > time.ago(7d)
| where p_alert_id == '00411934608291e0fccd928590194fd6'
| summarize instances = arrays.flatten(agg.make_set(p_any_aws_instance_ids)),
    mintime = agg.min(p_event_time),
    maxtime = agg.max(p_event_time);

union panther_logs.public.aws*
| where p_event_time between time.parse_timestamp(toscalar(alert_data | project mintime)) - 30m 
    .. time.parse_timestamp(toscalar(alert_data | project maxtime)) + 30m
| where arrays.overlap(p_any_aws_instance_ids, toscalar(alert_data | project instances))

The statements above leverage:

• let statement functionality

• Operators: union, project, summarize, where, and between

• Functions: arrays.flatten(), arrays.overlap(), agg.make_set(), agg.min(), agg.max(), time.ago(), time.parse_timestamp(), and toscalar()

Pulling IPs from one table to search in another

While threat hunting, it's common to need to pull values from one table and pivot to searching those values in another table. The queries below fetch IP addresses from the VPC Flow logs table, then search for them in Okta System logs.

let IPs = panther_logs.public.aws_vpcflow
| where p_event_time > time.ago(5d)
| where flowDirection == 'ingress'
| summarize IP = agg.make_set( re.substr(srcAddr, '(\\d{1,3})\\.(\\d{1,3})\\.(\\d{1,3})\\.(\\d{1,3})') );

panther_logs.public.okta_systemlog
| where p_event_time > time.ago(1d)
| where arrays.overlap(toscalar(IPs), p_any_ip_addresses)

The statements above leverage:

• let statement functionality

• Operators: summarize and where

• Functions: arrays.overlap(), agg.make_set(), re.substr(), time.ago(), and toscalar()

CIDR matching with a regular expression

This query searches AWS logs for an IP address that matches a regex expression.

union aws_alb , amazon_eks_audit, aws_cloudtrail, aws_s3serveraccess
| where p_event_time > time.ago(7d)

| extend ip = coalesce(clientIp, sourceIPs, sourceIPAddress, remoteip)
| summarize events=agg.count() by ip, p_log_type

| where re.matches(ip, "^34\\.222\\..+\\..+$")
| sort events desc

The statements above leverage:

• Operators: union, where, extend, summarize, and sort

• Functions: coalesce(), agg.count(), and re.matches()

Results:

Investigating an alert for API key creation

In this scenario, we've received an alert for a match on the Panther-managed AWS User API Key Created detection, which runs over CloudTrail data and alerts when an AWS API key is created for an AWS user by another user.

The event tells us that an actor named ariel.ropek has created API keys for a new user called snidely-whiplash. Let's investigate to see if this behavior is a false positive or real compromise.

1. First, let's look at ariel.ropek's activity during the hour surrounding the alert:

   Copy

   panther_logs.public.aws_cloudtrail
   | where p_event_time between time.parse_timestamp('2024-11-13 19:00') .. time.parse_timestamp('2024-11-13 20:00')
   | extend p_actor = strings.split(userIdentity.arn, '/')[arrays.len(strings.split(userIdentity.arn, '/'))-1]
   | where p_actor == 'ariel.ropek'
   | sort p_event_time desc

   Results:

   

   Interesting! We see that not only did ariel.ropek create a new user named snidely-whiplash, but that they attached an AdministratorAccess policy to it:

   
2. Let’s add snidely-whiplash to our query to view their activity:

   Copy

   panther_logs.public.aws_cloudtrail
   | where p_event_time between time.parse_timestamp('2024-11-13 19:00') .. time.parse_timestamp('2024-11-13 20:00')
   | extend p_actor = strings.split(userIdentity.arn, '/')[arrays.len(strings.split(userIdentity.arn, '/'))-1]
   | where p_actor in ['ariel.ropek', 'snidely-whiplash']
   | sort p_event_time desc

   Results:

   

   There are many more results when snidely-whiplash is included—we can see that user ran commands in EKS. They created a new role, associated the AmazonEKSClusterAdminPolicy to it, then assumed the role under a new session name, snidely-whiplash-session.

   
3. Now that we know the user took action in EKS, we need to use union to expand our search to include EKS Audit and Authenticator logs. CloudTrail, EKS Audit, and EKS Authenticator logs each have a different schema; however, we can use coalesce() to create a data model that maps these log sources to common actor and action fields:

   Copy

   union panther_logs.public.aws_cloudtrail, panther_logs.public.amazon_eks_audit, panther_logs.public.amazon_eks_authenticator
   | where p_event_time between time.parse_timestamp('2024-11-13 19:00') .. time.parse_timestamp('2024-11-13 20:00')
   | extend p_aws_arn = coalesce(userIdentity.arn, user.username, arn)
   | extend p_actor = strings.split(p_aws_arn, '/')[arrays.len(strings.split(p_aws_arn, '/'))-1]
   | extend p_action = coalesce(eventName, strings.cat(verb, ' ', objectRef.resource), path)
   | where p_actor in ['ariel.ropek', 'snidely-whiplash', 'snidely-whiplash-session']
   | sort p_event_time desc

   Results:

   

   Looking at the p_action column, we can see that using snidely-whiplash-session, the user created a Kubernetes pod (create pods). Clicking into the full event, we see the pod they created is privileged:

   

   At this point, we can conclude that this is very likely a malicious actor. In sum, we've discovered:

   1. An AWS user (ariel-ropek) created a new user (snidley-whiplash) with admin privileges.

   2. snidley-whiplash pivoted into EKS and elevated their privileges there, becoming a cluster admin.

   3. They created a privileged pod in EKS.

4. We can view the full attack chain in a chart with visualize:

   Copy

   union panther_logs.public.aws_cloudtrail, panther_logs.public.amazon_eks_audit, panther_logs.public.amazon_eks_authenticator
   | where p_event_time between time.parse_timestamp('2024-11-13 19:00') .. time.parse_timestamp('2024-11-13 20:00')
   | extend p_aws_arn = coalesce(userIdentity.arn, user.username, arn)
   | extend p_actor = strings.split(p_aws_arn, '/')[arrays.len(strings.split(p_aws_arn, '/'))-1]
   | extend p_action = coalesce(eventName, strings.cat(verb, ' ', objectRef.resource), path)
   | where p_actor in ['ariel.ropek', 'snidely-whiplash', 'snidely-whiplash-session']
   | summarize events = agg.count() by p_actor, p_action
   | visualize bar xcolumn = p_action
   | sort p_actor

   Results:

   

   In the visualization above, the attack chain can be tracked from right to left, starting with ariel.ropek's actions.