build.main.Main._get_data() - Code Metrics - Inspection of "Merge pull request #54 from kytos-ng/fix/issue_53_..." - kytos-ng/of_lldp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( de6073...127cfb )

by Vinicius

created 2022-07-18 17:54 UTC

build.main.Main._get_data() A

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	5
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	4
CRAP Score	1

Importance

Changes

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

"""NApp responsible to discover new switches and hosts."""
import struct
import time

import requests
from flask import jsonify, request
from pyof.foundation.basic_types import DPID, UBInt16, UBInt32
from pyof.foundation.network_types import LLDP, VLAN, Ethernet, EtherType
from pyof.v0x01.common.action import ActionOutput as AO10
from pyof.v0x01.common.phy_port import Port as Port10
from pyof.v0x01.controller2switch.packet_out import PacketOut as PO10
from pyof.v0x04.common.action import ActionOutput as AO13
from pyof.v0x04.common.port import PortNo as Port13
from pyof.v0x04.controller2switch.packet_out import PacketOut as PO13

from kytos.core import KytosEvent, KytosNApp, log, rest
from kytos.core.helpers import listen_to
from napps.kytos.of_core.msg_prios import of_msg_prio
from napps.kytos.of_lldp import constants, settings
from napps.kytos.of_lldp.loop_manager import LoopManager, LoopState
from napps.kytos.of_lldp.utils import get_cookie


class Main(KytosNApp):
    """Main OF_LLDP NApp Class."""

    def setup(self):
        """Make this NApp run in a loop."""
        self.vlan_id = None
        self.polling_time = settings.POLLING_TIME
        if hasattr(settings, "FLOW_VLAN_VID"):
            self.vlan_id = settings.FLOW_VLAN_VID
        self.execute_as_loop(self.polling_time)
        self.loop_manager = LoopManager(self.controller)

    def execute(self):
        """Send LLDP Packets every 'POLLING_TIME' seconds to all switches."""
        switches = list(self.controller.switches.values())
        for switch in switches:
            try:
                of_version = switch.connection.protocol.version
            except AttributeError:
                of_version = None

            if not switch.is_connected():
                continue

            if of_version == 0x01:
                port_type = UBInt16
                local_port = Port10.OFPP_LOCAL
            elif of_version == 0x04:
                port_type = UBInt32
                local_port = Port13.OFPP_LOCAL
            else:
                # skip the current switch with unsupported OF version
                continue

            interfaces = list(switch.interfaces.values())
            for interface in interfaces:
                # Interface marked to receive lldp packet
                # Only send LLDP packet to active interface
                if(not interface.lldp or not interface.is_active()
                   or not interface.is_enabled()):
                    continue
                # Avoid the interface that connects to the controller.
                if interface.port_number == local_port:
                    continue

                lldp = LLDP()
                lldp.chassis_id.sub_value = DPID(switch.dpid)
                lldp.port_id.sub_value = port_type(interface.port_number)

                ethernet = Ethernet()
                ethernet.ether_type = EtherType.LLDP
                ethernet.source = interface.address
                ethernet.destination = constants.LLDP_MULTICAST_MAC
                ethernet.data = lldp.pack()
                # self.vlan_id == None will result in a packet with no VLAN.
                ethernet.vlans.append(VLAN(vid=self.vlan_id))

                packet_out = self._build_lldp_packet_out(
                                    of_version,
                                    interface.port_number, ethernet.pack())

                if packet_out is None:
                    continue

                event_out = KytosEvent(
                    name='kytos/of_lldp.messages.out.ofpt_packet_out',
                    priority=of_msg_prio(packet_out.header.message_type.value),
                    content={
                            'destination': switch.connection,
                            'message': packet_out})

                self.controller.buffers.msg_out.put(event_out)
                log.debug(
                    "Sending a LLDP PacketOut to the switch %s",
                    switch.dpid)

                msg = 'Switch: %s (%s)'
                msg += ' Interface: %s'
                msg += ' -- LLDP PacketOut --'
                msg += ' Ethernet: eth_type (%s) | src (%s) | dst (%s) /'
                msg += ' LLDP: Switch (%s) | portno (%s)'

                log.debug(
                    msg,
                    switch.connection, switch.dpid,
                    interface.id, ethernet.ether_type,
                    ethernet.source, ethernet.destination,
                    switch.dpid, interface.port_number)

        self.try_to_publish_stopped_loops()

    def try_to_publish_stopped_loops(self):
        """Try to publish current stopped loops."""
        for dpid, port_pairs in self.loop_manager.get_stopped_loops().items():
            try:
                switch = self.controller.get_switch_by_dpid(dpid)
                for port_pair in port_pairs:
                    interface_a = switch.interfaces[port_pair[0]]
                    interface_b = switch.interfaces[port_pair[1]]
                    self.loop_manager.publish_loop_state(
                        interface_a, interface_b, LoopState.stopped.value
                    )
            except (KeyError, AttributeError) as exc:
                log.error("try_to_publish_stopped_loops failed with switch:"
                          f"{dpid}, port_pair: {port_pair}. {str(exc)}")

    @listen_to('kytos/topology.switch.(enabled|disabled)')
    def handle_lldp_flows(self, event):
        """Install or remove flows in a switch.

        Install a flow to send LLDP packets to the controller. The proactive
        flow is installed whenever a switch is enabled. If the switch is
        disabled the flow is removed.

        Args:
            event (:class:`~kytos.core.events.KytosEvent`):
                Event with new switch information.

        """
        self._handle_lldp_flows(event)

    @listen_to("kytos/of_lldp.loop.action.log")
    def on_lldp_loop_log_action(self, event):
        """Handle LLDP loop log action."""
        interface_a = event.content["interface_a"]
        interface_b = event.content["interface_b"]
        self.loop_manager.handle_log_action(interface_a, interface_b)

    @listen_to("kytos/of_lldp.loop.action.disable")
    def on_lldp_loop_disable_action(self, event):
        """Handle LLDP loop disable action."""
        interface_a = event.content["interface_a"]
        interface_b = event.content["interface_b"]
        self.loop_manager.handle_disable_action(interface_a, interface_b)

    @listen_to("kytos/of_lldp.loop.detected")
    def on_lldp_loop_detected(self, event):
        """Handle LLDP loop detected."""
        interface_id = event.content["interface_id"]
        dpid = event.content["dpid"]
        port_pair = event.content["port_numbers"]
        self.loop_manager.handle_loop_detected(interface_id, dpid, port_pair)

    @listen_to("kytos/of_lldp.loop.stopped")
    def on_lldp_loop_stopped(self, event):
        """Handle LLDP loop stopped."""
        dpid = event.content["dpid"]
        port_pair = event.content["port_numbers"]
        try:
            switch = self.controller.get_switch_by_dpid(dpid)
            interface_a = switch.interfaces[port_pair[0]]
            interface_b = switch.interfaces[port_pair[1]]
            self.loop_manager.handle_loop_stopped(interface_a, interface_b)
        except (KeyError, AttributeError) as exc:
            log.error("on_lldp_loop_stopped failed with: "
                      f"{event.content} {str(exc)}")

    @listen_to("kytos/topology.topology_loaded")
    def on_topology_loaded(self, event):
        """Handle on topology loaded."""
        topology = event.content["topology"]
        self.loop_manager.handle_topology_loaded(topology)

    @listen_to("kytos/topology.switches.metadata.(added|removed)")
    def on_switches_metadata_changed(self, event):
        """Handle on switches metadata changed."""
        switch = event.content["switch"]
        self.loop_manager.handle_switch_metadata_changed(switch)

    def _handle_lldp_flows(self, event):
        """Install or remove flows in a switch.

        Install a flow to send LLDP packets to the controller. The proactive
        flow is installed whenever a switch is enabled. If the switch is
        disabled the flow is removed.
        """
        try:
            dpid = event.content['dpid']
            switch = self.controller.get_switch_by_dpid(dpid)
            of_version = switch.connection.protocol.version

        except AttributeError:
            of_version = None

        def _retry_if_status_code(response, endpoint, data, status_codes,
                                  retries=3, wait=2):
            """Retry if the response is in the status_codes."""
            if response.status_code not in status_codes:
                return
            if retries - 1 <= 0:
                return
            data = dict(data)
            data["force"] = True
            res = requests.post(endpoint, json=data)
            method = res.request.method
            if res.status_code != 202:
                log.error(f"Failed to retry on {endpoint}, error: {res.text},"
                          f" status: {res.status_code}, method: {method},"
                          f" data: {data}")
                time.sleep(wait)
                return _retry_if_status_code(response, endpoint, data,
                                             status_codes, retries - 1, wait)
            log.info(f"Successfully forced {method} flows to {endpoint}")

        flow = self._build_lldp_flow(of_version, get_cookie(switch.dpid))
        if flow:
            destination = switch.id
            endpoint = f'{settings.FLOW_MANAGER_URL}/flows/{destination}'
            data = {'flows': [flow]}
            if event.name == 'kytos/topology.switch.enabled':
                flow.pop("cookie_mask")
                res = requests.post(endpoint, json=data)
                if res.status_code != 202:
                    log.error(f"Failed to push flows on {destination},"
                              f" error: {res.text}, status: {res.status_code},"
                              f" data: {data}")
                _retry_if_status_code(res, endpoint, data, [424, 500])
            else:
                res = requests.delete(endpoint, json=data)
                if res.status_code != 202:
                    log.error(f"Failed to delete flows on {destination},"
                              f" error: {res.text}, status: {res.status_code}",
                              f" data: {data}")
                _retry_if_status_code(res, endpoint, data, [424, 500])

    @listen_to('kytos/of_core.v0x0[14].messages.in.ofpt_packet_in')
    def on_ofpt_packet_in(self, event):
        """Dispatch two KytosEvents to notify identified NNI interfaces.

        Args:
            event (:class:`~kytos.core.events.KytosEvent`):
                Event with an LLDP packet as data.

        """
        self.notify_uplink_detected(event)

    def notify_uplink_detected(self, event):
        """Dispatch two KytosEvents to notify identified NNI interfaces.

        Args:
            event (:class:`~kytos.core.events.KytosEvent`):
                Event with an LLDP packet as data.

        """
        ethernet = self._unpack_non_empty(Ethernet, event.message.data)
        if ethernet.ether_type == EtherType.LLDP:
            try:
                lldp = self._unpack_non_empty(LLDP, ethernet.data)
                dpid = self._unpack_non_empty(DPID, lldp.chassis_id.sub_value)
            except struct.error:
                #: If we have a LLDP packet but we cannot unpack it, or the
                #: unpacked packet does not contain the dpid attribute, then
                #: we are dealing with a LLDP generated by someone else. Thus
                #: this packet is not useful for us and we may just ignore it.
                return

            switch_a = event.source.switch
            port_a = event.message.in_port
            switch_b = None
            port_b = None

            # in_port is currently a UBInt16 in v0x01 and an Int in v0x04.
            if isinstance(port_a, int):
                port_a = UBInt32(port_a)

            try:
                switch_b = self.controller.get_switch_by_dpid(dpid.value)
                of_version = switch_b.connection.protocol.version
                port_type = UBInt16 if of_version == 0x01 else UBInt32
                port_b = self._unpack_non_empty(port_type,
                                                lldp.port_id.sub_value)
            except AttributeError:
                log.debug("Couldn't find datapath %s.", dpid.value)

            # Return if any of the needed information are not available
            if not (switch_a and port_a and switch_b and port_b):
                return

            interface_a = switch_a.get_interface_by_port_no(port_a.value)
            interface_b = switch_b.get_interface_by_port_no(port_b.value)

            self.loop_manager.process_if_looped(interface_a, interface_b)
            event_out = KytosEvent(name='kytos/of_lldp.interface.is.nni',
                                   content={'interface_a': interface_a,
                                            'interface_b': interface_b})
            self.controller.buffers.app.put(event_out)

    def notify_lldp_change(self, state, interface_ids):
        """Dispatch a KytosEvent to notify changes to the LLDP status."""
        content = {'attribute': 'LLDP',
                   'state': state,
                   'interface_ids': interface_ids}
        event_out = KytosEvent(name='kytos/of_lldp.network_status.updated',
                               content=content)
        self.controller.buffers.app.put(event_out)

    def shutdown(self):
        """End of the application."""
        log.debug('Shutting down...')

    @staticmethod
    def _build_lldp_packet_out(version, port_number, data):
        """Build a LLDP PacketOut message.

        Args:
            version (int): OpenFlow version
            port_number (int): Switch port number where the packet must be
                forwarded to.
            data (bytes): Binary data to be sent through the port.

        Returns:
            PacketOut message for the specific given OpenFlow version, if it
                is supported.
            None if the OpenFlow version is not supported.

        """
        if version == 0x01:
            action_output_class = AO10
            packet_out_class = PO10
        elif version == 0x04:
            action_output_class = AO13
            packet_out_class = PO13
        else:
            log.info('Openflow version %s is not yet supported.', version)
            return None

        output_action = action_output_class()
        output_action.port = port_number

        packet_out = packet_out_class()
        packet_out.data = data
        packet_out.actions.append(output_action)

        return packet_out

    def _build_lldp_flow(self, version, cookie,
                         cookie_mask=0xffffffffffffffff):
        """Build a Flow message to send LLDP to the controller.

        Args:
            version (int): OpenFlow version.

        Returns:
            Flow dictionary message for the specific given OpenFlow version,
            if it is supported.
            None if the OpenFlow version is not supported.

        """
        flow = {}
        match = {}
        flow['priority'] = settings.FLOW_PRIORITY
        flow['table_id'] = settings.TABLE_ID
        flow['cookie'] = cookie
        flow['cookie_mask'] = cookie_mask
        match['dl_type'] = EtherType.LLDP
        if self.vlan_id:
            match['dl_vlan'] = self.vlan_id
        flow['match'] = match

        if version == 0x01:
            flow['actions'] = [{'action_type': 'output',
                                'port': Port10.OFPP_CONTROLLER}]
        elif version == 0x04:
            flow['actions'] = [{'action_type': 'output',
                                'port': Port13.OFPP_CONTROLLER}]
        else:
            flow = None

        return flow

    @staticmethod
    def _unpack_non_empty(desired_class, data):
        """Unpack data using an instance of desired_class.

        Args:
            desired_class (class): The class to be used to unpack data.
            data (bytes): bytes to be unpacked.

        Return:
            An instance of desired_class class with data unpacked into it.

        Raises:
            UnpackException if the unpack could not be performed.

        """
        obj = desired_class()

        if hasattr(data, 'value'):
            data = data.value

        obj.unpack(data)

        return obj

    @staticmethod
    def _get_data(req):
        """Get request data."""
        data = req.get_json()  # Valid format { "interfaces": [...] }
        return data.get('interfaces', [])

    def _get_interfaces(self):
        """Get all interfaces."""
        interfaces = []
        for switch in list(self.controller.switches.values()):
            interfaces += list(switch.interfaces.values())
        return interfaces

    @staticmethod
    def _get_interfaces_dict(interfaces):
        """Return a dict of interfaces."""
        return {inter.id: inter for inter in interfaces}

    def _get_lldp_interfaces(self):
        """Get interfaces enabled to receive LLDP packets."""
        return [inter.id for inter in self._get_interfaces() if inter.lldp]

    @rest('v1/interfaces', methods=['GET'])
    def get_lldp_interfaces(self):
        """Return all the interfaces that have LLDP traffic enabled."""
        return jsonify({"interfaces": self._get_lldp_interfaces()}), 200

    @rest('v1/interfaces/disable', methods=['POST'])

    def disable_lldp(self):
        """Disables an interface to receive LLDP packets."""
        interface_ids = self._get_data(request)
        error_list = []  # List of interfaces that were not activated.
        changed_interfaces = []
        interface_ids = filter(None, interface_ids)
        interfaces = self._get_interfaces()
        if not interfaces:
            return jsonify("No interfaces were found."), 404
        interfaces = self._get_interfaces_dict(interfaces)
        for id_ in interface_ids:
            interface = interfaces.get(id_)
            if interface:
                interface.lldp = False
                changed_interfaces.append(id_)
            else:
                error_list.append(id_)
        if changed_interfaces:
            self.notify_lldp_change('disabled', changed_interfaces)
        if not error_list:
            return jsonify(
                "All the requested interfaces have been disabled."), 200

        # Return a list of interfaces that couldn't be disabled
        msg_error = "Some interfaces couldn't be found and deactivated: "
        return jsonify({msg_error:
                        error_list}), 400

    @rest('v1/interfaces/enable', methods=['POST'])

    def enable_lldp(self):
        """Enable an interface to receive LLDP packets."""
        interface_ids = self._get_data(request)
        error_list = []  # List of interfaces that were not activated.
        changed_interfaces = []
        interface_ids = filter(None, interface_ids)
        interfaces = self._get_interfaces()
        if not interfaces:
            return jsonify("No interfaces were found."), 404
        interfaces = self._get_interfaces_dict(interfaces)
        for id_ in interface_ids:
            interface = interfaces.get(id_)
            if interface:
                interface.lldp = True
                changed_interfaces.append(id_)
            else:
                error_list.append(id_)
        if changed_interfaces:
            self.notify_lldp_change('enabled', changed_interfaces)
        if not error_list:
            return jsonify(
                "All the requested interfaces have been enabled."), 200

        # Return a list of interfaces that couldn't be enabled
        msg_error = "Some interfaces couldn't be found and activated: "
        return jsonify({msg_error:
                        error_list}), 400

    @rest('v1/polling_time', methods=['GET'])
    def get_time(self):
        """Get LLDP polling time in seconds."""
        return jsonify({"polling_time": self.polling_time}), 200

    @rest('v1/polling_time', methods=['POST'])
    def set_time(self):
        """Set LLDP polling time."""
        # pylint: disable=attribute-defined-outside-init
        try:
            payload = request.get_json()
            polling_time = int(payload['polling_time'])
            if polling_time <= 0:
                raise ValueError(f"invalid polling_time {polling_time}, "
                                 "must be greater than zero")
            self.polling_time = polling_time
            self.execute_as_loop(self.polling_time)
            log.info("Polling time has been updated to %s"
                     " second(s), but this change will not be saved"
                     " permanently.", self.polling_time)
            return jsonify("Polling time has been updated."), 200
        except (ValueError, KeyError) as error:
            msg = f"This operation is not completed: {error}"
            return jsonify(msg), 400


1			"""NApp responsible to discover new switches and hosts."""
2	1		import struct
3	1		import time
4
5	1		import requests
6	1		from flask import jsonify, request
7	1		from pyof.foundation.basic_types import DPID, UBInt16, UBInt32
8	1		from pyof.foundation.network_types import LLDP, VLAN, Ethernet, EtherType
9	1		from pyof.v0x01.common.action import ActionOutput as AO10
10	1		from pyof.v0x01.common.phy_port import Port as Port10
11	1		from pyof.v0x01.controller2switch.packet_out import PacketOut as PO10
12	1		from pyof.v0x04.common.action import ActionOutput as AO13
13	1		from pyof.v0x04.common.port import PortNo as Port13
14	1		from pyof.v0x04.controller2switch.packet_out import PacketOut as PO13
15
16	1		from kytos.core import KytosEvent, KytosNApp, log, rest
17	1		from kytos.core.helpers import listen_to
18	1		from napps.kytos.of_core.msg_prios import of_msg_prio
19	1		from napps.kytos.of_lldp import constants, settings
20	1		from napps.kytos.of_lldp.loop_manager import LoopManager, LoopState
21	1		from napps.kytos.of_lldp.utils import get_cookie
22
23
24	1		class Main(KytosNApp):
25			"""Main OF_LLDP NApp Class."""
26
27	1		def setup(self):
28			"""Make this NApp run in a loop."""
29	1		self.vlan_id = None
30	1		self.polling_time = settings.POLLING_TIME
31	1		if hasattr(settings, "FLOW_VLAN_VID"):
32	1		self.vlan_id = settings.FLOW_VLAN_VID
33	1		self.execute_as_loop(self.polling_time)
34	1		self.loop_manager = LoopManager(self.controller)
35
36	1		def execute(self):
37			"""Send LLDP Packets every 'POLLING_TIME' seconds to all switches."""
38	1		switches = list(self.controller.switches.values())
39	1		for switch in switches:
40	1		try:
41	1		of_version = switch.connection.protocol.version
42			except AttributeError:
43			of_version = None
44
45	1		if not switch.is_connected():
46			continue
47
48	1		if of_version == 0x01:
49	1		port_type = UBInt16
50	1		local_port = Port10.OFPP_LOCAL
51	1		elif of_version == 0x04:
52	1		port_type = UBInt32
53	1		local_port = Port13.OFPP_LOCAL
54			else:
55			# skip the current switch with unsupported OF version
56			continue
57
58	1		interfaces = list(switch.interfaces.values())
59	1		for interface in interfaces:
60			# Interface marked to receive lldp packet
61			# Only send LLDP packet to active interface
62	1		if(not interface.lldp or not interface.is_active()
63			or not interface.is_enabled()):
64			continue
65			# Avoid the interface that connects to the controller.
66	1		if interface.port_number == local_port:
67			continue
68
69	1		lldp = LLDP()
70	1		lldp.chassis_id.sub_value = DPID(switch.dpid)
71	1		lldp.port_id.sub_value = port_type(interface.port_number)
72
73	1		ethernet = Ethernet()
74	1		ethernet.ether_type = EtherType.LLDP
75	1		ethernet.source = interface.address
76	1		ethernet.destination = constants.LLDP_MULTICAST_MAC
77	1		ethernet.data = lldp.pack()
78			# self.vlan_id == None will result in a packet with no VLAN.
79	1		ethernet.vlans.append(VLAN(vid=self.vlan_id))
80
81	1		packet_out = self._build_lldp_packet_out(
82			of_version,
83			interface.port_number, ethernet.pack())
84
85	1		if packet_out is None:
86			continue
87
88	1		event_out = KytosEvent(
89			name='kytos/of_lldp.messages.out.ofpt_packet_out',
90			priority=of_msg_prio(packet_out.header.message_type.value),
91			content={
92			'destination': switch.connection,
93			'message': packet_out})
94
95	1		self.controller.buffers.msg_out.put(event_out)
96	1		log.debug(
97			"Sending a LLDP PacketOut to the switch %s",
98			switch.dpid)
99
100	1		msg = 'Switch: %s (%s)'
101	1		msg += ' Interface: %s'
102	1		msg += ' -- LLDP PacketOut --'
103	1		msg += ' Ethernet: eth_type (%s) \| src (%s) \| dst (%s) /'
104	1		msg += ' LLDP: Switch (%s) \| portno (%s)'
105
106	1		log.debug(
107			msg,
108			switch.connection, switch.dpid,
109			interface.id, ethernet.ether_type,
110			ethernet.source, ethernet.destination,
111			switch.dpid, interface.port_number)
112
113	1		self.try_to_publish_stopped_loops()
114
115	1		def try_to_publish_stopped_loops(self):
116			"""Try to publish current stopped loops."""
117			for dpid, port_pairs in self.loop_manager.get_stopped_loops().items():
118			try:
119			switch = self.controller.get_switch_by_dpid(dpid)
120			for port_pair in port_pairs:
121			interface_a = switch.interfaces[port_pair[0]]
122			interface_b = switch.interfaces[port_pair[1]]
123			self.loop_manager.publish_loop_state(
124			interface_a, interface_b, LoopState.stopped.value
125			)
126			except (KeyError, AttributeError) as exc:
127			log.error("try_to_publish_stopped_loops failed with switch:"
128			f"{dpid}, port_pair: {port_pair}. {str(exc)}")
129
130	1		@listen_to('kytos/topology.switch.(enabled\|disabled)')
131	1		def handle_lldp_flows(self, event):
132			"""Install or remove flows in a switch.
133
134			Install a flow to send LLDP packets to the controller. The proactive
135			flow is installed whenever a switch is enabled. If the switch is
136			disabled the flow is removed.
137
138			Args:
139			event (:class:`~kytos.core.events.KytosEvent`):
140			Event with new switch information.
141
142			"""
143	1		self._handle_lldp_flows(event)
144
145	1		@listen_to("kytos/of_lldp.loop.action.log")
146	1		def on_lldp_loop_log_action(self, event):
147			"""Handle LLDP loop log action."""
148			interface_a = event.content["interface_a"]
149			interface_b = event.content["interface_b"]
150			self.loop_manager.handle_log_action(interface_a, interface_b)
151
152	1		@listen_to("kytos/of_lldp.loop.action.disable")
153	1		def on_lldp_loop_disable_action(self, event):
154			"""Handle LLDP loop disable action."""
155			interface_a = event.content["interface_a"]
156			interface_b = event.content["interface_b"]
157			self.loop_manager.handle_disable_action(interface_a, interface_b)
158
159	1		@listen_to("kytos/of_lldp.loop.detected")
160	1		def on_lldp_loop_detected(self, event):
161			"""Handle LLDP loop detected."""
162			interface_id = event.content["interface_id"]
163			dpid = event.content["dpid"]
164			port_pair = event.content["port_numbers"]
165			self.loop_manager.handle_loop_detected(interface_id, dpid, port_pair)
166
167	1		@listen_to("kytos/of_lldp.loop.stopped")
168	1		def on_lldp_loop_stopped(self, event):
169			"""Handle LLDP loop stopped."""
170			dpid = event.content["dpid"]
171			port_pair = event.content["port_numbers"]
172			try:
173			switch = self.controller.get_switch_by_dpid(dpid)
174			interface_a = switch.interfaces[port_pair[0]]
175			interface_b = switch.interfaces[port_pair[1]]
176			self.loop_manager.handle_loop_stopped(interface_a, interface_b)
177			except (KeyError, AttributeError) as exc:
178			log.error("on_lldp_loop_stopped failed with: "
179			f"{event.content} {str(exc)}")
180
181	1		@listen_to("kytos/topology.topology_loaded")
182	1		def on_topology_loaded(self, event):
183			"""Handle on topology loaded."""
184			topology = event.content["topology"]
185			self.loop_manager.handle_topology_loaded(topology)
186
187	1		@listen_to("kytos/topology.switches.metadata.(added\|removed)")
188	1		def on_switches_metadata_changed(self, event):
189			"""Handle on switches metadata changed."""
190			switch = event.content["switch"]
191			self.loop_manager.handle_switch_metadata_changed(switch)
192
193	1		def _handle_lldp_flows(self, event):
194			"""Install or remove flows in a switch.
195
196			Install a flow to send LLDP packets to the controller. The proactive
197			flow is installed whenever a switch is enabled. If the switch is
198			disabled the flow is removed.
199			"""
200	1		try:
201	1		dpid = event.content['dpid']
202	1		switch = self.controller.get_switch_by_dpid(dpid)
203	1		of_version = switch.connection.protocol.version
204
205			except AttributeError:
206			of_version = None
207
208	1		def _retry_if_status_code(response, endpoint, data, status_codes,
209			retries=3, wait=2):
210			"""Retry if the response is in the status_codes."""
211	1		if response.status_code not in status_codes:
212	1		return
213	1		if retries - 1 <= 0:
214	1		return
215	1		data = dict(data)
216	1		data["force"] = True
217	1		res = requests.post(endpoint, json=data)
218	1		method = res.request.method
219	1		if res.status_code != 202:
220	1		log.error(f"Failed to retry on {endpoint}, error: {res.text},"
221			f" status: {res.status_code}, method: {method},"
222			f" data: {data}")
223	1		time.sleep(wait)
224	1		return _retry_if_status_code(response, endpoint, data,
225			status_codes, retries - 1, wait)
226			log.info(f"Successfully forced {method} flows to {endpoint}")
227
228	1		flow = self._build_lldp_flow(of_version, get_cookie(switch.dpid))
229	1		if flow:
230	1		destination = switch.id
231	1		endpoint = f'{settings.FLOW_MANAGER_URL}/flows/{destination}'
232	1		data = {'flows': [flow]}
233	1		if event.name == 'kytos/topology.switch.enabled':
234	1		flow.pop("cookie_mask")
235	1		res = requests.post(endpoint, json=data)
236	1		if res.status_code != 202:
237	1		log.error(f"Failed to push flows on {destination},"
238			f" error: {res.text}, status: {res.status_code},"
239			f" data: {data}")
240	1		_retry_if_status_code(res, endpoint, data, [424, 500])
241			else:
242	1		res = requests.delete(endpoint, json=data)
243	1		if res.status_code != 202:
244	1		log.error(f"Failed to delete flows on {destination},"
245			f" error: {res.text}, status: {res.status_code}",
246			f" data: {data}")
247			_retry_if_status_code(res, endpoint, data, [424, 500])
248
249	1		@listen_to('kytos/of_core.v0x0[14].messages.in.ofpt_packet_in')
250	1		def on_ofpt_packet_in(self, event):
251			"""Dispatch two KytosEvents to notify identified NNI interfaces.
252
253			Args:
254			event (:class:`~kytos.core.events.KytosEvent`):
255			Event with an LLDP packet as data.
256
257			"""
258			self.notify_uplink_detected(event)
259
260	1		def notify_uplink_detected(self, event):
261			"""Dispatch two KytosEvents to notify identified NNI interfaces.
262
263			Args:
264			event (:class:`~kytos.core.events.KytosEvent`):
265			Event with an LLDP packet as data.
266
267			"""
268	1		ethernet = self._unpack_non_empty(Ethernet, event.message.data)
269	1		if ethernet.ether_type == EtherType.LLDP:
270	1		try:
271	1		lldp = self._unpack_non_empty(LLDP, ethernet.data)
272	1		dpid = self._unpack_non_empty(DPID, lldp.chassis_id.sub_value)
273			except struct.error:
274			#: If we have a LLDP packet but we cannot unpack it, or the
275			#: unpacked packet does not contain the dpid attribute, then
276			#: we are dealing with a LLDP generated by someone else. Thus
277			#: this packet is not useful for us and we may just ignore it.
278			return
279
280	1		switch_a = event.source.switch
281	1		port_a = event.message.in_port
282	1		switch_b = None
283	1		port_b = None
284
285			# in_port is currently a UBInt16 in v0x01 and an Int in v0x04.
286	1		if isinstance(port_a, int):
287	1		port_a = UBInt32(port_a)
288
289	1		try:
290	1		switch_b = self.controller.get_switch_by_dpid(dpid.value)
291	1		of_version = switch_b.connection.protocol.version
292	1		port_type = UBInt16 if of_version == 0x01 else UBInt32
293	1		port_b = self._unpack_non_empty(port_type,
294			lldp.port_id.sub_value)
295			except AttributeError:
296			log.debug("Couldn't find datapath %s.", dpid.value)
297
298			# Return if any of the needed information are not available
299	1		if not (switch_a and port_a and switch_b and port_b):
300			return
301
302	1		interface_a = switch_a.get_interface_by_port_no(port_a.value)
303	1		interface_b = switch_b.get_interface_by_port_no(port_b.value)
304
305	1		self.loop_manager.process_if_looped(interface_a, interface_b)
306	1		event_out = KytosEvent(name='kytos/of_lldp.interface.is.nni',
307			content={'interface_a': interface_a,
308			'interface_b': interface_b})
309	1		self.controller.buffers.app.put(event_out)
310
311	1		def notify_lldp_change(self, state, interface_ids):
312			"""Dispatch a KytosEvent to notify changes to the LLDP status."""
313	1		content = {'attribute': 'LLDP',
314			'state': state,
315			'interface_ids': interface_ids}
316	1		event_out = KytosEvent(name='kytos/of_lldp.network_status.updated',
317			content=content)
318	1		self.controller.buffers.app.put(event_out)
319
320	1		def shutdown(self):
321			"""End of the application."""
322			log.debug('Shutting down...')
323
324	1		@staticmethod
325	1		def _build_lldp_packet_out(version, port_number, data):
326			"""Build a LLDP PacketOut message.
327
328			Args:
329			version (int): OpenFlow version
330			port_number (int): Switch port number where the packet must be
331			forwarded to.
332			data (bytes): Binary data to be sent through the port.
333
334			Returns:
335			PacketOut message for the specific given OpenFlow version, if it
336			is supported.
337			None if the OpenFlow version is not supported.
338
339			"""
340	1		if version == 0x01:
341	1		action_output_class = AO10
342	1		packet_out_class = PO10
343	1		elif version == 0x04:
344	1		action_output_class = AO13
345	1		packet_out_class = PO13
346			else:
347	1		log.info('Openflow version %s is not yet supported.', version)
348	1		return None
349
350	1		output_action = action_output_class()
351	1		output_action.port = port_number
352
353	1		packet_out = packet_out_class()
354	1		packet_out.data = data
355	1		packet_out.actions.append(output_action)
356
357	1		return packet_out
358
359	1		def _build_lldp_flow(self, version, cookie,
360			cookie_mask=0xffffffffffffffff):
361			"""Build a Flow message to send LLDP to the controller.
362
363			Args:
364			version (int): OpenFlow version.
365
366			Returns:
367			Flow dictionary message for the specific given OpenFlow version,
368			if it is supported.
369			None if the OpenFlow version is not supported.
370
371			"""
372	1		flow = {}
373	1		match = {}
374	1		flow['priority'] = settings.FLOW_PRIORITY
375	1		flow['table_id'] = settings.TABLE_ID
376	1		flow['cookie'] = cookie
377	1		flow['cookie_mask'] = cookie_mask
378	1		match['dl_type'] = EtherType.LLDP
379	1		if self.vlan_id:
380	1		match['dl_vlan'] = self.vlan_id
381	1		flow['match'] = match
382
383	1		if version == 0x01:
384	1		flow['actions'] = [{'action_type': 'output',
385			'port': Port10.OFPP_CONTROLLER}]
386	1		elif version == 0x04:
387	1		flow['actions'] = [{'action_type': 'output',
388			'port': Port13.OFPP_CONTROLLER}]
389			else:
390			flow = None
391
392	1		return flow
393
394	1		@staticmethod
395	1		def _unpack_non_empty(desired_class, data):
396			"""Unpack data using an instance of desired_class.
397
398			Args:
399			desired_class (class): The class to be used to unpack data.
400			data (bytes): bytes to be unpacked.
401
402			Return:
403			An instance of desired_class class with data unpacked into it.
404
405			Raises:
406			UnpackException if the unpack could not be performed.
407
408			"""
409	1		obj = desired_class()
410
411	1		if hasattr(data, 'value'):
412	1		data = data.value
413
414	1		obj.unpack(data)
415
416	1		return obj
417
418	1		@staticmethod
419	1		def _get_data(req):
420			"""Get request data."""
421	1		data = req.get_json() # Valid format { "interfaces": [...] }
422	1		return data.get('interfaces', [])
423
424	1		def _get_interfaces(self):
425			"""Get all interfaces."""
426	1		interfaces = []
427	1		for switch in list(self.controller.switches.values()):
428	1		interfaces += list(switch.interfaces.values())
429	1		return interfaces
430
431	1		@staticmethod
432	1		def _get_interfaces_dict(interfaces):
433			"""Return a dict of interfaces."""
434	1		return {inter.id: inter for inter in interfaces}
435
436	1		def _get_lldp_interfaces(self):
437			"""Get interfaces enabled to receive LLDP packets."""
438	1		return [inter.id for inter in self._get_interfaces() if inter.lldp]
439
440	1		@rest('v1/interfaces', methods=['GET'])
441	1		def get_lldp_interfaces(self):
442			"""Return all the interfaces that have LLDP traffic enabled."""
443	1		return jsonify({"interfaces": self._get_lldp_interfaces()}), 200
444
445	1	View Code Duplication	@rest('v1/interfaces/disable', methods=['POST'])
			0 ignored issues – show Duplication introduced 2021-11-30 20:15 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
446	1		def disable_lldp(self):
447			"""Disables an interface to receive LLDP packets."""
448	1		interface_ids = self._get_data(request)
449	1		error_list = [] # List of interfaces that were not activated.
450	1		changed_interfaces = []
451	1		interface_ids = filter(None, interface_ids)
452	1		interfaces = self._get_interfaces()
453	1		if not interfaces:
454	1		return jsonify("No interfaces were found."), 404
455	1		interfaces = self._get_interfaces_dict(interfaces)
456	1		for id_ in interface_ids:
457	1		interface = interfaces.get(id_)
458	1		if interface:
459	1		interface.lldp = False
460	1		changed_interfaces.append(id_)
461			else:
462	1		error_list.append(id_)
463	1		if changed_interfaces:
464	1		self.notify_lldp_change('disabled', changed_interfaces)
465	1		if not error_list:
466	1		return jsonify(
467			"All the requested interfaces have been disabled."), 200
468
469			# Return a list of interfaces that couldn't be disabled
470	1		msg_error = "Some interfaces couldn't be found and deactivated: "
471	1		return jsonify({msg_error:
472			error_list}), 400
473
474	1	View Code Duplication	@rest('v1/interfaces/enable', methods=['POST'])
			0 ignored issues – show Duplication introduced 2021-11-30 20:15 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
475	1		def enable_lldp(self):
476			"""Enable an interface to receive LLDP packets."""
477	1		interface_ids = self._get_data(request)
478	1		error_list = [] # List of interfaces that were not activated.
479	1		changed_interfaces = []
480	1		interface_ids = filter(None, interface_ids)
481	1		interfaces = self._get_interfaces()
482	1		if not interfaces:
483	1		return jsonify("No interfaces were found."), 404
484	1		interfaces = self._get_interfaces_dict(interfaces)
485	1		for id_ in interface_ids:
486	1		interface = interfaces.get(id_)
487	1		if interface:
488	1		interface.lldp = True
489	1		changed_interfaces.append(id_)
490			else:
491	1		error_list.append(id_)
492	1		if changed_interfaces:
493	1		self.notify_lldp_change('enabled', changed_interfaces)
494	1		if not error_list:
495	1		return jsonify(
496			"All the requested interfaces have been enabled."), 200
497
498			# Return a list of interfaces that couldn't be enabled
499	1		msg_error = "Some interfaces couldn't be found and activated: "
500	1		return jsonify({msg_error:
501			error_list}), 400
502
503	1		@rest('v1/polling_time', methods=['GET'])
504	1		def get_time(self):
505			"""Get LLDP polling time in seconds."""
506	1		return jsonify({"polling_time": self.polling_time}), 200
507
508	1		@rest('v1/polling_time', methods=['POST'])
509	1		def set_time(self):
510			"""Set LLDP polling time."""
511			# pylint: disable=attribute-defined-outside-init
512	1		try:
513	1		payload = request.get_json()
514	1		polling_time = int(payload['polling_time'])
515	1		if polling_time <= 0:
516			raise ValueError(f"invalid polling_time {polling_time}, "
517			"must be greater than zero")
518	1		self.polling_time = polling_time
519	1		self.execute_as_loop(self.polling_time)
520	1		log.info("Polling time has been updated to %s"
521			" second(s), but this change will not be saved"
522			" permanently.", self.polling_time)
523	1		return jsonify("Polling time has been updated."), 200
524	1		except (ValueError, KeyError) as error:
525	1		msg = f"This operation is not completed: {error}"
526			return jsonify(msg), 400
527

kytos-ng / of_lldp

Push — master ( de6073...127cfb )

build.main.Main._get_data() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like