<commit_before>#!/usr/bin/env python
#  Author:
#  <NAME> (<EMAIL>)
#  <NAME> (Networked Systems Group ETH Zurich)
#  <NAME> (Princeton)

import os
import json
import sys
import argparse
import time

from multiprocessing.connection import Listener, Client

import Queue
from threading import Thread,Event
from server import server as Server
from core import XRS, XRSPeer



LOG = False

class route_server():

    def __init__(self, config_file):
        print "Initialize the Route Server"

        # Init the Route Server
        self.server = None
        self.ah_socket = None
        self.participants = {}

        # Several useful mappings
        self.port_2_participant = {}
        self.participant_2_port = {}
        self.portip_2_participant = {}
        self.participant_2_portip = {}
        self.portmac_2_participant = {}
        self.participant_2_portmac = {}
        self.asn_2_participant = {}
        self.participant_2_asn = {}

        ## Parse Config
        self.parse_config(config_file)

        # Initialize a XRS Server
        self.server = Server()
        self.run = True

        """
        Start the announcement Listener which will receive announcements
        from participants's controller and put them in XRS's sender queue.
        """
        self.set_announcement_handler()


    def start(self):
        print "Starting the Server to handle incoming BGP Updates"
        self.server.start()

        while self.run:
            # get BGP messages from ExaBGP via stdin
            try:
                route = self.server.receiver_queue.get(True, 1)
                route = json.loads(route)

                # Received BGP route advertisement from ExaBGP
                for id, peer in self.participants.iteritems():
                    # Apply the filtering logic
                    advertiser_ip = route['neighbor']
                    advertise_id = self.portip_2_participant[advertiser_ip]
                    if id in self.participants[advertise_id].peers_out and advertise_id in self.participants[id].peers_in:
                        # Now send this route to participant `id`'s controller'
                        self.send_update(id, route)

            except Queue.Empty:
                if LOG:
                    print "Empty Queue"


    def parse_config(self, config_file):
        # loading config file
        config = json.load(open(config_file, 'r'))

        self.ah_socket = tuple(config["Route Server"]["AH_SOCKET"])

        for participant_name in config["Participants"]:
            participant = config["Participants"][participant_name]

            # adding asn and mappings
            asn = participant["ASN"]
            self.asn_2_participant[participant["ASN"]] = int(participant_name)
            self.participant_2_asn[int(participant_name)] = participant["ASN"]

            self.participant_2_port[int(participant_name)] = []
            self.participant_2_portip[int(participant_name)] = []
            self.participant_2_portmac[int(participant_name)] = []

            for i in range(0, len(participant["Ports"])):
                self.port_2_participant[participant["Ports"][i]['Id']] = int(participant_name)
                self.portip_2_participant[participant["Ports"][i]['IP']] = int(participant_name)
                self.portmac_2_participant[participant["Ports"][i]['MAC']] = int(participant_name)
                self.participant_2_port[int(participant_name)].append(participant["Ports"][i]['Id'])
                self.participant_2_portip[int(participant_name)].append(participant["Ports"][i]['IP'])
                self.participant_2_portmac[int(participant_name)].append(participant["Ports"][i]['MAC'])

            # adding ports and mappings
            ports = [{"ID": participant["Ports"][i]['Id'],
                         "MAC": participant["Ports"][i]['MAC'],
                         "IP": participant["Ports"][i]['IP']}
                         for i in range(0, len(participant["Ports"]))]

            peers_out = [peer for peer in participant["Peers"]]
            # TODO: Make sure this is not an insane assumption
            peers_in = peers_out

            eh_socket = tuple(participant["EH_SOCKET"])

            # create peer and add it to the route server environment
            self.participants[int(participant_name)] = XRSPeer(asn, ports, peers_in, peers_out, eh_socket)


    def set_announcement_handler(self):
        '''Start the listener socket for BGP Announcements'''
        #print "set_announcement_handler() called"
        self.listener_eh = Listener(self.ah_socket, authkey=None)
        ps_thread = Thread(target=self.start_ah)
        ps_thread.daemon = True
        ps_thread.start()


    def start_ah(self):
        '''Announcement Handler '''
        print "Announcement Handler started "
        while True:
            conn_ah = self.listener_eh.accept()
            tmp = conn.recv()
            announcement = json.loads(tmp)
            self.server.sender_queue.put(announcement)
            reply = "Announcement processed"
            conn_ah.send(reply)
            conn_ah.close()


    def send_update(self, id, route):
        # TODO: Explore what is better, persistent client sockets or
        # new socket for each BGP update
        "Send this BGP route to participant id's controller"
        conn = Client(self.participants[id].eh_socket, authkey = None)
        data = {}
        data['bgp'] = route
        conn.send(json.dumps(data))
        recv = conn.recv()
        conn.close()


    def stop(self):
        self.run = False
        self.sdx_ap.stop()
        self.ap_thread.join()


def main(argv):
    # locate config file
    base_path = os.path.abspath(os.path.join(os.path.dirname(os.path.realpath(__file__)),"..","examples",args.dir,"controller","sdx_config"))
    config_file = os.path.join(base_path, "sdx_global.cfg")

    # start route server
    sdx_rs = route_server(config_file)
    rs_thread = Thread(target=sdx_rs.start)
    rs_thread.daemon = True
    rs_thread.start()

    while rs_thread.is_alive():
        try:
            rs_thread.join(1)<commit_msg>partial bugfix<commit_after>#!/usr/bin/env python
#  Author:
#  <NAME> (muham<EMAIL>)
#  <NAME> (Networked Systems Group ETH Zurich)
#  <NAME> (Princeton)

import os
import json
import sys
import argparse
import time

from multiprocessing.connection import Listener, Client

import Queue
from threading import Thread,Event
from server import server as Server
from core import XRS, XRSPeer



LOG = False

class route_server():

    def __init__(self, config_file):
        print "Initialize the Route Server"

        # Init the Route Server
        self.server = None
        self.ah_socket = None
        self.participants = {}

        # Several useful mappings
        self.port_2_participant = {}
        self.participant_2_port = {}
        self.portip_2_participant = {}
        self.participant_2_portip = {}
        self.portmac_2_participant = {}
        self.participant_2_portmac = {}
        self.asn_2_participant = {}
        self.participant_2_asn = {}

        ## Parse Config
        self.parse_config(config_file)

        # Initialize a XRS Server
        self.server = Server()
        self.run = True

        """
        Start the announcement Listener which will receive announcements
        from participants's controller and put them in XRS's sender queue.
        """
        self.set_announcement_handler()


    def start(self):
        print "Starting the Server to handle incoming BGP Updates"
        self.server.start()

        while self.run:
            # get BGP messages from ExaBGP via stdin
            try:
                route = self.server.receiver_queue.get(True, 1)
                route = json.loads(route)

                # Received BGP route advertisement from ExaBGP
                for id, peer in self.participants.iteritems():
                    # Apply the filtering logic
                    advertiser_ip = route['neighbor']['ip']
                    print "Advertiser IP:", advertiser_ip
                    advertise_id = self.portip_2_participant[advertiser_ip]
                    if id in self.participants[advertise_id].peers_out and advertise_id in self.participants[id].peers_in:
                        # Now send this route to participant `id`'s controller'
                        self.send_update(id, route)

            except Queue.Empty:
                if LOG:
                    print "Empty Queue"


    def parse_config(self, config_file):
        # loading config file
        config = json.load(open(config_file, 'r'))

        self.ah_socket = tuple(config["Route Server"]["AH_SOCKET"])

        for participant_name in config["Participants"]:
            participant = config["Participants"][participant_name]

            # adding asn and mappings
            asn = participant["ASN"]
            self.asn_2_participant[participant["ASN"]] = int(participant_name)
            self.participant_2_asn[int(participant_name)] = participant["ASN"]

            self.participant_2_port[int(participant_name)] = []
            self.participant_2_portip[int(participant_name)] = []
            self.participant_2_portmac[int(participant_name)] = []

            for i in range(0, len(participant["Ports"])):
                self.port_2_participant[participant["Ports"][i]['Id']] = int(participant_name)
                self.portip_2_participant[participant["Ports"][i]['IP']] = int(participant_name)
                self.portmac_2_participant[participant["Ports"][i]['MAC']] = int(participant_name)
                self.participant_2_port[int(participant_name)].append(participant["Ports"][i]['Id'])
                self.participant_2_portip[int(participant_name)].append(participant["Ports"][i]['IP'])
                self.participant_2_portmac[int(participant_name)].append(participant["Ports"][i]['MAC'])

            # adding ports and mappings
            ports = [{"ID": participant["Ports"][i]['Id'],
                         "MAC": participant["Ports"][i]['MAC'],
                         "IP": participant["Ports"][i]['IP']}
                         for i in range(0, len(participant["Ports"]))]

            peers_out = [peer for peer in participant["Peers"]]
            # TODO: Make sure this is not an insane assumption
            peers_in = peers_out

            temp = participant["EH_SOCKET"]
            eh_socket = (str(temp[0]), int(temp[1]))

            # create peer and add it to the route server environment
            self.participants[int(participant_name)] = XRSPeer(asn, ports, peers_in, peers_out, eh_socket)


    def set_announcement_handler(self):
        '''Start the listener socket for BGP Announcements'''
        #print "set_announcement_handler() called"
        self.listener_eh = Listener(self.ah_socket, authkey=None)
        ps_thread = Thread(target=self.start_ah)
        ps_thread.daemon = True
        ps_thread.start()


    def start_ah(self):
        '''Announcement Handler '''
        print "Announcement Handler started "
        while True:
            conn_ah = self.listener_eh.accept()
            tmp = conn.recv()
            announcement = json.loads(tmp)
            self.server.sender_queue.put(announcement)
            reply = "Announcement processed"
            conn_ah.send(reply)
            conn_ah.close()


    def send_update(self, id, route):
        # TODO: Explore what is better, persistent client sockets or
        # new socket for each BGP update
        "Send this BGP route to participant id's controller"
        print "EH Socket:", self.participants[id].eh_socket
        conn = Client(tuple(self.participants[id].eh_socket), authkey = None)
        data = {}
        data['bgp'] = route
        conn.send(json.dumps(data))
        recv = conn.recv()
        conn.close()


    def stop(self):
        self.run = False
        self.sdx_ap.stop()
        self.ap_thread.join()


def main(argv):
    # locate config file
    base_path = os.path.abspath(os.path.join(os.path.dirname(os.path.realpath(__file__)),"..","examples",args.dir,"controller","sdx_config"))
    config_file = os.path.join(base_path, "sdx_global.cfg")

    # start route server
    sdx_rs = route_server(config_file)
    rs_thread = Thread(target=sdx_rs.start)
    rs_thread.daemon = True
    rs_thread.start()

    while rs_thread.is_alive():
        try:
            rs_thread.join(1)
{'fraction_non_alphanumeric': 0.10663240725871206, 'fraction_numerical': 0.004014774369680424, 'mean_word_length': 2.3355650776647026, 'pattern_counts': {'":': 6, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '192302', 'n_tokens_mistral': 3822, 'n_tokens_neox': 3563, 'n_words': 779}