exabgp.configuration.announce.ip.AnnounceIP._check() - Code Metrics - Inspection of "large change required to perform better route vali..." - Exa-Networks/exabgp - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 25ce4f...b9bc4f )

by Thomas

created 2019-09-24 00:20 UTC

) A

↳ Parent: exabgp.configuration.announce.ip.AnnounceIP.check()

Complexity

Conditions

Size

Total Lines	4
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	4
nop	1
dl	0
loc	4
rs	10
c	0
b	0
f	0

# encoding: utf-8
"""
announce/ip.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 exabgp.protocol.ip import NoNextHop

from exabgp.rib.change import Change

from exabgp.bgp.message import OUT

from exabgp.protocol.family import AFI
from exabgp.protocol.family import SAFI

from exabgp.bgp.message.update.nlri.inet import INET
from exabgp.bgp.message.update.nlri.cidr import CIDR
from exabgp.bgp.message.update.attribute import Attributes

from exabgp.configuration.announce import ParseAnnounce

from exabgp.configuration.static.parser import prefix
# from exabgp.configuration.static.parser import inet
from exabgp.configuration.static.parser import attribute
from exabgp.configuration.static.parser import next_hop
from exabgp.configuration.static.parser import origin
from exabgp.configuration.static.parser import med
from exabgp.configuration.static.parser import as_path
from exabgp.configuration.static.parser import local_preference
from exabgp.configuration.static.parser import atomic_aggregate
from exabgp.configuration.static.parser import aggregator
from exabgp.configuration.static.parser import originator_id
from exabgp.configuration.static.parser import cluster_list
from exabgp.configuration.static.parser import community
from exabgp.configuration.static.parser import large_community
from exabgp.configuration.static.parser import extended_community
from exabgp.configuration.static.parser import aigp
from exabgp.configuration.static.parser import name as named
from exabgp.configuration.static.parser import split
from exabgp.configuration.static.parser import watchdog
from exabgp.configuration.static.parser import withdraw


class AnnounceIP (ParseAnnounce):
	# put next-hop first as it is a requirement atm
	definition = [
		'next-hop <ip>',
		'origin IGP|EGP|INCOMPLETE',
		'as-path [ <asn>.. ]',
		'med <16 bits number>',
		'local-preference <16 bits number>',
		'atomic-aggregate',
		'community <16 bits number>',
		'large-community <96 bits number>',
		'extended-community target:<16 bits number>:<ipv4 formated number>',
		'originator-id <ipv4>',
		'cluster-list <ipv4>',
		'label <15 bits number>',
		'bgp-prefix-sid [ 32 bits number> ] | [ <32 bits number>, [ ( <24 bits number>,<24 bits number> ) ]]',
		'aggregator ( <asn16>:<ipv4> )',
		'aigp <40 bits number>',
		'attribute [ generic attribute format ]'
		'name <mnemonic>',
		'split /<mask>',
		'watchdog <watchdog-name>',
		'withdraw',
	]

	syntax = \
		'<safi> <ip>/<netmask> { ' \
		'\n   ' + ' ;\n   '.join(definition) + '\n}'

	known = {
		'attribute':           attribute,
		'next-hop':            next_hop,
		'origin':              origin,
		'med':                 med,
		'as-path':             as_path,
		'local-preference':    local_preference,
		'atomic-aggregate':    atomic_aggregate,
		'aggregator':          aggregator,
		'originator-id':       originator_id,
		'cluster-list':        cluster_list,
		'community':           community,
		'large-community':     large_community,
		'extended-community':  extended_community,
		'aigp':                aigp,
		'name':                named,
		'split':               split,
		'watchdog':            watchdog,
		'withdraw':            withdraw,
	}

	action = {
		'attribute':           'attribute-add',
		'next-hop':            'nexthop-and-attribute',
		'origin':              'attribute-add',
		'med':                 'attribute-add',
		'as-path':             'attribute-add',
		'local-preference':    'attribute-add',
		'atomic-aggregate':    'attribute-add',
		'aggregator':          'attribute-add',
		'originator-id':       'attribute-add',
		'cluster-list':        'attribute-add',
		'community':           'attribute-add',
		'large-community':     'attribute-add',
		'extended-community':  'attribute-add',
		'name':                'attribute-add',
		'split':               'attribute-add',
		'watchdog':            'attribute-add',
		'withdraw':            'attribute-add',
		'aigp':                'attribute-add',
	}

	assign = {
	}

	name = 'ip'

	def __init__ (self, tokeniser, scope, error, logger):
		ParseAnnounce.__init__(self,tokeniser,scope,error,logger)

	def clear (self):
		return True

	def pre (self):
		return True

	def post (self):
		return self._check()

	@staticmethod
	def check (change,afi):
		if change.nlri.action == OUT.ANNOUNCE \
			and change.nlri.nexthop is NoNextHop \
			and change.nlri.afi == afi \
			and change.nlri.safi in (SAFI.unicast,SAFI.multicast):
			return False

		return True


def ip (tokeniser,afi,safi):

	action = OUT.ANNOUNCE if tokeniser.announce else OUT.WITHDRAW
	ipmask = prefix(tokeniser)

	nlri = INET(afi, safi, action)
	nlri.cidr = CIDR(ipmask.pack(),ipmask.mask)

	change = Change(
		nlri,
		Attributes()
	)

	while True:
		command = tokeniser()

		if not command:
			break

		action = AnnounceIP.action.get(command,'')

		if action == 'attribute-add':
			change.attributes.add(AnnounceIP.known[command](tokeniser))
		elif action == 'nlri-set':
			change.nlri.assign(AnnounceIP.assign[command],AnnounceIP.known[command](tokeniser))
		elif action == 'nexthop-and-attribute':
			nexthop,attribute = AnnounceIP.known[command](tokeniser)
			change.nlri.nexthop = nexthop
			change.attributes.add(attribute)
		else:
			raise ValueError('unknown command "%s"' % command)

	if not AnnounceIP.check(change,afi):
		raise ValueError('invalid announcement (missing next-hop ?)')

	return [change]


def ip_multicast (tokeniser,afi,safi):
	action = OUT.ANNOUNCE if tokeniser.announce else OUT.WITHDRAW
	ipmask = prefix(tokeniser)

	nlri = INET(afi,safi,action)
	nlri.cidr = CIDR(ipmask.pack(),ipmask.mask)

	change = Change(
		nlri,
		Attributes()
	)

	while True:
		command = tokeniser()

		if not command:
			break

		action = AnnounceIP.action.get(command,'')

		if action == 'attribute-add':
			change.attributes.add(AnnounceIP.known[command](tokeniser))
		elif action == 'nlri-set':
			change.nlri.assign(AnnounceIP.assign[command],AnnounceIP.known[command](tokeniser))
		elif action == 'nexthop-and-attribute':
			nexthop,attribute = AnnounceIP.known[command](tokeniser)
			change.nlri.nexthop = nexthop
			change.attributes.add(attribute)
		else:
			raise ValueError('unknown command "%s"' % command)

	return [change]


@ParseAnnounce.register('multicast','extend-name','ipv4')
def multicast_v4 (tokeniser):
	return ip_multicast(tokeniser,AFI.ipv4,SAFI.multicast)


@ParseAnnounce.register('multicast','extend-name','ipv6')
def multicast_v6 (tokeniser):
	return ip_multicast(tokeniser,AFI.ipv6,SAFI.multicast)


1		# encoding: utf-8
2		"""
3		announce/ip.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 exabgp.protocol.ip import NoNextHop
11
12		from exabgp.rib.change import Change
13
14		from exabgp.bgp.message import OUT
15
16		from exabgp.protocol.family import AFI
17		from exabgp.protocol.family import SAFI
18
19		from exabgp.bgp.message.update.nlri.inet import INET
20		from exabgp.bgp.message.update.nlri.cidr import CIDR
21		from exabgp.bgp.message.update.attribute import Attributes
22
23		from exabgp.configuration.announce import ParseAnnounce
24
25		from exabgp.configuration.static.parser import prefix
26		# from exabgp.configuration.static.parser import inet
27		from exabgp.configuration.static.parser import attribute
28		from exabgp.configuration.static.parser import next_hop
29		from exabgp.configuration.static.parser import origin
30		from exabgp.configuration.static.parser import med
31		from exabgp.configuration.static.parser import as_path
32		from exabgp.configuration.static.parser import local_preference
33		from exabgp.configuration.static.parser import atomic_aggregate
34		from exabgp.configuration.static.parser import aggregator
35		from exabgp.configuration.static.parser import originator_id
36		from exabgp.configuration.static.parser import cluster_list
37		from exabgp.configuration.static.parser import community
38		from exabgp.configuration.static.parser import large_community
39		from exabgp.configuration.static.parser import extended_community
40		from exabgp.configuration.static.parser import aigp
41		from exabgp.configuration.static.parser import name as named
42		from exabgp.configuration.static.parser import split
43		from exabgp.configuration.static.parser import watchdog
44		from exabgp.configuration.static.parser import withdraw
45
46
47		class AnnounceIP (ParseAnnounce):
48		# put next-hop first as it is a requirement atm
49		definition = [
50		'next-hop <ip>',
51		'origin IGP\|EGP\|INCOMPLETE',
52		'as-path [ <asn>.. ]',
53		'med <16 bits number>',
54		'local-preference <16 bits number>',
55		'atomic-aggregate',
56		'community <16 bits number>',
57		'large-community <96 bits number>',
58		'extended-community target:<16 bits number>:<ipv4 formated number>',
59		'originator-id <ipv4>',
60		'cluster-list <ipv4>',
61		'label <15 bits number>',
62		'bgp-prefix-sid [ 32 bits number> ] \| [ <32 bits number>, [ ( <24 bits number>,<24 bits number> ) ]]',
63		'aggregator ( <asn16>:<ipv4> )',
64		'aigp <40 bits number>',
65		'attribute [ generic attribute format ]'
66		'name <mnemonic>',
67		'split /<mask>',
68		'watchdog <watchdog-name>',
69		'withdraw',
70		]
71
72		syntax = \
73		'<safi> <ip>/<netmask> { ' \
74		'\n ' + ' ;\n '.join(definition) + '\n}'
75
76		known = {
77		'attribute': attribute,
78		'next-hop': next_hop,
79		'origin': origin,
80		'med': med,
81		'as-path': as_path,
82		'local-preference': local_preference,
83		'atomic-aggregate': atomic_aggregate,
84		'aggregator': aggregator,
85		'originator-id': originator_id,
86		'cluster-list': cluster_list,
87		'community': community,
88		'large-community': large_community,
89		'extended-community': extended_community,
90		'aigp': aigp,
91		'name': named,
92		'split': split,
93		'watchdog': watchdog,
94		'withdraw': withdraw,
95		}
96
97		action = {
98		'attribute': 'attribute-add',
99		'next-hop': 'nexthop-and-attribute',
100		'origin': 'attribute-add',
101		'med': 'attribute-add',
102		'as-path': 'attribute-add',
103		'local-preference': 'attribute-add',
104		'atomic-aggregate': 'attribute-add',
105		'aggregator': 'attribute-add',
106		'originator-id': 'attribute-add',
107		'cluster-list': 'attribute-add',
108		'community': 'attribute-add',
109		'large-community': 'attribute-add',
110		'extended-community': 'attribute-add',
111		'name': 'attribute-add',
112		'split': 'attribute-add',
113		'watchdog': 'attribute-add',
114		'withdraw': 'attribute-add',
115		'aigp': 'attribute-add',
116		}
117
118		assign = {
119		}
120
121		name = 'ip'
122
123		def __init__ (self, tokeniser, scope, error, logger):
124		ParseAnnounce.__init__(self,tokeniser,scope,error,logger)
125
126		def clear (self):
127		return True
128
129		def pre (self):
130		return True
131
132		def post (self):
133		return self._check()
134
135		@staticmethod
136		def check (change,afi):
137		if change.nlri.action == OUT.ANNOUNCE \
138		and change.nlri.nexthop is NoNextHop \
139		and change.nlri.afi == afi \
140		and change.nlri.safi in (SAFI.unicast,SAFI.multicast):
141		return False
142
143		return True
144
145
146	View Code Duplication	def ip (tokeniser,afi,safi):
		0 ignored issues – show Duplication introduced 2019-09-24 00:32 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
147		action = OUT.ANNOUNCE if tokeniser.announce else OUT.WITHDRAW
148		ipmask = prefix(tokeniser)
149
150		nlri = INET(afi, safi, action)
151		nlri.cidr = CIDR(ipmask.pack(),ipmask.mask)
152
153		change = Change(
154		nlri,
155		Attributes()
156		)
157
158		while True:
159		command = tokeniser()
160
161		if not command:
162		break
163
164		action = AnnounceIP.action.get(command,'')
165
166		if action == 'attribute-add':
167		change.attributes.add(AnnounceIP.known[command](tokeniser))
168		elif action == 'nlri-set':
169		change.nlri.assign(AnnounceIP.assign[command],AnnounceIP.known[command](tokeniser))
170		elif action == 'nexthop-and-attribute':
171		nexthop,attribute = AnnounceIP.known[command](tokeniser)
172		change.nlri.nexthop = nexthop
173		change.attributes.add(attribute)
174		else:
175		raise ValueError('unknown command "%s"' % command)
176
177		if not AnnounceIP.check(change,afi):
178		raise ValueError('invalid announcement (missing next-hop ?)')
179
180		return [change]
181
182
183		def ip_multicast (tokeniser,afi,safi):
184		action = OUT.ANNOUNCE if tokeniser.announce else OUT.WITHDRAW
185		ipmask = prefix(tokeniser)
186
187		nlri = INET(afi,safi,action)
188		nlri.cidr = CIDR(ipmask.pack(),ipmask.mask)
189
190		change = Change(
191		nlri,
192		Attributes()
193		)
194
195		while True:
196		command = tokeniser()
197
198		if not command:
199		break
200
201		action = AnnounceIP.action.get(command,'')
202
203		if action == 'attribute-add':
204		change.attributes.add(AnnounceIP.known[command](tokeniser))
205		elif action == 'nlri-set':
206		change.nlri.assign(AnnounceIP.assign[command],AnnounceIP.known[command](tokeniser))
207		elif action == 'nexthop-and-attribute':
208		nexthop,attribute = AnnounceIP.known[command](tokeniser)
209		change.nlri.nexthop = nexthop
210		change.attributes.add(attribute)
211		else:
212		raise ValueError('unknown command "%s"' % command)
213
214		return [change]
215
216
217		@ParseAnnounce.register('multicast','extend-name','ipv4')
218		def multicast_v4 (tokeniser):
219		return ip_multicast(tokeniser,AFI.ipv4,SAFI.multicast)
220
221
222		@ParseAnnounce.register('multicast','extend-name','ipv6')
223		def multicast_v6 (tokeniser):
224		return ip_multicast(tokeniser,AFI.ipv6,SAFI.multicast)
225

Exa-Networks / exabgp

Push — master ( 25ce4f...b9bc4f )

) A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like