<s> """<STR_LIT>""" <EOL> from struct import pack <EOL> from exabgp . protocol . ip import IP <EOL> from exabgp . protocol . ip import IPSelf <EOL> from exabgp . protocol . family import AFI <EOL> from exabgp . bgp . message import OUT <EOL> from exabgp . bgp . message . update . nlri import CIDR <EOL> from exabgp . bgp . message . update . nlri import INET <EOL> from exabgp . bgp . message . update . nlri import IPVPN <EOL> from exabgp . bgp . message . open import ASN <EOL> from exabgp . bgp . message . open import RouterID <EOL> from exabgp . bgp . message . update . attribute import Attribute <EOL> from exabgp . bgp . message . update . attribute import Attributes <EOL> from exabgp . bgp . message . update . attribute import NextHop <EOL> from exabgp . bgp . message . update . attribute import NextHopSelf <EOL> from exabgp . bgp . message . update . attribute import Origin <EOL> from exabgp . bgp . message . update . attribute import MED <EOL> from exabgp . bgp . message . update . attribute import ASPath <EOL> from exabgp . bgp . message . update . attribute import LocalPreference <EOL> from exabgp . bgp . message . update . attribute import AtomicAggregate <EOL> from exabgp . bgp . message . update . attribute import Aggregator <EOL> from exabgp . bgp . message . update . attribute import Aggregator4 <EOL> from exabgp . bgp . message . update . attribute import OriginatorID <EOL> from exabgp . bgp . message . update . attribute import ClusterID <EOL> from exabgp . bgp . message . update . attribute import ClusterList <EOL> from exabgp . bgp . message . update . attribute import AIGP <EOL> from exabgp . bgp . message . update . attribute import GenericAttribute <EOL> from exabgp . bgp . message . update . attribute . community . community import Community <EOL> from exabgp . bgp . message . update . attribute . community . communities import Communities <EOL> from exabgp . bgp . message . update . attribute . community . extended import ExtendedCommunity <EOL> from exabgp . bgp . message . update . attribute . community . extended import ExtendedCommunities <EOL> from exabgp . bgp . message . update . nlri . qualifier import PathInfo <EOL> from exabgp . rib . change import Change <EOL> class Range ( IP ) : <EOL> def __init__ ( self , ip , packed , mask ) : <EOL> IP . init ( self , ip , packed ) <EOL> self . mask = mask <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( self . top ( ) , self . mask ) <EOL> def prefix ( tokeniser ) : <EOL> ip = tokeniser ( ) <EOL> try : <EOL> ip , mask = ip . split ( '<STR_LIT:/>' ) <EOL> mask = int ( mask ) <EOL> except ValueError : <EOL> mask = <NUM_LIT:32> <EOL> tokeniser . afi = IP . toafi ( ip ) <EOL> return Range ( ip , IP . pton ( ip ) , mask ) <EOL> def path_information ( tokeniser ) : <EOL> pi = tokeniser ( ) <EOL> if pi . isdigit ( ) : <EOL> return PathInfo ( integer = int ( pi ) ) <EOL> else : <EOL> return PathInfo ( ip = pi ) <EOL> def next_hop ( tokeniser ) : <EOL> value = tokeniser ( ) <EOL> if value . lower ( ) == '<STR_LIT>' : <EOL> return IPSelf ( tokeniser . afi ) , NextHopSelf ( tokeniser . afi ) <EOL> else : <EOL> ip = IP . create ( value ) <EOL> if ip . afi == AFI . ipv4 : <EOL> return ip , NextHop ( ip . top ( ) ) <EOL> return ip , None <EOL> def inet ( tokeniser ) : <EOL> ipmask = prefix ( tokeniser ) <EOL> inet = INET ( <EOL> afi = IP . toafi ( ipmask . top ( ) ) , <EOL> safi = IP . tosafi ( ipmask . top ( ) ) , <EOL> action = OUT . UNSET <EOL> ) <EOL> inet . cidr = CIDR ( ipmask . ton ( ) , ipmask . mask ) <EOL> return Change ( <EOL> inet , <EOL> Attributes ( ) <EOL> ) <EOL> def mpls ( tokeniser ) : <EOL> ipmask = prefix ( tokeniser ) <EOL> mpls = IPVPN ( <EOL> afi = IP . toafi ( ipmask . top ( ) ) , <EOL> safi = IP . tosafi ( ipmask . top ( ) ) , <EOL> action = OUT . UNSET <EOL> ) <EOL> mpls . cidr = CIDR ( ipmask . ton ( ) , ipmask . mask ) <EOL> return Change ( <EOL> mpls , <EOL> Attributes ( ) <EOL> ) <EOL> def attribute ( tokeniser ) : <EOL> start = tokeniser ( ) <EOL> if start != '<STR_LIT:[>' : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> code = tokeniser ( ) . lower ( ) <EOL> if not code . startswith ( '<STR_LIT>' ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> try : <EOL> code = int ( code , <NUM_LIT:16> ) <EOL> except ValueError : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> flag = tokeniser ( ) . lower ( ) <EOL> if not flag . startswith ( '<STR_LIT>' ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> try : <EOL> flag = int ( flag , <NUM_LIT:16> ) <EOL> except ValueError : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> data = tokeniser ( ) . lower ( ) <EOL> if not data . startswith ( '<STR_LIT>' ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if len ( data ) % <NUM_LIT:2> : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> data = '<STR_LIT>' . join ( chr ( int ( data [ _ : _ + <NUM_LIT:2> ] , <NUM_LIT:16> ) ) for _ in range ( <NUM_LIT:2> , len ( data ) , <NUM_LIT:2> ) ) <EOL> end = tokeniser ( ) <EOL> if end != '<STR_LIT:]>' : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> return GenericAttribute ( code , flag , data ) <EOL> def aigp ( tokeniser ) : <EOL> if not tokeniser . tokens : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> value = tokeniser ( ) <EOL> base = <NUM_LIT:16> if value . lower ( ) . startswith ( '<STR_LIT>' ) else <NUM_LIT:10> <EOL> try : <EOL> number = int ( value , base ) <EOL> except ValueError : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> return AIGP ( '<STR_LIT>' + pack ( '<STR_LIT>' , number ) ) <EOL> def origin ( tokeniser ) : <EOL> value = tokeniser ( ) . lower ( ) <EOL> if value == '<STR_LIT>' : <EOL> return Origin ( Origin . IGP ) <EOL> if value == '<STR_LIT>' : <EOL> return Origin ( Origin . EGP ) <EOL> if value == '<STR_LIT>' : <EOL> return Origin ( Origin . INCOMPLETE ) <EOL> raise ValueError ( '<STR_LIT>' % value ) <EOL> def med ( tokeniser ) : <EOL> value = tokeniser ( ) <EOL> if not value . isdigit ( ) : <EOL> raise ValueError ( '<STR_LIT>' % value ) <EOL> return MED ( int ( value ) ) <EOL> def as_path ( tokeniser ) : <EOL> as_seq = [ ] <EOL> as_set = [ ] <EOL> value = tokeniser ( ) <EOL> inset = False <EOL> try : <EOL> if value == '<STR_LIT:[>' : <EOL> while True : <EOL> value = tokeniser ( ) <EOL> if value == '<STR_LIT:U+002C>' : <EOL> continue <EOL> if value in ( '<STR_LIT:(>' , '<STR_LIT:[>' ) : <EOL> inset = True <EOL> while True : <EOL> value = tokeniser ( ) <EOL> if value == '<STR_LIT:)>' : <EOL> break <EOL> as_set . append ( ASN . from_string ( value ) ) <EOL> if value == '<STR_LIT:)>' : <EOL> inset = False <EOL> continue <EOL> if value == '<STR_LIT:]>' : <EOL> if inset : <EOL> inset = False <EOL> continue <EOL> break <EOL> as_seq . append ( ASN . from_string ( value ) ) <EOL> else : <EOL> as_seq . append ( ASN . from_string ( value ) ) <EOL> except ValueError : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> return ASPath ( as_seq , as_set ) <EOL> def local_preference ( tokeniser ) : <EOL> value = tokeniser ( ) <EOL> if not value . isdigit ( ) : <EOL> raise ValueError ( '<STR_LIT>' % value ) <EOL> return LocalPreference ( int ( value ) ) <EOL> def atomic_aggregate ( tokeniser ) : <EOL> return AtomicAggregate ( ) <EOL> def aggregator ( tokeniser ) : <EOL> first = tokeniser . peek ( ) <EOL> eat = True if first == '<STR_LIT:(>' else False <EOL> if eat : <EOL> tokeniser ( ) <EOL> try : <EOL> as_number , address = tokeniser ( ) . split ( '<STR_LIT::>' ) <EOL> local_as = ASN . from_string ( as_number ) <EOL> local_address = RouterID ( address ) <EOL> except ( ValueError , IndexError ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if eat : <EOL> if tokeniser ( ) != '<STR_LIT:)>' : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if local_as . asn4 ( ) : <EOL> return Aggregator4 ( local_as , local_address ) <EOL> return Aggregator ( local_as , local_address ) <EOL> def originator_id ( tokeniser ) : <EOL> value = tokeniser ( ) <EOL> if value . count ( '<STR_LIT:.>' ) != <NUM_LIT:3> : <EOL> raise ValueError ( '<STR_LIT>' % value ) <EOL> if not all ( _ . isdigit ( ) for _ in value . split ( '<STR_LIT:.>' ) ) : <EOL> raise ValueError ( '<STR_LIT>' % value ) <EOL> return OriginatorID ( value ) <EOL> def cluster_list ( tokeniser ) : <EOL> clusterids = [ ] <EOL> value = tokeniser ( ) <EOL> try : <EOL> if value == '<STR_LIT:[>' : <EOL> while True : <EOL> value = tokeniser ( ) <EOL> if value == '<STR_LIT:]>' : <EOL> break <EOL> clusterids . append ( ClusterID ( value ) ) <EOL> else : <EOL> clusterids . append ( ClusterID ( value ) ) <EOL> if not clusterids : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> return ClusterList ( clusterids ) <EOL> except ValueError : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> def _community ( value ) : <EOL> separator = value . find ( '<STR_LIT::>' ) <EOL> if separator > <NUM_LIT:0> : <EOL> prefix = value [ : separator ] <EOL> suffix = value [ separator + <NUM_LIT:1> : ] <EOL> if not prefix . isdigit ( ) or not suffix . isdigit ( ) : <EOL> raise ValueError ( '<STR_LIT>' % value ) <EOL> prefix , suffix = int ( prefix ) , int ( suffix ) <EOL> if prefix > Community . MAX : <EOL> raise ValueError ( '<STR_LIT>' % value ) <EOL> if suffix > Community . MAX : <EOL> raise ValueError ( '<STR_LIT>' % value ) <EOL> return Community ( pack ( '<STR_LIT>' , ( prefix << <NUM_LIT:16> ) + suffix ) ) <EOL> elif value [ : <NUM_LIT:2> ] . lower ( ) == '<STR_LIT>' : <EOL> number = long ( value , <NUM_LIT:16> ) <EOL> if number > Community . MAX : <EOL> raise ValueError ( '<STR_LIT>' % value ) <EOL> return Community ( pack ( '<STR_LIT>' , number ) ) <EOL> else : <EOL> low = value . lower ( ) <EOL> if low == '<STR_LIT>' : <EOL> return Community ( Community . NO_EXPORT ) <EOL> elif low == '<STR_LIT>' : <EOL> return Community ( Community . NO_ADVERTISE ) <EOL> elif low == '<STR_LIT>' : <EOL> return Community ( Community . NO_EXPORT_SUBCONFED ) <EOL> elif low == '<STR_LIT>' or low == '<STR_LIT>' : <EOL> return Community ( Community . NO_PEER ) <EOL> elif value . isdigit ( ) : <EOL> number = int ( value ) <EOL> if number > Community . MAX : <EOL> raise ValueError ( '<STR_LIT>' % value ) <EOL> return Community ( pack ( '<STR_LIT>' , number ) ) <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' % value ) <EOL> def community ( tokeniser ) : <EOL> communities = Communities ( ) <EOL> value = tokeniser ( ) <EOL> if value == '<STR_LIT:[>' : <EOL> while True : <EOL> value = tokeniser ( ) <EOL> if value == '<STR_LIT:]>' : <EOL> break <EOL> communities . add ( _community ( value ) ) <EOL> else : <EOL> communities . add ( _community ( value ) ) <EOL> return communities <EOL> _HEADER = { <EOL> '<STR_LIT:target>' : chr ( <NUM_LIT> ) + chr ( <NUM_LIT> ) , <EOL> '<STR_LIT>' : chr ( <NUM_LIT> ) + chr ( <NUM_LIT> ) , <EOL> '<STR_LIT>' : chr ( <NUM_LIT> ) + chr ( <NUM_LIT> ) , <EOL> '<STR_LIT>' : chr ( <NUM_LIT> ) + chr ( <NUM_LIT> ) , <EOL> '<STR_LIT>' : chr ( <NUM_LIT> ) + chr ( <NUM_LIT> ) , <EOL> '<STR_LIT>' : chr ( <NUM_LIT> ) + chr ( <NUM_LIT> ) , <EOL> '<STR_LIT>' : chr ( <NUM_LIT> ) + chr ( <NUM_LIT> ) , <EOL> } <EOL> _SIZE = { <EOL> '<STR_LIT:target>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : <NUM_LIT:4> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> } <EOL> _SIZE_H = <NUM_LIT> <EOL> def _extended_community ( value ) : <EOL> if value [ : <NUM_LIT:2> ] . lower ( ) == '<STR_LIT>' : <EOL> if len ( value ) % <NUM_LIT:2> : <EOL> raise ValueError ( '<STR_LIT>' % value ) <EOL> raw = '<STR_LIT>' . join ( [ chr ( int ( value [ _ : _ + <NUM_LIT:2> ] , <NUM_LIT:16> ) ) for _ in range ( <NUM_LIT:2> , len ( value ) , <NUM_LIT:2> ) ] ) <EOL> return ExtendedCommunity . unpack ( raw ) <EOL> elif value . count ( '<STR_LIT::>' ) : <EOL> components = value . split ( '<STR_LIT::>' ) <EOL> command = '<STR_LIT:target>' if len ( components ) == <NUM_LIT:2> else components . pop ( <NUM_LIT:0> ) <EOL> if command not in _HEADER : <EOL> raise ValueError ( '<STR_LIT>' % command ) <EOL> if len ( components ) != _SIZE [ command ] : <EOL> raise ValueError ( '<STR_LIT>' % ( command , len ( components ) ) ) <EOL> header = _HEADER . get ( command , None ) <EOL> if header is None : <EOL> raise ValueError ( '<STR_LIT>' % command ) <EOL> if command == '<STR_LIT>' : <EOL> return ExtendedCommunity . unpack ( header + pack ( '<STR_LIT>' , * [ int ( _ ) for _ in components ] ) ) <EOL> _ga , _la = components <EOL> ga , la = _ga . upper ( ) , _la . upper ( ) <EOL> if command in ( '<STR_LIT:target>' , '<STR_LIT>' ) : <EOL> if '<STR_LIT:.>' in ga or '<STR_LIT:.>' in la : <EOL> gc = ga . count ( '<STR_LIT:.>' ) <EOL> lc = la . count ( '<STR_LIT:.>' ) <EOL> if gc == <NUM_LIT:0> and lc == <NUM_LIT:3> : <EOL> return ExtendedCommunity . unpack ( header + pack ( '<STR_LIT>' , int ( ga ) , * [ int ( _ ) for _ in la . split ( '<STR_LIT:.>' ) ] ) ) <EOL> if gc == <NUM_LIT:3> and lc == <NUM_LIT:0> : <EOL> return ExtendedCommunity . unpack ( header + pack ( '<STR_LIT>' , * [ int ( _ ) for _ in ga . split ( '<STR_LIT:.>' ) ] + [ int ( la ) ] ) ) <EOL> iga = int ( ga [ : - <NUM_LIT:1> ] ) if '<STR_LIT:L>' in ga else int ( ga ) <EOL> ila = int ( la [ : - <NUM_LIT:1> ] ) if '<STR_LIT:L>' in la else int ( la ) <EOL> if command == '<STR_LIT:target>' : <EOL> if ga . endswith ( '<STR_LIT:L>' ) or iga > _SIZE_H : <EOL> return ExtendedCommunity . unpack ( _HEADER [ '<STR_LIT>' ] + pack ( '<STR_LIT>' , iga , ila ) , None ) <EOL> else : <EOL> return ExtendedCommunity . unpack ( header + pack ( '<STR_LIT>' , iga , ila ) , None ) <EOL> if command == '<STR_LIT>' : <EOL> if ga . endswith ( '<STR_LIT:L>' ) or iga > _SIZE_H : <EOL> return ExtendedCommunity . unpack ( _HEADER [ '<STR_LIT>' ] + pack ( '<STR_LIT>' , iga , ila ) , None ) <EOL> else : <EOL> return ExtendedCommunity . unpack ( header + pack ( '<STR_LIT>' , iga , ila ) , None ) <EOL> if command == '<STR_LIT>' : <EOL> return ExtendedCommunity . unpack ( _HEADER [ '<STR_LIT>' ] + pack ( '<STR_LIT>' , iga , ila ) , None ) <EOL> if command == '<STR_LIT>' : <EOL> return ExtendedCommunity . unpack ( _HEADER [ '<STR_LIT>' ] + pack ( '<STR_LIT>' , iga , ila ) , None ) <EOL> if command in ( '<STR_LIT>' , ) : <EOL> return ExtendedCommunity . unpack ( header + pack ( '<STR_LIT>' , iga , ila ) , None ) <EOL> raise ValueError ( '<STR_LIT>' % command ) <EOL> elif value == '<STR_LIT>' : <EOL> header = _HEADER [ value ] <EOL> return ExtendedCommunity . unpack ( header + pack ( '<STR_LIT>' , <NUM_LIT:0> , <NUM_LIT:0> ) , None ) <EOL> else : <EOL> raise ValueError ( '<STR_LIT>' % value ) <EOL> def extended_community ( tokeniser ) : <EOL> communities = ExtendedCommunities ( ) <EOL> value = tokeniser ( ) <EOL> if value == '<STR_LIT:[>' : <EOL> while True : <EOL> value = tokeniser ( ) <EOL> if value == '<STR_LIT:]>' : <EOL> break <EOL> communities . add ( _extended_community ( value ) ) <EOL> else : <EOL> communities . add ( _extended_community ( value ) ) <EOL> return communities <EOL> def name ( tokeniser ) : <EOL> class Name ( str ) : <EOL> ID = Attribute . CODE . INTERNAL_NAME <EOL> return Name ( tokeniser ( ) ) <EOL> def split ( tokeniser ) : <EOL> class Split ( int ) : <EOL> ID = Attribute . CODE . INTERNAL_SPLIT <EOL> cidr = tokeniser ( ) <EOL> if not cidr or cidr [ <NUM_LIT:0> ] != '<STR_LIT:/>' : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> size = cidr [ <NUM_LIT:1> : ] <EOL> if not size . isdigit ( ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> return Split ( int ( size ) ) <EOL> def watchdog ( tokeniser ) : <EOL> class Watchdog ( str ) : <EOL> ID = Attribute . CODE . INTERNAL_WATCHDOG <EOL> command = tokeniser ( ) <EOL> if command . lower ( ) in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> raise ValueError ( '<STR_LIT>' % command ) <EOL> return Watchdog ( command ) <EOL> def withdraw ( tokeniser = None ) : <EOL> class Withdrawn ( object ) : <EOL> ID = Attribute . CODE . INTERNAL_WITHDRAW <EOL> return Withdrawn ( ) </s>
