This commit extends the analyzer script with two new flags: * Adding --no-dns disables hostname DNS resolution, improving speed * Adding --match-host <IP address> filters all flows not matching the IP Additional small things were changed, the script is still work in progress. Especially the "pairing" of two flows will be removed in future versions.
328 lines
12 KiB
Python
Executable file
328 lines
12 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
|
|
"""
|
|
Example analyzing script for saved exports (by main.py, as JSON).
|
|
This file belongs to https://github.com/bitkeks/python-netflow-v9-softflowd.
|
|
|
|
Copyright 2017-2020 Dominik Pataky <dev@bitkeks.eu>
|
|
Licensed under MIT License. See LICENSE.
|
|
"""
|
|
|
|
import argparse
|
|
from collections import namedtuple
|
|
import contextlib
|
|
from datetime import datetime
|
|
import functools
|
|
import gzip
|
|
import ipaddress
|
|
import json
|
|
import logging
|
|
import os.path
|
|
import socket
|
|
import sys
|
|
|
|
|
|
IP_PROTOCOLS = {
|
|
1: "ICMP",
|
|
6: "TCP",
|
|
17: "UDP",
|
|
58: "ICMPv6"
|
|
}
|
|
|
|
Pair = namedtuple('Pair', ['src', 'dest'])
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def printv(message, *args_, **kwargs):
|
|
if args.verbose:
|
|
print(message.format(*args_, **kwargs))
|
|
|
|
|
|
@functools.lru_cache(maxsize=None)
|
|
def resolve_hostname(ip: str) -> str:
|
|
if args.no_dns:
|
|
# If no DNS resolution is requested, simply return the IP string
|
|
return ip
|
|
# else resolve the IP address to a hostname and return the hostname
|
|
return socket.getfqdn(ip)
|
|
|
|
|
|
def fallback(d, keys):
|
|
for k in keys:
|
|
if k in d:
|
|
return d[k]
|
|
raise KeyError(", ".join(keys))
|
|
|
|
|
|
def human_size(size_bytes):
|
|
# Calculate a human readable size of the flow
|
|
if size_bytes < 1024:
|
|
return "%dB" % size_bytes
|
|
elif size_bytes / 1024. < 1024:
|
|
return "%.2fK" % (size_bytes / 1024.)
|
|
elif size_bytes / 1024.**2 < 1024:
|
|
return "%.2fM" % (size_bytes / 1024.**2)
|
|
else:
|
|
return "%.2fG" % (size_bytes / 1024.**3)
|
|
|
|
|
|
def human_duration(seconds):
|
|
# Calculate human readable duration times
|
|
if seconds < 60:
|
|
# seconds
|
|
return "%d sec" % seconds
|
|
if seconds / 60 > 60:
|
|
# hours
|
|
return "%d:%02d.%02d hours" % (seconds / 60**2, seconds % 60**2 / 60, seconds % 60)
|
|
# minutes
|
|
return "%02d:%02d min" % (seconds / 60, seconds % 60)
|
|
|
|
|
|
class Connection:
|
|
"""Connection model for two flows.
|
|
The direction of the data flow can be seen by looking at the size.
|
|
|
|
'src' describes the peer which sends more data towards the other. This
|
|
does NOT have to mean that 'src' was the initiator of the connection.
|
|
"""
|
|
def __init__(self, flow1, flow2):
|
|
if not flow1 or not flow2:
|
|
raise Exception("A connection requires two flows")
|
|
|
|
# Assume the size that sent the most data is the source
|
|
# TODO: this might not always be right, maybe use earlier timestamp?
|
|
size1 = fallback(flow1, ['IN_BYTES', 'IN_OCTETS'])
|
|
size2 = fallback(flow2, ['IN_BYTES', 'IN_OCTETS'])
|
|
if size1 >= size2:
|
|
src = flow1
|
|
dest = flow2
|
|
else:
|
|
src = flow2
|
|
dest = flow1
|
|
|
|
# TODO: this next approach uses the lower port as the service identifier
|
|
# port1 = fallback(flow1, ['L4_SRC_PORT', 'SRC_PORT'])
|
|
# port2 = fallback(flow2, ['L4_SRC_PORT', 'SRC_PORT'])
|
|
#
|
|
# src = flow1
|
|
# dest = flow2
|
|
# if port1 > port2:
|
|
# src = flow2
|
|
# dest = flow1
|
|
|
|
self.src_flow = src
|
|
self.dest_flow = dest
|
|
ips = self.get_ips(src)
|
|
self.src = ips.src
|
|
self.dest = ips.dest
|
|
self.src_port = fallback(src, ['L4_SRC_PORT', 'SRC_PORT'])
|
|
self.dest_port = fallback(dest, ['L4_DST_PORT', 'DST_PORT'])
|
|
self.size = fallback(src, ['IN_BYTES', 'IN_OCTETS'])
|
|
|
|
# Duration is given in milliseconds
|
|
self.duration = src['LAST_SWITCHED'] - src['FIRST_SWITCHED']
|
|
if self.duration < 0:
|
|
# 32 bit int has its limits. Handling overflow here
|
|
# TODO: Should be handled in the collection phase
|
|
self.duration = (2**32 - src['FIRST_SWITCHED']) + src['LAST_SWITCHED']
|
|
|
|
def __repr__(self):
|
|
return "<Connection from {} to {}, size {}>".format(
|
|
self.src, self.dest, self.human_size)
|
|
|
|
@staticmethod
|
|
def get_ips(flow):
|
|
# IPv4
|
|
if flow.get('IP_PROTOCOL_VERSION') == 4 or \
|
|
'IPV4_SRC_ADDR' in flow or 'IPV4_DST_ADDR' in flow:
|
|
return Pair(
|
|
ipaddress.ip_address(flow['IPV4_SRC_ADDR']),
|
|
ipaddress.ip_address(flow['IPV4_DST_ADDR'])
|
|
)
|
|
|
|
# IPv6
|
|
return Pair(
|
|
ipaddress.ip_address(flow['IPV6_SRC_ADDR']),
|
|
ipaddress.ip_address(flow['IPV6_DST_ADDR'])
|
|
)
|
|
|
|
@property
|
|
def human_size(self):
|
|
return human_size(self.size)
|
|
|
|
@property
|
|
def human_duration(self):
|
|
duration = self.duration // 1000 # uptime in milliseconds, floor it
|
|
return human_duration(duration)
|
|
|
|
@property
|
|
def hostnames(self):
|
|
# Resolve the IPs of this flows to their hostname
|
|
src_hostname = resolve_hostname(self.src.compressed)
|
|
dest_hostname = resolve_hostname(self.dest.compressed)
|
|
return Pair(src_hostname, dest_hostname)
|
|
|
|
@property
|
|
def service(self):
|
|
# Resolve ports to their services, if known
|
|
default = "({} {})".format(self.src_port, self.dest_port)
|
|
with contextlib.suppress(OSError):
|
|
return socket.getservbyport(self.src_port)
|
|
with contextlib.suppress(OSError):
|
|
return socket.getservbyport(self.dest_port)
|
|
return default
|
|
|
|
@property
|
|
def total_packets(self):
|
|
return self.src_flow["IN_PKTS"] + self.dest_flow["IN_PKTS"]
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description="Output a basic analysis of NetFlow data")
|
|
parser.add_argument("-f", "--file", dest="file", type=str, default=sys.stdin,
|
|
help="The file to analyze (defaults to stdin if not provided)")
|
|
parser.add_argument("-p", "--packets", dest="packets_threshold", type=int, default=10,
|
|
help="Number of packets representing the lower bound in connections to be processed")
|
|
parser.add_argument("-v", "--verbose", dest="verbose", action="store_true",
|
|
help="Enable verbose output.")
|
|
parser.add_argument("--match-host", dest="match_host", type=str, default=None,
|
|
help="Filter output by matching on the given host (matches source or destination)")
|
|
parser.add_argument("-n", "--no-dns", dest="no_dns", action="store_true", help="Disable DNS resolving of IP addresses")
|
|
args = parser.parse_args()
|
|
|
|
# Sanity check for IP address
|
|
if args.match_host:
|
|
try:
|
|
match_host = ipaddress.ip_address(args.match_host)
|
|
except ValueError as ex:
|
|
exit("IP address '{}' is neither IPv4 nor IPv6".format(args.match_host))
|
|
|
|
# Using a file and using stdin differ in their further usage for gzip.open
|
|
file = args.file
|
|
mode = "rb" # reading files
|
|
if file != sys.stdin and not os.path.exists(file):
|
|
exit("File {} does not exist!".format(file))
|
|
|
|
if file == sys.stdin:
|
|
file = sys.stdin.buffer
|
|
mode = "rt" # reading from stdin
|
|
|
|
data = {}
|
|
|
|
with gzip.open(file, mode) as gzipped:
|
|
# "for line in" lazy-loads all lines in the file
|
|
for line in gzipped:
|
|
entry = json.loads(line)
|
|
if len(entry.keys()) != 1:
|
|
logger.warning("Line \"{}\" does not have exactly one timestamp key.")
|
|
|
|
try:
|
|
ts = list(entry)[0] # timestamp from key
|
|
except KeyError:
|
|
logger.error("Saved line \"{}\" has no timestamp key!".format(line))
|
|
continue
|
|
|
|
data[ts] = entry[ts]
|
|
|
|
# Go through data and dissect every flow saved inside the dump
|
|
|
|
# The following dict holds flows which are looking for a peer, to analyze a duplex 'Connection'.
|
|
# For each flow, the destination address is looked up. If the peer is not in the list of pending peers,
|
|
# insert this flow, waiting for its peer. If found, take the waiting peer and create a Connection object.
|
|
pending = {}
|
|
skipped = 0
|
|
skipped_threshold = args.packets_threshold
|
|
|
|
first_line = True # print header line before first line
|
|
|
|
for key in sorted(data):
|
|
timestamp = datetime.fromtimestamp(float(key)).strftime("%Y-%m-%d %H:%M.%S")
|
|
client = data[key]["client"]
|
|
flows = data[key]["flows"]
|
|
|
|
for flow in sorted(flows, key=lambda x: x["FIRST_SWITCHED"]):
|
|
first_switched = flow["FIRST_SWITCHED"]
|
|
|
|
if first_switched - 1 in pending:
|
|
# TODO: handle fitting, yet mismatching (here: 1 second) pairs
|
|
pass
|
|
|
|
# Find the peer for this connection
|
|
if "IPV4_SRC_ADDR" in flow or flow.get("IP_PROTOCOL_VERSION") == 4:
|
|
local_peer = flow["IPV4_SRC_ADDR"]
|
|
remote_peer = flow["IPV4_DST_ADDR"]
|
|
else:
|
|
local_peer = flow["IPV6_SRC_ADDR"]
|
|
remote_peer = flow["IPV6_DST_ADDR"]
|
|
|
|
# Match on host filter passed in as argument
|
|
if args.match_host and not any([local_peer == args.match_host, remote_peer == args.match_host]):
|
|
# If a match_host is given but neither local_peer nor remote_peer match
|
|
continue
|
|
|
|
if first_switched not in pending:
|
|
pending[first_switched] = {}
|
|
|
|
# Match peers
|
|
if remote_peer in pending[first_switched]:
|
|
# The destination peer put itself into the pending dict, getting and removing entry
|
|
peer_flow = pending[first_switched].pop(remote_peer)
|
|
if len(pending[first_switched]) == 0:
|
|
del pending[first_switched]
|
|
else:
|
|
# Flow did not find a matching, pending peer - inserting itself
|
|
pending[first_switched][local_peer] = flow
|
|
continue
|
|
|
|
con = Connection(flow, peer_flow)
|
|
if con.total_packets < skipped_threshold:
|
|
skipped += 1
|
|
continue
|
|
|
|
if first_line:
|
|
print("{:19} | {:14} | {:8} | {:9} | {:7} | Involved hosts".format("Timestamp", "Service", "Size", "Duration", "Packets"))
|
|
print("-" * 100)
|
|
first_line = False
|
|
|
|
print("{timestamp} | {service:<14} | {size:8} | {duration:9} | {packets:7} | "
|
|
"Between {src_host} ({src}) and {dest_host} ({dest})" \
|
|
.format(timestamp=timestamp, service=con.service.upper(), src_host=con.hostnames.src, src=con.src,
|
|
dest_host=con.hostnames.dest, dest=con.dest, size=con.human_size, duration=con.human_duration,
|
|
packets=con.total_packets))
|
|
|
|
if skipped > 0:
|
|
print(f"{skipped} connections skipped, because they had less than {skipped_threshold} packets (this value can be set with the -p flag).")
|
|
|
|
if not args.verbose:
|
|
# Exit here if no debugging session was wanted
|
|
exit(0)
|
|
|
|
if len(pending) > 0:
|
|
print(f"\nThere are {len(pending)} first_switched entries left in the pending dict!")
|
|
all_noise = True
|
|
for first_switched, flows in sorted(pending.items(), key=lambda x: x[0]):
|
|
for peer, flow in flows.items():
|
|
# Ignore all pings, SYN scans and other noise to find only those peers left over which need a fix
|
|
if flow["IN_PKTS"] < skipped_threshold:
|
|
continue
|
|
all_noise = False
|
|
|
|
src = flow.get("IPV4_SRC_ADDR") or flow.get("IPV6_SRC_ADDR")
|
|
src_host = resolve_hostname(src)
|
|
src_text = f"{src}" if src == src_host else f"{src_host} ({src})"
|
|
dst = flow.get("IPV4_DST_ADDR") or flow.get("IPV6_DST_ADDR")
|
|
dst_host = resolve_hostname(dst)
|
|
dst_text = f"{dst}" if dst == dst_host else f"{dst_host} ({dst})"
|
|
proto = flow["PROTOCOL"]
|
|
size = flow["IN_BYTES"]
|
|
packets = flow["IN_PKTS"]
|
|
src_port = flow.get("L4_SRC_PORT", 0)
|
|
dst_port = flow.get("L4_DST_PORT", 0)
|
|
|
|
print(f"From {src_text}:{src_port} to {dst_text}:{dst_port} with "
|
|
f"proto {IP_PROTOCOLS.get(proto, 'UNKNOWN')} and size {human_size(size)}"
|
|
f" ({packets} packets)")
|
|
|
|
if all_noise:
|
|
print("They were all noise!")
|