add extended community support for evpn

Signed-off-by: Peng Xiao <xiaoquwl@gmail.com>

yabgp/message/attribute/extcommunity.py

@@ -132,6 +132,22 @@ def parse(cls, value):
                 mark = struct.unpack('!B', value_tmp[-1])[0]
                 ext_community.append([bgp_cons.BGP_EXT_TRA_MARK, mark])
 
+            # Transitive Opaque
+            elif comm_code == bgp_cons.BGP_EXT_COM_ENCAP:
+                ext_community.append([bgp_cons.BGP_EXT_COM_ENCAP, struct.unpack('!I', value_tmp[2:])[0]])
+            # EVPN
+            elif comm_code == bgp_cons.BGP_EXT_COM_EVPN_ES_IMPORT:
+                mac = str(netaddr.EUI(int(binascii.b2a_hex(value_tmp), 16)))
+                ext_community.append([comm_code, mac])
+            elif comm_code == bgp_cons.BGP_EXT_COM_EVPN_MAC_MOBIL:
+                flag = ord(value_tmp[0:1])
+                seq = struct.unpack('!I', value_tmp[2:])[0]
+                ext_community.append([comm_code, flag, seq])
+            elif comm_code == bgp_cons.BGP_EXT_COM_EVPN_ESI_MPLS_LABEL:
+                flag = ord(value_tmp[0:1])
+                label = struct.unpack('!L', b'\00'+value_tmp[3:])[0]
+                label >>= 4
+                ext_community.append([comm_code, flag, label])
             else:
                 ext_community.append([bgp_cons.BGP_EXT_COM_UNKNOW, repr(value_tmp)])
                 LOG.warn('unknow bgp extended community, type=%s, value=%s', comm_code, repr(value_tmp))
@@ -190,6 +206,21 @@ def construct(cls, value):
                 asn, rate = item[1].split(':')
                 ext_community_hex += struct.pack('!HHf', bgp_cons.BGP_EXT_TRA_RATE, int(asn), int(rate))
 
+            # Transitive Opaque
+            elif item[0] == bgp_cons.BGP_EXT_COM_ENCAP:
+                ext_community_hex += struct.pack('!HHI', bgp_cons.BGP_EXT_COM_ENCAP, 0, item[1])
+            # EVPN
+            elif item[0] == bgp_cons.BGP_EXT_COM_EVPN_ES_IMPORT:
+                mac = b''.join([struct.pack('!B', (int(i, 16))) for i in item[1].split("-")])
+                ext_community_hex += struct.pack('!H', item[0]) + mac
+            elif item[0] == bgp_cons.BGP_EXT_COM_EVPN_ESI_MPLS_LABEL:
+                flag = struct.pack('!B', item[1])
+                label = struct.pack('!L', (item[2] << 4 | 1))[1:]
+                ext_community_hex += struct.pack('!H', item[0]) + flag + b'\x00\x00' + label
+            elif item[0] == bgp_cons.BGP_EXT_COM_EVPN_MAC_MOBIL:
+                flag = struct.pack('!B', item[1])
+                seq = struct.pack('!I', item[2])
+                ext_community_hex += struct.pack('!H', item[0]) + flag + b'\x00' + seq
             else:
                 LOG.warn('unknow bgp extended community for construct, type=%s, value=%s', item[0], item[1])
 

yabgp/tests/unit/message/attribute/test_extcommunity.py

@@ -111,5 +111,29 @@ def test_parse_construct_tarffic_rate(self):
         self.assertEqual([[bgp_cons.BGP_EXT_TRA_RATE, "100:6250000"]],
                          ExtCommunity.parse(value=b'\x80\x06\x00dJ\xbe\xbc '))
 
+    def test_parse_construct_transitive_opaque_encap(self):
+        community_list = [[bgp_cons.BGP_EXT_COM_ENCAP, 8]]
+        community_hex = b'\x03\x0c\x00\x00\x00\x00\x00\x08'
+        self.assertEqual(community_list, ExtCommunity.parse(community_hex))
+        self.assertEqual(community_hex, ExtCommunity.construct(community_list)[3:])
+
+    def test_parse_construct_es_import(self):
+        community_list = [[bgp_cons.BGP_EXT_COM_EVPN_ES_IMPORT, '00-11-22-33-44-55']]
+        community_hex = b'\x06\x02\x00\x11\x22\x33\x44\x55'
+        self.assertEqual(community_list, ExtCommunity.parse(community_hex))
+        self.assertEqual(community_hex, ExtCommunity.construct(community_list)[3:])
+
+    def test_parse_construct_els_label(self):
+        community_list = [[bgp_cons.BGP_EXT_COM_EVPN_ESI_MPLS_LABEL, 1, 20]]
+        community_hex = b'\x06\x01\x01\x00\x00\x00\x01\x41'
+        self.assertEqual(community_hex, ExtCommunity.construct(community_list)[3:])
+        self.assertEqual(community_list, ExtCommunity.parse(community_hex))
+
+    def test_parse_construct_mac_mobil(self):
+        community_list = [[bgp_cons.BGP_EXT_COM_EVPN_MAC_MOBIL, 1, 500]]
+        community_hex = b'\x06\x00\x01\x00\x00\x00\x01\xf4'
+        self.assertEqual(community_hex, ExtCommunity.construct(community_list)[3:])
+        self.assertEqual(community_list, ExtCommunity.parse(community_hex))
+
 if __name__ == '__main__':
     unittest.main()