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)
99 Generate human readable error message
101 Concatenate text comment, colon with space
102 and hexlified message. Use a placeholder string
103 if the message turns out to be empty.
105 message = binascii.hexlify(self.msg)
107 message = "(empty message)"
108 return self.text + ': ' + message
111 def read_open_message(bgp_socket):
112 """Receive message, perform some validation, return the raw message."""
113 msg_in = bgp_socket.recv(65535) # TODO: Is smaller buffer size safe?
114 # TODO: Is it possible for incoming open message to be split in more than one packet?
116 if len(msg_in) < 37: # 37 is minimal length of open message with 4-byte AS number.
117 raise MessageError("Got something else than open with 4-byte AS number", msg_in)
118 # TODO: We could check BGP marker, but it is defined only later; decide what to do.
119 reported_length = get_short_int_from_message(msg_in)
120 if len(msg_in) != reported_length:
121 raise MessageError("Message length is not " + reported_length + " in message", msg_in)
122 print "Open message received"
126 class MessageGenerator(object):
127 """Class with methods returning messages and state holding configuration data required to do it properly."""
129 # TODO: Define bgp marker as class (constant) variable.
130 def __init__(self, args):
131 """Initialize data according to command-line args."""
132 # Various auxiliary variables.
133 # Hack: 4-byte AS number uses the same "int to packed" encoding as IPv4 addresses.
134 asnumber_4bytes = ipaddr.v4_int_to_packed(args.asnumber)
135 asnumber_2bytes = "\x5b\xa0" # AS_TRANS value, 23456 decadic.
136 if args.asnumber < 65536: # AS number is mappable to 2 bytes
137 asnumber_2bytes = asnumber_4bytes[2:4]
138 # From now on, attribute docsrings are used.
139 self.int_nextprefix = int(args.firstprefix)
140 """Prefix IP address for next update message, as integer."""
141 self.updates_to_send = args.amount
142 """Number of update messages left to be sent."""
143 # All information ready, so we can define messages. Mostly copied from play.py by Jozef Behran.
144 # The following attributes are constant.
145 self.bgp_marker = "\xFF" * 16
146 """Every message starts with this, see rfc4271#section-4.1"""
147 self.keepalive_message = self.bgp_marker + (
149 "\x04" # Type KEEPALIVE
151 """KeepAlive message, see rfc4271#section-4.4"""
152 # TODO: Notification for hold timer expiration can be handy.
153 self.eor_message = self.bgp_marker + (
155 "\x02" # Type (UPDATE)
156 "\x00\x00" # Withdrawn routes length (0)
157 "\x00\x00" # Total Path Attributes Length (0)
159 """End-of-RIB marker, see rfc4724#section-2"""
160 self.update_message_without_prefix = self.bgp_marker + (
162 "\x02" # Type (UPDATE)
163 "\x00\x00" # Withdrawn routes length (0)
164 "\x00\x14" # Total Path Attributes Length (20)
165 "\x40" # Flags ("Well-Known")
166 "\x01" # Type (ORIGIN)
169 "\x40" # Flags ("Well-Known")
170 "\x02" # Type (AS_PATH)
172 "\x02" # AS segment type (AS_SEQUENCE)
173 "\x01" # AS segment length (1)
174 + asnumber_4bytes + # AS segment (4 bytes)
175 "\x40" # Flags ("Well-Known")
176 "\x03" # Type (NEXT_HOP)
178 + args.nexthop.packed + # IP address of the next hop (4 bytes)
179 "\x1c" # IPv4 prefix length, see RFC 4271, page 20. This tool uses Network Mask: 255.255.255.240
181 """The IP address prefix (4 bytes) has to be appended to complete Update message, see rfc4271#section-4.3."""
182 self.open_message = self.bgp_marker + (
185 "\x04" # BGP Varsion (4)
186 + asnumber_2bytes + # My Autonomous System
187 # FIXME: The following hold time is hardcoded separately. Compute from initial hold_time value.
188 "\x00\xb4" # Hold Time (180)
189 + args.myip.packed + # BGP Identifer
190 "\x10" # Optional parameters length
191 "\x02" # Param type ("Capability Ad")
192 "\x06" # Length (6 bytes)
193 "\x01" # Capability type (NLRI Unicast), see RFC 4760, secton 8
194 "\x04" # Capability value length
195 "\x00\x01" # AFI (Ipv4)
197 "\x01" # SAFI (Unicast)
198 "\x02" # Param type ("Capability Ad")
199 "\x06" # Length (6 bytes)
200 "\x41" # "32 bit AS Numbers Support" (see RFC 6793, section 3)
201 "\x04" # Capability value length
202 + asnumber_4bytes # My AS in 32 bit format
204 """Open message, see rfc4271#section-4.2"""
207 def compose_update_message(self):
208 """Return update message, prepare next prefix, decrease amount without checking it."""
209 prefix_packed = ipaddr.v4_int_to_packed(self.int_nextprefix)
210 # print "DEBUG: prefix", self.int_nextprefix, "packed to", binascii.hexlify(prefix_packed)
211 msg_out = self.update_message_without_prefix + prefix_packed
212 self.int_nextprefix += 16 # Hardcoded, as open message specifies such netmask.
213 self.updates_to_send -= 1
217 class TimeTracker(object):
218 """Class for tracking timers, both for my keepalives and peer's hold time."""
220 def __init__(self, msg_in):
221 """Initialize config, based on hardcoded defaults and open message from peer."""
222 # Note: Relative time is always named timedelta, to stress that (non-delta) time is absolute.
223 self.report_timedelta = 1.0 # In seconds. TODO: Configurable?
224 """Upper bound for being stuck in the same state, we should at least report something before continuing."""
225 # Negotiate the hold timer by taking the smaller of the 2 values (mine and the peer's).
226 hold_timedelta = 240 # Not an attribute of self yet.
227 # TODO: Make the default value configurable, default value could mirror what peer said.
228 peer_hold_timedelta = get_short_int_from_message(msg_in, offset=22)
229 if hold_timedelta > peer_hold_timedelta:
230 hold_timedelta = peer_hold_timedelta
231 if hold_timedelta < 3:
232 raise ValueError("FIXME: Zero (or invalid) hold timedelta is not supported: ", hold_timedelta)
233 self.hold_timedelta = hold_timedelta # only now the final value is visible from outside
234 """If we do not hear from peer this long, we assume it has died."""
235 self.keepalive_timedelta = int(hold_timedelta / 3.0)
236 """Upper limit for duration between messages, to avoid being declared dead."""
237 self.snapshot_time = time.time() # The same as calling snapshot(), but also declares a field.
238 """Sometimes we need to store time. This is where to get the value from afterwards."""
239 self.peer_hold_time = self.snapshot_time + self.hold_timedelta # time_keepalive may be too strict
240 """At this time point, peer will be declared dead."""
241 self.my_keepalive_time = None # to be set later
242 """At this point, we should be sending keepalive message."""
245 """Store current time in instance data to use later."""
246 self.snapshot_time = time.time() # Read as time before something interesting was called.
248 def reset_peer_hold_time(self):
249 """Move hold time to future as peer has just proven it still lives."""
250 self.peer_hold_time = time.time() + self.hold_timedelta
252 # Some methods could rely on self.snapshot_time, but it is better to require user to provide it explicitly.
253 def reset_my_keepalive_time(self, keepalive_time):
254 """Move KA timer to future based on given time from before sending."""
255 self.my_keepalive_time = keepalive_time + self.keepalive_timedelta
257 def check_peer_hold_time(self, snapshot_time):
258 """Raise error if nothing was read from peer until specified time."""
259 if snapshot_time > self.peer_hold_time: # time.time() may be too strict
260 raise RuntimeError("Peer has overstepped the hold timer.") # TODO: Include hold_timedelta?
261 # TODO: Add notification sending (attempt). That means move to write tracker.
264 class ReadTracker(object):
265 """Class for tracking read of mesages chunk by chunk and for idle waiting."""
267 def __init__(self, bgp_socket, timer):
268 """Set initial state."""
269 # References to outside objects.
270 self.socket = bgp_socket
273 self.header_length = 18
274 """BGP marker length plus length field length.""" # TODO: make it class (constant) attribute
275 self.reading_header = True
276 """Computation of where next chunk ends depends on whether we are beyond length field."""
277 self.bytes_to_read = self.header_length
278 """Countdown towards next size computation."""
280 """Incremental buffer for message under read."""
282 def read_message_chunk(self):
283 """Read up to one message, do not return anything."""
284 # TODO: We also could return the whole message, but currently nobody cares.
285 # We assume the socket is readable.
286 chunk_message = self.socket.recv(self.bytes_to_read)
287 self.msg_in += chunk_message
288 self.bytes_to_read -= len(chunk_message)
289 if not self.bytes_to_read: # TODO: bytes_to_read < 0 is not possible, right?
290 # Finished reading a logical block.
291 if self.reading_header:
292 # The logical block was a BGP header. Now we know size of message.
293 self.reading_header = False
294 self.bytes_to_read = get_short_int_from_message(self.msg_in)
295 else: # We have finished reading the body of the message.
296 # Peer has just proven it is still alive.
297 self.timer.reset_peer_hold_time()
298 # TODO: Do we want to count received messages?
299 # This version ignores the received message.
300 # TODO: Should we do validation and exit on anything besides update or keepalive?
301 # Prepare state for reading another message.
303 self.reading_header = True
304 self.bytes_to_read = self.header_length
305 # We should not act upon peer_hold_time if we are reading something right now.
308 def wait_for_read(self):
309 """When we know there are no more updates to send, we use this to avoid busy-wait."""
310 # First, compute time to first predictable state change (or report event)
311 event_time = min(self.timer.my_keepalive_time, self.timer.peer_hold_time)
312 # And wait for event or something to read.
313 select.select([self.socket], [], [self.socket], event_time - time.time()) # snapshot_time would be imprecise
314 # Not checking anything, that will be done in next iteration.
318 class WriteTracker(object):
319 """Class tracking enqueueing messages and sending chunks of them."""
321 def __init__(self, bgp_socket, generator, timer):
322 """Set initial state."""
323 # References to outside objects,
324 self.socket = bgp_socket
325 self.generator = generator
328 # TODO: Would attribute docstrings add anything substantial?
329 self.sending_message = False
330 self.bytes_to_send = 0
333 def enqueue_message_for_sending(self, message):
334 """Change write state to include the message."""
335 self.msg_out += message
336 self.bytes_to_send += len(message)
337 self.sending_message = True
339 def send_message_chunk_is_whole(self):
340 """Perform actions related to sending (chunk of) message, return whether message was completed."""
341 # We assume there is a msg_out to send and socket is writable.
342 # print 'going to send', repr(self.msg_out)
343 self.timer.snapshot()
344 bytes_sent = self.socket.send(self.msg_out)
345 self.msg_out = self.msg_out[bytes_sent:] # Forget the part of message that was sent.
346 self.bytes_to_send -= bytes_sent
347 if not self.bytes_to_send:
348 # TODO: Is it possible to hit negative bytes_to_send?
349 self.sending_message = False
350 # We should have reset hold timer on peer side.
351 self.timer.reset_my_keepalive_time(self.timer.snapshot_time)
352 # Which means the possible reason for not prioritizing reads is gone.
357 class StateTracker(object):
358 """Main loop has state so complex it warrants this separate class."""
360 def __init__(self, bgp_socket, generator, timer):
361 """Set the initial state according to existing socket and generator."""
362 # References to outside objects.
363 self.socket = bgp_socket
364 self.generator = generator
367 self.reader = ReadTracker(bgp_socket, timer)
368 self.writer = WriteTracker(bgp_socket, generator, timer)
369 # Prioritization state.
370 self.prioritize_writing = False
372 In general, we prioritize reading over writing. But in order to not get blocked by neverending reads,
373 we should check whether we are not risking running out of holdtime.
374 So in some situations, this field is set to True to attempt finishing sending a message,
375 after which this field resets back to False.
377 # TODO: Alternative is to switch fairly between reading and writing (called round robin from now on).
378 # Message counting is done in generator.
380 def perform_one_loop_iteration(self):
381 """Calculate priority, resolve all ifs, call appropriate method, return to caller to repeat."""
382 self.timer.snapshot()
383 if not self.prioritize_writing:
384 if self.timer.snapshot_time >= self.timer.my_keepalive_time:
385 if not self.writer.sending_message:
386 # We need to schedule a keepalive ASAP.
387 self.writer.enqueue_message_for_sending(self.generator.keepalive_message)
388 # We are sending a message now, so prioritize finishing it.
389 self.prioritize_writing = True
390 # Now we know what our priorities are, we have to check which actions are available.
391 # socket.socket() returns three lists, we store them to list of lists.
392 list_list = select.select([self.socket], [self.socket], [self.socket], self.timer.report_timedelta)
393 read_list, write_list, except_list = list_list
394 # Lists are unpacked, each is either [] or [self.socket], so we will test them as boolean.
396 raise RuntimeError("Exceptional state on socket", self.socket)
397 # We will do either read or write.
398 if not (self.prioritize_writing and write_list):
399 # Either we have no reason to rush writes, or the socket is not writable.
400 # We are focusing on reading here.
401 if read_list: # there is something to read indeed
402 # In this case we want to read chunk of message and repeat the select,
403 self.reader.read_message_chunk()
405 # We were focusing on reading, but nothing to read was there.
406 # Good time to check peer for hold timer.
407 self.timer.check_peer_hold_time(self.timer.snapshot_time)
408 # Things are quiet on the read front, we can go on and attempt to write.
410 # Either we really want to reset peer's view of our hold timer, or there was nothing to read.
411 if self.writer.sending_message: # We were in the middle of sending a message.
412 whole = self.writer.send_message_chunk_is_whole() # Was it the end of a message?
413 if self.prioritize_writing and whole: # We were pressed to send something and we did it.
414 self.prioritize_writing = False # We prioritize reading again.
416 # Finally, we can look if there is some update message for us to generate.
417 if self.generator.updates_to_send:
418 msg_out = self.generator.compose_update_message()
419 if not self.generator.updates_to_send: # We have just finished update generation, end-of-rib is due.
420 msg_out += self.generator.eor_message
421 self.writer.enqueue_message_for_sending(msg_out)
422 return # Attempt for the actual sending will be done in next iteration.
423 # Nothing to write anymore, except occasional keepalives.
424 # To avoid busy loop, we do idle waiting here.
425 self.reader.wait_for_read()
427 # We can neither read nor write.
428 print 'Input and output both blocked for', self.timer.report_timedelta, 'seconds.'
429 # FIXME: Are we sure select has been really waiting the whole period?
434 """Establish BGP connection and enter main loop for sending updates."""
435 arguments = parse_arguments()
436 bgp_socket = establish_connection(arguments)
437 # Initial handshake phase. TODO: Can it be also moved to StateTracker?
438 # Receive open message before sending anything.
439 # FIXME: Add parameter to send default open message first, to work with "you first" peers.
440 msg_in = read_open_message(bgp_socket)
441 timer = TimeTracker(msg_in)
442 generator = MessageGenerator(arguments)
443 msg_out = generator.open_message
444 # print "DEBUG: going to send open:", binascii.hexlify(msg_out)
445 # Send our open message to the peer.
446 bgp_socket.send(msg_out)
447 # Wait for confirming keepalive.
448 # TODO: Surely in just one packet?
449 msg_in = bgp_socket.recv(19) # Using exact keepalive length to not see possible updates.
450 if msg_in != generator.keepalive_message:
451 raise MessageError("Open not confirmed by keepalive, instead got", msg_in)
452 timer.reset_peer_hold_time()
453 # Send the keepalive to indicate the connection is accepted.
454 timer.snapshot() # Remember this time.
455 bgp_socket.send(generator.keepalive_message)
456 timer.reset_my_keepalive_time(timer.snapshot_time) # Use the remembered time.
457 # End of initial handshake phase.
458 state = StateTracker(bgp_socket, generator, timer)
459 while True: # main reactor loop
460 state.perform_one_loop_iteration()
462 if __name__ == "__main__":