Script to detect possible/similar DNS traffic generated by DNSStager tool

dnsstager_dns_traffic_filter_detect.py

#!/usr/bin/env python
# coding: utf-8

# In[1]:


import pandas as pd
import requests
import json


# In[10]:


# Define some data members
ip_info_api_key = 'API_KEY'
ip_info_api_url ='https://ipinfo.io/'

# Function defention
'''
Check a given IP address if it's genuine using https://ipinfo.io
Returns true: if the IP is genuine
'''
def is_valid_ip(ip):
    url = "{}{}/?token={}".format(ip_info_api_url, ip, ip_info_api_key)
    
    r = requests.get(url)
    
    j = json.loads(r.text)
    
    # Suspect: bogon, anycast, only ip returned
    if len(j) == 1 and 'ip' in j:
        return False
    
    if 'bogon' in j and j['bogon'] is True:
        return False
    
    if 'anycast' in j and j['anycast'] is True:
        return False
    
    # invalid ip? check these entry why we got 404!
    if 'status' in j and j['status'] == 404:
        return False
    
    return True



# Read the logs from 'dns_logs.csv
pd.set_option('display.max_columns', 100)
df = pd.read_csv('dns_logs.csv')
df = df.reset_index()

# Filter the data frame
df = df[df['QueryResults'].map(len) > 20]


print("QueryName\t\t\tComputer\t\t\tuser\t\tprocess_path")
for index, row in df.iterrows():
    queryResult = row['QueryResults']
    
    if queryResult.startswith("::1") or queryResult.startswith("::ffff:") or queryResult.startswith("::1;::ffff:") or queryResult.startswith("type:  "):
        continue
    
    if queryResult.startswith("::1") or queryResult.startswith("fe80::") or queryResult.startswith("::ffff:") or queryResult.startswith("::1;::ffff:") or queryResult.startswith("type:  "):
        continue
        
    # IPv4 type in IPv6. skip
    if queryResult.startswith('::ffff:'):
        continue
        
    if len(queryResult.rstrip(';').split('.')) == 4:
        continue
    
    # multi-ips seprated by comma
    parts = queryResult.split(';')
    
    for ip in parts:
        ip = ip.rstrip(';')
        if ip is None or len(ip) <= 1:
            continue
            
        if not is_valid_ip(ip):
            x = "{}\t{}\t\t{}\t{}".format(row["QueryName"], row["Computer"], row["user"], row["process_path"])
            print(x)
            continue