See the latest slide deck about Flowtuples presented in 2021.

Overview

FlowTuple data is an aggregated representation of traffic captured at the network telescope that enables a more efficient processing and analysis for many research use cases that do not need access to the full packet contents. In addition, the FlowTuple format includes meta-data associated with the corresponding source IP address of each FlowTuple, such as IP geolocation, IP2ASN, or whether we categorize the source address as spoofed.

Definition

A STARDUST flowtuple has a slightly different definition compared to the typical flowtuple definition (i.e., the 5-tuple of source IP, dest IP, source port, dest port and transport protocol). The STARDUST flowtuple takes into account additional header fields (e.g., TCP flags) because of the characterstics of various componenets found in unsolicited one-way traffic and the most common types of analyses that research perform.

Properties

FlowTuples are stored in the Avro data format. A single flow entry is distinguished by the timestamp, source IP address, destination IP network (within a /24 subnet mask), destination port, and protocol.

Data fields in a FlowTuple record the following:

1. uniq_<data field>: the number of unique values seen for the data field
2. common_<data field>: an array containing any data values which occur frequently for the flow
3. common_<data field>_freqs: an array containing the frequency at which the values shown in the preceding array occurred.
   • A note about required frequencies: To be considered a “frequently occurring” value, the value must be seen in at least 20% of packets that are matched to a given flow. For flows where the packet count is low, the ratio is increased to compensate for the low sample size.

   • Total Flow Packets	Minimum Ratio
     1 - 4	1.0
     5 - 6	0.5
     7 - 14	0.33
     15+	0.2

Contents of a FlowTuple v4 Record

User Guide

FlowTuple data is stored in Openstack Swift using the Apache Avro data format. Specifically, data for each year is saved in a separate Swift container:

• telescope-ucsdnt-avro-flowtuple-v4-2021
• telescope-ucsdnt-avro-flowtuple-v4-2020
• telescope-ucsdnt-avro-flowtuple-v4-...

Users can read the Avro files directly using existing Avro-compatible tools or libraries, if desired, but the best way to work with the flowtuple data is by writing Python scripts using PyAvro-STARDUST or the Pyspark STARDUST API.

Note: Flowtuple File Name Formats

• August 2008 to July 2020 : v2_5min_<timestamp>.ft4.avro
• August 2020 to Aug 2021 : v3_5min_<timestamp>.ft4.avro
• September 2021 to Present : ucsd-nt.<timestamp>.flowtuple-v4.avro

Processing FlowTuples with PyAvro-STARDUST

You can write Python analysis code that will process the flowtuple Avro files in the STARDUST object store directly using the PyAvro-STARDUST module. This module provides a simple interface that aims to be faster and easier to use than existing Python Avro libraries (such as fastavro).

Documentation on how to install and use PyAvro-STARDUST can be found in the GitHub repository.

An example script for processing flowtuple v4 data can be found here.

Processing FlowTuples with the STARDUST Pyspark API

For larger scale processing, we have developed an API for writing Python scripts that can run flowtuple analysis jobs on an Apache Spark cluster. The API allows you to specify a time period that you are interested in, then run methods that can perform common filtering or data processing tasks (e.g. filtering flowtuples by source address prefix, or finding the top 10 more frequent values for a metric.

You can also run raw Spark SQL queries directly against a set of flowtuples, if the exact analysis that you want to perform is not implemented as an API method.

This API is still a work in progress, but our implementation thus far is available here.

Some example flowtuple analysis code written using the STARDUST Pyspark API is given below:

from stardust import StardustPysparkHelper

# creating an instance of a helper
sd = StardustPysparkHelper("telescope-ucsdnt-avro-flowtuple-v4-2021",
        "ucsd-nt", "v3_5min")

# starting a spark session with the name "test", using 4 partitions (CPUs)
sd.startSparkSession("test", 4)

# getting flowtuple records from a specific time range
range_recs = sd.getFlowtuplesByTimeRange(1612130100, 1612133700)


# filtering a set of flowtuples based on a source IP prefix
# NOTE: the prefix must be a unicode string
prefix_df = sd.filterFlowtuplesByPrefix(lastday_recs, u"1.0.0.0/8")

# get the number of flows that matched our filter prefix
print(prefix_df.count())

# print the first 20 flows that matched our query
# ALWAYS use collect() to convert a data frame into a list
# of rows if you want to inspect the results
results = prefix_df.limit(20).collect()
for r in results:
# this prints the raw row object, which is fine for debugging but
# you'll want to do some proper formatting for usable output
    print(r)
print()


# finding flowtuples that match a set of filtering criteria
# base will contain flowtuples with a Russian source IP,
# sect will contain flowtuples with a Russian source IP that
# use multiple TTLs.
sect, base = sd.createFlowtupleIntersection(range_recs,
        ["netacq_country == 'RU'", "uniq_ttls > 1"])

# generate aggregated report-style time series data for the intersection
# we just generated
sect_report = sd.generateReportOutputFromFlowtuples(sect, "example",
        "ftintersect", "RU_multiTTL")

# again, use collect() to convert our dataframe in rows that we can
# write as output
sect_results = sect_report.collect()
base_results = base_report.collect()

# dump the time series to stdout as pairs of Row objects
# this is a bit lazy -- if you were doing this for real,
# you would want better error checking and output formatting
for i in range(0, sect_report.count()):
    print(sect_results[i], base_results[i])


# get all flowtuples where the TTL falls between 20 and 30 OR 70 and 80
q_res = sd.filterFlowtuplesByCommonValue(range_recs, "common_ttls",
        [(20, 30), (70, 80)])

# show the top 10 destination ports (by flow count) for Russian source IPs
# setting the last parameter to True will also include an "Other" category
topn = sd.getTopValuesByFlowCount(base, "netacq_country", 10, True)
for k,v in topn.items():
    print(v)

# there are also other handy methods, see the module documentation
# (can be accessed using help())