exabgp.configuration.check.check_generation() - Code Metrics - Exa-Networks/exabgp - Measure and Improve Code Quality continuously with Scrutinizer

exabgp.configuration.check.check_generation() F
last analyzed 2020-07-19 19:06 UTC

↳ Parent: exabgp.configuration.check

Complexity

Conditions

Size

Total Lines	115
Code Lines	83

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	17
eloc	83
nop	1
dl	0
loc	115
rs	1.8
c	0
b	0
f	0

How to fix Long Method Complexity

# encoding: utf-8
"""
check.py

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

# common

import copy

from exabgp.bgp.message import Update
from exabgp.bgp.message import Open
from exabgp.bgp.message.open import Version
from exabgp.bgp.message.open import ASN
from exabgp.bgp.message.open import RouterID
from exabgp.bgp.message.open import HoldTime
from exabgp.bgp.message.open.capability import Capabilities
from exabgp.bgp.message.open.capability import Capability
from exabgp.bgp.message.open.capability import Negotiated
from exabgp.bgp.message import Notify
from exabgp.bgp.message.update.nlri import NLRI

from exabgp.logger import log
from exabgp.logger import option

# check_neighbor

from exabgp.util.od import od
from exabgp.rib.change import Change

# check_update

from exabgp.reactor.api.response import Response

# check_notification

from exabgp.bgp.message import Notification

# JSON version

from exabgp.version import json as json_version


# =============================================================== check_neighbor
# ...


def check_generation(neighbors):
    option.enabled['parser'] = True

    for name in neighbors.keys():
        neighbor = copy.deepcopy(neighbors[name])
        neighbor.local_as = neighbor.peer_as

        path = {}
        for f in NLRI.known_families():
            if neighbor.add_path:
                path[f] = neighbor.add_path

        capa = Capabilities().new(neighbor, False)
        if path:
            capa[Capability.CODE.ADD_PATH] = path
        capa[Capability.CODE.MULTIPROTOCOL] = neighbor.families()

        routerid_1 = str(neighbor.router_id)
        routerid_2 = '.'.join(str((int(_) + 1) % 250) for _ in str(neighbor.router_id).split('.', -1))

        o1 = Open(Version(4), ASN(neighbor.local_as), HoldTime(180), RouterID(routerid_1), capa)
        o2 = Open(Version(4), ASN(neighbor.peer_as), HoldTime(180), RouterID(routerid_2), capa)
        negotiated = Negotiated(neighbor)
        negotiated.sent(o1)
        negotiated.received(o2)
        # grouped = False

        for _ in neighbor.rib.outgoing.updates(False):
            pass

        for change1 in neighbor.rib.outgoing.cached_changes():
            str1 = change1.extensive()
            packed = list(Update([change1.nlri], change1.attributes).messages(negotiated))
            pack1 = packed[0]

            log.debug('parsed route requires %d updates' % len(packed), 'parser')
            log.debug('update size is %d' % len(pack1), 'parser')

            log.debug('parsed route %s' % str1, 'parser')
            log.debug('parsed hex   %s' % od(pack1), 'parser')

            # This does not take the BGP header - let's assume we will not break that :)
            try:
                log.debug('')  # new line

                pack1s = pack1[19:] if pack1.startswith(b'\xFF' * 16) else pack1
                update = Update.unpack_message(pack1s, negotiated)

                change2 = Change(update.nlris[0], update.attributes)
                str2 = change2.extensive()
                pack2 = list(Update([update.nlris[0]], update.attributes).messages(negotiated))[0]

                log.debug('recoded route %s' % str2, 'parser')
                log.debug('recoded hex   %s' % od(pack2), 'parser')

                str1 = str1.replace('attribute [ 0x04 0x80 0x00000064 ]', 'med 100')
                str1r = (
                    str1.lower().replace(' med 100', '').replace(' local-preference 100', '').replace(' origin igp', '')
                )
                str2r = (
                    str2.lower().replace(' med 100', '').replace(' local-preference 100', '').replace(' origin igp', '')
                )
                str2r = str2r.replace(
                    'large-community [ 1:2:3 10:11:12 ]',
                    'attribute [ 0x20 0xc0 0x0000000100000002000000030000000a0000000b0000000c ]',
                )

                if 'next-hop self' in str1r:
                    if ':' in str1r:
                        str1r = str1r.replace('next-hop self', 'next-hop ::1')
                    else:
                        str1r = str1r.replace('next-hop self', 'next-hop %s' % neighbor.local_address)

                if ' name ' in str1r:
                    parts = str1r.split(' ')
                    pos = parts.index('name')
                    str1r = ' '.join(parts[:pos] + parts[pos + 2 :])

                skip = False

                if str1r != str2r:
                    if 'attribute [' in str1r and ' 0x00 ' in str1r:
                        # we do not decode non-transitive attributes
                        log.debug('skipping string check on update with non-transitive attribute(s)', 'parser')
                        skip = True
                    else:
                        log.debug('strings are different:', 'parser')
                        log.debug('[%s]' % (str1r), 'parser')
                        log.debug('[%s]' % (str2r), 'parser')
                        return False
                else:
                    log.debug('strings are fine', 'parser')

                if skip:
                    log.debug('skipping encoding for update with non-transitive attribute(s)', 'parser')
                elif pack1 != pack2:
                    log.debug('encoding are different', 'parser')
                    log.debug('[%s]' % (od(pack1)), 'parser')
                    log.debug('[%s]' % (od(pack2)), 'parser')
                    return False
                else:
                    log.debug('encoding is fine', 'parser')
                    log.debug('----------------------------------------', 'parser')

                log.debug('JSON nlri %s' % change1.nlri.json(), 'parser')
                log.debug('JSON attr %s' % change1.attributes.json(), 'parser')

            except Notify as exc:
                log.debug('----------------------------------------', 'parser')
                log.debug(str(exc), 'parser')
                log.debug('----------------------------------------', 'parser')
                return False
        neighbor.rib.clear()

    return True


# ================================================================ check_message
#


def check_message(neighbor, message):
    message = message.replace(':', '')
    msghexa = [message[i * 2 : (i * 2) + 2] for i in range(len(message) // 2)]
    raw = bytes([int(_, 16) for _ in msghexa])

    if raw.startswith(b'\xff' * 16):
        kind = raw[18]
        # XXX: FIXME: check size
        # size = (raw[16] << 16) + raw[17]

        if kind == 1:
            return check_open(neighbor, raw[18:])
        elif kind == 2:
            return check_update(neighbor, raw)
        elif kind == 3:
            return check_notification(raw)
    else:
        return check_update(neighbor, raw)


# ================================================================= check_update
#


def check_open(neighbor, raw):
    pass


# ================================================================= check_update
#


def check_update(neighbor, raw):
    option.enabled['parser'] = True

    neighbor = neighbor[list(neighbor)[0]]

    path = {}
    for f in NLRI.known_families():
        if neighbor.add_path:
            path[f] = neighbor.add_path

    capa = Capabilities().new(neighbor, False)
    capa[Capability.CODE.ADD_PATH] = path
    capa[Capability.CODE.MULTIPROTOCOL] = neighbor.families()
    # capa[Capability.CODE.FOUR_BYTES_ASN] = True

    routerid_1 = str(neighbor.router_id)
    routerid_2 = '.'.join(str((int(_) + 1) % 250) for _ in str(neighbor.router_id).split('.', -1))

    o1 = Open(Version(4), ASN(neighbor.local_as), HoldTime(180), RouterID(routerid_1), capa)
    o2 = Open(Version(4), ASN(neighbor.peer_as), HoldTime(180), RouterID(routerid_2), capa)
    negotiated = Negotiated(neighbor)
    negotiated.sent(o1)
    negotiated.received(o2)
    # grouped = False

    while raw:
        if raw.startswith(b'\xff' * 16):
            kind = raw[18]
            size = (raw[16] << 16) + raw[17]

            injected, raw = raw[19:size], raw[size:]

            if kind == 2:
                log.debug('the message is an update', 'parser')
                decoding = 'update'
            else:
                log.debug('the message is not an update (%d) - aborting' % kind, 'parser')
                return False
        else:
            log.debug('header missing, assuming this message is ONE update', 'parser')
            decoding = 'update'
            injected, raw = raw, ''

        try:
            # This does not take the BGP header - let's assume we will not break that :)
            update = Update.unpack_message(injected, negotiated)
        except Notify:
            import traceback

            log.error('could not parse the message', 'parser')
            log.error(traceback.format_exc(), 'parser')
            return False
        except Exception:
            import traceback

            log.error('could not parse the message', 'parser')
            log.error(traceback.format_exc(), 'parser')
            return False

        log.debug('', 'parser')  # new line
        for number in range(len(update.nlris)):
            change = Change(update.nlris[number], update.attributes)
            log.info('decoded %s %s %s' % (decoding, change.nlri.action, change.extensive()), 'parser')
        log.info('update json %s' % Response.JSON(json_version).update(neighbor, 'in', update, None, '', ''), 'parser')

    return True


# ================================================================= check_update
#


def check_notification(raw):
    notification = Notification.unpack_message(raw[18:], None)
    # XXX: FIXME: should be using logger here
    print(notification)
    return True


1			# encoding: utf-8
2			"""
3			check.py
4
5			Created by Thomas Mangin on 2009-08-25.
6			Copyright (c) 2009-2017 Exa Networks. All rights reserved.
7			License: 3-clause BSD. (See the COPYRIGHT file)
8			"""
9
10			# common
11
12			import copy
13
14			from exabgp.bgp.message import Update
15			from exabgp.bgp.message import Open
16			from exabgp.bgp.message.open import Version
17			from exabgp.bgp.message.open import ASN
18			from exabgp.bgp.message.open import RouterID
19			from exabgp.bgp.message.open import HoldTime
20			from exabgp.bgp.message.open.capability import Capabilities
21			from exabgp.bgp.message.open.capability import Capability
22			from exabgp.bgp.message.open.capability import Negotiated
23			from exabgp.bgp.message import Notify
24			from exabgp.bgp.message.update.nlri import NLRI
25
26			from exabgp.logger import log
27			from exabgp.logger import option
28
29			# check_neighbor
30
31			from exabgp.util.od import od
32			from exabgp.rib.change import Change
33
34			# check_update
35
36			from exabgp.reactor.api.response import Response
37
38			# check_notification
39
40			from exabgp.bgp.message import Notification
41
42			# JSON version
43
44			from exabgp.version import json as json_version
45
46
47			# =============================================================== check_neighbor
48			# ...
49
50
51			def check_generation(neighbors):
52			option.enabled['parser'] = True
53
54			for name in neighbors.keys():
55			neighbor = copy.deepcopy(neighbors[name])
56			neighbor.local_as = neighbor.peer_as
57
58			path = {}
59			for f in NLRI.known_families():
60			if neighbor.add_path:
61			path[f] = neighbor.add_path
62
63			capa = Capabilities().new(neighbor, False)
64			if path:
65			capa[Capability.CODE.ADD_PATH] = path
66			capa[Capability.CODE.MULTIPROTOCOL] = neighbor.families()
67
68			routerid_1 = str(neighbor.router_id)
69			routerid_2 = '.'.join(str((int(_) + 1) % 250) for _ in str(neighbor.router_id).split('.', -1))
70
71			o1 = Open(Version(4), ASN(neighbor.local_as), HoldTime(180), RouterID(routerid_1), capa)
72			o2 = Open(Version(4), ASN(neighbor.peer_as), HoldTime(180), RouterID(routerid_2), capa)
73			negotiated = Negotiated(neighbor)
74			negotiated.sent(o1)
75			negotiated.received(o2)
76			# grouped = False
77
78			for _ in neighbor.rib.outgoing.updates(False):
79			pass
80
81			for change1 in neighbor.rib.outgoing.cached_changes():
82			str1 = change1.extensive()
83			packed = list(Update([change1.nlri], change1.attributes).messages(negotiated))
84			pack1 = packed[0]
85
86			log.debug('parsed route requires %d updates' % len(packed), 'parser')
87			log.debug('update size is %d' % len(pack1), 'parser')
88
89			log.debug('parsed route %s' % str1, 'parser')
90			log.debug('parsed hex %s' % od(pack1), 'parser')
91
92			# This does not take the BGP header - let's assume we will not break that :)
93			try:
94			log.debug('') # new line
95
96			pack1s = pack1[19:] if pack1.startswith(b'\xFF' * 16) else pack1
97			update = Update.unpack_message(pack1s, negotiated)
98
99			change2 = Change(update.nlris[0], update.attributes)
100			str2 = change2.extensive()
101			pack2 = list(Update([update.nlris[0]], update.attributes).messages(negotiated))[0]
102
103			log.debug('recoded route %s' % str2, 'parser')
104			log.debug('recoded hex %s' % od(pack2), 'parser')
105
106			str1 = str1.replace('attribute [ 0x04 0x80 0x00000064 ]', 'med 100')
107			str1r = (
108			str1.lower().replace(' med 100', '').replace(' local-preference 100', '').replace(' origin igp', '')
109			)
110			str2r = (
111			str2.lower().replace(' med 100', '').replace(' local-preference 100', '').replace(' origin igp', '')
112			)
113			str2r = str2r.replace(
114			'large-community [ 1:2:3 10:11:12 ]',
115			'attribute [ 0x20 0xc0 0x0000000100000002000000030000000a0000000b0000000c ]',
116			)
117
118			if 'next-hop self' in str1r:
119			if ':' in str1r:
120			str1r = str1r.replace('next-hop self', 'next-hop ::1')
121			else:
122			str1r = str1r.replace('next-hop self', 'next-hop %s' % neighbor.local_address)
123
124			if ' name ' in str1r:
125			parts = str1r.split(' ')
126			pos = parts.index('name')
127			str1r = ' '.join(parts[:pos] + parts[pos + 2 :])
128
129			skip = False
130
131			if str1r != str2r:
132			if 'attribute [' in str1r and ' 0x00 ' in str1r:
133			# we do not decode non-transitive attributes
134			log.debug('skipping string check on update with non-transitive attribute(s)', 'parser')
135			skip = True
136			else:
137			log.debug('strings are different:', 'parser')
138			log.debug('[%s]' % (str1r), 'parser')
139			log.debug('[%s]' % (str2r), 'parser')
140			return False
141			else:
142			log.debug('strings are fine', 'parser')
143
144			if skip:
145			log.debug('skipping encoding for update with non-transitive attribute(s)', 'parser')
146			elif pack1 != pack2:
147			log.debug('encoding are different', 'parser')
148			log.debug('[%s]' % (od(pack1)), 'parser')
149			log.debug('[%s]' % (od(pack2)), 'parser')
150			return False
151			else:
152			log.debug('encoding is fine', 'parser')
153			log.debug('----------------------------------------', 'parser')
154
155			log.debug('JSON nlri %s' % change1.nlri.json(), 'parser')
156			log.debug('JSON attr %s' % change1.attributes.json(), 'parser')
157
158			except Notify as exc:
159			log.debug('----------------------------------------', 'parser')
160			log.debug(str(exc), 'parser')
161			log.debug('----------------------------------------', 'parser')
162			return False
163			neighbor.rib.clear()
164
165			return True
166
167
168			# ================================================================ check_message
169			#
170
171
172			def check_message(neighbor, message):
173			message = message.replace(':', '')
174			msghexa = [message[i * 2 : (i * 2) + 2] for i in range(len(message) // 2)]
175			raw = bytes([int(_, 16) for _ in msghexa])
176
177			if raw.startswith(b'\xff' * 16):
178			kind = raw[18]
179			# XXX: FIXME: check size
180			# size = (raw[16] << 16) + raw[17]
181
182			if kind == 1:
183			return check_open(neighbor, raw[18:])
184			elif kind == 2:
185			return check_update(neighbor, raw)
186			elif kind == 3:
187			return check_notification(raw)
188			else:
189			return check_update(neighbor, raw)
190
191
192			# ================================================================= check_update
193			#
194
195
196			def check_open(neighbor, raw):
197			pass
198
199
200			# ================================================================= check_update
201			#
202
203
204			def check_update(neighbor, raw):
205			option.enabled['parser'] = True
206
207			neighbor = neighbor[list(neighbor)[0]]
208
209			path = {}
210			for f in NLRI.known_families():
211			if neighbor.add_path:
212			path[f] = neighbor.add_path
213
214			capa = Capabilities().new(neighbor, False)
215			capa[Capability.CODE.ADD_PATH] = path
216			capa[Capability.CODE.MULTIPROTOCOL] = neighbor.families()
217			# capa[Capability.CODE.FOUR_BYTES_ASN] = True
218
219			routerid_1 = str(neighbor.router_id)
220			routerid_2 = '.'.join(str((int(_) + 1) % 250) for _ in str(neighbor.router_id).split('.', -1))
221
222			o1 = Open(Version(4), ASN(neighbor.local_as), HoldTime(180), RouterID(routerid_1), capa)
223			o2 = Open(Version(4), ASN(neighbor.peer_as), HoldTime(180), RouterID(routerid_2), capa)
224			negotiated = Negotiated(neighbor)
225			negotiated.sent(o1)
226			negotiated.received(o2)
227			# grouped = False
228
229			while raw:
230			if raw.startswith(b'\xff' * 16):
231			kind = raw[18]
232			size = (raw[16] << 16) + raw[17]
233
234			injected, raw = raw[19:size], raw[size:]
235
236			if kind == 2:
237			log.debug('the message is an update', 'parser')
238			decoding = 'update'
239			else:
240			log.debug('the message is not an update (%d) - aborting' % kind, 'parser')
241			return False
242			else:
243			log.debug('header missing, assuming this message is ONE update', 'parser')
244			decoding = 'update'
245			injected, raw = raw, ''
246
247			try:
248			# This does not take the BGP header - let's assume we will not break that :)
249			update = Update.unpack_message(injected, negotiated)
250			except Notify:
251			import traceback
252
253			log.error('could not parse the message', 'parser')
254			log.error(traceback.format_exc(), 'parser')
255			return False
256			except Exception:
257			import traceback
258
259			log.error('could not parse the message', 'parser')
260			log.error(traceback.format_exc(), 'parser')
261			return False
262
263			log.debug('', 'parser') # new line
264			for number in range(len(update.nlris)):
265			change = Change(update.nlris[number], update.attributes)
266			log.info('decoded %s %s %s' % (decoding, change.nlri.action, change.extensive()), 'parser')
267			log.info('update json %s' % Response.JSON(json_version).update(neighbor, 'in', update, None, '', ''), 'parser')
268
269			return True
270
271
272			# ================================================================= check_update
273			#
274
275
276			def check_notification(raw):
277			notification = Notification.unpack_message(raw[18:], None)
278			# XXX: FIXME: should be using logger here
279			print(notification)
280			return True
281

Exa-Networks / exabgp

exabgp.configuration.check.check_generation() 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

exabgp.configuration.check.check_generation() F
last analyzed 2020-07-19 19:06 UTC