jc/jc/parsers/netstat_linux.py

"""jc - JSON CLI output utility Linux netstat Parser"""
import string


def normalize_headers(header):
    header = header.lower()
    header = header.replace('local address', 'local_address')
    header = header.replace('foreign address', 'foreign_address')
    header = header.replace('pid/program name', 'program_name')
    header = header.replace('security context', 'security_context')
    header = header.replace('i-node', 'inode')
    header = header.replace('-', '_')

    return header


def parse_network(headers, entry):
    # Count words in header
    # if len of line is one less than len of header, then insert None in field 5
    entry = entry.split(maxsplit=len(headers) - 1)

    if len(entry) == len(headers) - 1:
        entry.insert(5, None)

    output_line = dict(zip(headers, entry))
    output_line['kind'] = 'network'

    return output_line


def parse_socket(header_text, headers, entry):
    # get the column # of first letter of "state"
    state_col = header_text.find('state')
    # get the program name column area
    pn_start = header_text.find('program_name')
    pn_end = header_text.find('path') - 1

    # remove [ and ] from each line
    entry = entry.replace('[ ]', '---')
    entry = entry.replace('[', ' ').replace(']', ' ')

    # find program_name column area and substitute spaces with \u2063 there
    old_pn = entry[pn_start:pn_end]
    new_pn = old_pn.replace(' ', '\u2063')
    entry = entry.replace(old_pn, new_pn)

    entry_list = entry.split(maxsplit=len(headers) - 1)
    # check column # to see if state column is populated
    if entry[state_col] in string.whitespace:
        entry_list.insert(4, None)

    output_line = dict(zip(headers, entry_list))
    output_line['kind'] = 'socket'

    # fix program_name field to turn \u2063 back to spaces
    if 'program_name' in output_line:
        if output_line['program_name']:
            old_d_pn = output_line['program_name']
            new_d_pn = old_d_pn.replace('\u2063', ' ')
            output_line['program_name'] = new_d_pn

    return output_line


def parse_post(raw_data):
    # clean up trailing whitespace on each item in each entry
    # flags --- = null
    # program_name - = null
    # split pid and program name and ip addresses and ports
    # create network and transport protocol fields

    for entry in raw_data:
        for item in entry:
            try:
                entry[item] = entry[item].rstrip()
            except (AttributeError):
                # skips trying to rstrip Null entries
                pass

        if 'flags' in entry:
            if entry['flags'] == '---':
                entry['flags'] = None

        if 'program_name' in entry:
            entry['program_name'] = entry['program_name'].strip()
            if entry['program_name'] == '-':
                entry['program_name'] = None

            if entry['program_name']:
                pid = entry['program_name'].split('/', maxsplit=1)[0]
                name = entry['program_name'].split('/', maxsplit=1)[1]
                entry['pid'] = pid
                entry['program_name'] = name

        if 'local_address' in entry:
            if entry['local_address']:
                ladd = entry['local_address'].rsplit(':', maxsplit=1)[0]
                lport = entry['local_address'].rsplit(':', maxsplit=1)[1]
                entry['local_address'] = ladd
                entry['local_port'] = lport

        if 'foreign_address' in entry:
            if entry['foreign_address']:
                fadd = entry['foreign_address'].rsplit(':', maxsplit=1)[0]
                fport = entry['foreign_address'].rsplit(':', maxsplit=1)[1]
                entry['foreign_address'] = fadd
                entry['foreign_port'] = fport

        if 'proto' in entry and 'kind' in entry:
            if entry['kind'] == 'network':
                if 'tcp' in entry['proto']:
                    entry['transport_protocol'] = 'tcp'
                elif 'udp' in entry['proto']:
                    entry['transport_protocol'] = 'udp'
                else:
                    entry['transport_protocol'] = None

                if '6' in entry['proto']:
                    entry['network_protocol'] = 'ipv6'
                else:
                    entry['network_protocol'] = 'ipv4'

    return raw_data


def parse(cleandata):
    """
    Main text parsing function for OSX netstat

    Parameters:

        cleandata:   (string)  text data to parse

    Returns:

        List of dictionaries. Raw structured data.
    """
    raw_output = []
    network = False
    socket = False
    bluetooth = False
    headers = None

    for line in cleandata:

        if line.startswith('Active Internet'):
            network = True
            socket = False
            bluetooth = False
            continue

        if line.startswith('Active UNIX'):
            network = False
            socket = True
            bluetooth = False
            continue

        if line.startswith('Active Bluetooth'):
            network = False
            socket = False
            bluetooth = True
            continue

        if line.startswith('Proto'):
            header_text = normalize_headers(line)
            headers = header_text.split()
            continue

        if network:
            raw_output.append(parse_network(headers, line))
            continue

        if socket:
            raw_output.append(parse_socket(header_text, headers, line))
            continue

        if bluetooth:
            # not implemented
            continue

    return parse_post(raw_output)