parsePackets.py

#!/usr/bin/env python3.6

"""
Parse a capture file with probe requests and export as JSON.
Precondition: A capture split into multiple files must first be merged using
a tool such as mergecap.
"""

__author__ = "Richard Cosgrove"
__license__ = "MIT"

from collections import defaultdict
import hashlib
import os
from pprint import pprint
import pyshark
import re
import sys
from datetime import datetime
import urllib.request

# Local imports
from utilities import export_compressed_json

TOTAL_PACKETS = 3_049_699 # Hardcoded as packets are lazily evaluated.
EPOCH_DATETIME = datetime(2013, 3, 28, 17, 53, 46, 769103)
OUI_LIST_URL = "https://code.wireshark.org/review/gitweb?p=wireshark.git;a=blob_plain;f=manuf"

class Fingerprint(object):
    """Probe fingerprint generated by information element values."""

    def __init__(self, information_elements):
        self.values = self._extract(information_elements)

    def __repr__(self):
        """SHA256 hexdigest of specific information elements.""" 
        return hashlib.sha256(str(self.values).encode("UTF-8")).hexdigest()

    def _extract(self, ie):
        """Return tuple of specific information elements."""
        ordered_tag_fields = ie.get_field("wlan_mgt.tag.number").all_fields
        ordered_numbers = [field.int_value for field in ordered_tag_fields]

        return (
            ie.get_field_value("wlan_mgt.vs.ht.capabilities"), # HT capabilities info
            tuple(ordered_numbers), # Ordered list of tags numbers
            ie.get_field_value("wlan_mgt.extcap"), # Extended capabilities
            ie.get_field_value("wlan_mgt.vs.ht.ampduparam"), # HT A-MPDU parameters

            # HT MCS set bitmask
            ie.get_field_value("wlan_mgt.ht.mcsset.rxbitmask.0to7"),
            ie.get_field_value("wlan_mgt.ht.mcsset.rxbitmask.16to23"),
            ie.get_field_value("wlan_mgt.ht.mcsset.rxbitmask.24to31"),
            ie.get_field_value("wlan_mgt.ht.mcsset.rxbitmask.32"),
            ie.get_field_value("wlan_mgt.ht.mcsset.rxbitmask.33to38"),
            ie.get_field_value("wlan_mgt.ht.mcsset.rxbitmask.39to52"),
            ie.get_field_value("wlan_mgt.ht.mcsset.rxbitmask.53to76"),
            ie.get_field_value("wlan_mgt.ht.mcsset.rxbitmask.8to15"),

            ie.get_field_value("wlan_mgt.supported_rates"), # Supported Rates
            ie.get_field_value("wlan_mgt.extended_supported_rates"), # Extended supported rates

            ie.get_field_value("wps.uuid_e"), # WPS UUID
            ie.get_field_value("wlan_mgt.htex.capabilities"), # HT extended capabilities
            ie.get_field_value("wlan_mgt.vs.txbf"), # HT TxBeam Forming Cap.
            ie.get_field_value("wlan_mgt.vs.asel"), # HT Antenna Selection Cap.
            )

def extract_ssid(val):
    """Extract SSID identifier from probe info."""
    match = re.search("(?<=SSID_)\w+", val)
    if match:
        return match.group(0)
    else:
        return 0

def parse_packet_with_fingerprint(packet):
    """Parse packet into a probe dict with fingerprint."""
    dt_obj = packet.sniff_time
    return packet.wlan.sa, {
        "timestamp": (dt_obj - EPOCH_DATETIME).total_seconds(),
        "ssid": extract_ssid(packet.wlan_mgt.get_field_value("wlan_mgt.ssid")),
        "fingerprint": str(Fingerprint(packet.wlan_mgt))
    }


def parse_packet(packet):
    """Parse packet into a probe dict."""
    return packet.source, {
        "timestamp": packet.time,
        "ssid": extract_ssid(packet.info)
    }

def parse_probes(capture_file, oui_list, only_summaries=True):
    """Parse a capture file of probes into a dictionary of probes."""
    mac_to_probe = defaultdict(list)

    # Do not keep packets to stop memory leak.
    capture = pyshark.FileCapture(capture_file, only_summaries=only_summaries, keep_packets=False)

    num_of_packets = 0
    filtered_packets = 0
    deferred_errors = []

    for packet in capture:
        num_of_packets += 1

        # # Debugging
        # if num_of_packets > 20000:
        #     break

        if num_of_packets % 10000 == 0:
            i = round(num_of_packets / float(TOTAL_PACKETS) * 20)
            sys.stdout.write("\r")
            sys.stdout.write("[%-20s] %d%% (%i)" % ("="*i, 5*i, num_of_packets))
            sys.stdout.flush()

        try:
            if only_summaries:
                mac, probe_request = parse_packet(packet)
            else:
                mac, probe_request = parse_packet_with_fingerprint(packet)

        except Exception as e:
            """Not a valid probe request."""
            deferred_errors.append(e)
            continue

        # Filter any MAC that does not exist in the OUI list
        # as this may be already randomised.
        if mac[:8].upper() in oui_list:
            mac_to_probe[mac].append(probe_request)
        else:
            filtered_packets+=1

    sys.stdout.write("\n")
    print("Processed %i packets with %i errors and %i filtered packets." % (
        num_of_packets, len(deferred_errors), filtered_packets))
    pprint(deferred_errors)
    return mac_to_probe

def get_oui_list(file_path="int/oui.txt"):
    """To be used to filter out already randomised addresses.
    Try to download if not found locally
    """ 
    if not os.path.exists(file_path):
        print("OUI List not found - downloading...")
        os.makedirs(os.path.dirname(file_path), exist_ok=True)
        with open(file_path, "wb") as f, urllib.request.urlopen(OUI_LIST_URL) as url:
            contents = url.read()
            f.write(contents)

    oui_list = set()
    with open(file_path, "r") as f:
        for line in f:
            line_split = line.split()
            
            if not line_split:
                continue

            # E.g. 12:34:56
            match = re.search(r"^([0-9A-Fa-f]{2}[:-]){2}([0-9A-Fa-f]{2})$", line_split[0])
            if match:
                oui_list.add(line_split[0].upper())

    assert(oui_list) # Sanity check
    return oui_list

def main(include_fingerprints=False):
    """Pcap file to be exported as JSON.
    Including fingerprints signficantly slows down the parsing
    as additional XML files have to be processed via TShark.
    """
    oui_list = get_oui_list()
    script_dir = os.path.dirname(__file__)
    capture_file_path = os.path.join(script_dir, "universitycaps.pcap")
    
    # Parse probes
    only_summaries = (not include_fingerprints)
    mac_to_probe = parse_probes(capture_file_path, oui_list, only_summaries)

    # Export
    if include_fingerprints:
        export_compressed_json(mac_to_probe, "int/mac_to_probe_inc_fingerprint.json.gz")
    else:
        export_compressed_json(mac_to_probe, "int/mac_to_probe.json.gz")
    
if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--fingerprint", help="Include fingerprint (much slower).",
                        action="store_true")
    args = parser.parse_args()
    main(include_fingerprints=args.fingerprint)