build.main.Main.update_circuits() - Code Metrics - Inspection of "Merge pull request #94 from kytos-ng/chore/starlet..." - kytos-ng/sdntrace_cp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 58121b...21a39b )

by Vinicius

created 2023-05-03 18:05 UTC

build.main.Main.update_circuits() A

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	5
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	3
CRAP Score	2

Importance

Changes

Metric	Value
cc	2
eloc	3
nop	1
dl	0
loc	5
ccs	3
cts	3
cp	1
crap	2
rs	10
c	0
b	0
f	0

"""Main module of amlight/sdntrace_cp Kytos Network Application.

Run tracepaths on OpenFlow in the Control Plane
"""

import ipaddress
import pathlib
from datetime import datetime

from kytos.core import KytosNApp, log, rest
from kytos.core.helpers import load_spec, validate_openapi
from kytos.core.rest_api import (HTTPException, JSONResponse, Request,
                                 get_json_or_400)
from napps.amlight.sdntrace_cp import settings
from napps.amlight.sdntrace_cp.automate import Automate
from napps.amlight.sdntrace_cp.utils import (convert_entries,
                                             convert_list_entries,
                                             find_endpoint, get_stored_flows,
                                             match_field_dl_vlan, prepare_json)


class Main(KytosNApp):
    """Main class of amlight/sdntrace_cp NApp.

    This application gets the list of flows from the switches
    and uses it to trace paths without using the data plane.
    """

    spec = load_spec(pathlib.Path(__file__).parent / "openapi.yml")

    def setup(self):
        """Replace the '__init__' method for the KytosNApp subclass.

        The setup method is automatically called by the controller when your
        application is loaded.

        """
        log.info("Starting Kytos SDNTrace CP App!")

        self.automate = Automate(self)
        self.automate.schedule_traces()
        self.automate.schedule_important_traces()

    def execute(self):
        """This method is executed right after the setup method execution.

        You can also use this method in loop mode if you add to the above setup
        method a line like the following example:

            self.execute_as_loop(30)  # 30-second interval.
        """

    def shutdown(self):
        """This method is executed when your napp is unloaded.

        If you have some cleanup procedure, insert it here.
        """
        self.automate.unschedule_ids()
        self.automate.sheduler_shutdown(wait=False)

    @rest('/v1/trace', methods=['PUT'])
    @validate_openapi(spec)
    def trace(self, request: Request) -> JSONResponse:
        """Trace a path."""
        result = []
        data = get_json_or_400(request, self.controller.loop)
        entries = convert_entries(data)
        if not entries:
            raise HTTPException(400, "Empty entries")
        stored_flows = get_stored_flows()
        result = self.tracepath(entries, stored_flows)
        return JSONResponse(prepare_json(result))

    @rest('/v1/traces', methods=['PUT'])
    @validate_openapi(spec)
    def get_traces(self, request: Request) -> JSONResponse:
        """For bulk requests."""
        data = get_json_or_400(request, self.controller.loop)
        entries = convert_list_entries(data)
        stored_flows = get_stored_flows()
        results = []
        for entry in entries:
            results.append(self.tracepath(entry, stored_flows))
        return JSONResponse(prepare_json(results))

    def tracepath(self, entries, stored_flows):
        """Trace a path for a packet represented by entries."""
        # pylint: disable=too-many-branches
        trace_result = []
        trace_type = 'starting'
        do_trace = True
        while do_trace:
            if 'dpid' not in entries or 'in_port' not in entries:
                break
            trace_step = {'in': {'dpid': entries['dpid'],
                                 'port': entries['in_port'],
                                 'time': str(datetime.now()),
                                 'type': trace_type}}
            if 'dl_vlan' in entries:
                trace_step['in'].update({'vlan': entries['dl_vlan'][-1]})

            switch = self.controller.get_switch_by_dpid(entries['dpid'])
            if not switch:
                trace_step['in']['type'] = 'last'
                trace_result.append(trace_step)
                break
            result = self.trace_step(switch, entries, stored_flows)
            if result:
                out = {'port': result['out_port']}
                if 'dl_vlan' in result['entries']:
                    out.update({'vlan': result['entries']['dl_vlan'][-1]})
                trace_step.update({
                    'out': out
                })
                if 'dpid' in result:
                    next_step = {'dpid': result['dpid'],
                                 'port': result['in_port']}
                    entries = result['entries']
                    entries['dpid'] = result['dpid']
                    entries['in_port'] = result['in_port']
                    if self.has_loop(next_step, trace_result):
                        trace_step['in']['type'] = 'loop'
                        do_trace = False
                    else:
                        trace_type = 'intermediary'
                else:
                    trace_step['in']['type'] = 'last'
                    do_trace = False
            else:
                trace_step['in']['type'] = 'incomplete'
                do_trace = False
            if 'out' in trace_step and trace_step['out']:
                if self.check_loop_trace_step(trace_step, trace_result):
                    do_trace = False
            trace_result.append(trace_step)
        return trace_result

    @staticmethod
    def check_loop_trace_step(trace_step, trace_result):
        """Check if there is a loop in the trace and add the step."""
        # outgoing interface is the same as the input interface
        if not trace_result and \
                trace_step['in']['type'] == 'last' and \
                trace_step['in']['port'] == trace_step['out']['port']:
            trace_step['in']['type'] = 'loop'
            return True
        if trace_result and \
                trace_result[0]['in']['dpid'] == trace_step['in']['dpid'] and \
                trace_result[0]['in']['port'] == trace_step['out']['port']:
            trace_step['in']['type'] = 'loop'
            return True
        return False

    @staticmethod
    def has_loop(trace_step, trace_result):
        """Check if there is a loop in the trace result."""
        for trace in trace_result:
            if trace['in']['dpid'] == trace_step['dpid'] and \
                            trace['in']['port'] == trace_step['port']:
                return True
        return False

    def trace_step(self, switch, entries, stored_flows):
        """Perform a trace step.

        Match the given fields against the switch's list of flows."""
        flow, entries, port = self.match_and_apply(
                                                    switch,
                                                    entries,
                                                    stored_flows
                                                )

        if not flow or not port:
            return None

        endpoint = find_endpoint(switch, port)
        if endpoint is None:
            log.warning(f"Port {port} not found on switch {switch}")
            return None
        endpoint = endpoint['endpoint']
        if endpoint is None:
            return {'out_port': port,
                    'entries': entries}

        return {'dpid': endpoint.switch.dpid,
                'in_port': endpoint.port_number,
                'out_port': port,
                'entries': entries}

    def update_circuits(self):
        """Update the list of circuits after a flow change."""
        # pylint: disable=unused-argument
        if settings.FIND_CIRCUITS_IN_FLOWS:
            self.automate.find_circuits()

    @classmethod
    def do_match(cls, flow, args):
        """Match a packet against this flow (OF1.3)."""
        # pylint: disable=consider-using-dict-items
        # pylint: disable=too-many-return-statements
        if ('match' not in flow['flow']) or (len(flow['flow']['match']) == 0):
            return False
        for name in flow['flow']['match']:
            field_flow = flow['flow']['match'][name]
            if name == 'dl_vlan':
                field = args.get(name)
                if field:
                    field = field[-1]
                if not match_field_dl_vlan(field, field_flow):
                    return False
                continue
            if name not in args:
                return False
            field = args[name]
            if name not in ('ipv4_src', 'ipv4_dst', 'ipv6_src', 'ipv6_dst'):
                if field_flow != field:
                    return False
            else:
                packet_ip = int(ipaddress.ip_address(field))
                ip_addr = flow['flow']['match'][name]
                if packet_ip & ip_addr.netmask != ip_addr.address:
                    return False
        return flow

    def match_flows(self, switch, args, stored_flows, many=True):
        # pylint: disable=bad-staticmethod-argument
        """
        Match the packet in request against the stored flows from flow_manager.
        Try the match with each flow, in other. If many is True, tries the
        match with all flows, if False, tries until the first match.
        :param args: packet data
        :param many: Boolean, indicating whether to continue after matching the
                first flow or not
        :return: If many, the list of matched flows, or the matched flow
        """
        if switch.dpid not in stored_flows:
            return None
        response = []
        if switch.dpid not in stored_flows:
            return None
        try:
            for flow in stored_flows[switch.dpid]:
                match = Main.do_match(flow, args)
                if match:
                    if many:
                        response.append(match)
                    else:
                        response = match
                        break
        except AttributeError:
            return None
        if not many and isinstance(response, list):
            return None
        return response

    # pylint: disable=redefined-outer-name
    def match_and_apply(self, switch, args, stored_flows):
        # pylint: disable=bad-staticmethod-argument
        """Match flows and apply actions.
        Match given packet (in args) against
        the stored flows (from flow_manager) and,
        if a match flow is found, apply its actions."""
        flow = self.match_flows(switch, args, stored_flows, False)
        port = None
        actions = []
        # pylint: disable=too-many-nested-blocks
        if not flow or switch.ofp_version != '0x04':
            return flow, args, port
        if 'actions' in flow['flow']:
            actions = flow['flow']['actions']
        for action in actions:
            action_type = action['action_type']
            if action_type == 'output':
                port = action['port']
            if action_type == 'push_vlan':
                if 'dl_vlan' not in args:
                    args['dl_vlan'] = []
                args['dl_vlan'].append(0)
            if action_type == 'pop_vlan':
                if 'dl_vlan' in args:
                    args['dl_vlan'].pop()
                    if len(args['dl_vlan']) == 0:
                        del args['dl_vlan']
            if action_type == 'set_vlan':
                args['dl_vlan'][-1] = action['vlan_id']
        return flow, args, port


1		"""Main module of amlight/sdntrace_cp Kytos Network Application.
2
3		Run tracepaths on OpenFlow in the Control Plane
4		"""
5
6	1	import ipaddress
7	1	import pathlib
8	1	from datetime import datetime
9
10	1	from kytos.core import KytosNApp, log, rest
11	1	from kytos.core.helpers import load_spec, validate_openapi
12	1	from kytos.core.rest_api import (HTTPException, JSONResponse, Request,
13		get_json_or_400)
14	1	from napps.amlight.sdntrace_cp import settings
15	1	from napps.amlight.sdntrace_cp.automate import Automate
16	1	from napps.amlight.sdntrace_cp.utils import (convert_entries,
17		convert_list_entries,
18		find_endpoint, get_stored_flows,
19		match_field_dl_vlan, prepare_json)
20
21
22	1	class Main(KytosNApp):
23		"""Main class of amlight/sdntrace_cp NApp.
24
25		This application gets the list of flows from the switches
26		and uses it to trace paths without using the data plane.
27		"""
28
29	1	spec = load_spec(pathlib.Path(__file__).parent / "openapi.yml")
30
31	1	def setup(self):
32		"""Replace the '__init__' method for the KytosNApp subclass.
33
34		The setup method is automatically called by the controller when your
35		application is loaded.
36
37		"""
38	1	log.info("Starting Kytos SDNTrace CP App!")
39
40	1	self.automate = Automate(self)
41	1	self.automate.schedule_traces()
42	1	self.automate.schedule_important_traces()
43
44	1	def execute(self):
45		"""This method is executed right after the setup method execution.
46
47		You can also use this method in loop mode if you add to the above setup
48		method a line like the following example:
49
50		self.execute_as_loop(30) # 30-second interval.
51		"""
52
53	1	def shutdown(self):
54		"""This method is executed when your napp is unloaded.
55
56		If you have some cleanup procedure, insert it here.
57		"""
58		self.automate.unschedule_ids()
59		self.automate.sheduler_shutdown(wait=False)
60
61	1	@rest('/v1/trace', methods=['PUT'])
62	1	@validate_openapi(spec)
63	1	def trace(self, request: Request) -> JSONResponse:
64		"""Trace a path."""
65	1	result = []
66	1	data = get_json_or_400(request, self.controller.loop)
67	1	entries = convert_entries(data)
68	1	if not entries:
69		raise HTTPException(400, "Empty entries")
70	1	stored_flows = get_stored_flows()
71	1	result = self.tracepath(entries, stored_flows)
72	1	return JSONResponse(prepare_json(result))
73
74	1	@rest('/v1/traces', methods=['PUT'])
75	1	@validate_openapi(spec)
76	1	def get_traces(self, request: Request) -> JSONResponse:
77		"""For bulk requests."""
78	1	data = get_json_or_400(request, self.controller.loop)
79	1	entries = convert_list_entries(data)
80	1	stored_flows = get_stored_flows()
81	1	results = []
82	1	for entry in entries:
83	1	results.append(self.tracepath(entry, stored_flows))
84	1	return JSONResponse(prepare_json(results))
85
86	1	def tracepath(self, entries, stored_flows):
87		"""Trace a path for a packet represented by entries."""
88		# pylint: disable=too-many-branches
89	1	trace_result = []
90	1	trace_type = 'starting'
91	1	do_trace = True
92	1	while do_trace:
93	1	if 'dpid' not in entries or 'in_port' not in entries:
94		break
95	1	trace_step = {'in': {'dpid': entries['dpid'],
96		'port': entries['in_port'],
97		'time': str(datetime.now()),
98		'type': trace_type}}
99	1	if 'dl_vlan' in entries:
100	1	trace_step['in'].update({'vlan': entries['dl_vlan'][-1]})
101
102	1	switch = self.controller.get_switch_by_dpid(entries['dpid'])
103	1	if not switch:
104	1	trace_step['in']['type'] = 'last'
105	1	trace_result.append(trace_step)
106	1	break
107	1	result = self.trace_step(switch, entries, stored_flows)
108	1	if result:
109	1	out = {'port': result['out_port']}
110	1	if 'dl_vlan' in result['entries']:
111	1	out.update({'vlan': result['entries']['dl_vlan'][-1]})
112	1	trace_step.update({
113		'out': out
114		})
115	1	if 'dpid' in result:
116	1	next_step = {'dpid': result['dpid'],
117		'port': result['in_port']}
118	1	entries = result['entries']
119	1	entries['dpid'] = result['dpid']
120	1	entries['in_port'] = result['in_port']
121	1	if self.has_loop(next_step, trace_result):
122	1	trace_step['in']['type'] = 'loop'
123	1	do_trace = False
124		else:
125	1	trace_type = 'intermediary'
126		else:
127	1	trace_step['in']['type'] = 'last'
128	1	do_trace = False
129		else:
130	1	trace_step['in']['type'] = 'incomplete'
131	1	do_trace = False
132	1	if 'out' in trace_step and trace_step['out']:
133	1	if self.check_loop_trace_step(trace_step, trace_result):
134	1	do_trace = False
135	1	trace_result.append(trace_step)
136	1	return trace_result
137
138	1	@staticmethod
139	1	def check_loop_trace_step(trace_step, trace_result):
140		"""Check if there is a loop in the trace and add the step."""
141		# outgoing interface is the same as the input interface
142	1	if not trace_result and \
143		trace_step['in']['type'] == 'last' and \
144		trace_step['in']['port'] == trace_step['out']['port']:
145	1	trace_step['in']['type'] = 'loop'
146	1	return True
147	1	if trace_result and \
148		trace_result[0]['in']['dpid'] == trace_step['in']['dpid'] and \
149		trace_result[0]['in']['port'] == trace_step['out']['port']:
150	1	trace_step['in']['type'] = 'loop'
151	1	return True
152	1	return False
153
154	1	@staticmethod
155	1	def has_loop(trace_step, trace_result):
156		"""Check if there is a loop in the trace result."""
157	1	for trace in trace_result:
158	1	if trace['in']['dpid'] == trace_step['dpid'] and \
159		trace['in']['port'] == trace_step['port']:
160	1	return True
161	1	return False
162
163	1	def trace_step(self, switch, entries, stored_flows):
164		"""Perform a trace step.
165
166		Match the given fields against the switch's list of flows."""
167	1	flow, entries, port = self.match_and_apply(
168		switch,
169		entries,
170		stored_flows
171		)
172
173	1	if not flow or not port:
174	1	return None
175
176	1	endpoint = find_endpoint(switch, port)
177	1	if endpoint is None:
178		log.warning(f"Port {port} not found on switch {switch}")
179		return None
180	1	endpoint = endpoint['endpoint']
181	1	if endpoint is None:
182	1	return {'out_port': port,
183		'entries': entries}
184
185	1	return {'dpid': endpoint.switch.dpid,
186		'in_port': endpoint.port_number,
187		'out_port': port,
188		'entries': entries}
189
190	1	def update_circuits(self):
191		"""Update the list of circuits after a flow change."""
192		# pylint: disable=unused-argument
193	1	if settings.FIND_CIRCUITS_IN_FLOWS:
194	1	self.automate.find_circuits()
195
196	1	@classmethod
197	1	def do_match(cls, flow, args):
198		"""Match a packet against this flow (OF1.3)."""
199		# pylint: disable=consider-using-dict-items
200		# pylint: disable=too-many-return-statements
201	1	if ('match' not in flow['flow']) or (len(flow['flow']['match']) == 0):
202	1	return False
203	1	for name in flow['flow']['match']:
204	1	field_flow = flow['flow']['match'][name]
205	1	if name == 'dl_vlan':
206	1	field = args.get(name)
207	1	if field:
208	1	field = field[-1]
209	1	if not match_field_dl_vlan(field, field_flow):
210	1	return False
211		continue
212	1	if name not in args:
213		return False
214	1	field = args[name]
215	1	if name not in ('ipv4_src', 'ipv4_dst', 'ipv6_src', 'ipv6_dst'):
216	1	if field_flow != field:
217	1	return False
218		else:
219		packet_ip = int(ipaddress.ip_address(field))
220		ip_addr = flow['flow']['match'][name]
221		if packet_ip & ip_addr.netmask != ip_addr.address:
222		return False
223	1	return flow
224
225	1	def match_flows(self, switch, args, stored_flows, many=True):
226		# pylint: disable=bad-staticmethod-argument
227		"""
228		Match the packet in request against the stored flows from flow_manager.
229		Try the match with each flow, in other. If many is True, tries the
230		match with all flows, if False, tries until the first match.
231		:param args: packet data
232		:param many: Boolean, indicating whether to continue after matching the
233		first flow or not
234		:return: If many, the list of matched flows, or the matched flow
235		"""
236	1	if switch.dpid not in stored_flows:
237		return None
238	1	response = []
239	1	if switch.dpid not in stored_flows:
240		return None
241	1	try:
242	1	for flow in stored_flows[switch.dpid]:
243	1	match = Main.do_match(flow, args)
244	1	if match:
245	1	if many:
246		response.append(match)
247		else:
248	1	response = match
249	1	break
250		except AttributeError:
251		return None
252	1	if not many and isinstance(response, list):
253	1	return None
254	1	return response
255
256		# pylint: disable=redefined-outer-name
257	1	def match_and_apply(self, switch, args, stored_flows):
258		# pylint: disable=bad-staticmethod-argument
259		"""Match flows and apply actions.
260		Match given packet (in args) against
261		the stored flows (from flow_manager) and,
262		if a match flow is found, apply its actions."""
263	1	flow = self.match_flows(switch, args, stored_flows, False)
264	1	port = None
265	1	actions = []
266		# pylint: disable=too-many-nested-blocks
267	1	if not flow or switch.ofp_version != '0x04':
268	1	return flow, args, port
269	1	if 'actions' in flow['flow']:
270	1	actions = flow['flow']['actions']
271	1	for action in actions:
272	1	action_type = action['action_type']
273	1	if action_type == 'output':
274	1	port = action['port']
275	1	if action_type == 'push_vlan':
276	1	if 'dl_vlan' not in args:
277		args['dl_vlan'] = []
278	1	args['dl_vlan'].append(0)
279	1	if action_type == 'pop_vlan':
280	1	if 'dl_vlan' in args:
281	1	args['dl_vlan'].pop()
282	1	if len(args['dl_vlan']) == 0:
283	1	del args['dl_vlan']
284	1	if action_type == 'set_vlan':
285	1	args['dl_vlan'][-1] = action['vlan_id']
286		return flow, args, port
287

kytos-ng / sdntrace_cp

Push — master ( 58121b...21a39b )

build.main.Main.update_circuits() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like