build.main.Main.tracepath() - Code Metrics - Inspection of "Temp" - kytos-ng/sdntrace_cp - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Retried Inspection

Pull Request — master (#47)

unknown

created 2022-10-13 20:14 UTC

build.main.Main.tracepath() B

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	42
Code Lines	36

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	26
CRAP Score	7.0024

Importance

Changes

Metric	Value
cc	7
eloc	36
nop	2
dl	0
loc	42
rs	7.616
c	0
b	0
f	0
ccs	26
cts	27
cp	0.963
crap	7.0024

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

Run tracepaths on OpenFlow in the Control Plane
"""

import ipaddress
from datetime import datetime

import requests
from flask import jsonify, request
from kytos.core import KytosNApp, log, rest
from kytos.core.helpers import listen_to
from napps.amlight.sdntrace_cp import settings
from napps.amlight.sdntrace_cp.automate import Automate
from napps.amlight.sdntrace_cp.utils import (convert_list_entries,
                                             find_endpoint, prepare_json,
                                             prepare_list_json)
from napps.kytos.of_core.v0x04.flow import Action


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.
    """

    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.traces = {}
        self.last_id = 30000
        self.automate = Automate(self)
        self.automate.schedule_traces()
        self.automate.schedule_important_traces()
        self.stored_flows = None

    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('/trace', methods=['PUT'])
    @rest('/traces', methods=['PUT'])
    def trace(self):
        """Trace a path."""
        entries = request.get_json()
        return_list = True
        if isinstance(entries, dict):
            entries = [entries]
            return_list = False
        entries = convert_list_entries(entries)
        self.get_stored_flows_and_map()
        if return_list:
            results = {}
            list_ready = []
            for entry in entries:
                if (entry['dpid'], entry['in_port']) in list_ready:
                    continue
                list_ready.append((entry['dpid'], entry['in_port']))
                dpid = entry['dpid']
                if dpid not in results:
                    results[dpid] = []
                result = prepare_list_json(self.tracepath(entry))
                results[dpid].append(result)
            return jsonify(results)
        result = self.tracepath(entries[0])
        return jsonify(prepare_json(result))

    def tracepath(self, entries):
        """Trace a path for a packet represented by entries."""
        self.last_id += 1
        trace_id = self.last_id
        trace_result = []
        trace_type = 'starting'
        do_trace = True
        while do_trace:
            trace_step = {'in': {'dpid': entries['dpid'],
                                 'port': entries['in_port'],
                                 'time': str(datetime.now()),
                                 'type': trace_type}}
            if 'vlan_vid' in entries:
                trace_step['in'].update({'vlan': entries['vlan_vid'][-1]})
            switch = self.controller.get_switch_by_dpid(entries['dpid'])
            result = self.trace_step(switch, entries)
            if result:
                out = {'port': result['out_port']}
                if 'vlan_vid' in result['entries']:
                    out.update({'vlan': result['entries']['vlan_vid'][-1]})
                trace_step.update({
                    'out': out
                })
                if 'dpid' in result:
                    next_step = {'dpid': result['dpid'],
                                 'port': result['in_port']}
                    if self.has_loop(next_step, trace_result):
                        do_trace = False
                    else:
                        entries = result['entries']
                        entries['dpid'] = result['dpid']
                        entries['in_port'] = result['in_port']
                        trace_type = 'trace'
                else:
                    do_trace = False
            else:
                do_trace = False
            trace_result.append(trace_step)
        self.traces.update({
            trace_id: trace_result
        })
        return trace_result

    @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):
        """Perform a trace step.

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

        if not flow or not port:
            return None

        endpoint = find_endpoint(switch, port)
        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}

    @listen_to('amlight/flow_stats.flows_updated')
    def update_circuits(self, event):
        """Update the list of circuits after a flow change."""
        # pylint: disable=unused-argument
        if settings.FIND_CIRCUITS_IN_FLOWS:
            self.automate.find_circuits()

    def map_flows(self, switches=None):
        """Map the flows in memory given the stored flows"""
        flows = {}
        if not switches:
            switches = self.controller.switches.copy().values()
        for switch in switches:
            flows[switch.dpid] = []
            flows_sw_dict = {flow_sw.id: flow_sw for flow_sw in switch.flows}
            for flow_item in self.stored_flows[switch.dpid]:
                id_ = flow_item['flow_id']
                if id_ in flows_sw_dict:
                    flow = flows_sw_dict[id_]
                    flow_item = flow_item['flow']
                    flow.cookie = flow_item['cookie']
                    flow.hard_timeout = flow_item['hard_timeout']
                    flow.idle_timeout = flow_item['idle_timeout']
                    flow.priority = flow_item['priority']
                    flow.table_id = flow_item['table_id']
                    flow.actions = []
                    for action in flow_item['actions']:
                        action = Action.from_dict(action)
                        flow.actions.append(action)
                    flows[switch.dpid].append(flow)
        # pylint: disable=attribute-defined-outside-init
        self.stored_flows = flows

    @staticmethod
    def get_stored_flows(dpids: list = None, state: str = None):
        """Get stored flows from flow_manager napps."""
        api_url = f'{settings.FLOW_MANAGER_URL}/stored_flows'
        if dpids:
            str_dpids = ''
            for dpid in dpids:
                str_dpids += f'&dpid={dpid}'
            api_url += '/?'+str_dpids[1:]
        if state:
            char = '&' if dpids else '/?'
            api_url += char+f'state={state}'
        result = requests.get(api_url)
        flows_from_manager = result.json()
        return flows_from_manager

    def get_stored_flows_and_map(self, dpids: list = None, state: str = None):
        """Get stored flows from flow_manager napp and
        map witn flows on switches."""
        # pylint: disable=attribute-defined-outside-init
        self.stored_flows = Main.get_stored_flows(dpids, state)
        self.map_flows()

    @classmethod
    def do_match(cls, flow, args):
        """Match a packet against this flow (OF1.3)."""
        # pylint: disable=consider-using-dict-items
        if len(flow.as_dict()['match']) == 0:
            return False
        for name in flow.as_dict()['match']:
            field_flow = flow.as_dict()['match'][name]
            if name == 'dl_vlan':
                name = 'vlan_vid'
            if name not in args:
                return False
            if name == 'vlan_vid':
                field = args[name][-1]
            else:
                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.as_dict()['match'][name]
                if packet_ip & ip_addr.netmask != ip_addr.address:
                    return False
        return flow

    def match_flows(self, switch, args, many=True):
        # pylint: disable=bad-staticmethod-argument
        """
        Match the packet in request against the flows installed
        in the switch and stored according 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 not self.stored_flows:
            self.get_stored_flows_and_map()
        response = []
        try:
            for flow in self.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):
        # pylint: disable=bad-staticmethod-argument
        """Match flows and apply actions.
        Match given packet (in args) against the switch flows (stored) and,
        if a match flow is found, apply its actions."""
        flow = self.match_flows(switch, args, False)
        port = None
        actions = None
        # pylint: disable=too-many-nested-blocks
        if flow:
            actions = flow.actions
            if switch.ofp_version == '0x04':
                for action in actions:
                    action_type = action.action_type
                    if action_type == 'output':
                        port = action.port
                    if action_type == 'push_vlan':
                        if 'vlan_vid' not in args:
                            args['vlan_vid'] = []
                        args['vlan_vid'].append(0)
                    if action_type == 'pop_vlan':
                        if 'vlan_vid' in args:
                            args['vlan_vid'].pop()
                            if len(args['vlan_vid']) == 0:
                                del args['vlan_vid']
                    if action_type == 'set_vlan':
                        args['vlan_vid'][-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		from datetime import datetime
8	1
9	1	import requests
10	1	from flask import jsonify, request
11	1	from kytos.core import KytosNApp, log, rest
12	1	from kytos.core.helpers import listen_to
13	1	from napps.amlight.sdntrace_cp import settings
14	1	from napps.amlight.sdntrace_cp.automate import Automate
15		from napps.amlight.sdntrace_cp.utils import (convert_list_entries,
16		find_endpoint, prepare_json,
17		prepare_list_json)
18	1	from napps.kytos.of_core.v0x04.flow import Action
19
20
21		class Main(KytosNApp):
22		"""Main class of amlight/sdntrace_cp NApp.
23
24		This application gets the list of flows from the switches
25	1	and uses it to trace paths without using the data plane.
26		"""
27
28		def setup(self):
29		"""Replace the '__init__' method for the KytosNApp subclass.
30
31		The setup method is automatically called by the controller when your
32	1	application is loaded.
33
34	1	"""
35	1	log.info("Starting Kytos SDNTrace CP App!")
36	1
37	1	self.traces = {}
38	1	self.last_id = 30000
39		self.automate = Automate(self)
40	1	self.automate.schedule_traces()
41		self.automate.schedule_important_traces()
42		self.stored_flows = None
43
44		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	1
50		self.execute_as_loop(30) # 30-second interval.
51		"""
52
53		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	1	"""
58	1	self.automate.unschedule_ids()
59		self.automate.sheduler_shutdown(wait=False)
60	1
61	1	@rest('/trace', methods=['PUT'])
62	1	@rest('/traces', methods=['PUT'])
63		def trace(self):
64	1	"""Trace a path."""
65		entries = request.get_json()
66	1	return_list = True
67	1	if isinstance(entries, dict):
68	1	entries = [entries]
69	1	return_list = False
70	1	entries = convert_list_entries(entries)
71		self.get_stored_flows_and_map()
72	1	if return_list:
73	1	results = {}
74	1	list_ready = []
75		for entry in entries:
76		if (entry['dpid'], entry['in_port']) in list_ready:
77		continue
78	1	list_ready.append((entry['dpid'], entry['in_port']))
79	1	dpid = entry['dpid']
80	1	if dpid not in results:
81	1	results[dpid] = []
82	1	result = prepare_list_json(self.tracepath(entry))
83	1	results[dpid].append(result)
84	1	return jsonify(results)
85		result = self.tracepath(entries[0])
86	1	return jsonify(prepare_json(result))
87
88		def tracepath(self, entries):
89	1	"""Trace a path for a packet represented by entries."""
90	1	self.last_id += 1
91		trace_id = self.last_id
92	1	trace_result = []
93	1	trace_type = 'starting'
94		do_trace = True
95	1	while do_trace:
96	1	trace_step = {'in': {'dpid': entries['dpid'],
97	1	'port': entries['in_port'],
98	1	'time': str(datetime.now()),
99		'type': trace_type}}
100		if 'vlan_vid' in entries:
101		trace_step['in'].update({'vlan': entries['vlan_vid'][-1]})
102	1	switch = self.controller.get_switch_by_dpid(entries['dpid'])
103	1	result = self.trace_step(switch, entries)
104	1	if result:
105		out = {'port': result['out_port']}
106		if 'vlan_vid' in result['entries']:
107	1	out.update({'vlan': result['entries']['vlan_vid'][-1]})
108		trace_step.update({
109	1	'out': out
110	1	})
111		if 'dpid' in result:
112	1	next_step = {'dpid': result['dpid'],
113	1	'port': result['in_port']}
114		if self.has_loop(next_step, trace_result):
115	1	do_trace = False
116	1	else:
117		entries = result['entries']
118	1	entries['dpid'] = result['dpid']
119	1	entries['in_port'] = result['in_port']
120		trace_type = 'trace'
121		else:
122		do_trace = False
123	1	else:
124	1	do_trace = False
125	1	trace_result.append(trace_step)
126		self.traces.update({
127	1	trace_id: trace_result
128	1	})
129	1	return trace_result
130
131		@staticmethod
132	1	def has_loop(trace_step, trace_result):
133		"""Check if there is a loop in the trace result."""
134		for trace in trace_result:
135		if trace['in']['dpid'] == trace_step['dpid'] and \
136		trace['in']['port'] == trace_step['port']:
137	1	return True
138	1	return False
139
140		def trace_step(self, switch, entries):
141	1	"""Perform a trace step.
142	1
143		Match the given fields against the switch's list of flows."""
144		flow, entries, port = self.match_and_apply(switch, entries)
145
146		if not flow or not port:
147		return None
148
149		endpoint = find_endpoint(switch, port)
150		if endpoint is None:
151		return {'out_port': port,
152		'entries': entries}
153
154		return {'dpid': endpoint.switch.dpid,
155		'in_port': endpoint.port_number,
156		'out_port': port,
157		'entries': entries}
158
159		@listen_to('amlight/flow_stats.flows_updated')
160		def update_circuits(self, event):
161		"""Update the list of circuits after a flow change."""
162		# pylint: disable=unused-argument
163		if settings.FIND_CIRCUITS_IN_FLOWS:
164		self.automate.find_circuits()
165
166		def map_flows(self, switches=None):
167		"""Map the flows in memory given the stored flows"""
168		flows = {}
169		if not switches:
170		switches = self.controller.switches.copy().values()
171		for switch in switches:
172		flows[switch.dpid] = []
173		flows_sw_dict = {flow_sw.id: flow_sw for flow_sw in switch.flows}
174		for flow_item in self.stored_flows[switch.dpid]:
175		id_ = flow_item['flow_id']
176		if id_ in flows_sw_dict:
177		flow = flows_sw_dict[id_]
178		flow_item = flow_item['flow']
179		flow.cookie = flow_item['cookie']
180		flow.hard_timeout = flow_item['hard_timeout']
181		flow.idle_timeout = flow_item['idle_timeout']
182		flow.priority = flow_item['priority']
183		flow.table_id = flow_item['table_id']
184		flow.actions = []
185		for action in flow_item['actions']:
186		action = Action.from_dict(action)
187		flow.actions.append(action)
188		flows[switch.dpid].append(flow)
189		# pylint: disable=attribute-defined-outside-init
190		self.stored_flows = flows
191
192		@staticmethod
193		def get_stored_flows(dpids: list = None, state: str = None):
194		"""Get stored flows from flow_manager napps."""
195		api_url = f'{settings.FLOW_MANAGER_URL}/stored_flows'
196		if dpids:
197		str_dpids = ''
198		for dpid in dpids:
199		str_dpids += f'&dpid={dpid}'
200		api_url += '/?'+str_dpids[1:]
201		if state:
202		char = '&' if dpids else '/?'
203		api_url += char+f'state={state}'
204		result = requests.get(api_url)
205		flows_from_manager = result.json()
206		return flows_from_manager
207
208		def get_stored_flows_and_map(self, dpids: list = None, state: str = None):
209		"""Get stored flows from flow_manager napp and
210		map witn flows on switches."""
211		# pylint: disable=attribute-defined-outside-init
212		self.stored_flows = Main.get_stored_flows(dpids, state)
213		self.map_flows()
214
215		@classmethod
216		def do_match(cls, flow, args):
217		"""Match a packet against this flow (OF1.3)."""
218		# pylint: disable=consider-using-dict-items
219		if len(flow.as_dict()['match']) == 0:
220		return False
221		for name in flow.as_dict()['match']:
222		field_flow = flow.as_dict()['match'][name]
223		if name == 'dl_vlan':
224		name = 'vlan_vid'
225		if name not in args:
226		return False
227		if name == 'vlan_vid':
228		field = args[name][-1]
229		else:
230		field = args[name]
231		if name not in ('ipv4_src', 'ipv4_dst', 'ipv6_src', 'ipv6_dst'):
232		if field_flow != field:
233		return False
234		else:
235		packet_ip = int(ipaddress.ip_address(field))
236		ip_addr = flow.as_dict()['match'][name]
237		if packet_ip & ip_addr.netmask != ip_addr.address:
238		return False
239		return flow
240
241		def match_flows(self, switch, args, many=True):
242		# pylint: disable=bad-staticmethod-argument
243		"""
244		Match the packet in request against the flows installed
245		in the switch and stored according flow_manager.
246		Try the match with each flow, in other. If many is True, tries the
247		match with all flows, if False, tries until the first match.
248		:param args: packet data
249		:param many: Boolean, indicating whether to continue after matching the
250		first flow or not
251		:return: If many, the list of matched flows, or the matched flow
252		"""
253		if not self.stored_flows:
254		self.get_stored_flows_and_map()
255		response = []
256		try:
257		for flow in self.stored_flows[switch.dpid]:
258		match = Main.do_match(flow, args)
259		if match:
260		if many:
261		response.append(match)
262		else:
263		response = match
264		break
265		except AttributeError:
266		return None
267		if not many and isinstance(response, list):
268		return None
269		return response
270
271		# pylint: disable=redefined-outer-name
272		def match_and_apply(self, switch, args):
273		# pylint: disable=bad-staticmethod-argument
274		"""Match flows and apply actions.
275		Match given packet (in args) against the switch flows (stored) and,
276		if a match flow is found, apply its actions."""
277		flow = self.match_flows(switch, args, False)
278		port = None
279		actions = None
280		# pylint: disable=too-many-nested-blocks
281		if flow:
282		actions = flow.actions
283		if switch.ofp_version == '0x04':
284		for action in actions:
285		action_type = action.action_type
286		if action_type == 'output':
287		port = action.port
288		if action_type == 'push_vlan':
289		if 'vlan_vid' not in args:
290		args['vlan_vid'] = []
291		args['vlan_vid'].append(0)
292		if action_type == 'pop_vlan':
293		if 'vlan_vid' in args:
294		args['vlan_vid'].pop()
295		if len(args['vlan_vid']) == 0:
296		del args['vlan_vid']
297		if action_type == 'set_vlan':
298		args['vlan_vid'][-1] = action.vlan_id
299		return flow, args, port
300

kytos-ng / sdntrace_cp

Pull Request — master (#47)

build.main.Main.tracepath() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like