exabgp.configuration.static.parser._extended_community() - Code Metrics - Exa-Networks/exabgp - Measure and Improve Code Quality continuously with Scrutinizer

_extended_community() F
last analyzed 2020-07-19 19:06 UTC

↳ Parent: exabgp.configuration.static.parser

Complexity

Conditions

Size

Total Lines	75
Code Lines	52

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	29
eloc	52
nop	1
dl	0
loc	75
rs	0
c	0
b	0
f	0

How to fix Long Method Complexity

# encoding: utf-8
"""
inet/parser.py

Created by Thomas Mangin on 2015-06-04.
Copyright (c) 2009-2017 Exa Networks. All rights reserved.
License: 3-clause BSD. (See the COPYRIGHT file)
"""

from struct import pack
import sys

from exabgp.protocol.ip import IP
from exabgp.protocol.ip import IPSelf
from exabgp.protocol.ip import IPRange
from exabgp.protocol.family import AFI

# from exabgp.protocol.family import SAFI

from exabgp.bgp.message import OUT
from exabgp.bgp.message.update.nlri import CIDR
from exabgp.bgp.message.update.nlri import INET
from exabgp.bgp.message.update.nlri import IPVPN

from exabgp.bgp.message.open import ASN
from exabgp.bgp.message.open import RouterID
from exabgp.bgp.message.update.attribute import Attribute
from exabgp.bgp.message.update.attribute import Attributes
from exabgp.bgp.message.update.attribute import NextHop
from exabgp.bgp.message.update.attribute import NextHopSelf
from exabgp.bgp.message.update.attribute import Origin
from exabgp.bgp.message.update.attribute import MED
from exabgp.bgp.message.update.attribute import ASPath
from exabgp.bgp.message.update.attribute import LocalPreference
from exabgp.bgp.message.update.attribute import AtomicAggregate
from exabgp.bgp.message.update.attribute import Aggregator
from exabgp.bgp.message.update.attribute import Aggregator4
from exabgp.bgp.message.update.attribute import OriginatorID
from exabgp.bgp.message.update.attribute import ClusterID
from exabgp.bgp.message.update.attribute import ClusterList
from exabgp.bgp.message.update.attribute import AIGP
from exabgp.bgp.message.update.attribute import GenericAttribute

from exabgp.bgp.message.update.attribute.community import Community
from exabgp.bgp.message.update.attribute.community import Communities
from exabgp.bgp.message.update.attribute.community import LargeCommunity
from exabgp.bgp.message.update.attribute.community import LargeCommunities
from exabgp.bgp.message.update.attribute.community import ExtendedCommunity
from exabgp.bgp.message.update.attribute.community import ExtendedCommunities

from exabgp.bgp.message.update.nlri.qualifier import PathInfo

from exabgp.rib.change import Change


def prefix(tokeniser):
    # XXX: could raise
    ip = tokeniser()
    try:
        ip, mask = ip.split('/')
    except ValueError:
        mask = '32'
        if ':' in ip:
            mask = '128'

    tokeniser.afi = IP.toafi(ip)
    iprange = IPRange.create(ip, mask)

    if iprange.address() & iprange.mask.hostmask() != 0:
        raise ValueError('invalid network %s for netmask %s' % (ip, mask))

    return iprange


def path_information(tokeniser):
    pi = tokeniser()
    if pi.isdigit():
        return PathInfo(integer=int(pi))
    else:
        return PathInfo(ip=pi)


def next_hop(tokeniser):
    value = tokeniser()

    if value.lower() == 'self':
        return IPSelf(tokeniser.afi), NextHopSelf(tokeniser.afi)
    else:
        ip = IP.create(value)
        return ip, NextHop(ip.top())


# XXX: using OUT.UNSET should we use the following ?
# action = OUT.ANNOUNCE if tokeniser.announce else OUT.WITHDRAW


def inet(tokeniser):
    ipmask = prefix(tokeniser)
    inet = INET(afi=IP.toafi(ipmask.top()), safi=IP.tosafi(ipmask.top()), action=OUT.UNSET)
    inet.cidr = CIDR(ipmask.ton(), ipmask.mask)

    return Change(inet, Attributes())


# XXX: using OUT.ANNOUNCE should we use the following ?
# action = OUT.ANNOUNCE if tokeniser.announce else OUT.WITHDRAW


def mpls(tokeniser):
    ipmask = prefix(tokeniser)
    mpls = IPVPN(afi=IP.toafi(ipmask.top()), safi=IP.tosafi(ipmask.top()), action=OUT.ANNOUNCE)
    mpls.cidr = CIDR(ipmask.ton(), ipmask.mask)

    return Change(mpls, Attributes())


def attribute(tokeniser):
    start = tokeniser()
    if start != '[':
        raise ValueError('invalid attribute, does not starts with [')

    code = tokeniser().lower()
    if not code.startswith('0x'):
        raise ValueError('invalid attribute, code is not 0x hexadecimal')
    try:
        code = int(code, 16)
    except ValueError:
        raise ValueError('invalid attribute, code is not 0x hexadecimal')

    flag = tokeniser().lower()
    if not flag.startswith('0x'):
        raise ValueError('invalid attribute, flag is not 0x hexadecimal')
    try:
        flag = int(flag, 16)
    except ValueError:
        raise ValueError('invalid attribute, flag is not 0x hexadecimal')

    data = tokeniser().lower()
    if not data.startswith('0x'):
        raise ValueError('invalid attribute, data is not 0x hexadecimal')
    if len(data) % 2:
        raise ValueError('invalid attribute, data is not 0x hexadecimal')
    data = b''.join(bytes([int(data[_ : _ + 2], 16)]) for _ in range(2, len(data), 2))

    end = tokeniser()
    if end != ']':
        raise ValueError('invalid attribute, does not ends with ]')

    return GenericAttribute(code, flag, data)

    # for ((ID,flag),klass) in Attribute.registered_attributes.items():
    # 	length = len(data)
    # 	if code == ID and flag | Attribute.Flag.EXTENDED_LENGTH == klass.FLAG | Attribute.Flag.EXTENDED_LENGTH:
    # 		# if length > 0xFF or flag & Attribute.Flag.EXTENDED_LENGTH:
    # 		# 	raw = pack('!BBH',flag,code,length & (0xFF-Attribute.Flag.EXTENDED_LENGTH)) + data
    # 		# else:
    # 		# 	raw = pack('!BBB',flag,code,length) + data
    # 		return klass.unpack(data,None)


def aigp(tokeniser):
    if not tokeniser.tokens:
        raise ValueError('aigp requires number (decimal or hexadecimal 0x prefixed)')
    value = tokeniser()
    base = 16 if value.lower().startswith('0x') else 10
    try:
        number = int(value, base)
    except ValueError:
        raise ValueError('aigp requires number (decimal or hexadecimal 0x prefixed)')

    return AIGP(b'\x01\x00\x0b' + pack('!Q', number))


def origin(tokeniser):
    value = tokeniser().lower()
    if value == 'igp':
        return Origin(Origin.IGP)
    if value == 'egp':
        return Origin(Origin.EGP)
    if value == 'incomplete':
        return Origin(Origin.INCOMPLETE)
    raise ValueError('unknown origin %s' % value)


def med(tokeniser):
    value = tokeniser()
    if not value.isdigit():
        raise ValueError('invalid MED %s' % value)
    return MED(int(value))


def as_path(tokeniser):
    as_seq = []
    as_set = []
    value = tokeniser()
    inset = False
    try:
        if value == '[':
            while True:
                value = tokeniser()
                if value == ',':
                    continue
                if value in ('(', '['):
                    inset = True
                    while True:
                        value = tokeniser()
                        if value in (')', ']'):
                            break
                        as_set.append(ASN.from_string(value))
                if value == ')':
                    inset = False
                    continue
                if value == ']':
                    if inset:
                        inset = False
                        continue
                    break
                as_seq.append(ASN.from_string(value))
        else:
            as_seq.append(ASN.from_string(value))
    except ValueError:
        raise ValueError('could not parse as-path')
    return ASPath(as_seq, as_set)


def local_preference(tokeniser):
    value = tokeniser()
    if not value.isdigit():
        raise ValueError('invalid local preference %s' % value)
    return LocalPreference(int(value))


def atomic_aggregate(tokeniser):
    return AtomicAggregate()


def aggregator(tokeniser):
    agg = tokeniser()
    eat = True if agg == '(' else False

    if eat:
        agg = tokeniser()
        if agg.endswith(')'):
            eat = False
            agg = agg[:-1]
    elif agg.startswith('('):
        if agg.endswith(')'):
            eat = False
            agg = agg[1:-1]
        else:
            eat = True
            agg = agg[1:]

    try:
        as_number, address = agg.split(':')
        local_as = ASN.from_string(as_number)
        local_address = RouterID(address)
    except (ValueError, IndexError):
        raise ValueError('invalid aggregator')

    if eat:
        if tokeniser() != ')':
            raise ValueError('invalid aggregator')

    return Aggregator(local_as, local_address)


def originator_id(tokeniser):
    value = tokeniser()
    if value.count('.') != 3:
        raise ValueError('invalid Originator ID %s' % value)
    if not all(_.isdigit() for _ in value.split('.')):
        raise ValueError('invalid Originator ID %s' % value)
    return OriginatorID(value)


def cluster_list(tokeniser):
    clusterids = []
    value = tokeniser()
    try:
        if value == '[':
            while True:
                value = tokeniser()
                if value == ']':
                    break
                clusterids.append(ClusterID(value))
        else:
            clusterids.append(ClusterID(value))
        if not clusterids:
            raise ValueError('no cluster-id in the cluster list')
        return ClusterList(clusterids)
    except ValueError:
        raise ValueError('invalud cluster list')


# XXX: Community does does not cache anymore .. we SHOULD really do it !


def _community(value):
    separator = value.find(':')
    if separator > 0:
        prefix = value[:separator]
        suffix = value[separator + 1 :]

        if not prefix.isdigit() or not suffix.isdigit():
            raise ValueError('invalid community %s' % value)

        prefix, suffix = int(prefix), int(suffix)

        if prefix > Community.MAX:
            raise ValueError('invalid community %s (prefix too large)' % value)

        if suffix > Community.MAX:
            raise ValueError('invalid community %s (suffix too large)' % value)

        return Community(pack('!L', (prefix << 16) + suffix))

    elif value[:2].lower() == '0x':
        number = int(value, 16)
        if number > Community.MAX:
            raise ValueError('invalid community %s (too large)' % value)
        return Community(pack('!L', number))

    else:
        low = value.lower()
        if low == 'no-export':
            return Community(Community.NO_EXPORT)
        elif low == 'no-advertise':
            return Community(Community.NO_ADVERTISE)
        elif low == 'no-export-subconfed':
            return Community(Community.NO_EXPORT_SUBCONFED)
        # no-peer is not a correct syntax but I am sure someone will make the mistake :)
        elif low == 'nopeer' or low == 'no-peer':
            return Community(Community.NO_PEER)
        elif low == 'blackhole':
            return Community(Community.BLACKHOLE)
        elif value.isdigit():
            number = int(value)
            if number > Community.MAX:
                raise ValueError('invalid community %s (too large)' % value)
            return Community(pack('!L', number))
        else:
            raise ValueError('invalid community name %s' % value)


def community(tokeniser):
    communities = Communities()

    value = tokeniser()
    if value == '[':
        while True:
            value = tokeniser()
            if value == ']':
                break
            communities.add(_community(value))
    else:
        communities.add(_community(value))

    return communities


def _large_community(value):
    separator = value.find(':')
    if separator > 0:
        prefix, affix, suffix = value.split(':')

        if not any(map(lambda c: c.isdigit(), [prefix, affix, suffix])):
            raise ValueError('invalid community %s' % value)

        prefix, affix, suffix = map(int, [prefix, affix, suffix])

        for i in [prefix, affix, suffix]:
            if i > LargeCommunity.MAX:
                raise ValueError('invalid community %i in %s too large' % (i, value))

        return LargeCommunity(pack('!LLL', prefix, affix, suffix))

    elif value[:2].lower() == '0x':
        number = int(value)
        if number > LargeCommunity.MAX:
            raise ValueError('invalid large community %s (too large)' % value)
        return LargeCommunity(pack('!LLL', number >> 64, (number >> 32) & 0xFFFFFFFF, number & 0xFFFFFFFF))

    else:
        low = value.lower()
        if value.isdigit():
            number = int(value)
            if number > LargeCommunity.MAX:
                raise ValueError('invalid large community %s (too large)' % value)
            return LargeCommunity(pack('!LLL', number >> 64, (number >> 32) & 0xFFFFFFFF, number & 0xFFFFFFFF))
        else:
            raise ValueError('invalid large community name %s' % value)


def large_community(tokeniser):
    large_communities = LargeCommunities()

    value = tokeniser()
    if value == '[':
        while True:
            value = tokeniser()
            if value == ']':
                break
            lc = _large_community(value)
            if lc in large_communities.communities:
                continue
            large_communities.add(lc)
    else:
        large_communities.add(_large_community(value))

    return large_communities


# fmt: off
_HEADER = {
    # header and subheader
    'target':   bytes([0x00, 0x02]),
    'target4':  bytes([0x02, 0x02]),
    'origin':   bytes([0x00, 0x03]),
    'origin4':  bytes([0x02, 0x03]),
    'redirect': bytes([0x80, 0x08]),
    'l2info':   bytes([0x80, 0x0A]),
    'redirect-to-nexthop': bytes([0x08, 0x00]),
    'bandwidth': bytes([0x40, 0x04]),
}

_SIZE = {
    # fmt: off
    'target':   2,
    'target4':  2,
    'origin':   2,
    'origin4':  2,
    'redirect': 2,
    'l2info':   4,
    'redirect-to-nexthop': 0,
    'bandwidth': 2,
}
# fmt: on

_SIZE_H = 0xFFFF


def _extended_community(value):
    if value[:2].lower() == '0x':
        # we could raise if the length is not 8 bytes (16 chars)
        if len(value) % 2:
            raise ValueError('invalid extended community %s' % value)
        raw = b''.join(bytes([int(value[_ : _ + 2], 16)]) for _ in range(2, len(value), 2))
        return ExtendedCommunity.unpack(raw)
    elif value.count(':'):
        components = value.split(':')
        command = 'target' if len(components) == 2 else components.pop(0)

        if command not in _HEADER:
            raise ValueError('invalid extended community %s (only origin,target or l2info are supported) ' % command)

        if len(components) != _SIZE[command]:
            raise ValueError('invalid extended community %s, expecting %d fields ' % (command, len(components)))

        header = _HEADER.get(command, None)

        if header is None:
            raise ValueError('unknown extended community %s' % command)

        if command == 'l2info':
            # encaps, control, mtu, site
            return ExtendedCommunity.unpack(header + pack('!BBHH', *[int(_) for _ in components]))

        _ga, _la = components
        ga, la = _ga.upper(), _la.upper()

        if command in ('target', 'origin'):
            # global admin, local admin
            if '.' in ga or '.' in la:
                gc = ga.count('.')
                lc = la.count('.')
                if gc == 0 and lc == 3:
                    # ASN first, IP second
                    return ExtendedCommunity.unpack(header + pack('!HBBBB', int(ga), *[int(_) for _ in la.split('.')]))
                if gc == 3 and lc == 0:
                    # IP first, ASN second
                    return ExtendedCommunity.unpack(
                        header + pack('!BBBBH', *[int(_) for _ in ga.split('.')] + [int(la)])
                    )

        iga = int(ga[:-1]) if 'L' in ga else int(ga)
        ila = int(la[:-1]) if 'L' in la else int(la)

        if command == 'target':
            if ga.endswith('L') or iga > _SIZE_H:
                return ExtendedCommunity.unpack(_HEADER['target4'] + pack('!LH', iga, ila), None)
            else:
                return ExtendedCommunity.unpack(header + pack('!HI', iga, ila), None)
        if command == 'origin':
            if ga.endswith('L') or iga > _SIZE_H:
                return ExtendedCommunity.unpack(_HEADER['origin4'] + pack('!LH', iga, ila), None)
            else:
                return ExtendedCommunity.unpack(header + pack('!HI', iga, ila), None)

        if command == 'target4':
            return ExtendedCommunity.unpack(_HEADER['target4'] + pack('!LH', iga, ila), None)

        if command == 'origin4':
            return ExtendedCommunity.unpack(_HEADER['origin4'] + pack('!LH', iga, ila), None)

        if command == 'redirect':
            return ExtendedCommunity.unpack(header + pack('!HL', iga, ila), None)

        if command == 'bandwidth':
            return ExtendedCommunity.unpack(_HEADER['bandwidth'] + pack('!Hf', iga, ila), None)

        raise ValueError('invalid extended community %s' % command)
    elif value == 'redirect-to-nexthop':
        header = _HEADER[value]
        return ExtendedCommunity.unpack(header + pack('!HL', 0, 0), None)
    else:
        raise ValueError('invalid extended community %s - lc+gc' % value)


# The previous code was extracting the extended-community class from the attributes
# And adding to it.


def extended_community(tokeniser):
    communities = ExtendedCommunities()

    value = tokeniser()
    if value == '[':
        while True:
            value = tokeniser()
            if value == ']':
                break
            communities.add(_extended_community(value))
    else:
        communities.add(_extended_community(value))

    return communities


# Duck class, faking part of the Attribute interface
# We add this to routes when when need o split a route in smaller route
# The value stored is the longer netmask we want to use
# As this is not a real BGP attribute this stays in the configuration file


def name(tokeniser):
    class Name(str):
        ID = Attribute.CODE.INTERNAL_NAME

    return Name(tokeniser())


def split(tokeniser):
    class Split(int):
        ID = Attribute.CODE.INTERNAL_SPLIT

    cidr = tokeniser()

    if not cidr or cidr[0] != '/':
        raise ValueError('split /<number>')

    size = cidr[1:]

    if not size.isdigit():
        raise ValueError('split /<number>')

    return Split(int(size))


def watchdog(tokeniser):
    class Watchdog(str):
        ID = Attribute.CODE.INTERNAL_WATCHDOG

    command = tokeniser()
    if command.lower() in ['announce', 'withdraw']:
        raise ValueError('invalid watchdog name %s' % command)
    return Watchdog(command)


def withdraw(tokeniser=None):
    class Withdrawn(object):
        ID = Attribute.CODE.INTERNAL_WITHDRAW

    return Withdrawn()


1			# encoding: utf-8
2			"""
3			inet/parser.py
4
5			Created by Thomas Mangin on 2015-06-04.
6			Copyright (c) 2009-2017 Exa Networks. All rights reserved.
7			License: 3-clause BSD. (See the COPYRIGHT file)
8			"""
9
10			from struct import pack
11			import sys
12
13			from exabgp.protocol.ip import IP
14			from exabgp.protocol.ip import IPSelf
15			from exabgp.protocol.ip import IPRange
16			from exabgp.protocol.family import AFI
17
18			# from exabgp.protocol.family import SAFI
19
20			from exabgp.bgp.message import OUT
21			from exabgp.bgp.message.update.nlri import CIDR
22			from exabgp.bgp.message.update.nlri import INET
23			from exabgp.bgp.message.update.nlri import IPVPN
24
25			from exabgp.bgp.message.open import ASN
26			from exabgp.bgp.message.open import RouterID
27			from exabgp.bgp.message.update.attribute import Attribute
28			from exabgp.bgp.message.update.attribute import Attributes
29			from exabgp.bgp.message.update.attribute import NextHop
30			from exabgp.bgp.message.update.attribute import NextHopSelf
31			from exabgp.bgp.message.update.attribute import Origin
32			from exabgp.bgp.message.update.attribute import MED
33			from exabgp.bgp.message.update.attribute import ASPath
34			from exabgp.bgp.message.update.attribute import LocalPreference
35			from exabgp.bgp.message.update.attribute import AtomicAggregate
36			from exabgp.bgp.message.update.attribute import Aggregator
37			from exabgp.bgp.message.update.attribute import Aggregator4
38			from exabgp.bgp.message.update.attribute import OriginatorID
39			from exabgp.bgp.message.update.attribute import ClusterID
40			from exabgp.bgp.message.update.attribute import ClusterList
41			from exabgp.bgp.message.update.attribute import AIGP
42			from exabgp.bgp.message.update.attribute import GenericAttribute
43
44			from exabgp.bgp.message.update.attribute.community import Community
45			from exabgp.bgp.message.update.attribute.community import Communities
46			from exabgp.bgp.message.update.attribute.community import LargeCommunity
47			from exabgp.bgp.message.update.attribute.community import LargeCommunities
48			from exabgp.bgp.message.update.attribute.community import ExtendedCommunity
49			from exabgp.bgp.message.update.attribute.community import ExtendedCommunities
50
51			from exabgp.bgp.message.update.nlri.qualifier import PathInfo
52
53			from exabgp.rib.change import Change
54
55
56			def prefix(tokeniser):
57			# XXX: could raise
58			ip = tokeniser()
59			try:
60			ip, mask = ip.split('/')
61			except ValueError:
62			mask = '32'
63			if ':' in ip:
64			mask = '128'
65
66			tokeniser.afi = IP.toafi(ip)
67			iprange = IPRange.create(ip, mask)
68
69			if iprange.address() & iprange.mask.hostmask() != 0:
70			raise ValueError('invalid network %s for netmask %s' % (ip, mask))
71
72			return iprange
73
74
75			def path_information(tokeniser):
76			pi = tokeniser()
77			if pi.isdigit():
78			return PathInfo(integer=int(pi))
79			else:
80			return PathInfo(ip=pi)
81
82
83			def next_hop(tokeniser):
84			value = tokeniser()
85
86			if value.lower() == 'self':
87			return IPSelf(tokeniser.afi), NextHopSelf(tokeniser.afi)
88			else:
89			ip = IP.create(value)
90			return ip, NextHop(ip.top())
91
92
93			# XXX: using OUT.UNSET should we use the following ?
94			# action = OUT.ANNOUNCE if tokeniser.announce else OUT.WITHDRAW
95
96
97			def inet(tokeniser):
98			ipmask = prefix(tokeniser)
99			inet = INET(afi=IP.toafi(ipmask.top()), safi=IP.tosafi(ipmask.top()), action=OUT.UNSET)
100			inet.cidr = CIDR(ipmask.ton(), ipmask.mask)
101
102			return Change(inet, Attributes())
103
104
105			# XXX: using OUT.ANNOUNCE should we use the following ?
106			# action = OUT.ANNOUNCE if tokeniser.announce else OUT.WITHDRAW
107
108
109			def mpls(tokeniser):
110			ipmask = prefix(tokeniser)
111			mpls = IPVPN(afi=IP.toafi(ipmask.top()), safi=IP.tosafi(ipmask.top()), action=OUT.ANNOUNCE)
112			mpls.cidr = CIDR(ipmask.ton(), ipmask.mask)
113
114			return Change(mpls, Attributes())
115
116
117			def attribute(tokeniser):
118			start = tokeniser()
119			if start != '[':
120			raise ValueError('invalid attribute, does not starts with [')
121
122			code = tokeniser().lower()
123			if not code.startswith('0x'):
124			raise ValueError('invalid attribute, code is not 0x hexadecimal')
125			try:
126			code = int(code, 16)
127			except ValueError:
128			raise ValueError('invalid attribute, code is not 0x hexadecimal')
129
130			flag = tokeniser().lower()
131			if not flag.startswith('0x'):
132			raise ValueError('invalid attribute, flag is not 0x hexadecimal')
133			try:
134			flag = int(flag, 16)
135			except ValueError:
136			raise ValueError('invalid attribute, flag is not 0x hexadecimal')
137
138			data = tokeniser().lower()
139			if not data.startswith('0x'):
140			raise ValueError('invalid attribute, data is not 0x hexadecimal')
141			if len(data) % 2:
142			raise ValueError('invalid attribute, data is not 0x hexadecimal')
143			data = b''.join(bytes([int(data[_ : _ + 2], 16)]) for _ in range(2, len(data), 2))
144
145			end = tokeniser()
146			if end != ']':
147			raise ValueError('invalid attribute, does not ends with ]')
148
149			return GenericAttribute(code, flag, data)
150
151			# for ((ID,flag),klass) in Attribute.registered_attributes.items():
152			# length = len(data)
153			# if code == ID and flag \| Attribute.Flag.EXTENDED_LENGTH == klass.FLAG \| Attribute.Flag.EXTENDED_LENGTH:
154			# # if length > 0xFF or flag & Attribute.Flag.EXTENDED_LENGTH:
155			# # raw = pack('!BBH',flag,code,length & (0xFF-Attribute.Flag.EXTENDED_LENGTH)) + data
156			# # else:
157			# # raw = pack('!BBB',flag,code,length) + data
158			# return klass.unpack(data,None)
159
160
161			def aigp(tokeniser):
162			if not tokeniser.tokens:
163			raise ValueError('aigp requires number (decimal or hexadecimal 0x prefixed)')
164			value = tokeniser()
165			base = 16 if value.lower().startswith('0x') else 10
166			try:
167			number = int(value, base)
168			except ValueError:
169			raise ValueError('aigp requires number (decimal or hexadecimal 0x prefixed)')
170
171			return AIGP(b'\x01\x00\x0b' + pack('!Q', number))
172
173
174			def origin(tokeniser):
175			value = tokeniser().lower()
176			if value == 'igp':
177			return Origin(Origin.IGP)
178			if value == 'egp':
179			return Origin(Origin.EGP)
180			if value == 'incomplete':
181			return Origin(Origin.INCOMPLETE)
182			raise ValueError('unknown origin %s' % value)
183
184
185			def med(tokeniser):
186			value = tokeniser()
187			if not value.isdigit():
188			raise ValueError('invalid MED %s' % value)
189			return MED(int(value))
190
191
192			def as_path(tokeniser):
193			as_seq = []
194			as_set = []
195			value = tokeniser()
196			inset = False
197			try:
198			if value == '[':
199			while True:
200			value = tokeniser()
201			if value == ',':
202			continue
203			if value in ('(', '['):
204			inset = True
205			while True:
206			value = tokeniser()
207			if value in (')', ']'):
208			break
209			as_set.append(ASN.from_string(value))
210			if value == ')':
211			inset = False
212			continue
213			if value == ']':
214			if inset:
215			inset = False
216			continue
217			break
218			as_seq.append(ASN.from_string(value))
219			else:
220			as_seq.append(ASN.from_string(value))
221			except ValueError:
222			raise ValueError('could not parse as-path')
223			return ASPath(as_seq, as_set)
224
225
226			def local_preference(tokeniser):
227			value = tokeniser()
228			if not value.isdigit():
229			raise ValueError('invalid local preference %s' % value)
230			return LocalPreference(int(value))
231
232
233			def atomic_aggregate(tokeniser):
234			return AtomicAggregate()
235
236
237			def aggregator(tokeniser):
238			agg = tokeniser()
239			eat = True if agg == '(' else False
240
241			if eat:
242			agg = tokeniser()
243			if agg.endswith(')'):
244			eat = False
245			agg = agg[:-1]
246			elif agg.startswith('('):
247			if agg.endswith(')'):
248			eat = False
249			agg = agg[1:-1]
250			else:
251			eat = True
252			agg = agg[1:]
253
254			try:
255			as_number, address = agg.split(':')
256			local_as = ASN.from_string(as_number)
257			local_address = RouterID(address)
258			except (ValueError, IndexError):
259			raise ValueError('invalid aggregator')
260
261			if eat:
262			if tokeniser() != ')':
263			raise ValueError('invalid aggregator')
264
265			return Aggregator(local_as, local_address)
266
267
268			def originator_id(tokeniser):
269			value = tokeniser()
270			if value.count('.') != 3:
271			raise ValueError('invalid Originator ID %s' % value)
272			if not all(_.isdigit() for _ in value.split('.')):
273			raise ValueError('invalid Originator ID %s' % value)
274			return OriginatorID(value)
275
276
277			def cluster_list(tokeniser):
278			clusterids = []
279			value = tokeniser()
280			try:
281			if value == '[':
282			while True:
283			value = tokeniser()
284			if value == ']':
285			break
286			clusterids.append(ClusterID(value))
287			else:
288			clusterids.append(ClusterID(value))
289			if not clusterids:
290			raise ValueError('no cluster-id in the cluster list')
291			return ClusterList(clusterids)
292			except ValueError:
293			raise ValueError('invalud cluster list')
294
295
296			# XXX: Community does does not cache anymore .. we SHOULD really do it !
297
298
299			def _community(value):
300			separator = value.find(':')
301			if separator > 0:
302			prefix = value[:separator]
303			suffix = value[separator + 1 :]
304
305			if not prefix.isdigit() or not suffix.isdigit():
306			raise ValueError('invalid community %s' % value)
307
308			prefix, suffix = int(prefix), int(suffix)
309
310			if prefix > Community.MAX:
311			raise ValueError('invalid community %s (prefix too large)' % value)
312
313			if suffix > Community.MAX:
314			raise ValueError('invalid community %s (suffix too large)' % value)
315
316			return Community(pack('!L', (prefix << 16) + suffix))
317
318			elif value[:2].lower() == '0x':
319			number = int(value, 16)
320			if number > Community.MAX:
321			raise ValueError('invalid community %s (too large)' % value)
322			return Community(pack('!L', number))
323
324			else:
325			low = value.lower()
326			if low == 'no-export':
327			return Community(Community.NO_EXPORT)
328			elif low == 'no-advertise':
329			return Community(Community.NO_ADVERTISE)
330			elif low == 'no-export-subconfed':
331			return Community(Community.NO_EXPORT_SUBCONFED)
332			# no-peer is not a correct syntax but I am sure someone will make the mistake :)
333			elif low == 'nopeer' or low == 'no-peer':
334			return Community(Community.NO_PEER)
335			elif low == 'blackhole':
336			return Community(Community.BLACKHOLE)
337			elif value.isdigit():
338			number = int(value)
339			if number > Community.MAX:
340			raise ValueError('invalid community %s (too large)' % value)
341			return Community(pack('!L', number))
342			else:
343			raise ValueError('invalid community name %s' % value)
344
345
346			def community(tokeniser):
347			communities = Communities()
348
349			value = tokeniser()
350			if value == '[':
351			while True:
352			value = tokeniser()
353			if value == ']':
354			break
355			communities.add(_community(value))
356			else:
357			communities.add(_community(value))
358
359			return communities
360
361
362			def _large_community(value):
363			separator = value.find(':')
364			if separator > 0:
365			prefix, affix, suffix = value.split(':')
366
367			if not any(map(lambda c: c.isdigit(), [prefix, affix, suffix])):
368			raise ValueError('invalid community %s' % value)
369
370			prefix, affix, suffix = map(int, [prefix, affix, suffix])
371
372			for i in [prefix, affix, suffix]:
373			if i > LargeCommunity.MAX:
374			raise ValueError('invalid community %i in %s too large' % (i, value))
375
376			return LargeCommunity(pack('!LLL', prefix, affix, suffix))
377
378			elif value[:2].lower() == '0x':
379			number = int(value)
380			if number > LargeCommunity.MAX:
381			raise ValueError('invalid large community %s (too large)' % value)
382			return LargeCommunity(pack('!LLL', number >> 64, (number >> 32) & 0xFFFFFFFF, number & 0xFFFFFFFF))
383
384			else:
385			low = value.lower()
386			if value.isdigit():
387			number = int(value)
388			if number > LargeCommunity.MAX:
389			raise ValueError('invalid large community %s (too large)' % value)
390			return LargeCommunity(pack('!LLL', number >> 64, (number >> 32) & 0xFFFFFFFF, number & 0xFFFFFFFF))
391			else:
392			raise ValueError('invalid large community name %s' % value)
393
394
395			def large_community(tokeniser):
396			large_communities = LargeCommunities()
397
398			value = tokeniser()
399			if value == '[':
400			while True:
401			value = tokeniser()
402			if value == ']':
403			break
404			lc = _large_community(value)
405			if lc in large_communities.communities:
406			continue
407			large_communities.add(lc)
408			else:
409			large_communities.add(_large_community(value))
410
411			return large_communities
412
413
414			# fmt: off
415			_HEADER = {
416			# header and subheader
417			'target': bytes([0x00, 0x02]),
418			'target4': bytes([0x02, 0x02]),
419			'origin': bytes([0x00, 0x03]),
420			'origin4': bytes([0x02, 0x03]),
421			'redirect': bytes([0x80, 0x08]),
422			'l2info': bytes([0x80, 0x0A]),
423			'redirect-to-nexthop': bytes([0x08, 0x00]),
424			'bandwidth': bytes([0x40, 0x04]),
425			}
426
427			_SIZE = {
428			# fmt: off
429			'target': 2,
430			'target4': 2,
431			'origin': 2,
432			'origin4': 2,
433			'redirect': 2,
434			'l2info': 4,
435			'redirect-to-nexthop': 0,
436			'bandwidth': 2,
437			}
438			# fmt: on
439
440			_SIZE_H = 0xFFFF
441
442
443			def _extended_community(value):
444			if value[:2].lower() == '0x':
445			# we could raise if the length is not 8 bytes (16 chars)
446			if len(value) % 2:
447			raise ValueError('invalid extended community %s' % value)
448			raw = b''.join(bytes([int(value[_ : _ + 2], 16)]) for _ in range(2, len(value), 2))
449			return ExtendedCommunity.unpack(raw)
450			elif value.count(':'):
451			components = value.split(':')
452			command = 'target' if len(components) == 2 else components.pop(0)
453
454			if command not in _HEADER:
455			raise ValueError('invalid extended community %s (only origin,target or l2info are supported) ' % command)
456
457			if len(components) != _SIZE[command]:
458			raise ValueError('invalid extended community %s, expecting %d fields ' % (command, len(components)))
459
460			header = _HEADER.get(command, None)
461
462			if header is None:
463			raise ValueError('unknown extended community %s' % command)
464
465			if command == 'l2info':
466			# encaps, control, mtu, site
467			return ExtendedCommunity.unpack(header + pack('!BBHH', *[int(_) for _ in components]))
468
469			_ga, _la = components
470			ga, la = _ga.upper(), _la.upper()
471
472			if command in ('target', 'origin'):
473			# global admin, local admin
474			if '.' in ga or '.' in la:
475			gc = ga.count('.')
476			lc = la.count('.')
477			if gc == 0 and lc == 3:
478			# ASN first, IP second
479			return ExtendedCommunity.unpack(header + pack('!HBBBB', int(ga), *[int(_) for _ in la.split('.')]))
480			if gc == 3 and lc == 0:
481			# IP first, ASN second
482			return ExtendedCommunity.unpack(
483			header + pack('!BBBBH', *[int(_) for _ in ga.split('.')] + [int(la)])
484			)
485
486			iga = int(ga[:-1]) if 'L' in ga else int(ga)
487			ila = int(la[:-1]) if 'L' in la else int(la)
488
489			if command == 'target':
490			if ga.endswith('L') or iga > _SIZE_H:
491			return ExtendedCommunity.unpack(_HEADER['target4'] + pack('!LH', iga, ila), None)
492			else:
493			return ExtendedCommunity.unpack(header + pack('!HI', iga, ila), None)
494			if command == 'origin':
495			if ga.endswith('L') or iga > _SIZE_H:
496			return ExtendedCommunity.unpack(_HEADER['origin4'] + pack('!LH', iga, ila), None)
497			else:
498			return ExtendedCommunity.unpack(header + pack('!HI', iga, ila), None)
499
500			if command == 'target4':
501			return ExtendedCommunity.unpack(_HEADER['target4'] + pack('!LH', iga, ila), None)
502
503			if command == 'origin4':
504			return ExtendedCommunity.unpack(_HEADER['origin4'] + pack('!LH', iga, ila), None)
505
506			if command == 'redirect':
507			return ExtendedCommunity.unpack(header + pack('!HL', iga, ila), None)
508
509			if command == 'bandwidth':
510			return ExtendedCommunity.unpack(_HEADER['bandwidth'] + pack('!Hf', iga, ila), None)
511
512			raise ValueError('invalid extended community %s' % command)
513			elif value == 'redirect-to-nexthop':
514			header = _HEADER[value]
515			return ExtendedCommunity.unpack(header + pack('!HL', 0, 0), None)
516			else:
517			raise ValueError('invalid extended community %s - lc+gc' % value)
518
519
520			# The previous code was extracting the extended-community class from the attributes
521			# And adding to it.
522
523
524			def extended_community(tokeniser):
525			communities = ExtendedCommunities()
526
527			value = tokeniser()
528			if value == '[':
529			while True:
530			value = tokeniser()
531			if value == ']':
532			break
533			communities.add(_extended_community(value))
534			else:
535			communities.add(_extended_community(value))
536
537			return communities
538
539
540			# Duck class, faking part of the Attribute interface
541			# We add this to routes when when need o split a route in smaller route
542			# The value stored is the longer netmask we want to use
543			# As this is not a real BGP attribute this stays in the configuration file
544
545
546			def name(tokeniser):
547			class Name(str):
548			ID = Attribute.CODE.INTERNAL_NAME
549
550			return Name(tokeniser())
551
552
553			def split(tokeniser):
554			class Split(int):
555			ID = Attribute.CODE.INTERNAL_SPLIT
556
557			cidr = tokeniser()
558
559			if not cidr or cidr[0] != '/':
560			raise ValueError('split /<number>')
561
562			size = cidr[1:]
563
564			if not size.isdigit():
565			raise ValueError('split /<number>')
566
567			return Split(int(size))
568
569
570			def watchdog(tokeniser):
571			class Watchdog(str):
572			ID = Attribute.CODE.INTERNAL_WATCHDOG
573
574			command = tokeniser()
575			if command.lower() in ['announce', 'withdraw']:
576			raise ValueError('invalid watchdog name %s' % command)
577			return Watchdog(command)
578
579
580			def withdraw(tokeniser=None):
581			class Withdrawn(object):
582			ID = Attribute.CODE.INTERNAL_WITHDRAW
583
584			return Withdrawn()
585

Exa-Networks / exabgp

_extended_community() F last analyzed 2020-07-19 19:06 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

_extended_community() F
last analyzed 2020-07-19 19:06 UTC