Rfc7854 compliance and bug fixes (#3)

* Changing Readme
Add new Stat types

* Update peer header

* Adds:
  - Support for 'A' flag in Per-Peer header
  - Support for Route Mirroring message type

* Add bitstring to requirements.txt

* Fixes two bugs:
- Initiation and Termination messages. These two message types don't have per
  peer header hence we need to pass the raw_body as their argument, instead of raw_body[42:]
- When BGP capabilities include the use of 4 byte ASNs, regular ASN_PATH type 2 attribute is
  used for Updates. This commit includes a temporary fix to avoid an AS_PATH attributes decoding
  error when the routers use 4 bytes ASNs. The code seems to use asn4=False by default and makes no attempt
  to change that based on the Peer_up notification message. I just changed to True for now we but need to
  find a way to use shared state in the connection to detect the use of 4Byte ASN capability and then
  pass the correct flag to Update.parse(asn4=True)

* Fix PEP8 formatting

* Don't add autogenerated files

.gitignore

@@ -57,4 +57,5 @@ docs/_build/
 target/
 
 /.idea
-etc/openbmp/openbmp.ini
+etc/openbmp/openbmp.ini
+.DS_Store

README.rst

@@ -12,7 +12,7 @@ Overview
 
 `YABMP` is a receiver-side implementation of the `BMP` (BGP Monitoring Protocol) in the Python language. It serves as a reference for how to step through the messages and write their contents to files.
 
-This implementation covers draft BGP Monitoring Protocol draft-ietf-grow-bmp-07
+This implementation covers RFC 7854 BGP Monitoring Protocol version 3.
 
 RFCs to read to help you understand the code better:
 

requirements.txt

@@ -3,3 +3,4 @@ netaddr>=0.7.12
 pbr==1.6
 oslo.config>=2.1.0
 Twisted==15.0.0
+bitstring==3.1.5

yabmp/common/constants.py

@@ -29,21 +29,26 @@
 MSG_TYPE_PEER_UP_NOTIFICATION = 3
 MSG_TYPE_INITIATION = 4
 MSG_TYPE_TERMINATION = 5
+MSG_TYPE_ROUTE_MIRRORING = 6
 MSG_TYPE_STR = {
     MSG_TYPE_ROUTE_MONITORING: "Route Monitoring",
     MSG_TYPE_STATISTICS_REPORT: "Statistics Report",
     MSG_TYPE_PEER_DOWN_NOTIFICATION: "Peer Down Notification",
     MSG_TYPE_PEER_UP_NOTIFICATION: "Peer Up Notification",
     MSG_TYPE_INITIATION: "Initiation Message",
     MSG_TYPE_TERMINATION: "Termination Message",
+    MSG_TYPE_ROUTE_MIRRORING: "Route Mirroring"
 }
 
 # Peer types.
 PEER_TYPE_GLOBAL = 0
-PEER_TYPE_L3_VPN = 1
+PEER_TYPE_RD_INSTANCE = 1
+PEER_TYPE_LOCAL = 2
 PEER_TYPE_STR = {PEER_TYPE_GLOBAL: "Global",
-                 PEER_TYPE_L3_VPN: "L3 VPN"}
+                 PEER_TYPE_RD_INSTANCE: "RD Instance",
+                 PEER_TYPE_LOCAL: "Local Instance"}
 
+PEER_FLAGS = ['V', 'L', 'A']
 
 BMP_STAT_TYPE = {
     0: 'Number of prefixes rejected by inbound policy',
@@ -55,6 +60,11 @@
     6: 'Number of updates invalidated due to AS_CONFED loop',
     7: 'Number of routes in Adj-RIBs-In',
     8: 'Number of routes in Loc-RIB',
+    9: 'Number of routes in per-AFI/SAFI Adj-RIB-In',
+    10: 'Number of routes in per-AFI/SAFI Loc-RIB',
+    11: 'Number of updates subjected to treat-as-withdraw',
+    12: 'Number of prefixes subjected to treat-as-withdraw',
+    13: 'Number of duplicate update messages received',
     32767: 'SRTT',
     32768: 'RTTO',
     32769: 'RTV',
@@ -80,5 +90,17 @@
     0: 'Session administratively closed',
     1: 'Unspecified reason',
     2: 'Out of resources',
-    3: 'Redundant connection'
+    3: 'Redundant connection',
+    4: 'Permanently administratively closed'
+}
+
+ROUTE_MIRRORING_TLV_TYPE = {
+    0: 'BGP Message TLV',
+    1: 'Information TLV'
+}
+
+ROUTE_MIRRORING_INFORMATION_TYPE_CODE = {
+    0: 'Errored PDU',
+    1: 'Message Lost'
+
 }

yabmp/core/protocol.py

@@ -153,7 +153,7 @@ def write_msg_file(self, msg_type, msg):
         """
         write msg to file
         """
-        if msg_type in [4, 5]:
+        if msg_type in [4, 5, 6]:
             return
         peer_ip = msg[0]['addr']
         if peer_ip not in self.bgp_peer_dict:

yabmp/message/bmp.py

@@ -17,8 +17,11 @@
 import binascii
 import logging
 import traceback
+import itertools
 
 import netaddr
+from bitstring import BitArray
+
 from yabgp.message.notification import Notification
 from yabgp.message.update import Update
 from yabgp.message.route_refresh import RouteRefresh
@@ -82,28 +85,30 @@ def parse_per_peer_header(raw_peer_header):
         LOG.debug('decode per-peer header')
         per_header_dict['type'] = struct.unpack('!B', raw_peer_header[0:1])[0]
         # Peer Type = 0: Global Instance Peer
-        # Peer Type = 1: L3 VPN Instance Peer
-        if per_header_dict['type'] not in [0, 1]:
+        # Peer Type = 1: RD Instance Peer
+        # Peer Type = 2: Local Instance Peer
+        if per_header_dict['type'] not in [0, 1, 2]:
             raise excp.UnknownPeerTypeValue(peer_type=per_header_dict['type'])
-
         LOG.debug('peer type: %s ' % per_header_dict['type'])
+
+        # Peer Flags
         peer_flags_value = binascii.b2a_hex(raw_peer_header[1:2])
-        if peer_flags_value == '80':
-            per_header_dict['flags'] = {'V': 1, 'L': 0}  # IPv6, pre-policy Adj-RIB-In
-        elif peer_flags_value == '00':
-            per_header_dict['flags'] = {'V': 0, 'L': 0}  # IPv4, pre-policy Adj-RIB-In
-        elif peer_flags_value == '40':
-            per_header_dict['flags'] = {'V': 0, 'L': 1}  # IPv4, post-policy Adj-RIB-In
-        elif peer_flags_value == 'c0':
-            per_header_dict['flags'] = {'V': 1, 'L': 1}  # IPv6, post-policy Adj-RIB-In
+        hex_rep = hex(int(peer_flags_value, 16))
+        bit_array = BitArray(hex_rep)
+        valid_flags = [''.join(item)+'00000' for item in itertools.product('01', repeat=3)]
+        valid_flags.append('0000')
+        if bit_array.bin in valid_flags:
+            flags = dict(zip(bmp_cons.PEER_FLAGS, bit_array.bin))
+            per_header_dict['flags'] = flags
+            LOG.debug('Per Peer header flags %s' % flags)
         else:
             raise excp.UnknownPeerFlagValue(peer_flags=peer_flags_value)
         LOG.debug('peer flag: %s ' % per_header_dict['flags'])
-        if per_header_dict['type'] == 1:
+        if per_header_dict['type'] in [1, 2]:
             per_header_dict['dist'] = int(binascii.b2a_hex(raw_peer_header[2:10]), 16)
+
         ip_value = int(binascii.b2a_hex(raw_peer_header[10:26]), 16)
         if per_header_dict['flags']['V']:
-
             per_header_dict['addr'] = str(netaddr.IPAddress(ip_value, version=6))
         else:
             per_header_dict['addr'] = str(netaddr.IPAddress(ip_value, version=4))
@@ -122,8 +127,7 @@ def parse_route_monitoring_msg(msg):
         """
             Route Monitoring messages are used for initial synchronization of
         ADJ-RIBs-In. They are also used for ongoing monitoring of received
-        advertisements and withdraws. This is discussed in more detail in
-        Section 5.
+        advertisements and withdraws.
         Following the common BMP header and per-peer header is a BGP Update
         PDU.
         :param msg:
@@ -135,7 +139,7 @@ def parse_route_monitoring_msg(msg):
         msg = msg[bgp_cons.HDR_LEN:]
         if bgp_msg_type == 2:
             # decode update message
-            results = Update().parse(None, msg)
+            results = Update().parse(None, msg, asn4=True)
             if results['sub_error']:
                 LOG.error('error: decode update message error!, error code: %s' % results['sub_error'])
                 LOG.error('Raw data: %s' % repr(results['hex']))
@@ -155,6 +159,73 @@ def parse_route_monitoring_msg(msg):
                                   'sub_type': bgp_route_refresh_msg[1],
                                   'safi': bgp_route_refresh_msg[2]}
 
+    @staticmethod
+    def parse_route_mirroring_msg(msg):
+        """
+            Route Mirroring messages are used for verbatim duplication of
+        messages as received. Following the common BMP header and per-peer
+        header is a set of TLVs that contain information about a message
+        or set of messages.
+        :param msg:
+        :return:
+        """
+        LOG.debug('decode route mirroring message')
+
+        msg_dict = {}
+        open_l = []
+        update = []
+        notification = []
+        route_refresh = []
+        while msg:
+            mirror_type, length = struct.unpack('!HH', msg[0:4])
+            mirror_value = msg[4: 4 + length]
+            msg = msg[4 + length:]
+            if mirror_type == 0:
+                # BGP message type
+                bgp_msg_type = struct.unpack('!B', mirror_value[18])[0]
+                LOG.debug('bgp message type=%s' % bgp_msg_type)
+                bgp_msg_body = mirror_value[bgp_cons.HDR_LEN:]
+                if bgp_msg_type == 2:
+                    # Update message
+                    bgp_update_msg = Update().parse(None, bgp_msg_body, asn4=True)
+                    if bgp_update_msg['sub_error']:
+                        LOG.error('error: decode update message error!, error code: %s' % bgp_update_msg['sub_error'])
+                        LOG.error('Raw data: %s' % repr(bgp_update_msg['hex']))
+                    else:
+                        update.append(bgp_update_msg)
+                elif bgp_msg_type == 5:
+                    # Route Refresh message
+                    bgp_route_refresh_msg = RouteRefresh().parse(msg=bgp_msg_body)
+                    LOG.debug('bgp route refresh message: afi=%s,res=%s,safi=%s' % (bgp_route_refresh_msg[0],
+                                                                                    bgp_route_refresh_msg[1],
+                                                                                    bgp_route_refresh_msg[2]))
+                    route_refresh.append(bgp_route_refresh_msg)
+                elif bgp_msg_type == 1:
+                    # Open message
+                    open_msg = Open().parse(bgp_msg_body)
+                    open_l.append(open_msg)
+                elif bgp_msg_type == 3:
+                    # Notification message
+                    notification_msg = Notification().parse(bgp_msg_body)
+                    notification.append(notification_msg)
+            elif mirror_type == 1:
+                # Information type.
+                # Amount of this TLV is not specified but we can assume
+                # only one per mirroring message is present.
+                info_code_type = struct.unpack('!H', mirror_value)[0]
+                msg_dict['1'] = info_code_type
+            else:
+                msg_dict[mirror_type] = binascii.unhexlify(binascii.hexlify(mirror_value))
+                LOG.info('unknow mirroring type, type = %s' % mirror_type)
+
+        msg_dict['0'] = {
+                        'update': update,
+                        'route_refresh': route_refresh,
+                        'open': open_l,
+                        'notification': notification
+                        }
+        return msg_dict
+
     @staticmethod
     def parse_statistic_report_msg(msg):
         """
@@ -356,7 +427,7 @@ def parse_termination_msg(msg):
 
     def consume(self):
 
-        if self.msg_type in [0, 1, 2, 3]:
+        if self.msg_type in [0, 1, 2, 3, 6]:
             try:
                 per_peer_header = self.parse_per_peer_header(self.raw_body[0:42])
                 self.msg_body = self.raw_body[42:]
@@ -368,6 +439,8 @@ def consume(self):
                     return per_peer_header, self.parse_peer_down_notification(self.msg_body)
                 elif self.msg_type == 3:
                     return per_peer_header, self.parse_peer_up_notification(self.msg_body, per_peer_header['flags'])
+                elif self.msg_type == 6:
+                    return per_peer_header, self.parse_route_mirroring_msg(self.msg_body)
             except Exception as e:
                 LOG.error(e)
                 error_str = traceback.format_exc()
@@ -376,6 +449,6 @@ def consume(self):
                 return None
 
         elif self.msg_type == 4:
-            return None, self.parse_initiation_msg(self.msg_body)
+            return None, self.parse_initiation_msg(self.raw_body)
         elif self.msg_type == 5:
-            return None, self.parse_termination_msg(self.msg_body)
+            return None, self.parse_termination_msg(self.raw_body)