Test Failed
Pull Request — master (#56)
by Carlos
02:11
created

build.main.Main._build_lldp_flow()   A

Complexity

Conditions 4

Size

Total Lines 31
Code Lines 16

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 16
CRAP Score 4

Importance

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