build.main.Main.lldp_loop_handler() - Code Metrics - Inspection of "[Feature] Loop detection via LLDP `ofpt_packet_in`" - kytos-ng/of_lldp - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#36)

by Vinicius

created 2022-01-07 22:32 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])

    @staticmethod
    def _is_port_looped(dpid_a, port_a, dpid_b, port_b):
        """Check if tuple (dpid_a, port_a, dpid_b, port_b) is looped."""
        if dpid_a == dpid_b and port_a != port_b:
            return True
        return False

    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

    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)

            dpid_a = switch_a.dpid
            dpid_b = switch_b.dpid
            port_a = port_a.value
            port_b = port_b.value
            if all(
                (
                    self._is_port_looped(dpid_a, port_a, dpid_b, port_b),
                    not self._is_loop_ignored(dpid_a, port_a, port_b),
                    port_a < port_b  # only enter one pair
                )
            ):
                self.lldp_loop_handler(switch_a, 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):
        """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		@staticmethod
183			def _is_port_looped(dpid_a, port_a, dpid_b, port_b):
184			"""Check if tuple (dpid_a, port_a, dpid_b, port_b) is looped."""
185	1		if dpid_a == dpid_b and port_a != port_b:
186	1		return True
187	1		return False
188
189	1		def _is_loop_ignored(self, dpid, port_a, port_b):
190			"""Check if a loop is ignored."""
191	1		if dpid not in self.ignored_loops:
192	1		return False
193	1		if any(
194			(
195			(port_a, port_b) in self.ignored_loops[dpid],
196			(port_b, port_a) in self.ignored_loops[dpid],
197			)
198			):
199	1		return True
200	1		return False
201
202	1		def lldp_loop_handler(
203			self,
204			switch,
205			interface_a,
206			interface_b,
207			action=settings.LLDP_LOOP_ACTION,
208			):
209			"""Handler to execute when a LLDP loop is found."""
210	1		if action == "log":
211	1		port_a = interface_a.port_number
212	1		port_b = interface_b.port_number
213	1		port_pair = (port_a, port_b)
214	1		log_every = settings.LOOP_LOG_EVERY
215	1		with self.counter_lock:
216	1		if port_pair not in self.loop_counter[switch.id]:
217	1		self.loop_counter[switch.id][port_pair] = 0
218			else:
219			self.loop_counter[switch.id][port_pair] += 1
220			self.loop_counter[switch.id][port_pair] %= log_every
221	1		count = self.loop_counter[switch.id][port_pair]
222	1		if count % settings.LOOP_LOG_EVERY != 0:
223			return
224
225	1		log.warning(
226			f"LLDP loop detected on switch: {switch.id}, "
227			f"interface_a: {interface_a.name}, number: {port_a}, "
228			f"interface_b: {interface_b.name}, number: {port_b}"
229			)
230
231	1		@listen_to('kytos/of_core.v0x0[14].messages.in.ofpt_packet_in')
232			def on_ofpt_packet_in(self, event):
233			"""Dispatch two KytosEvents to notify identified NNI interfaces.
234
235			Args:
236			event (:class:`~kytos.core.events.KytosEvent`):
237			Event with an LLDP packet as data.
238
239			"""
240			self.notify_uplink_detected(event)
241
242	1		def notify_uplink_detected(self, event):
243			"""Dispatch two KytosEvents to notify identified NNI interfaces.
244
245			Args:
246			event (:class:`~kytos.core.events.KytosEvent`):
247			Event with an LLDP packet as data.
248
249			"""
250	1		ethernet = self._unpack_non_empty(Ethernet, event.message.data)
251	1		if ethernet.ether_type == EtherType.LLDP:
252	1		try:
253	1		lldp = self._unpack_non_empty(LLDP, ethernet.data)
254	1		dpid = self._unpack_non_empty(DPID, lldp.chassis_id.sub_value)
255			except struct.error:
256			#: If we have a LLDP packet but we cannot unpack it, or the
257			#: unpacked packet does not contain the dpid attribute, then
258			#: we are dealing with a LLDP generated by someone else. Thus
259			#: this packet is not useful for us and we may just ignore it.
260			return
261
262	1		switch_a = event.source.switch
263	1		port_a = event.message.in_port
264	1		switch_b = None
265	1		port_b = None
266
267			# in_port is currently a UBInt16 in v0x01 and an Int in v0x04.
268	1		if isinstance(port_a, int):
269	1		port_a = UBInt32(port_a)
270
271	1		try:
272	1		switch_b = self.controller.get_switch_by_dpid(dpid.value)
273	1		of_version = switch_b.connection.protocol.version
274	1		port_type = UBInt16 if of_version == 0x01 else UBInt32
275	1		port_b = self._unpack_non_empty(port_type,
276			lldp.port_id.sub_value)
277			except AttributeError:
278			log.debug("Couldn't find datapath %s.", dpid.value)
279
280			# Return if any of the needed information are not available
281	1		if not (switch_a and port_a and switch_b and port_b):
282			return
283
284	1		interface_a = switch_a.get_interface_by_port_no(port_a.value)
285	1		interface_b = switch_b.get_interface_by_port_no(port_b.value)
286
287	1		dpid_a = switch_a.dpid
288	1		dpid_b = switch_b.dpid
289	1		port_a = port_a.value
290	1		port_b = port_b.value
291	1		if all(
292			(
293			self._is_port_looped(dpid_a, port_a, dpid_b, port_b),
294			not self._is_loop_ignored(dpid_a, port_a, port_b),
295			port_a < port_b # only enter one pair
296			)
297			):
298			self.lldp_loop_handler(switch_a, interface_a, interface_b)
299
300	1		event_out = KytosEvent(name='kytos/of_lldp.interface.is.nni',
301			content={'interface_a': interface_a,
302			'interface_b': interface_b})
303	1		self.controller.buffers.app.put(event_out)
304
305	1		def notify_lldp_change(self, state, interface_ids):
306			"""Dispatch a KytosEvent to notify changes to the LLDP status."""
307	1		content = {'attribute': 'LLDP',
308			'state': state,
309			'interface_ids': interface_ids}
310	1		event_out = KytosEvent(name='kytos/of_lldp.network_status.updated',
311			content=content)
312	1		self.controller.buffers.app.put(event_out)
313
314	1		def shutdown(self):
315			"""End of the application."""
316			log.debug('Shutting down...')
317
318	1		@staticmethod
319			def _build_lldp_packet_out(version, port_number, data):
320			"""Build a LLDP PacketOut message.
321
322			Args:
323			version (int): OpenFlow version
324			port_number (int): Switch port number where the packet must be
325			forwarded to.
326			data (bytes): Binary data to be sent through the port.
327
328			Returns:
329			PacketOut message for the specific given OpenFlow version, if it
330			is supported.
331			None if the OpenFlow version is not supported.
332
333			"""
334	1		if version == 0x01:
335	1		action_output_class = AO10
336	1		packet_out_class = PO10
337	1		elif version == 0x04:
338	1		action_output_class = AO13
339	1		packet_out_class = PO13
340			else:
341	1		log.info('Openflow version %s is not yet supported.', version)
342	1		return None
343
344	1		output_action = action_output_class()
345	1		output_action.port = port_number
346
347	1		packet_out = packet_out_class()
348	1		packet_out.data = data
349	1		packet_out.actions.append(output_action)
350
351	1		return packet_out
352
353	1		def _build_lldp_flow(self, version):
354			"""Build a Flow message to send LLDP to the controller.
355
356			Args:
357			version (int): OpenFlow version.
358
359			Returns:
360			Flow dictionary message for the specific given OpenFlow version,
361			if it is supported.
362			None if the OpenFlow version is not supported.
363
364			"""
365	1		flow = {}
366	1		match = {}
367	1		flow['priority'] = settings.FLOW_PRIORITY
368	1		flow['table_id'] = settings.TABLE_ID
369	1		match['dl_type'] = EtherType.LLDP
370	1		if self.vlan_id:
371	1		match['dl_vlan'] = self.vlan_id
372	1		flow['match'] = match
373
374	1		if version == 0x01:
375	1		flow['actions'] = [{'action_type': 'output',
376			'port': Port10.OFPP_CONTROLLER}]
377	1		elif version == 0x04:
378	1		flow['actions'] = [{'action_type': 'output',
379			'port': Port13.OFPP_CONTROLLER}]
380			else:
381			flow = None
382
383	1		return flow
384
385	1		@staticmethod
386			def _unpack_non_empty(desired_class, data):
387			"""Unpack data using an instance of desired_class.
388
389			Args:
390			desired_class (class): The class to be used to unpack data.
391			data (bytes): bytes to be unpacked.
392
393			Return:
394			An instance of desired_class class with data unpacked into it.
395
396			Raises:
397			UnpackException if the unpack could not be performed.
398
399			"""
400	1		obj = desired_class()
401
402	1		if hasattr(data, 'value'):
403	1		data = data.value
404
405	1		obj.unpack(data)
406
407	1		return obj
408
409	1		@staticmethod
410			def _get_data(req):
411			"""Get request data."""
412	1		data = req.get_json() # Valid format { "interfaces": [...] }
413	1		return data.get('interfaces', [])
414
415	1		def _get_interfaces(self):
416			"""Get all interfaces."""
417	1		interfaces = []
418	1		for switch in list(self.controller.switches.values()):
419	1		interfaces += list(switch.interfaces.values())
420	1		return interfaces
421
422	1		@staticmethod
423			def _get_interfaces_dict(interfaces):
424			"""Return a dict of interfaces."""
425	1		return {inter.id: inter for inter in interfaces}
426
427	1		def _get_lldp_interfaces(self):
428			"""Get interfaces enabled to receive LLDP packets."""
429	1		return [inter.id for inter in self._get_interfaces() if inter.lldp]
430
431	1		@rest('v1/interfaces', methods=['GET'])
432			def get_lldp_interfaces(self):
433			"""Return all the interfaces that have LLDP traffic enabled."""
434	1		return jsonify({"interfaces": self._get_lldp_interfaces()}), 200
435
436	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...
437			def disable_lldp(self):
438			"""Disables an interface to receive LLDP packets."""
439	1		interface_ids = self._get_data(request)
440	1		error_list = [] # List of interfaces that were not activated.
441	1		changed_interfaces = []
442	1		interface_ids = filter(None, interface_ids)
443	1		interfaces = self._get_interfaces()
444	1		if not interfaces:
445	1		return jsonify("No interfaces were found."), 404
446	1		interfaces = self._get_interfaces_dict(interfaces)
447	1		for id_ in interface_ids:
448	1		interface = interfaces.get(id_)
449	1		if interface:
450	1		interface.lldp = False
451	1		changed_interfaces.append(id_)
452			else:
453	1		error_list.append(id_)
454	1		if changed_interfaces:
455	1		self.notify_lldp_change('disabled', changed_interfaces)
456	1		if not error_list:
457	1		return jsonify(
458			"All the requested interfaces have been disabled."), 200
459
460			# Return a list of interfaces that couldn't be disabled
461	1		msg_error = "Some interfaces couldn't be found and deactivated: "
462	1		return jsonify({msg_error:
463			error_list}), 400
464
465	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...
466			def enable_lldp(self):
467			"""Enable an interface to receive LLDP packets."""
468	1		interface_ids = self._get_data(request)
469	1		error_list = [] # List of interfaces that were not activated.
470	1		changed_interfaces = []
471	1		interface_ids = filter(None, interface_ids)
472	1		interfaces = self._get_interfaces()
473	1		if not interfaces:
474	1		return jsonify("No interfaces were found."), 404
475	1		interfaces = self._get_interfaces_dict(interfaces)
476	1		for id_ in interface_ids:
477	1		interface = interfaces.get(id_)
478	1		if interface:
479	1		interface.lldp = True
480	1		changed_interfaces.append(id_)
481			else:
482	1		error_list.append(id_)
483	1		if changed_interfaces:
484	1		self.notify_lldp_change('enabled', changed_interfaces)
485	1		if not error_list:
486	1		return jsonify(
487			"All the requested interfaces have been enabled."), 200
488
489			# Return a list of interfaces that couldn't be enabled
490	1		msg_error = "Some interfaces couldn't be found and activated: "
491	1		return jsonify({msg_error:
492			error_list}), 400
493
494	1		@rest('v1/polling_time', methods=['GET'])
495			def get_time(self):
496			"""Get LLDP polling time in seconds."""
497	1		return jsonify({"polling_time": self.polling_time}), 200
498
499	1		@rest('v1/polling_time', methods=['POST'])
500			def set_time(self):
501			"""Set LLDP polling time."""
502			# pylint: disable=attribute-defined-outside-init
503	1		try:
504	1		payload = request.get_json()
505	1		polling_time = int(payload['polling_time'])
506	1		if polling_time <= 0:
507			raise ValueError(f"invalid polling_time {polling_time}, "
508			"must be greater than zero")
509	1		self.polling_time = polling_time
510	1		self.execute_as_loop(self.polling_time)
511	1		log.info("Polling time has been updated to %s"
512			" second(s), but this change will not be saved"
513			" permanently.", self.polling_time)
514	1		return jsonify("Polling time has been updated."), 200
515	1		except (ValueError, KeyError) as error:
516	1		msg = f"This operation is not completed: {error}"
517			return jsonify(msg), 400
518

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