Passed
Push — master ( acb8d9...8107f9 )
by Vinicius
03:43 queued 14s
created

build.main.Main._get_data()   A

Complexity

Conditions 1

Size

Total Lines 5
Code Lines 4

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 3
CRAP Score 1

Importance

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