build.main.Main.lldp_loop_handler() - Code Metrics - Inspection of "[Feature] Loop detection in the same switch via LL..." - kytos-ng/of_lldp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#36)

by Vinicius

created 2022-01-09 23:24 UTC

build.main.Main.lldp_loop_handler() B

↳ Parent: build.main

Complexity

Conditions

Size

Total Lines	25
Code Lines	21

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	12
CRAP Score	5.2

Importance

Changes

Metric	Value
cc	5
eloc	21
nop	5
dl	0
loc	25
rs	8.9093
c	0
b	0
f	0
ccs	12
cts	15
cp	0.8
crap	5.2

"""NApp responsible to discover new switches and hosts."""
import struct
import time
from collections import defaultdict
from threading import Lock

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_lldp import constants, settings


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
        self.ignored_loops = settings.LLDP_IGNORED_LOOPS
        self.counter_lock = Lock()
        self.loop_counter = defaultdict(dict)
        if hasattr(settings, "FLOW_VLAN_VID"):
            self.vlan_id = settings.FLOW_VLAN_VID
        self.execute_as_loop(self.polling_time)

    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',
                    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)

    @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)

    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)
        if flow:
            destination = switch.id
            endpoint = f'{settings.FLOW_MANAGER_URL}/flows/{destination}'
            data = {'flows': [flow]}
            if event.name == 'kytos/topology.switch.enabled':
                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])

    def _is_loop_ignored(self, dpid, port_a, port_b):
        """Check if a loop is ignored."""
        if dpid not in self.ignored_loops:
            return False
        if any(
            (
                (port_a, port_b) in self.ignored_loops[dpid],
                (port_b, port_a) in self.ignored_loops[dpid],
            )
        ):
            return True
        return False

    @staticmethod
    def _is_lldp_looped(dpid_a, port_a, dpid_b, port_b):
        """Check if LLDP is looped."""
        if all(
            (
                dpid_a == dpid_b,
                port_a <= port_b  # only enter one pair
            )
        ):
            return True
        return False

    def process_if_lldp_looped(
        self,
        switch_a,
        interface_a,
        switch_b,
        interface_b,
        action=settings.LLDP_LOOP_ACTION,
    ):
        """Process if received LLDP data is looped."""
        dpid_a = switch_a.dpid
        dpid_b = switch_b.dpid
        port_a = interface_a.port_number
        port_b = interface_b.port_number
        if all(
            (
                self._is_lldp_looped(dpid_a, port_a, dpid_b, port_b),
                not self._is_loop_ignored(dpid_a, port_a, port_b),
            )
        ):
            self.lldp_loop_handler(switch_a, interface_a, interface_b, action)

    def lldp_loop_handler(
        self,
        switch,
        interface_a,
        interface_b,
        action=settings.LLDP_LOOP_ACTION,
    ):
        """Handler to execute when a LLDP loop is found."""
        if action == "log":
            port_a = interface_a.port_number
            port_b = interface_b.port_number
            port_pair = (port_a, port_b)
            log_every = settings.LOOP_LOG_EVERY
            with self.counter_lock:
                if port_pair not in self.loop_counter[switch.id]:
                    self.loop_counter[switch.id][port_pair] = 0
                else:
                    self.loop_counter[switch.id][port_pair] += 1
                    self.loop_counter[switch.id][port_pair] %= log_every
                count = self.loop_counter[switch.id][port_pair]
                if count % settings.LOOP_LOG_EVERY != 0:
                    return

            log.warning(
                f"LLDP loop detected on switch: {switch.id}, "
                f"interface_a: {interface_a.name}, number: {port_a}, "
                f"interface_b: {interface_b.name}, number: {port_b}"
            )

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

kytos-ng / of_lldp

Pull Request — master (#36)

build.main.Main.lldp_loop_handler() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like