exabgp.configuration.static.mpls.prefix_sid_srv6() - Code Metrics - Inspection of "Support additional sub-type of BGP-Prefix-SID for..." - Exa-Networks/exabgp - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#946)

unknown

created 2020-01-12 13:28 UTC

exabgp.configuration.static.mpls.prefix_sid_srv6() C

↳ Parent: exabgp.configuration.static.mpls

Complexity

Conditions

Size

Total Lines	30
Code Lines	27

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	9
eloc	27
nop	1
dl	0
loc	30
rs	6.6666
c	0
b	0
f	0

# 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

from exabgp.util import character
from exabgp.util import concat_bytes_i

from exabgp.bgp.message.update.nlri.qualifier import Labels
from exabgp.bgp.message.update.nlri.qualifier import RouteDistinguisher
from exabgp.bgp.message.update.attribute.sr.prefixsid import PrefixSid
from exabgp.bgp.message.update.attribute.sr.labelindex import SrLabelIndex
from exabgp.bgp.message.update.attribute.sr.ipv6sid import SrV6Sid
from exabgp.bgp.message.update.attribute.sr.srv6vpnsid import Srv6VpnSid
from exabgp.bgp.message.update.attribute.sr.srv6l3vpnsid import Srv6L3vpnSid
from exabgp.bgp.message.update.attribute.sr.srgb import SrGb


def label (tokeniser):
	labels = []
	value = tokeniser()

	if value == '[':
		while True:
			value = tokeniser()
			if value == ']':
				break
			labels.append(int(value))
	else:
		labels.append(int(value))

	return Labels(labels)


def route_distinguisher (tokeniser):
	data = tokeniser()

	separator = data.find(':')
	if separator > 0:
		prefix = data[:separator]
		suffix = int(data[separator+1:])

	if '.' in prefix:

		data = [character(0),character(1)]
		data.extend([character(int(_)) for _ in prefix.split('.')])
		data.extend([character(suffix >> 8),character(suffix & 0xFF)])

		rtd = concat_bytes_i(data)
	else:
		number = int(prefix)
		if number < pow(2,16) and suffix < pow(2,32):
			rtd = character(0) + character(0) + pack('!H',number) + pack('!L',suffix)
		elif number < pow(2,32) and suffix < pow(2,16):
			rtd = character(0) + character(2) + pack('!L',number) + pack('!H',suffix)
		else:
			raise ValueError('invalid route-distinguisher %s' % data)

	return RouteDistinguisher(rtd)

# [ 300, [ ( 800000,100 ), ( 1000000,5000 ) ] ]
def prefix_sid (tokeniser):
	sr_attrs = []
	srgbs = []
	srgb_data = []
	value = tokeniser()
	get_range = False
	consume_extra = False
	try:
		if value == '[':
			label_sid = tokeniser()
			while True:
				value = tokeniser()
				if value == '[':
					consume_extra = True
					continue
				if value == ',':
					continue
				if value == '(':
					while True:
						value = tokeniser()
						if value == ')':
							break
						if value == ',':
							get_range = True
							continue
						if get_range:
							srange = value
							get_range = False
						else:
							base = value
				if value == ')':
					srgb_data.append((base,srange))

					continue
				if value == ']':
					break
		if consume_extra:
			tokeniser()
	except Exception as e:
		raise ValueError('could not parse BGP PrefixSid attribute: {}'.format(e))

	if int(label_sid) < pow(2,32):

		sr_attrs.append(SrLabelIndex(int(label_sid)))

	for srgb in srgb_data:
		if (len(srgb) == 2 and int(srgb[0]) < pow(2,24) and int(srgb[1]) < pow(2,24)):
			srgbs.append((int(srgb[0]),int(srgb[1])))
		else:
			raise ValueError('could not parse SRGB tupple')

	if srgbs:
		sr_attrs.append(SrGb(srgbs))

	return PrefixSid(sr_attrs)

# { ipv6 <ipv6-addr> | vpn <ipv6-addr> | l3vpn <ipv6-addr> }
def prefix_sid_srv6 (tokeniser):
	sr_attrs = []
	value = tokeniser()
	try:
		if value == '(':
			value = tokeniser()
			if value == 'ipv6':
				value = tokeniser()
				sr_attrs.append(SrV6Sid(value))
				value = tokeniser()
				if value == ')':
					return PrefixSid(sr_attrs)

			if value == 'vpn':
				value = tokeniser()
				sr_attrs.append(Srv6VpnSid(value))
				value = tokeniser()
				if value == ')':
					return PrefixSid(sr_attrs)

			if value == 'l3vpn':
				value = tokeniser()
				sr_attrs.append(Srv6L3vpnSid(value))
				value = tokeniser()
				if value == ')':
					return PrefixSid(sr_attrs)

		raise Exception("format error")
	except Exception as e:
		raise ValueError('could not parse BGP PrefixSid Srv6 attribute: {}'.format(e))


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
12		from exabgp.util import character
13		from exabgp.util import concat_bytes_i
14
15		from exabgp.bgp.message.update.nlri.qualifier import Labels
16		from exabgp.bgp.message.update.nlri.qualifier import RouteDistinguisher
17		from exabgp.bgp.message.update.attribute.sr.prefixsid import PrefixSid
18		from exabgp.bgp.message.update.attribute.sr.labelindex import SrLabelIndex
19		from exabgp.bgp.message.update.attribute.sr.ipv6sid import SrV6Sid
20		from exabgp.bgp.message.update.attribute.sr.srv6vpnsid import Srv6VpnSid
21		from exabgp.bgp.message.update.attribute.sr.srv6l3vpnsid import Srv6L3vpnSid
22		from exabgp.bgp.message.update.attribute.sr.srgb import SrGb
23
24
25		def label (tokeniser):
26		labels = []
27		value = tokeniser()
28
29		if value == '[':
30		while True:
31		value = tokeniser()
32		if value == ']':
33		break
34		labels.append(int(value))
35		else:
36		labels.append(int(value))
37
38		return Labels(labels)
39
40
41		def route_distinguisher (tokeniser):
42		data = tokeniser()
43
44		separator = data.find(':')
45		if separator > 0:
46		prefix = data[:separator]
47		suffix = int(data[separator+1:])
48
49	View Code Duplication	if '.' in prefix:
		0 ignored issues – show introduced 2019-07-17 07:30 UTC by Report Bug Copy Issue Report The variable `prefix` does not seem to be defined in case `separator > 0` on line `45` is `False`. Are you sure this can never be the case? Loading history... Duplication introduced 2019-07-17 07:30 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
50		data = [character(0),character(1)]
51		data.extend([character(int(_)) for _ in prefix.split('.')])
52		data.extend([character(suffix >> 8),character(suffix & 0xFF)])
		0 ignored issues – show introduced 2019-07-17 07:30 UTC by Report Bug Copy Issue Report The variable `suffix` does not seem to be defined in case `separator > 0` on line `45` is `False`. Are you sure this can never be the case? Loading history...
53		rtd = concat_bytes_i(data)
54		else:
55		number = int(prefix)
56		if number < pow(2,16) and suffix < pow(2,32):
57		rtd = character(0) + character(0) + pack('!H',number) + pack('!L',suffix)
58		elif number < pow(2,32) and suffix < pow(2,16):
59		rtd = character(0) + character(2) + pack('!L',number) + pack('!H',suffix)
60		else:
61		raise ValueError('invalid route-distinguisher %s' % data)
62
63		return RouteDistinguisher(rtd)
64
65		# [ 300, [ ( 800000,100 ), ( 1000000,5000 ) ] ]
66		def prefix_sid (tokeniser):
67		sr_attrs = []
68		srgbs = []
69		srgb_data = []
70		value = tokeniser()
71		get_range = False
72		consume_extra = False
73		try:
74		if value == '[':
75		label_sid = tokeniser()
76		while True:
77		value = tokeniser()
78		if value == '[':
79		consume_extra = True
80		continue
81		if value == ',':
82		continue
83		if value == '(':
84		while True:
85		value = tokeniser()
86		if value == ')':
87		break
88		if value == ',':
89		get_range = True
90		continue
91		if get_range:
92		srange = value
93		get_range = False
94		else:
95		base = value
96		if value == ')':
97		srgb_data.append((base,srange))
		0 ignored issues – show introduced 2019-07-17 07:30 UTC by Report Bug Copy Issue Report The variable `base` does not seem to be defined for all execution paths. Loading history... introduced 2019-07-17 07:30 UTC by Report Bug Copy Issue Report The variable `srange` does not seem to be defined for all execution paths. Loading history...
98		continue
99		if value == ']':
100		break
101		if consume_extra:
102		tokeniser()
103		except Exception as e:
104		raise ValueError('could not parse BGP PrefixSid attribute: {}'.format(e))
105
106		if int(label_sid) < pow(2,32):
		0 ignored issues – show introduced 2019-07-17 07:30 UTC by Report Bug Copy Issue Report The variable `label_sid` does not seem to be defined in case `value == '['` on line `74` is `False`. Are you sure this can never be the case? Loading history...
107		sr_attrs.append(SrLabelIndex(int(label_sid)))
108
109		for srgb in srgb_data:
110		if (len(srgb) == 2 and int(srgb[0]) < pow(2,24) and int(srgb[1]) < pow(2,24)):
111		srgbs.append((int(srgb[0]),int(srgb[1])))
112		else:
113		raise ValueError('could not parse SRGB tupple')
114
115		if srgbs:
116		sr_attrs.append(SrGb(srgbs))
117
118		return PrefixSid(sr_attrs)
119
120		# { ipv6 <ipv6-addr> \| vpn <ipv6-addr> \| l3vpn <ipv6-addr> }
121		def prefix_sid_srv6 (tokeniser):
122		sr_attrs = []
123		value = tokeniser()
124		try:
125		if value == '(':
126		value = tokeniser()
127		if value == 'ipv6':
128		value = tokeniser()
129		sr_attrs.append(SrV6Sid(value))
130		value = tokeniser()
131		if value == ')':
132		return PrefixSid(sr_attrs)
133
134		if value == 'vpn':
135		value = tokeniser()
136		sr_attrs.append(Srv6VpnSid(value))
137		value = tokeniser()
138		if value == ')':
139		return PrefixSid(sr_attrs)
140
141		if value == 'l3vpn':
142		value = tokeniser()
143		sr_attrs.append(Srv6L3vpnSid(value))
144		value = tokeniser()
145		if value == ')':
146		return PrefixSid(sr_attrs)
147
148		raise Exception("format error")
149		except Exception as e:
150		raise ValueError('could not parse BGP PrefixSid Srv6 attribute: {}'.format(e))
151

Exa-Networks / exabgp

Pull Request — master (#946)

exabgp.configuration.static.mpls.prefix_sid_srv6() C

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like