Track Outbound Traffic from your AWS NAT Gateways

I would like to mention a few pointers before diving deep into the topic.

  • It is expected that your VPC Flow Logs and Cloudfront Logs are being stored on an ELK Cluster.

  • You should have access to ec2:DescribeNatGateways on your current role/user.

  • Pre-configured awscli with above-mentioned access added on your current user/role.

  • If you are running this script from an EC2 instance, make sure you can access ELK from that subnet.

NAT Gateways, and why is it even important?

As the name suggests, NAT Gateways basically acts like a gateway for devices i.e EC2 Instance, RDS Instance, or VPC attached AWS Lambda from your private subnet to talk to the world, which means, only egress traffic from your devices will be allowed to go. For example, when the command sudo apt update is executed from your Ubuntu machine from a private subnet it still gets updated, whereas when one sends traffic to this instance from another device it will always be unreachable.

This is the essence behind NAT Gateways. It allows private resources to send internet traffic.

VPC Flow logs

VPC Flow Logs is a feature that enables you to capture information about the IP traffic going to and from network interfaces in your VPC.

VPC Flow Logs follow the following sequence in their logs

version vpc-id subnet-id instance-id interface-id account-id type srcaddr dstaddr srcport dstport pkt-srcaddr pkt-dstaddr protocol bytes packets start end action tcp-flags log-status

Below is an example of the type of data you get in VPC Flow logs:

2 123456789010 eni-1235b8ca123456789 203.0.113.12 172.31.16.139 0 0 1 4 336 1432917027 1432917142 ACCEPT OK

Cloudfront Logs

AWS Cloudfront is a web service that speeds up the distribution of your static and dynamic web content, like .html, .js, and .css files to your users. It uses the concept of using a Content Distribution Network.

  • We need to define an Origin Server while configuring a Cloudfront Distribution

  • The Edge Server retrieves content from the Origin Server and then caches it locally, so that the next time whenever a request for the static content comes in, the edge location simply pulls it out from the cache and sends it back to the user. This ensures Low Latency in content delivery.

When a user requests an object/file they are always logged in Cloudfront Access Logs, with the following sequence(adding only what is necessary for us) you can read more about different fields that CloudFront Access Logs have in their sequence.

  • cs-uri-stem: Shares information about the requested URI path for e.g /index.html

  • c-ip: Shares information related to the IP address of the requester. If the viewer made a using an HTTP proxy or a load balancer then the IP will be of the proxy/load balancer.

One can be creative after pulling out IP addresses from VPC Flow logs and query Cloudfront Access Logs to co-relate what kind of request was made.

Enumerating NAT Gateways

One can simply enumerate all the NAT Gateways in their AWS Env using boto3

import boto3

# Enumerates NAT gateways and returns a list of ENI-Ids for NAT Gateways
# ["eni-1234", "eni-098", "eni-6641"]
def enum_nat(session):
    nat = session.describe_nat_gateways()
    eni_ids = list(map(lambda x:x['NatGatewayAddresses'][0]['NetworkInterfaceId'],nat['NatGateways']))
    return eni_ids

if __name__ == "__main__":
    session = boto3.client('ec2', region_name='us-east-1')
    nat_eni_list = enum_nat(session)

Note: Region can vary according to workloads.

Following the code above returns us a list of Elastic Network Interface IDs(ENI). The requirement of ENI-Ids of NAT Gateways is essential, as, under VPC Flow Logs, the source is mentioned with an elastic network interface id and not as a resource id.

Querying the right endpoints!

Considering that you are ingesting your VPC Flow Logs onto an ELK Cluster, with the above-mentioned sequence.

Now that we know what to look for and where to let's quickly jump on how can we query VPC Flow logs ELK cluster.

We can have an example query as mentioned below.

{ 
  "query": {
    "bool":{
        "must":{
            "term": {
                "interface_id": {
                    "value": <ENI-ID>
                }
            }
        },
        "must_not": {
            "range":{
                "dstaddr":{
                    "gte": "10.0.0.0",
                    "lte": "10.255.255.255"
                }
            }
        }
    }
}, 
"aggs": {    
    "dstaddr":{      
        "terms":{
            "field":"dstaddr",
            "size":10000
        } 
    }    
}}

Quick break down of the query and let's try to understand what are we trying to pull here.

  • We are querying elastic for key-value where the key is interface_id, here we can supply the list of all the NAT Gateway's Elastic Network Interface(ENI) ids.

  • The Second thing to note here is that we have added a must_not clause for range on dstaddr.

    • Here we have hardcoded 172.16.0.0/12, it can also be 192.168.0.0/16 or 10.0.0.0/8.

    • The reason why we are doing this here is that we do not want any logs that have a destination address as a Private IP address this private IP CIDR will be dependent on what's your VPC's internal CIDR.

  • Lastly, we have added a clause to give us a count of the IP addresses

After the above step, the response will be something like this.

{
	"bucket":[
	{
		"key": "1.2.3.4",
		"doc_count": 100
	}, 
	{
		"key": "5.6.7.8",
		"doc_count": 400
	},
	{
		"key": "9.8.6.5",
		"doc_count": 1000
	}
  ]
}

Now that we have a list of IP addresses corresponding to NAT Gateways, one can write a simple wrapper around this query.

#! /usr/bin/python3
import json
import requests

# Function to process incoming ENI id and query ELK to pull the count of destination IP address
def enumerate_elk_eni(eni):
    print(f"[+]Enumerating ENI: {eni}")
    try:
        headers = {"Content-Type": "application/json"}
        data = { 
            "query": {
                "bool":{
                    "must":{
                        "term": {
                            "interface_id": {
                                "value": eni
                            }
                        }
                    },
                    "must_not": {
                        "range":{
                            "dstaddr":{
                                "gte": "10.0.0.0",
                                "lte": "10.255.255.255"
                            }
                        }
                    }
                }
            }, 
            "aggs": {    
                "dstaddr":{      
                    "terms":{
                        "field":"dstaddr",
                        "size":10000
                    } 
                }    
            }
        }
        
        # ELK endpoint
        url = "https://yourelkcluster/_search"
        r1 = requests.post(url=url, headers=headers, data=json.dumps(data), verify=False, timeout=None)
        if r1.status_code == 200:
            data = json.loads(r1.text)
            data_len = data['aggregations']['dstaddr']['buckets']
            if len(data_len) > 0:
                for ips in data_len:
                    ip = ips['key']
                    count = ips['doc_count']
                    # Final Processor to corelate with Cloudfront logs and write on csv
                    processor(eni, ip, count) 
            else:
                pass
        else:
            print(f'error: {r1.text}')
    except Exception as em:
        print(f"[+]got exception at enumerate_es_eni: {em}")
        pass

There is another function processor(eni, ip, count) on line number 51, this function takes one single IP, ENI and count and sends it Cloudfront ELK cluster to do correlation, geoIP Enrichment and Reverse DNS Lookup.

def processor(eni, ip, count):
    print(f"[+]Egress data from {eni} to {ip}: {count}, processing IP")
    key, doc_count = cloudfront_es(ip)
    # Condition if CloudFront ES finds data in that cluster
    if doc_count != None and key != None:
        print(f'The {ip} has made calls to the following endpoints {key}, {doc_count} times')
        geoip = get_geoip_details(ip)
        location = geo_ip_processor(geoip)
        reverse_dns = reverse_lookup(ip)
        print()
        print(f"[+]The IP {ip} was invoked from {location}")
        write_csv(eni, ip, count, location,reverse_dns, key)
    # Condition if CloudFront ES does not return any data, i.e could'nt find the IP in CloudFront ES
    else:
        geoip = get_geoip_details(ip)
        location = geo_ip_processor(geoip)
        reverse_dns = reverse_lookup(ip)
        print(f"[+]The IP {ip} was invoked from {location}")
        write_csv(eni, ip, count, location, reverse_dns)

Coming to cloudfront_es function we can run a sample query like.

{
"query": {
    "bool": {
	"must":{
            "wildcard":{
                "c_ip": {
                    "value": f"{ip}"
             }
        }
    }
}},
    "aggs": {    
        "cs_uri_stem":{      
            "terms":{
                "field":"cs_uri_stem.keyword",
                "size":10
        } 
    }    
}}

Breaking the query, and quickly going through what is going on here.

  • c_ip : We are sending the individual IP address that we got earlier from enumerate_elk_eni and checking if any request was made from this IP to our infrastructure on cloudfront access logs.

  • cs-uri-stem.keyword: This will give us a list of all the URI paths that were used for querying our infrastructure.

The Response from the above query for individual IP addresses comes like this.

{
"bucket":[
{
	"key": "/api/v1/sampleEndpoint/ThirdParty",
	"doc_count": 100
}, 
{
	"key": "/api/v1/accountDetails/ThirdParty",
	"doc_count": 400
},
{
	"key": "/api/v1/userDetails/ThirdParty",
	"doc_count": 1000
}]
}

We can simply write a function cloudfron_es which will help us to automate the above-mentioned step and send data back to the processor function, i.e in format of key and doc_count. 

def cloudfront_es(ip):
    try:
        url = "https://your_cloudfront_elastic_endpoint/_search"
        headers = {"Content-Type": "application/json"}
        data = {
                "query": {
                        "bool": {
                            "must":{
                                "wildcard":{
                                    "c_ip": {
                                        "value": f"{ip}"
                            }
                        }
                    }
                }
                },
                "aggs": {    
                        "cs_uri_stem":{      
                            "terms":{
                                "field":"cs_uri_stem.keyword",
                                "size":10
                        } 
                    }    
                }
            }
        r1 = requests.post(url=url, headers=headers, data=json.dumps(data), timeout=None)
        if r1.status_code == 200:
            data = json.loads(r1.text)
            data_len = data['aggregations']['cs_uri_stem']['buckets']
            if len(data_len) == 0:
                return None, None
            else:
                for key in data_len:
                    return key['key'], key['doc_count']
                         
    except Exception as em:
        print(f'[+]Exception at cloudfront_es: {em}')

Now that we have all the co-relation done, we can simply do GeoIP Enrichment, GeoLocation, and Reverse DNS Lookup for the IP and write results onto a CSV file. 

NAT Gateway	IP	Count	Location	Reverse DNS Lookup	Common Paths from CloudFront ES
eni-01234	1.2.3.4	100	Mumbai, India	google.com	        /api/userdata
eni-01234	2.3.4.5	200	Mumbai, India	youtube.com	        /api/metadata
eni-01233	3.4.5.6	300	Mumbai, India	gmail.com               /api/check

Now the above-mentioned CSV that was generated, we can figure out the following pointers

  • NAT Gateway that was used to send request to

  • What was the destination address where our traffic is going

  • Location of the IP address

  • Reverse DNS Lookup on IP address

  • What was the URI path used when this IP made the first call to our infrastructure.

You can download the source code for this from here.

Signoff Notes

I hope the above-mentioned exercise helped you to grab some basic ideas around how AWS works and how we can leverage boto3, ELK, and basic Networking concepts to fiddle around with data. Please don't hesitate to reach out if you feel stuck anywhere.

Thank you, everyone! Saransh Rana Infrastructure Security Engineer @ CRED

PreviousFirst things First!NextBypass Cloudtrail Detection using AWS Appstream

Last updated