Step 2: Move test folder to root

[integration/test.git] / tools / fastbgp / play.py

"""Utility for playing generated BGP data to ODL.

It needs to be run with sudo-able user when you want to use ports below 1024
as --myip. This utility is used to avoid excessive waiting times which EXABGP
exhibits when used with huge router tables and also avoid the memory cost of
EXABGP in this type of scenario."""

# Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
#
# This program and the accompanying materials are made available under the
# terms of the Eclipse Public License v1.0 which accompanies this distribution,
# and is available at http://www.eclipse.org/legal/epl-v10.html

__author__ = "Vratko Polak"
__copyright__ = "Copyright(c) 2015, Cisco Systems, Inc."
__license__ = "Eclipse Public License v1.0"
__email__ = "vrpolak@cisco.com"


import argparse
import binascii
import ipaddr
import select
import socket
import time


def parse_arguments():
    """Use argparse to get arguments, return args object."""
    parser = argparse.ArgumentParser()
    # TODO: Should we use --argument-names-with-spaces?
    str_help = 'Autonomous System number use in the stream (current default as in ODL: 64496).'
    parser.add_argument('--asnumber', default=64496, type=int, help=str_help)
    # FIXME: We are acting as iBGP peer, we should mirror AS number from peer's open message.
    str_help = 'Amount of IP prefixes to generate. Negative number means "practically infinite".'
    parser.add_argument('--amount', default='1', type=int, help=str_help)
    str_help = 'The first IPv4 prefix to announce, given as numeric IPv4 address.'
    parser.add_argument('--firstprefix', default='8.0.1.0', type=ipaddr.IPv4Address, help=str_help)
    str_help = 'If present, this tool will be listening for connection, instead of initiating it.'
    parser.add_argument('--listen', action='store_true', help=str_help)
    str_help = 'Numeric IP Address to bind to and derive BGP ID from. Default value only suitable for listening.'
    parser.add_argument('--myip', default='0.0.0.0', type=ipaddr.IPv4Address, help=str_help)
    str_help = 'TCP port to bind to when listening or initiating connection. Default only suitable for initiating.'
    parser.add_argument('--myport', default='0', type=int, help=str_help)
    str_help = 'The IP of the next hop to be placed into the update messages.'
    parser.add_argument('--nexthop', default='192.0.2.1', type=ipaddr.IPv4Address, dest="nexthop", help=str_help)
    str_help = 'Numeric IP Address to try to connect to. Currently no effect in listening mode.'
    parser.add_argument('--peerip', default='127.0.0.2', type=ipaddr.IPv4Address, help=str_help)
    str_help = 'TCP port to try to connect to. No effect in listening mode.'
    parser.add_argument('--peerport', default='179', type=int, help=str_help)
    # TODO: The step between IP previxes is currently hardcoded to 16. Should we make it configurable?
    # Yes, the argument list above is sorted alphabetically.
    arguments = parser.parse_args()
    # TODO: Are sanity checks (such as asnumber>=0) required?
    return arguments


def establish_connection(arguments):
    """Establish connection according to arguments, return socket."""
    if arguments.listen:
        # print "DEBUG: connecting in the listening case."
        listening_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        listening_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        listening_socket.bind((str(arguments.myip), arguments.myport))  # bind need single tuple as argument
        listening_socket.listen(1)
        bgp_socket, _ = listening_socket.accept()
        # TODO: Verify client IP is cotroller IP.
        listening_socket.close()
    else:
        # print "DEBUG: connecting in the talking case."
        talking_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        talking_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        talking_socket.bind((str(arguments.myip), arguments.myport))  # bind to force specified address and port
        talking_socket.connect((str(arguments.peerip), arguments.peerport))  # socket does not spead ipaddr, hence str()
        bgp_socket = talking_socket
    print 'Connected to ODL.'
    return bgp_socket


def get_short_int_from_message(message, offset=16):
    """Extract 2-bytes number from packed string, default offset is for BGP message size."""
    high_byte_int = ord(message[offset])
    low_byte_int = ord(message[offset + 1])
    short_int = high_byte_int * 256 + low_byte_int
    return short_int


class MessageError(ValueError):
    """Value error with logging optimized for hexlified messages."""

    def __init__(self, text, message, *args):
        """Store and call super init for textual comment, store raw message which caused it."""
        self.text = text
        self.msg = message
        super(MessageError, self).__init__(text, message, *args)

    def __str__(self):
        """
        Generate human readable error message

        Concatenate text comment, colon with space
        and hexlified message. Use a placeholder string
        if the message turns out to be empty.
        """
        message = binascii.hexlify(self.msg)
        if message == "":
            message = "(empty message)"
        return self.text + ': ' + message


def read_open_message(bgp_socket):
    """Receive message, perform some validation, return the raw message."""
    msg_in = bgp_socket.recv(65535)  # TODO: Is smaller buffer size safe?
    # TODO: Is it possible for incoming open message to be split in more than one packet?
    # Some validation.
    if len(msg_in) < 37:  # 37 is minimal length of open message with 4-byte AS number.
        raise MessageError("Got something else than open with 4-byte AS number", msg_in)
    # TODO: We could check BGP marker, but it is defined only later; decide what to do.
    reported_length = get_short_int_from_message(msg_in)
    if len(msg_in) != reported_length:
        raise MessageError("Message length is not " + reported_length + " in message", msg_in)
    print "Open message received"
    return msg_in


class MessageGenerator(object):
    """Class with methods returning messages and state holding configuration data required to do it properly."""

    # TODO: Define bgp marker as class (constant) variable.
    def __init__(self, args):
        """Initialize data according to command-line args."""
        # Various auxiliary variables.
        # Hack: 4-byte AS number uses the same "int to packed" encoding as IPv4 addresses.
        asnumber_4bytes = ipaddr.v4_int_to_packed(args.asnumber)
        asnumber_2bytes = "\x5b\xa0"  # AS_TRANS value, 23456 decadic.
        if args.asnumber < 65536:  # AS number is mappable to 2 bytes
            asnumber_2bytes = asnumber_4bytes[2:4]
        # From now on, attribute docsrings are used.
        self.int_nextprefix = int(args.firstprefix)
        """Prefix IP address for next update message, as integer."""
        self.updates_to_send = args.amount
        """Number of update messages left to be sent."""
        # All information ready, so we can define messages. Mostly copied from play.py by Jozef Behran.
        # The following attributes are constant.
        self.bgp_marker = "\xFF" * 16
        """Every message starts with this, see rfc4271#section-4.1"""
        self.keepalive_message = self.bgp_marker + (
            "\x00\x13"  # Size
            "\x04"  # Type KEEPALIVE
        )
        """KeepAlive message, see rfc4271#section-4.4"""
        # TODO: Notification for hold timer expiration can be handy.
        self.eor_message = self.bgp_marker + (
            "\x00\x17"  # Size
            "\x02"  # Type (UPDATE)
            "\x00\x00"  # Withdrawn routes length (0)
            "\x00\x00"  # Total Path Attributes Length (0)
        )
        """End-of-RIB marker, see rfc4724#section-2"""
        self.update_message_without_prefix = self.bgp_marker + (
            "\x00\x30"  # Size
            "\x02"  # Type (UPDATE)
            "\x00\x00"  # Withdrawn routes length (0)
            "\x00\x14"  # Total Path Attributes Length (20)
            "\x40"  # Flags ("Well-Known")
            "\x01"  # Type (ORIGIN)
            "\x01"  # Length (1)
            "\x00"  # Origin: IGP
            "\x40"  # Flags ("Well-Known")
            "\x02"  # Type (AS_PATH)
            "\x06"  # Length (6)
            "\x02"  # AS segment type (AS_SEQUENCE)
            "\x01"  # AS segment length (1)
            + asnumber_4bytes +  # AS segment (4 bytes)
            "\x40"  # Flags ("Well-Known")
            "\x03"  # Type (NEXT_HOP)
            "\x04"  # Length (4)
            + args.nexthop.packed +  # IP address of the next hop (4 bytes)
            "\x1c"  # IPv4 prefix length, see RFC 4271, page 20. This tool uses Network Mask: 255.255.255.240
        )
        """The IP address prefix (4 bytes) has to be appended to complete Update message, see rfc4271#section-4.3."""
        self.open_message = self.bgp_marker + (
            "\x00\x2d"  # Size
            "\x01"  # Type (OPEN)
            "\x04"  # BGP Varsion (4)
            + asnumber_2bytes +  # My Autonomous System
            # FIXME: The following hold time is hardcoded separately. Compute from initial hold_time value.
            "\x00\xb4"  # Hold Time (180)
            + args.myip.packed +  # BGP Identifer
            "\x10"  # Optional parameters length
            "\x02"  # Param type ("Capability Ad")
            "\x06"  # Length (6 bytes)
            "\x01"  # Capability type (NLRI Unicast), see RFC 4760, secton 8
            "\x04"  # Capability value length
            "\x00\x01"  # AFI (Ipv4)
            "\x00"  # (reserved)
            "\x01"  # SAFI (Unicast)
            "\x02"  # Param type ("Capability Ad")
            "\x06"  # Length (6 bytes)
            "\x41"  # "32 bit AS Numbers Support" (see RFC 6793, section 3)
            "\x04"  # Capability value length
            + asnumber_4bytes  # My AS in 32 bit format
        )
        """Open message, see rfc4271#section-4.2"""
        # __init__ ends

    def compose_update_message(self):
        """Return update message, prepare next prefix, decrease amount without checking it."""
        prefix_packed = ipaddr.v4_int_to_packed(self.int_nextprefix)
        # print "DEBUG: prefix", self.int_nextprefix, "packed to", binascii.hexlify(prefix_packed)
        msg_out = self.update_message_without_prefix + prefix_packed
        self.int_nextprefix += 16  # Hardcoded, as open message specifies such netmask.
        self.updates_to_send -= 1
        return msg_out


class TimeTracker(object):
    """Class for tracking timers, both for my keepalives and peer's hold time."""

    def __init__(self, msg_in):
        """Initialize config, based on hardcoded defaults and open message from peer."""
        # Note: Relative time is always named timedelta, to stress that (non-delta) time is absolute.
        self.report_timedelta = 1.0  # In seconds. TODO: Configurable?
        """Upper bound for being stuck in the same state, we should at least report something before continuing."""
        # Negotiate the hold timer by taking the smaller of the 2 values (mine and the peer's).
        hold_timedelta = 240  # Not an attribute of self yet.
        # TODO: Make the default value configurable, default value could mirror what peer said.
        peer_hold_timedelta = get_short_int_from_message(msg_in, offset=22)
        if hold_timedelta > peer_hold_timedelta:
            hold_timedelta = peer_hold_timedelta
        if hold_timedelta != 0 and hold_timedelta < 3:
            raise ValueError("Invalid hold timedelta value: ", hold_timedelta)
        self.hold_timedelta = hold_timedelta  # only now the final value is visible from outside
        """If we do not hear from peer this long, we assume it has died."""
        self.keepalive_timedelta = int(hold_timedelta / 3.0)
        """Upper limit for duration between messages, to avoid being declared dead."""
        self.snapshot_time = time.time()  # The same as calling snapshot(), but also declares a field.
        """Sometimes we need to store time. This is where to get the value from afterwards."""
        self.peer_hold_time = self.snapshot_time + self.hold_timedelta  # time_keepalive may be too strict
        """At this time point, peer will be declared dead."""
        self.my_keepalive_time = None  # to be set later
        """At this point, we should be sending keepalive message."""

    def snapshot(self):
        """Store current time in instance data to use later."""
        self.snapshot_time = time.time()  # Read as time before something interesting was called.

    def reset_peer_hold_time(self):
        """Move hold time to future as peer has just proven it still lives."""
        self.peer_hold_time = time.time() + self.hold_timedelta

    # Some methods could rely on self.snapshot_time, but it is better to require user to provide it explicitly.
    def reset_my_keepalive_time(self, keepalive_time):
        """Move KA timer to future based on given time from before sending."""
        self.my_keepalive_time = keepalive_time + self.keepalive_timedelta

    def is_time_for_my_keepalive(self):
        if self.hold_timedelta == 0:
            return False
        return self.snapshot_time >= self.my_keepalive_time

    def get_next_event_time(self):
        if self.hold_timedelta == 0:
            return self.snapshot_time + 86400
        return min(self.my_keepalive_time, self.peer_hold_time)

    def check_peer_hold_time(self, snapshot_time):
        """Raise error if nothing was read from peer until specified time."""
        if self.hold_timedelta != 0:  # Hold time = 0 means keepalive checking off.
            if snapshot_time > self.peer_hold_time:  # time.time() may be too strict
                raise RuntimeError("Peer has overstepped the hold timer.")  # TODO: Include hold_timedelta?
                # TODO: Add notification sending (attempt). That means move to write tracker.


class ReadTracker(object):
    """Class for tracking read of mesages chunk by chunk and for idle waiting."""

    def __init__(self, bgp_socket, timer):
        """Set initial state."""
        # References to outside objects.
        self.socket = bgp_socket
        self.timer = timer
        # Really new fields.
        self.header_length = 18
        """BGP marker length plus length field length."""  # TODO: make it class (constant) attribute
        self.reading_header = True
        """Computation of where next chunk ends depends on whether we are beyond length field."""
        self.bytes_to_read = self.header_length
        """Countdown towards next size computation."""
        self.msg_in = ""
        """Incremental buffer for message under read."""

    def read_message_chunk(self):
        """Read up to one message, do not return anything."""
        # TODO: We also could return the whole message, but currently nobody cares.
        # We assume the socket is readable.
        chunk_message = self.socket.recv(self.bytes_to_read)
        self.msg_in += chunk_message
        self.bytes_to_read -= len(chunk_message)
        if not self.bytes_to_read:  # TODO: bytes_to_read < 0 is not possible, right?
            # Finished reading a logical block.
            if self.reading_header:
                # The logical block was a BGP header. Now we know size of message.
                self.reading_header = False
                self.bytes_to_read = get_short_int_from_message(self.msg_in)
            else:  # We have finished reading the body of the message.
                # Peer has just proven it is still alive.
                self.timer.reset_peer_hold_time()
                # TODO: Do we want to count received messages?
                # This version ignores the received message.
                # TODO: Should we do validation and exit on anything besides update or keepalive?
                # Prepare state for reading another message.
                self.msg_in = ""
                self.reading_header = True
                self.bytes_to_read = self.header_length
        # We should not act upon peer_hold_time if we are reading something right now.
        return

    def wait_for_read(self):
        """When we know there are no more updates to send, we use this to avoid busy-wait."""
        # First, compute time to first predictable state change (or report event)
        event_time = self.timer.get_next_event_time()
        wait_timedelta = event_time - time.time()  # snapshot_time would be imprecise
        if wait_timedelta < 0:
            # The program got around to waiting to an event in "very near
            # future" so late that it became a "past" event, thus tell
            # "select" to not wait at all. Passing negative timedelta to
            # select() would lead to either waiting forever (for -1) or
            # select.error("Invalid parameter") (for everything else).
            wait_timedelta = 0
        # And wait for event or something to read.
        select.select([self.socket], [], [self.socket], wait_timedelta)
        # Not checking anything, that will be done in next iteration.
        return


class WriteTracker(object):
    """Class tracking enqueueing messages and sending chunks of them."""

    def __init__(self, bgp_socket, generator, timer):
        """Set initial state."""
        # References to outside objects,
        self.socket = bgp_socket
        self.generator = generator
        self.timer = timer
        # Really new fields.
        # TODO: Would attribute docstrings add anything substantial?
        self.sending_message = False
        self.bytes_to_send = 0
        self.msg_out = ""

    def enqueue_message_for_sending(self, message):
        """Change write state to include the message."""
        self.msg_out += message
        self.bytes_to_send += len(message)
        self.sending_message = True

    def send_message_chunk_is_whole(self):
        """Perform actions related to sending (chunk of) message, return whether message was completed."""
        # We assume there is a msg_out to send and socket is writable.
        # print 'going to send', repr(self.msg_out)
        self.timer.snapshot()
        bytes_sent = self.socket.send(self.msg_out)
        self.msg_out = self.msg_out[bytes_sent:]  # Forget the part of message that was sent.
        self.bytes_to_send -= bytes_sent
        if not self.bytes_to_send:
            # TODO: Is it possible to hit negative bytes_to_send?
            self.sending_message = False
            # We should have reset hold timer on peer side.
            self.timer.reset_my_keepalive_time(self.timer.snapshot_time)
            # Which means the possible reason for not prioritizing reads is gone.
            return True
        return False


class StateTracker(object):
    """Main loop has state so complex it warrants this separate class."""

    def __init__(self, bgp_socket, generator, timer):
        """Set the initial state according to existing socket and generator."""
        # References to outside objects.
        self.socket = bgp_socket
        self.generator = generator
        self.timer = timer
        # Sub-trackers.
        self.reader = ReadTracker(bgp_socket, timer)
        self.writer = WriteTracker(bgp_socket, generator, timer)
        # Prioritization state.
        self.prioritize_writing = False
        """
        In general, we prioritize reading over writing. But in order to not get blocked by neverending reads,
        we should check whether we are not risking running out of holdtime.
        So in some situations, this field is set to True to attempt finishing sending a message,
        after which this field resets back to False.
        """
        # TODO: Alternative is to switch fairly between reading and writing (called round robin from now on).
        # Message counting is done in generator.

    def perform_one_loop_iteration(self):
        """Calculate priority, resolve all ifs, call appropriate method, return to caller to repeat."""
        self.timer.snapshot()
        if not self.prioritize_writing:
            if self.timer.is_time_for_my_keepalive():
                if not self.writer.sending_message:
                    # We need to schedule a keepalive ASAP.
                    self.writer.enqueue_message_for_sending(self.generator.keepalive_message)
                # We are sending a message now, so prioritize finishing it.
                self.prioritize_writing = True
        # Now we know what our priorities are, we have to check which actions are available.
        # socket.socket() returns three lists, we store them to list of lists.
        list_list = select.select([self.socket], [self.socket], [self.socket], self.timer.report_timedelta)
        read_list, write_list, except_list = list_list
        # Lists are unpacked, each is either [] or [self.socket], so we will test them as boolean.
        if except_list:
            raise RuntimeError("Exceptional state on socket", self.socket)
        # We will do either read or write.
        if not (self.prioritize_writing and write_list):
            # Either we have no reason to rush writes, or the socket is not writable.
            # We are focusing on reading here.
            if read_list:  # there is something to read indeed
                # In this case we want to read chunk of message and repeat the select,
                self.reader.read_message_chunk()
                return
            # We were focusing on reading, but nothing to read was there.
            # Good time to check peer for hold timer.
            self.timer.check_peer_hold_time(self.timer.snapshot_time)
            # Things are quiet on the read front, we can go on and attempt to write.
        if write_list:
            # Either we really want to reset peer's view of our hold timer, or there was nothing to read.
            if self.writer.sending_message:  # We were in the middle of sending a message.
                whole = self.writer.send_message_chunk_is_whole()  # Was it the end of a message?
                if self.prioritize_writing and whole:  # We were pressed to send something and we did it.
                    self.prioritize_writing = False  # We prioritize reading again.
                return
            # Finally, we can look if there is some update message for us to generate.
            if self.generator.updates_to_send:
                msg_out = self.generator.compose_update_message()
                if not self.generator.updates_to_send:  # We have just finished update generation, end-of-rib is due.
                    msg_out += self.generator.eor_message
                self.writer.enqueue_message_for_sending(msg_out)
                return  # Attempt for the actual sending will be done in next iteration.
            # Nothing to write anymore, except occasional keepalives.
            # To avoid busy loop, we do idle waiting here.
            self.reader.wait_for_read()
            return
        # We can neither read nor write.
        print 'Input and output both blocked for', self.timer.report_timedelta, 'seconds.'
        # FIXME: Are we sure select has been really waiting the whole period?
        return


def main():
    """Establish BGP connection and enter main loop for sending updates."""
    arguments = parse_arguments()
    bgp_socket = establish_connection(arguments)
    # Initial handshake phase. TODO: Can it be also moved to StateTracker?
    # Receive open message before sending anything.
    # FIXME: Add parameter to send default open message first, to work with "you first" peers.
    msg_in = read_open_message(bgp_socket)
    timer = TimeTracker(msg_in)
    generator = MessageGenerator(arguments)
    msg_out = generator.open_message
    # print "DEBUG: going to send open:", binascii.hexlify(msg_out)
    # Send our open message to the peer.
    bgp_socket.send(msg_out)
    # Wait for confirming keepalive.
    # TODO: Surely in just one packet?
    msg_in = bgp_socket.recv(19)  # Using exact keepalive length to not see possible updates.
    if msg_in != generator.keepalive_message:
        raise MessageError("Open not confirmed by keepalive, instead got", msg_in)
    timer.reset_peer_hold_time()
    # Send the keepalive to indicate the connection is accepted.
    timer.snapshot()  # Remember this time.
    bgp_socket.send(generator.keepalive_message)
    timer.reset_my_keepalive_time(timer.snapshot_time)  # Use the remembered time.
    # End of initial handshake phase.
    state = StateTracker(bgp_socket, generator, timer)
    while True:  # main reactor loop
        state.perform_one_loop_iteration()

if __name__ == "__main__":
    main()