pimwatch.py

#!/usr/bin/python3
import sys
import subprocess
import logging
import ipaddress
import re
import datetime
import random
import time
import argparse
from enum import Enum

'''
sample from tcpdump -n -vvv pim:
05:13:35.225804 IP (tos 0x0, ttl 1, id 920, offset 0, flags [none], proto PIM (103), length 54)
    10.8.1.1 > 224.0.0.13: PIMv2, length 34
	Join / Prune, cksum 0xccc9 (correct), upstream-neighbor: 10.8.1.2
	  1 group(s), holdtime: 3m30s
	    group #1: 233.44.15.9, joined sources: 1, pruned sources: 0
	      joined source #1: 129.174.131.51(S)
'''
#logger = logging.getLogger('pim2joinfile')
logger = None
upstream_interface = None
downstream_interface = None
joinfile = None

class PimNotice(object):
    def __init__(self, hold_time):
        self.joins = []
        self.prunes = []
        self.hold_time = hold_time

    def __repr__(self):
        joins = ','.join('%s->%s' % (s,g) for s,g in self.joins)
        prunes = ','.join('%s->%s' % (s,g) for s,g in self.prunes)
        return 'Joins[%s]; Prunes[%s]' % (joins, prunes)

def pimdump_lines(ifname):
    cmd = ['/usr/bin/stdbuf', '-oL', '-eL', '/usr/sbin/tcpdump', '-i', ifname, '-vvv', '-n', '-Qin', 'pim']
    popen = subprocess.Popen(cmd, stdout=subprocess.PIPE, universal_newlines=True)
    for stdout_line in iter(popen.stdout.readline, ""):
        yield stdout_line
    popen.stdout.close()
    return_code = popen.wait()
    if return_code:
        raise subprocess.CalledProcessError(return_code, cmd)

def sg_joinprune_watch(ifname):
    '''
    watches ifname with tcpdump, yields PimNotice objects.
    '''

    global logger
    # TBD: notice timeout from not getting an update before holdtime
    # (e.g. neighbor powers off abruptly)
    # Join / Prune, cksum 0xccc9 (correct), upstream-neighbor: 10.8.1.2
    # 1 group(s), holdtime: 3m30s
    # group #1: 233.44.15.9, joined sources: 0, pruned sources: 1
    # joined source #1: 129.174.131.51(S)
    # pruned source #1: 129.174.131.51(S)
    jp_re = re.compile(r'\s*Join / Prune, cksum .*, upstream-neighbor: (?P<nbr>\S+)')
    summ_re = re.compile(r'\s*(?P<grp_count>\d+) group\(s\), holdtime: (?P<hold>\S+)\s*')
    grp_re = re.compile(r'\s*group #(?P<idx>\d+): (?P<grp>[^,]+), joined sources: (?P<join_cnt>\d+), pruned sources: (?P<prune_cnt>\d+)\s*')
    src_re = re.compile(r'\s*(?P<jp>joined|pruned) source #(?P<idx>\d+): (?P<src>[^(]+)\(S\)\s*')

    reset_parse_state = True
    for line in pimdump_lines(ifname):
        if reset_parse_state:
            reset_parse_state = False
            cur_nbr = None
            cur_notice = None
            cur_grp = None
            in_pkt = False
            expect_groups, expect_joins, expect_prunes = 0,0,0

        line = line.strip()
        jp = jp_re.match(line)
        if jp:
            if in_pkt or expect_groups != 0 \
                    or expect_joins != 0 or expect_prunes != 0:
                logger.warning('got new packet start without finishing last:' +
                        ('in_pkt: %s' % in_pkt) + 
                        (',expect_groups: %s' % expect_groups) +
                        (',expect_joins: %s' % expect_joins) +
                        (',expect_prunes: %s' % expect_prunes))
            in_pkt = True
            expect_groups, expect_joins, expect_prunes = 0,0,0
            cur_nbr = jp.group('nbr')
            continue

        if not in_pkt:
            #logger.debug('ignoring: %s' % line)
            continue

        if expect_groups == 0:
            summ = summ_re.match(line)
            if not summ:
                logger.warning('unexpected line instead of summary: "%s"' % line)
                reset_parse_state = True
                continue
            grp_str = summ.group('grp_count')
            hold_str = summ.group('hold')
            try:
                expect_groups = int(grp_str)
            except ValueError as e:
                logger.error('%s parsing group line: "%s"' % (e, line))
                reset_parse_state = True
                continue

            if expect_groups <= 0:
                logger.warning('saw join/prune packet with groups==%d' % expect_groups)
                reset_parse_state = True
                continue

            # TBD: parse hold time: "3m30s". Be forgiving (with warnings)...
            hold_time = datetime.timedelta(seconds=210)
            cur_notice = PimNotice(hold_time)
            continue

        if not cur_notice:
            logger.error('cur_notice unset when expect_groups = %d' % expect_groups)
            reset_parse_state = True
            continue

        if expect_joins == 0 and expect_prunes == 0:
            grp = grp_re.match(line)
            if not grp:
                logger.warning('unexpected line instead of group desc: "%s"' % line)
                reset_parse_state = True
                continue

            cur_grp_str = grp.group('grp')
            join_src_str = grp.group('join_cnt')
            prune_src_str = grp.group('prune_cnt')
            try:
                cur_grp = ipaddress.ip_address(cur_grp_str)
                expect_joins = int(join_src_str)
                expect_prunes = int(prune_src_str)
            except ValueError as e:
                logger.error('%s: bad vals from line: "%s"\n  cur_grp=%s,joins=%s,prunes=%s' % (e, line, cur_grp_str, join_src_str, prune_src_str))
                reset_parse_state = True
                continue

            if expect_joins < 0 or expect_prunes < 0:
                logger.error('group desc with bad values: joins=%d, prunes=%d' % (expect_joins, expect_prunes))
                reset_parse_state = True
                continue

            if expect_joins == 0 and expect_prunes == 0:
                logger.warning('saw group with no sources joined or pruned')
            continue

        src = src_re.match(line)
        if not src:
            logger.warning('unexpected line instead of source desc: "%s"' % line)
            reset_parse_state = True
            continue

        jp_str = src.group('jp')
        src_str = src.group('src')
        try:
            src_ip = ipaddress.ip_address(src_str)
        except ValueError as e:
            logger.error('%s: bad src ip "%s" from line: "%s"' % (e, src_str, line))
            reset_parse_state = True
            continue

        if jp_str == "joined":
            if expect_joins <= 0:
                logger.error('got another join when out of expected joins in line "%s"' % line)
                reset_parse_state = True
                continue
            expect_joins -= 1
            cur_notice.joins.append((src_ip, cur_grp))
        elif jp_str == "pruned":
            if expect_prunes <= 0:
                logger.error('got another prune when out of expected prunes in line "%s"' % line)
                reset_parse_state = True
                continue
            expect_prunes -= 1
            cur_notice.prunes.append((src_ip, cur_grp))
        else:
            logger.error('unknown operation "%s" in line "%s"' % (jp_str, line))
            reset_parse_state = True
            continue

        if expect_joins == 0 and expect_prunes == 0:
            expect_groups -= 1

        if expect_groups == 0:
            logger.debug('cleanly finished join/prune packet')
            yield cur_notice
            reset_parse_state = True

class LiveSG(object):
    def __init__(self, source_ip, group_ip, expire_time):
        self.source = ipaddress.ip_address(source_ip)
        self.group = ipaddress.ip_address(group_ip)
        if not self.group.is_multicast:
            raise ValueError('non-multicast group for %s: "%s"' % (source, group))
        self.expire_time = expire_time

    def __repr__(self):
        return '%s->%s' % (self.source, self.group)

class ChannelManager(object):

    #dkr = '/snap/bin/docker'
    dkr = '/usr/bin/docker'

    def __init__(self, upstream, downstream):
        self.upstream = upstream
        self.downstream = downstream
        self.live_sgs = {}  # (src_ip,grp_ip) -> LiveSG

    def launch_sg_join(self, sg, expire_time):
        global logger
        source, group = sg
        assert(not sg in self.live_sgs)

        logger.info('launching join for %s' % (sg,))

        source = ipaddress.ip_address(source)
        group = ipaddress.ip_address(group)

        # the current way to do it is to add an smcroutectl route from the
        # upstream interface to the downstream interface for the group.
        # (the upstream join is done by the driad-ingest)
        # additionally, change the joinfile so that the upstream
        # driad-ingest (or cbacc) will handle it
        live_sg = LiveSG(source, group, expire_time)
        self.live_sgs[sg] = live_sg
        joinfile_out = '\n'.join([f'{s},{g}' for (s,g) in self.live_sgs.keys()])
        with open(joinfile, 'w') as f:
            print(joinfile_out, file=f)

        return live_sg

        '''
        # the next oldest way to do it was to tell frr to do the igmp join and rely on frr's pim forwarding
        cmd = ['/usr/bin/vtysh']
        in_stdio = "config term\ninterface eth1\nip igmp join %s %s\nexit\nexit\n" % (group, source)
        logger.info('running %s <<EOF\n%sEOF' % (' '.join(cmd), in_stdio))
        launch_p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE, universal_newlines=True)
        out, err = launch_p.communicate(input=in_stdio)
        '''

        '''
        # the old way was to launch another docker container just to
        # run iperf-ssm.  The new way is the above attempt to do an igmp
        # join inside vtysh
        nwname = mcast_nwname
        contname = 'pimwatch-join-%s-%s' % (source.exploded, group.exploded)
        imagename = 'grumpyoldtroll/iperf-ssm:latest'
        # TBD: support v6--iperf command is different
        # TBD: iperf-ssm is overkill and cruelly doomed to wait (or worse,
        # catch packets if they go to the right port). all i need is a
        # program that stays joined in the gateway's data network.
        cmd = [ChannelManager.dkr, 'run', '-d', '--network', nwname, '--rm', '--name', contname, imagename, '--server', '--udp', '--bind', str(group), '--source', str(source), '--interval', '1', '--len', '1500', '--interface', 'eth0']
        launch_p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
        out, err = launch_p.communicate(cmd)

        retcode = launch_p.wait()
        if retcode != 0:
            logger.error('return code %s from %s, out="%s", err="%s", failed joiner launch' % (retcode, cmd, out, err))
            return None
        live_sg = LiveSG(source, group, expire_time)
        self.live_sgs[sg] = live_sg
        return live_sg
        '''

    def stop_sg(self, sg):
        global logger, joinfile

        logger.info('stopping sg %s' % (sg,))

        source, group = sg.source, sg.group
        ip_sg = (sg.source, sg.group)
        if ip_sg not in self.live_sgs:
            logger.error('internal error: %s not in self.live_sgs in stop_sg' % (ip_sg,))
        else:
            del(self.live_sgs[ip_sg])

        joinfile_out = '\n'.join([f'{s},{g}' for (s,g) in self.live_sgs.keys()])
        with open(joinfile, 'w') as f:
            print(joinfile_out, file=f)


        '''
        source = ipaddress.ip_address(source)
        group = ipaddress.ip_address(group)
        cmd = ['/usr/bin/vtysh']
        in_stdio = "config term\ninterface eth1\nno ip igmp join %s %s\nexit\nexit\n" % (group, source)
        logger.info('running %s <<EOF\n%sEOF' % (' '.join(cmd), in_stdio))
        launch_p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE, universal_newlines=True)
        out, err = launch_p.communicate(input=in_stdio)
        '''

        '''
        cmd = [ChannelManager.dkr, 'container', 'stop', sg.contname]
        stopret = subprocess.run(cmd)
        logger.info('stopped container: %s' % (stopret))

        ip_sg = (sg.source, sg.group)
        if ip_sg not in self.live_sgs:
            logger.error('internal error: %s not in self.live_sgs in stop_sg' % (ip_sg,))
        else:
            del(self.live_sgs[ip_sg])
        '''

    def add_or_refresh_sg(self, sg, notice_time, hold_time):
        # TBD: grace period if we took too long, maybe not just notice+hold
        expire_time = notice_time + hold_time
        live_sg = self.live_sgs.get(sg)
        if live_sg:
            logger.info('live sg refreshed: %s' % (live_sg))
            live_sg.expire_time = expire_time
            return

        src_ip, grp_ip = sg
        logger.info('adding new sg: %s->%s' % (src_ip, grp_ip))

        while not live_sg:
            live_sg = self.launch_sg_join(sg, expire_time)
            if not live_sg:
                logger.error('failed to launch sg %s' % (sg,))

    def remove_sg(self, sg):
        live_sg = self.live_sgs.get(sg)
        if not live_sg:
            logger.info('ignored pruning non-live sg: %s' % (sg,))
            return
        logger.info('removing live sg: %s' % (live_sg))
        self.stop_sg(live_sg)

def get_logger(name, verbosity=0):
    log_level = logging.WARNING
    if verbosity > 1:
        log_level = logging.DEBUG
    elif verbosity > 0:
        log_level = logging.INFO

    # python logging wtf: logger.setLevel doesn't work the obvious way:
    # https://stackoverflow.com/a/59705351/3427357 (-jake 2020-07)
    handler = logging.StreamHandler()
    #formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    formatter = logging.Formatter('%(asctime)s[%(levelname)s]: %(message)s')
    handler.setFormatter(formatter)
    _logger = logging.getLogger(name)
    _logger.addHandler(handler)
    _logger.setLevel(log_level)
    return _logger

def main(args_in):
    global logger, joinfile, upstream_interface, downstream_interface

    parser = argparse.ArgumentParser(
        description='This is an implementation of an egress node in draft-jholland-mboned-mnat.')

    parser.add_argument('-v', '--verbose', action='count', default=0)
    parser.add_argument('-u', '--upstream', required=True,
            help='this is the upstream interface, for routes to be added based on downstream join/prunes')
    parser.add_argument('-d', '--downstream', required=True,
            help='this is the downstream interface with hopefully a connection to a pim network, monitored for joins and prunes')
    parser.add_argument('-j', '--joinfile', required=True,
        help='Name of the file inside the container where (S,G)s will be updated (needs absolute path within the container)')

    args = parser.parse_args(args_in[1:])

    logger = get_logger('pimwatch', args.verbose)

    if False: # TBD: cmd line flag?  running from service under journalctl this isn't needed but standalone it is. --jake 2020-09
        handler = RotatingFileHandler("pimwatch.log", maxBytes=10000000, backupCount=5)
        logger.addHandler(handler)

    joinfile = args.joinfile
    upstream_interface = args.upstream
    downstream_interface = args.downstream

    logger.info(f'started pimwatch, downstream={downstream_interface} upstream={upstream_interface}')
    channels = ChannelManager(upstream_interface, downstream_interface)
    '''
    if ret != 0:
        logger.error('prequisites check failed')
        exit(ret)
    '''
    with open(joinfile, 'w') as f:
        print('', file=f)

    for sgpkt in sg_joinprune_watch(downstream_interface):
        logger.debug('saw sg notice: "%s"', sgpkt)
        notice_time = datetime.datetime.now()
        for sg in sgpkt.joins:
            channels.add_or_refresh_sg(sg, notice_time, sgpkt.hold_time)
        for sg in sgpkt.prunes:
            channels.remove_sg(sg)
    return 0

if __name__=="__main__":
    retval = main(sys.argv)
    sys.exit(retval)