1# encoding: utf-8 2""" 3community.py 4 5Created by Thomas Mangin on 2009-11-05. 6Copyright (c) 2009-2017 Exa Networks. All rights reserved. 7License: 3-clause BSD. (See the COPYRIGHT file) 8""" 9 10# ============================================================== Communities (8) 11# https://www.iana.org/assignments/bgp-extended-communities 12 13from exabgp.util import ordinal 14from exabgp.util import concat_bytes_i 15 16from exabgp.bgp.message.update.attribute import Attribute 17from exabgp.bgp.message.update.attribute.community.initial.community import Community 18 19from exabgp.bgp.message.notification import Notify 20 21 22@Attribute.register() 23class Communities(Attribute): 24 ID = Attribute.CODE.COMMUNITY 25 FLAG = Attribute.Flag.TRANSITIVE | Attribute.Flag.OPTIONAL 26 27 # __slots__ = ['communities'] 28 29 def __init__(self, communities=None): 30 # Must be None as = param is only evaluated once 31 if communities: 32 self.communities = communities 33 else: 34 self.communities = [] 35 36 def add(self, data): 37 self.communities.append(data) 38 return self 39 40 def pack(self, negotiated=None): 41 if len(self.communities): 42 return self._attribute(concat_bytes_i(c.pack() for c in self.communities)) 43 return b'' 44 45 def __iter__(self): 46 return iter(self.communities) 47 48 def __repr__(self): 49 lc = len(self.communities) 50 if lc > 1: 51 return "[ %s ]" % " ".join(repr(community) for community in sorted(self.communities)) 52 if lc == 1: 53 return repr(self.communities[0]) 54 return "" 55 56 def json(self): 57 return "[ %s ]" % ", ".join(community.json() for community in self.communities) 58 59 @staticmethod 60 def unpack(data, negotiated): 61 communities = Communities() 62 while data: 63 if data and len(data) < 4: 64 raise Notify(3, 1, 'could not decode community %s' % str([hex(ordinal(_)) for _ in data])) 65 communities.add(Community.unpack(data[:4], negotiated)) 66 data = data[4:] 67 return communities 68