1 """Utility for playing generated BGP data to ODL.
3 It needs to be run with sudo-able user when you want to use ports below 1024
4 as --myip. This utility is used to avoid excessive waiting times which EXABGP
5 exhibits when used with huge router tables and also avoid the memory cost of
6 EXABGP in this type of scenario."""
8 # Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
10 # This program and the accompanying materials are made available under the
11 # terms of the Eclipse Public License v1.0 which accompanies this distribution,
12 # and is available at http://www.eclipse.org/legal/epl-v10.html
14 __author__ = "Vratko Polak"
15 __copyright__ = "Copyright(c) 2015, Cisco Systems, Inc."
16 __license__ = "Eclipse Public License v1.0"
17 __email__ = "vrpolak@cisco.com"
28 def parse_arguments():
29 """Use argparse to get arguments, return args object."""
30 parser = argparse.ArgumentParser()
31 # TODO: Should we use --argument-names-with-spaces?
32 str_help = 'Autonomous System number use in the stream (current default as in ODL: 64496).'
33 parser.add_argument('--asnumber', default=64496, type=int, help=str_help)
34 # FIXME: We are acting as iBGP peer, we should mirror AS number from peer's open message.
35 str_help = 'Amount of IP prefixes to generate. Negative number means "practically infinite".'
36 parser.add_argument('--amount', default='1', type=int, help=str_help)
37 str_help = 'The first IPv4 prefix to announce, given as numeric IPv4 address.'
38 parser.add_argument('--firstprefix', default='8.0.1.0', type=ipaddr.IPv4Address, help=str_help)
39 str_help = 'If present, this tool will be listening for connection, instead of initiating it.'
40 parser.add_argument('--listen', action='store_true', help=str_help)
41 str_help = 'Numeric IP Address to bind to and derive BGP ID from. Default value only suitable for listening.'
42 parser.add_argument('--myip', default='0.0.0.0', type=ipaddr.IPv4Address, help=str_help)
43 str_help = 'TCP port to bind to when listening or initiating connection. Default only suitable for initiating.'
44 parser.add_argument('--myport', default='0', type=int, help=str_help)
45 str_help = 'The IP of the next hop to be placed into the update messages.'
46 parser.add_argument('--nexthop', default='192.0.2.1', type=ipaddr.IPv4Address, dest="nexthop", help=str_help)
47 str_help = 'Numeric IP Address to try to connect to. Currently no effect in listening mode.'
48 parser.add_argument('--peerip', default='127.0.0.2', type=ipaddr.IPv4Address, help=str_help)
49 str_help = 'TCP port to try to connect to. No effect in listening mode.'
50 parser.add_argument('--peerport', default='179', type=int, help=str_help)
51 # TODO: The step between IP previxes is currently hardcoded to 16. Should we make it configurable?
52 # Yes, the argument list above is sorted alphabetically.
53 arguments = parser.parse_args()
54 # TODO: Are sanity checks (such as asnumber>=0) required?
58 def establish_connection(arguments):
59 """Establish connection according to arguments, return socket."""
61 # print "DEBUG: connecting in the listening case."
62 listening_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
63 listening_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
64 listening_socket.bind((str(arguments.myip), arguments.myport)) # bind need single tuple as argument
65 listening_socket.listen(1)
66 bgp_socket, _ = listening_socket.accept()
67 # TODO: Verify client IP is cotroller IP.
68 listening_socket.close()
70 # print "DEBUG: connecting in the talking case."
71 talking_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
72 talking_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
73 talking_socket.bind((str(arguments.myip), arguments.myport)) # bind to force specified address and port
74 talking_socket.connect((str(arguments.peerip), arguments.peerport)) # socket does not spead ipaddr, hence str()
75 bgp_socket = talking_socket
76 print 'Connected to ODL.'
80 def get_short_int_from_message(message, offset=16):
81 """Extract 2-bytes number from packed string, default offset is for BGP message size."""
82 high_byte_int = ord(message[offset])
83 low_byte_int = ord(message[offset + 1])
84 short_int = high_byte_int * 256 + low_byte_int
88 class MessageError(ValueError):
89 """Value error with logging optimized for hexlified messages."""
91 def __init__(self, text, message, *args):
92 """Store and call super init for textual comment, store raw message which caused it."""
95 super(MessageError, self).__init__(text, message, *args)
98 """Concatenate text comment, space and hexlified message."""
99 return self.text + ' ' + binascii.hexlify(self.msg)
102 def read_open_message(bgp_socket):
103 """Receive message, perform some validation, return the raw message."""
104 msg_in = bgp_socket.recv(65535) # TODO: Is smaller buffer size safe?
105 # TODO: Is it possible for incoming open message to be split in more than one packet?
107 if len(msg_in) < 37: # 37 is minimal length of open message with 4-byte AS number.
108 raise MessageError("Got something else than open with 4-byte AS number", msg_in)
109 # TODO: We could check BGP marker, but it is defined only later; decide what to do.
110 reported_length = get_short_int_from_message(msg_in)
111 if len(msg_in) != reported_length:
112 raise MessageError("Message length is not " + reported_length + " in message", msg_in)
113 print "Open message received"
117 class MessageGenerator(object):
118 """Class with methods returning messages and state holding configuration data required to do it properly."""
120 # TODO: Define bgp marker as class (constant) variable.
121 def __init__(self, args):
122 """Initialize data according to command-line args."""
123 # Various auxiliary variables.
124 # Hack: 4-byte AS number uses the same "int to packed" encoding as IPv4 addresses.
125 asnumber_4bytes = ipaddr.v4_int_to_packed(args.asnumber)
126 asnumber_2bytes = "\x5b\xa0" # AS_TRANS value, 23456 decadic.
127 if args.asnumber < 65536: # AS number is mappable to 2 bytes
128 asnumber_2bytes = asnumber_4bytes[2:4]
129 # From now on, attribute docsrings are used.
130 self.int_nextprefix = int(args.firstprefix)
131 """Prefix IP address for next update message, as integer."""
132 self.updates_to_send = args.amount
133 """Number of update messages left to be sent."""
134 # All information ready, so we can define messages. Mostly copied from play.py by Jozef Behran.
135 # The following attributes are constant.
136 self.bgp_marker = "\xFF" * 16
137 """Every message starts with this, see rfc4271#section-4.1"""
138 self.keepalive_message = self.bgp_marker + (
140 "\x04" # Type KEEPALIVE
142 """KeepAlive message, see rfc4271#section-4.4"""
143 # TODO: Notification for hold timer expiration can be handy.
144 self.eor_message = self.bgp_marker + (
146 "\x02" # Type (UPDATE)
147 "\x00\x00" # Withdrawn routes length (0)
148 "\x00\x00" # Total Path Attributes Length (0)
150 """End-of-RIB marker, see rfc4724#section-2"""
151 self.update_message_without_prefix = self.bgp_marker + (
153 "\x02" # Type (UPDATE)
154 "\x00\x00" # Withdrawn routes length (0)
155 "\x00\x14" # Total Path Attributes Length (20)
156 "\x40" # Flags ("Well-Known")
157 "\x01" # Type (ORIGIN)
160 "\x40" # Flags ("Well-Known")
161 "\x02" # Type (AS_PATH)
163 "\x02" # AS segment type (AS_SEQUENCE)
164 "\x01" # AS segment length (1)
165 + asnumber_4bytes + # AS segment (4 bytes)
166 "\x40" # Flags ("Well-Known")
167 "\x03" # Type (NEXT_HOP)
169 + args.nexthop.packed + # IP address of the next hop (4 bytes)
170 "\x1c" # IPv4 prefix length, see RFC 4271, page 20. This tool uses Network Mask: 255.255.255.240
172 """The IP address prefix (4 bytes) has to be appended to complete Update message, see rfc4271#section-4.3."""
173 self.open_message = self.bgp_marker + (
176 "\x04" # BGP Varsion (4)
177 + asnumber_2bytes + # My Autonomous System
178 # FIXME: The following hold time is hardcoded separately. Compute from initial hold_time value.
179 "\x00\xb4" # Hold Time (180)
180 + args.myip.packed + # BGP Identifer
181 "\x10" # Optional parameters length
182 "\x02" # Param type ("Capability Ad")
183 "\x06" # Length (6 bytes)
184 "\x01" # Capability type (NLRI Unicast), see RFC 4760, secton 8
185 "\x04" # Capability value length
186 "\x00\x01" # AFI (Ipv4)
188 "\x01" # SAFI (Unicast)
189 "\x02" # Param type ("Capability Ad")
190 "\x06" # Length (6 bytes)
191 "\x41" # "32 bit AS Numbers Support" (see RFC 6793, section 3)
192 "\x04" # Capability value length
193 + asnumber_4bytes # My AS in 32 bit format
195 """Open message, see rfc4271#section-4.2"""
198 def compose_update_message(self):
199 """Return update message, prepare next prefix, decrease amount without checking it."""
200 prefix_packed = ipaddr.v4_int_to_packed(self.int_nextprefix)
201 # print "DEBUG: prefix", self.int_nextprefix, "packed to", binascii.hexlify(prefix_packed)
202 msg_out = self.update_message_without_prefix + prefix_packed
203 self.int_nextprefix += 16 # Hardcoded, as open message specifies such netmask.
204 self.updates_to_send -= 1
208 class TimeTracker(object):
209 """Class for tracking timers, both for my keepalives and peer's hold time."""
211 def __init__(self, msg_in):
212 """Initialize config, based on hardcoded defaults and open message from peer."""
213 # Note: Relative time is always named timedelta, to stress that (non-delta) time is absolute.
214 self.report_timedelta = 1.0 # In seconds. TODO: Configurable?
215 """Upper bound for being stuck in the same state, we should at least report something before continuing."""
216 # Negotiate the hold timer by taking the smaller of the 2 values (mine and the peer's).
217 hold_timedelta = 240 # Not an attribute of self yet.
218 # TODO: Make the default value configurable, default value could mirror what peer said.
219 peer_hold_timedelta = get_short_int_from_message(msg_in, offset=22)
220 if hold_timedelta > peer_hold_timedelta:
221 hold_timedelta = peer_hold_timedelta
222 if hold_timedelta < 3:
223 raise ValueError("FIXME: Zero (or invalid) hold timedelta is not supported: ", hold_timedelta)
224 self.hold_timedelta = hold_timedelta # only now the final value is visible from outside
225 """If we do not hear from peer this long, we assume it has died."""
226 self.keepalive_timedelta = int(hold_timedelta / 3.0)
227 """Upper limit for duration between messages, to avoid being declared dead."""
228 self.snapshot_time = time.time() # The same as calling snapshot(), but also declares a field.
229 """Sometimes we need to store time. This is where to get the value from afterwards."""
230 self.peer_hold_time = self.snapshot_time + self.hold_timedelta # time_keepalive may be too strict
231 """At this time point, peer will be declared dead."""
232 self.my_keepalive_time = None # to be set later
233 """At this point, we should be sending keepalive message."""
236 """Store current time in instance data to use later."""
237 self.snapshot_time = time.time() # Read as time before something interesting was called.
239 def reset_peer_hold_time(self):
240 """Move hold time to future as peer has just proven it still lives."""
241 self.peer_hold_time = time.time() + self.hold_timedelta
243 # Some methods could rely on self.snapshot_time, but it is better to require user to provide it explicitly.
244 def reset_my_keepalive_time(self, keepalive_time):
245 """Move KA timer to future based on given time from before sending."""
246 self.my_keepalive_time = keepalive_time + self.keepalive_timedelta
248 def check_peer_hold_time(self, snapshot_time):
249 """Raise error if nothing was read from peer until specified time."""
250 if snapshot_time > self.peer_hold_time: # time.time() may be too strict
251 raise RuntimeError("Peer has overstepped the hold timer.") # TODO: Include hold_timedelta?
252 # TODO: Add notification sending (attempt). That means move to write tracker.
255 class ReadTracker(object):
256 """Class for tracking read of mesages chunk by chunk and for idle waiting."""
258 def __init__(self, bgp_socket, timer):
259 """Set initial state."""
260 # References to outside objects.
261 self.socket = bgp_socket
264 self.header_length = 18
265 """BGP marker length plus length field length.""" # TODO: make it class (constant) attribute
266 self.reading_header = True
267 """Computation of where next chunk ends depends on whether we are beyond length field."""
268 self.bytes_to_read = self.header_length
269 """Countdown towards next size computation."""
271 """Incremental buffer for message under read."""
273 def read_message_chunk(self):
274 """Read up to one message, do not return anything."""
275 # TODO: We also could return the whole message, but currently nobody cares.
276 # We assume the socket is readable.
277 chunk_message = self.socket.recv(self.bytes_to_read)
278 self.msg_in += chunk_message
279 self.bytes_to_read -= len(chunk_message)
280 if not self.bytes_to_read: # TODO: bytes_to_read < 0 is not possible, right?
281 # Finished reading a logical block.
282 if self.reading_header:
283 # The logical block was a BGP header. Now we know size of message.
284 self.reading_header = False
285 self.bytes_to_read = get_short_int_from_message(self.msg_in)
286 else: # We have finished reading the body of the message.
287 # Peer has just proven it is still alive.
288 self.timer.reset_peer_hold_time()
289 # TODO: Do we want to count received messages?
290 # This version ignores the received message.
291 # TODO: Should we do validation and exit on anything besides update or keepalive?
292 # Prepare state for reading another message.
294 self.reading_header = True
295 self.bytes_to_read = self.header_length
296 # We should not act upon peer_hold_time if we are reading something right now.
299 def wait_for_read(self):
300 """When we know there are no more updates to send, we use this to avoid busy-wait."""
301 # First, compute time to first predictable state change (or report event)
302 event_time = min(self.timer.my_keepalive_time, self.timer.peer_hold_time)
303 # And wait for event or something to read.
304 select.select([self.socket], [], [self.socket], event_time - time.time()) # snapshot_time would be imprecise
305 # Not checking anything, that will be done in next iteration.
309 class WriteTracker(object):
310 """Class tracking enqueueing messages and sending chunks of them."""
312 def __init__(self, bgp_socket, generator, timer):
313 """Set initial state."""
314 # References to outside objects,
315 self.socket = bgp_socket
316 self.generator = generator
319 # TODO: Would attribute docstrings add anything substantial?
320 self.sending_message = False
321 self.bytes_to_send = 0
324 def enqueue_message_for_sending(self, message):
325 """Change write state to include the message."""
326 self.msg_out += message
327 self.bytes_to_send += len(message)
328 self.sending_message = True
330 def send_message_chunk_is_whole(self):
331 """Perform actions related to sending (chunk of) message, return whether message was completed."""
332 # We assume there is a msg_out to send and socket is writable.
333 # print 'going to send', repr(self.msg_out)
334 self.timer.snapshot()
335 bytes_sent = self.socket.send(self.msg_out)
336 self.msg_out = self.msg_out[bytes_sent:] # Forget the part of message that was sent.
337 self.bytes_to_send -= bytes_sent
338 if not self.bytes_to_send:
339 # TODO: Is it possible to hit negative bytes_to_send?
340 self.sending_message = False
341 # We should have reset hold timer on peer side.
342 self.timer.reset_my_keepalive_time(self.timer.snapshot_time)
343 # Which means the possible reason for not prioritizing reads is gone.
348 class StateTracker(object):
349 """Main loop has state so complex it warrants this separate class."""
351 def __init__(self, bgp_socket, generator, timer):
352 """Set the initial state according to existing socket and generator."""
353 # References to outside objects.
354 self.socket = bgp_socket
355 self.generator = generator
358 self.reader = ReadTracker(bgp_socket, timer)
359 self.writer = WriteTracker(bgp_socket, generator, timer)
360 # Prioritization state.
361 self.prioritize_writing = False
363 In general, we prioritize reading over writing. But in order to not get blocked by neverending reads,
364 we should check whether we are not risking running out of holdtime.
365 So in some situations, this field is set to True to attempt finishing sending a message,
366 after which this field resets back to False.
368 # TODO: Alternative is to switch fairly between reading and writing (called round robin from now on).
369 # Message counting is done in generator.
371 def perform_one_loop_iteration(self):
372 """Calculate priority, resolve all ifs, call appropriate method, return to caller to repeat."""
373 self.timer.snapshot()
374 if not self.prioritize_writing:
375 if self.timer.snapshot_time >= self.timer.my_keepalive_time:
376 if not self.writer.sending_message:
377 # We need to schedule a keepalive ASAP.
378 self.writer.enqueue_message_for_sending(self.generator.keepalive_message)
379 # We are sending a message now, so prioritize finishing it.
380 self.prioritize_writing = True
381 # Now we know what our priorities are, we have to check which actions are available.
382 # socket.socket() returns three lists, we store them to list of lists.
383 list_list = select.select([self.socket], [self.socket], [self.socket], self.timer.report_timedelta)
384 read_list, write_list, except_list = list_list
385 # Lists are unpacked, each is either [] or [self.socket], so we will test them as boolean.
387 raise RuntimeError("Exceptional state on socket", self.socket)
388 # We will do either read or write.
389 if not (self.prioritize_writing and write_list):
390 # Either we have no reason to rush writes, or the socket is not writable.
391 # We are focusing on reading here.
392 if read_list: # there is something to read indeed
393 # In this case we want to read chunk of message and repeat the select,
394 self.reader.read_message_chunk()
396 # We were focusing on reading, but nothing to read was there.
397 # Good time to check peer for hold timer.
398 self.timer.check_peer_hold_time(self.timer.snapshot_time)
399 # Things are quiet on the read front, we can go on and attempt to write.
401 # Either we really want to reset peer's view of our hold timer, or there was nothing to read.
402 if self.writer.sending_message: # We were in the middle of sending a message.
403 whole = self.writer.send_message_chunk_is_whole() # Was it the end of a message?
404 if self.prioritize_writing and whole: # We were pressed to send something and we did it.
405 self.prioritize_writing = False # We prioritize reading again.
407 # Finally, we can look if there is some update message for us to generate.
408 if self.generator.updates_to_send:
409 msg_out = self.generator.compose_update_message()
410 if not self.generator.updates_to_send: # We have just finished update generation, end-of-rib is due.
411 msg_out += self.generator.eor_message
412 self.writer.enqueue_message_for_sending(msg_out)
413 return # Attempt for the actual sending will be done in next iteration.
414 # Nothing to write anymore, except occasional keepalives.
415 # To avoid busy loop, we do idle waiting here.
416 self.reader.wait_for_read()
418 # We can neither read nor write.
419 print 'Input and output both blocked for', self.timer.report_timedelta, 'seconds.'
420 # FIXME: Are we sure select has been really waiting the whole period?
425 """Establish BGP connection and enter main loop for sending updates."""
426 arguments = parse_arguments()
427 bgp_socket = establish_connection(arguments)
428 # Initial handshake phase. TODO: Can it be also moved to StateTracker?
429 # Receive open message before sending anything.
430 # FIXME: Add parameter to send default open message first, to work with "you first" peers.
431 msg_in = read_open_message(bgp_socket)
432 timer = TimeTracker(msg_in)
433 generator = MessageGenerator(arguments)
434 msg_out = generator.open_message
435 # print "DEBUG: going to send open:", binascii.hexlify(msg_out)
436 # Send our open message to the peer.
437 bgp_socket.send(msg_out)
438 # Wait for confirming keepalive.
439 # TODO: Surely in just one packet?
440 msg_in = bgp_socket.recv(19) # Using exact keepalive length to not see possible updates.
441 if msg_in != generator.keepalive_message:
442 raise MessageError("Open not confirmed by keepalive, instead got", msg_in)
443 timer.reset_peer_hold_time()
444 # Send the keepalive to indicate the connection is accepted.
445 timer.snapshot() # Remember this time.
446 bgp_socket.send(generator.keepalive_message)
447 timer.reset_my_keepalive_time(timer.snapshot_time) # Use the remembered time.
448 # End of initial handshake phase.
449 state = StateTracker(bgp_socket, generator, timer)
450 while True: # main reactor loop
451 state.perform_one_loop_iteration()
453 if __name__ == "__main__":