IPv4.pack() - Code Metrics - Inspection of "Changing unpack method of ErrorMsg class" - kytos/python-openflow - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#368)

by macartur

created 2017-06-09 19:51 UTC

IPv4.pack() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	6

Importance

Changes

Metric	Value
dl	0
loc	20
ccs	0
cts	0
cp	0
rs	9.4285
c	0
b	0
f	0
cc	2
crap	6

"""Basic Network packet types.

Defines and Implements Basic Network packet types , such as Ethertnet and LLDP.
"""

from pyof.foundation.base import GenericStruct
from pyof.foundation.basic_types import (
    BinaryData, HWAddress, IPAddress, UBInt8, UBInt16)
from pyof.foundation.exceptions import PackException

__all__ = ('Ethernet', 'GenericTLV', 'IPv4', 'TLVWithSubType', 'LLDP')


class Ethernet(GenericStruct):
    """Ethernet "struct".

    Objects of this class represents an ethernet packet. It contains the
    'Ethernet header', composed by destination (MAC), source (MAC), type
    (EtherType)[1] and the payload of the packet, as binary data.

    This class does not consider the Ethernet 'Preamble' or the 'CRC'.

    There is also a get_hash method, that hashes the binary representation of
    the object so we can have a unique representation of the ethernet packet,
    so we can keep a track of ethernet packets being flooded over the network.

    [1] EtherTypes:
    http://www.iana.org/assignments/ieee-802-numbers/ieee-802-numbers.xhtml#ieee-802-numbers-1
    """

    #: destination (:class:`HWAddress`): The final destination MAC address.
    destination = HWAddress()
    #: source (:class:`HWAddress`): The source MAC address of the packet.
    source = HWAddress()
    #: ether_type (:class:`UBInt16`): The EtherType of the packet.
    ether_type = UBInt16()
    #: data (:class:`BinaryData`): The content of the packet in binary format.
    data = BinaryData()

    def __init__(self, destination=None, source=None, ether_type=None,
                 data=b''):
        """Create an instance and set its attributes."""
        super().__init__()
        self.destination = destination
        self.source = source
        self.ether_type = ether_type
        self.data = data

    def get_hash(self):
        """Return a hash that identifies this instance."""
        return hash(self.pack())


class GenericTLV(GenericStruct):
    """TLV structure of LLDP packets.

    This is a Type, Length and Value (TLV) struct.

    The LLDP/TLV definition states that the Type field have 7 bits, while
    the length have 9 bits. The Value must be between 0-511 octets.

    Internally, on the instances of this class, the Type is a integer
    (0-127) and the Length is dynamically calculated based on the current
    type and value.
    """

    def __init__(self, tlv_type=127, value=BinaryData()):
        """Create an instance and set its attributes."""
        super().__init__()
        self.tlv_type = tlv_type
        self._value = value

    @property
    def value(self):
        """Return the value stored by GenericTLV."""
        return self._value

    @property
    def length(self):
        """Struct length in bytes."""
        return len(self.value.pack())

    @property
    def header(self):
        """Header of the TLV Packet.

        The header is composed by the Type (7 bits) and Length (9 bits),
        summing up 16 bits. To achieve that, we need to do some bitshift
        operations.
        """
        return UBInt16(((self.tlv_type & 127) << 9) | (self.length & 511))

    def pack(self, value=None):
        """Pack the TLV in a binary representation.

        Returns:
            bytes: Binary representation of the struct object.

        Raises:
            :exc:`~.exceptions.ValidationError`: If validation fails.
        """
        if value is None:
            output = self.header.pack()
            output += self.value.pack()
            return output

        elif isinstance(value, type(self)):
            return value.pack()
        else:
            msg = "{} is not an instance of {}".format(value,
                                                       type(self).__name__)
            raise PackException(msg)

    def unpack(self, buffer, offset=0):
        """Unpack a binary message into this object's attributes.

        Unpack the binary value *buff* and update this object attributes based
        on the results.

        Args:
            buff (bytes): Binary data package to be unpacked.
            offset (int): Where to begin unpacking.

        Raises:
            Exception: If there is a struct unpacking error.
        """
        header = UBInt16()
        header.unpack(buffer[offset:offset+2])
        self.tlv_type = header.value >> 9
        length = header.value & 511
        begin, end = offset + 2, offset + 2 + length
        self._value = BinaryData(buffer[begin:end])

    def get_size(self, value=None):
        """Return struct size."""
        if isinstance(value, type(self)):
            return value.get_size()

        return 2 + self.length


class IPv4(GenericStruct):
    """IPv4 packet "struct".

    Contains all fields of an IP version 4 packet header, plus the upper layer
    content as binary data.
    Some of the fields were merged together because of their size being
    inferior to 8 bits. They are represented as a single class attribute, but
    pack/unpack methods will take into account the values in individual
    instance attributes.
    """

    #: _version_ihl (:class:`UBInt8`): IP protocol version + Internet Header
    #: Length (words)
    _version_ihl = UBInt8()
    #: _dscp_ecn (:class:`UBInt8`): Differentiated Services Code Point
    #: (ToS - Type of Service) + Explicit Congestion Notification
    _dscp_ecn = UBInt8()
    #: length (:class:`UBInt16`): IP packet length (bytes)
    length = UBInt16()
    #: identification (:class:`UBInt16`): Packet ID - common to all fragments
    identification = UBInt16()
    #: _flags_offset (:class:`UBInt16`): Fragmentation flags + fragmentation
    #: offset
    _flags_offset = UBInt16()
    #: ttl (:class:`UBInt8`): Packet time-to-live
    ttl = UBInt8()
    #: protocol (:class:`UBInt8`): Upper layer protocol number
    protocol = UBInt8()
    #: checksum (:class:`UBInt16`): Header checksum
    checksum = UBInt16()
    #: source (:class:`IPAddress`): Source IPv4 address
    source = IPAddress()
    #: destination (:class:`IPAddress`): Destination IPv4 address
    destination = IPAddress()
    #: options (:class:`BinaryData`): IP Options - up to 320 bits, always
    #: padded to 32 bits
    options = BinaryData()
    #: data (:class:`BinaryData`): Packet data
    data = BinaryData()

    def __init__(self, version=4, ihl=5, dscp=0, ecn=0, length=0, # noqa
                 identification=0, flags=0, offset=0, ttl=255, protocol=0,
                 checksum=0, source="0.0.0.0", destination="0.0.0.0",
                 options=b'', data=b''):
        """Create the Packet and set instance attributes."""
        super().__init__()
        self.version = version
        self.ihl = ihl
        self.dscp = dscp
        self.ecn = ecn
        self.length = length
        self.identification = identification
        self.flags = flags
        self.offset = offset
        self.ttl = ttl
        self.protocol = protocol
        self.checksum = checksum
        self.source = source
        self.destination = destination
        self.options = options
        self.data = data

    def _update_checksum(self):
        """Update the packet checksum to enable integrity check."""
        source_list = [int(octet) for octet in self.source.split(".")]
        destination_list = [int(octet) for octet in
                            self.destination.split(".")]
        source_upper = (source_list[0] << 8) + source_list[1]
        source_lower = (source_list[2] << 8) + source_list[3]
        destination_upper = (destination_list[0] << 8) + destination_list[1]
        destination_lower = (destination_list[2] << 8) + destination_list[3]

        block_sum = ((self._version_ihl << 8 | self._dscp_ecn) + self.length +
                     self.identification + self._flags_offset +
                     (self.ttl << 8 | self.protocol) + source_upper +
                     source_lower + destination_upper + destination_lower)

        while block_sum > 65535:
            carry = block_sum >> 16
            block_sum = (block_sum & 65535) + carry

        self.checksum = ~block_sum & 65535

    def pack(self, value=None):
        """Pack the struct in a binary representation.

        Merge some fields to ensure correct packing.
        """
        # Set the correct IHL based on options size
        if self.options:
            self.ihl += int(len(self.options) / 4)

        # Set the correct packet length based on header length and data
        self.length = int(self.ihl * 4 + len(self.data))

        self._version_ihl = self.version << 4 | self.ihl
        self._dscp_ecn = self.dscp << 2 | self.ecn
        self._flags_offset = self.flags << 13 | self.offset

        # Set the checksum field before packing
        self._update_checksum()

        return super().pack()

    def unpack(self, buff, offset=0):
        """Unpack a binary struct into this object's attributes.

        Return the values instead of the lib's basic types.
        """
        super().unpack(buff, offset)

        self.version = self._version_ihl.value >> 4
        self.ihl = self._version_ihl.value & 15
        self.dscp = self._dscp_ecn.value >> 2
        self.ecn = self._dscp_ecn.value & 3
        self.length = self.length.value
        self.identification = self.identification.value
        self.flags = self._flags_offset.value >> 13
        self.offset = self._flags_offset.value & 8191
        self.ttl = self.ttl.value
        self.protocol = self.protocol.value
        self.checksum = self.checksum.value
        self.source = self.source.value
        self.destination = self.destination.value

        if self.ihl > 5:
            options_size = (self.ihl - 5) * 4
            self.data = self.options.value[options_size:]
            self.options = self.options.value[:options_size]
        else:
            self.data = self.options.value
            self.options = b''


class TLVWithSubType(GenericTLV):
    """Modify the :class:`GenericTLV` to a Organization Specific TLV structure.

    Beyond the standard TLV (type, length, value), we can also have a more
    specific structure, with the *value* field being splitted into a *sub_type*
    field and a new "*sub_value*" field.
    """

    def __init__(self, tlv_type=1, sub_type=7, sub_value=BinaryData()):
        """Create an instance and set its attributes."""
        super().__init__(tlv_type)
        self.sub_type = sub_type
        self.sub_value = sub_value

    @property
    def value(self):
        """Return sub type and sub value as binary data."""
        binary = UBInt8(self.sub_type).pack() + self.sub_value.pack()
        return BinaryData(binary)

    def unpack(self, buffer, offset=0):
        """Unpack a binary message into this object's attributes.

        Unpack the binary value *buff* and update this object attributes based
        on the results.

        Args:
            buff (bytes): Binary data package to be unpacked.
            offset (int): Where to begin unpacking.

        Raises:
            Exception: If there is a struct unpacking error.
        """
        header = UBInt16()
        header.unpack(buffer[offset:offset+2])
        self.tlv_type = header.value >> 9
        length = header.value & 511
        begin, end = offset + 2, offset + 2 + length
        sub_type = UBInt8()
        sub_type.unpack(buffer[begin:begin+1])
        self.sub_type = sub_type.value
        self.sub_value = BinaryData(buffer[begin+1:end])


class LLDP(GenericStruct):
    """LLDP class.

    Build a LLDP packet with TLVSubtype and Generic Subtypes.

    It contains a chassis_id TLV, a port_id TLV, a TTL (Time to live) and
    another TVL to represent the end of the LLDP Packet.
    """

    chassis_id = TLVWithSubType(tlv_type=1, sub_type=7)
    port_id = TLVWithSubType(tlv_type=2, sub_type=7)
    #: TTL time is given in seconds, between 0 and 65535
    ttl = GenericTLV(tlv_type=3, value=UBInt16(120))
    # We are not using list of tlvs for now
    # tlvs = ListOfTLVs()
    end = GenericTLV(tlv_type=0)


1		"""Basic Network packet types.
2
3		Defines and Implements Basic Network packet types , such as Ethertnet and LLDP.
4		"""
5
6	1	from pyof.foundation.base import GenericStruct
7	1	from pyof.foundation.basic_types import (
8	1	BinaryData, HWAddress, IPAddress, UBInt8, UBInt16)
9		from pyof.foundation.exceptions import PackException
10	1
11		__all__ = ('Ethernet', 'GenericTLV', 'IPv4', 'TLVWithSubType', 'LLDP')
12
13	1
14		class Ethernet(GenericStruct):
15		"""Ethernet "struct".
16
17		Objects of this class represents an ethernet packet. It contains the
18		'Ethernet header', composed by destination (MAC), source (MAC), type
19		(EtherType)[1] and the payload of the packet, as binary data.
20
21		This class does not consider the Ethernet 'Preamble' or the 'CRC'.
22
23		There is also a get_hash method, that hashes the binary representation of
24		the object so we can have a unique representation of the ethernet packet,
25		so we can keep a track of ethernet packets being flooded over the network.
26
27		[1] EtherTypes:
28		http://www.iana.org/assignments/ieee-802-numbers/ieee-802-numbers.xhtml#ieee-802-numbers-1
29		"""
30
31	1	#: destination (:class:`HWAddress`): The final destination MAC address.
32		destination = HWAddress()
33	1	#: source (:class:`HWAddress`): The source MAC address of the packet.
34		source = HWAddress()
35	1	#: ether_type (:class:`UBInt16`): The EtherType of the packet.
36		ether_type = UBInt16()
37	1	#: data (:class:`BinaryData`): The content of the packet in binary format.
38		data = BinaryData()
39	1
40		def __init__(self, destination=None, source=None, ether_type=None,
41		data=b''):
42		"""Create an instance and set its attributes."""
43		super().__init__()
44		self.destination = destination
45		self.source = source
46		self.ether_type = ether_type
47		self.data = data
48	1
49		def get_hash(self):
50		"""Return a hash that identifies this instance."""
51		return hash(self.pack())
52
53	1
54		class GenericTLV(GenericStruct):
55		"""TLV structure of LLDP packets.
56
57		This is a Type, Length and Value (TLV) struct.
58
59		The LLDP/TLV definition states that the Type field have 7 bits, while
60		the length have 9 bits. The Value must be between 0-511 octets.
61
62		Internally, on the instances of this class, the Type is a integer
63		(0-127) and the Length is dynamically calculated based on the current
64		type and value.
65		"""
66	1
67		def __init__(self, tlv_type=127, value=BinaryData()):
68	1	"""Create an instance and set its attributes."""
69	1	super().__init__()
70	1	self.tlv_type = tlv_type
71		self._value = value
72	1
73		@property
74		def value(self):
75	1	"""Return the value stored by GenericTLV."""
76		return self._value
77	1
78		@property
79		def length(self):
80	1	"""Struct length in bytes."""
81		return len(self.value.pack())
82	1
83		@property
84		def header(self):
85		"""Header of the TLV Packet.
86
87		The header is composed by the Type (7 bits) and Length (9 bits),
88		summing up 16 bits. To achieve that, we need to do some bitshift
89		operations.
90	1	"""
91		return UBInt16(((self.tlv_type & 127) << 9) \| (self.length & 511))
92	1
93		def pack(self, value=None):
94		"""Pack the TLV in a binary representation.
95
96		Returns:
97		bytes: Binary representation of the struct object.
98
99		Raises:
100		:exc:`~.exceptions.ValidationError`: If validation fails.
101	1	"""
102	1	if value is None:
103	1	output = self.header.pack()
104	1	output += self.value.pack()
105		return output
106
107		elif isinstance(value, type(self)):
108		return value.pack()
109		else:
110		msg = "{} is not an instance of {}".format(value,
111		type(self).__name__)
112		raise PackException(msg)
113	1
114		def unpack(self, buffer, offset=0):
115		"""Unpack a binary message into this object's attributes.
116
117		Unpack the binary value buff and update this object attributes based
118		on the results.
119
120		Args:
121		buff (bytes): Binary data package to be unpacked.
122		offset (int): Where to begin unpacking.
123
124		Raises:
125		Exception: If there is a struct unpacking error.
126	1	"""
127	1	header = UBInt16()
128	1	header.unpack(buffer[offset:offset+2])
129	1	self.tlv_type = header.value >> 9
130	1	length = header.value & 511
131	1	begin, end = offset + 2, offset + 2 + length
132		self._value = BinaryData(buffer[begin:end])
133	1
134		def get_size(self, value=None):
135		"""Return struct size."""
136		if isinstance(value, type(self)):
137		return value.get_size()
138
139		return 2 + self.length
140
141	1
142		class IPv4(GenericStruct):
143		"""IPv4 packet "struct".
144
145		Contains all fields of an IP version 4 packet header, plus the upper layer
146		content as binary data.
147		Some of the fields were merged together because of their size being
148		inferior to 8 bits. They are represented as a single class attribute, but
149	1	pack/unpack methods will take into account the values in individual
150		instance attributes.
151	1	"""
152	1
153	1	#: _version_ihl (:class:`UBInt8`): IP protocol version + Internet Header
154		#: Length (words)
155	1	_version_ihl = UBInt8()
156		#: _dscp_ecn (:class:`UBInt8`): Differentiated Services Code Point
157		#: (ToS - Type of Service) + Explicit Congestion Notification
158		_dscp_ecn = UBInt8()
159		#: length (:class:`UBInt16`): IP packet length (bytes)
160		length = UBInt16()
161	1	#: identification (:class:`UBInt16`): Packet ID - common to all fragments
162		identification = UBInt16()
163		#: _flags_offset (:class:`UBInt16`): Fragmentation flags + fragmentation
164		#: offset
165		_flags_offset = UBInt16()
166		#: ttl (:class:`UBInt8`): Packet time-to-live
167		ttl = UBInt8()
168		#: protocol (:class:`UBInt8`): Upper layer protocol number
169		protocol = UBInt8()
170		#: checksum (:class:`UBInt16`): Header checksum
171		checksum = UBInt16()
172		#: source (:class:`IPAddress`): Source IPv4 address
173		source = IPAddress()
174		#: destination (:class:`IPAddress`): Destination IPv4 address
175		destination = IPAddress()
176		#: options (:class:`BinaryData`): IP Options - up to 320 bits, always
177		#: padded to 32 bits
178		options = BinaryData()
179		#: data (:class:`BinaryData`): Packet data
180		data = BinaryData()
181
182		def __init__(self, version=4, ihl=5, dscp=0, ecn=0, length=0, # noqa
183		identification=0, flags=0, offset=0, ttl=255, protocol=0,
184		checksum=0, source="0.0.0.0", destination="0.0.0.0",
185	1	options=b'', data=b''):
186		"""Create the Packet and set instance attributes."""
187		super().__init__()
188		self.version = version
189		self.ihl = ihl
190		self.dscp = dscp
191		self.ecn = ecn
192		self.length = length
193		self.identification = identification
194	1	self.flags = flags
195	1	self.offset = offset
196		self.ttl = ttl
197	1	self.protocol = protocol
198		self.checksum = checksum
199		self.source = source
200	1	self.destination = destination
201		self.options = options
202		self.data = data
203
204		def _update_checksum(self):
205		"""Update the packet checksum to enable integrity check."""
206		source_list = [int(octet) for octet in self.source.split(".")]
207		destination_list = [int(octet) for octet in
208		self.destination.split(".")]
209		source_upper = (source_list[0] << 8) + source_list[1]
210		source_lower = (source_list[2] << 8) + source_list[3]
211		destination_upper = (destination_list[0] << 8) + destination_list[1]
212		destination_lower = (destination_list[2] << 8) + destination_list[3]
213
214		block_sum = ((self._version_ihl << 8 \| self._dscp_ecn) + self.length +
215		self.identification + self._flags_offset +
216		(self.ttl << 8 \| self.protocol) + source_upper +
217		source_lower + destination_upper + destination_lower)
218
219		while block_sum > 65535:
220		carry = block_sum >> 16
221		block_sum = (block_sum & 65535) + carry
222
223		self.checksum = ~block_sum & 65535
224
225		def pack(self, value=None):
226		"""Pack the struct in a binary representation.
227
228		Merge some fields to ensure correct packing.
229		"""
230		# Set the correct IHL based on options size
231		if self.options:
232		self.ihl += int(len(self.options) / 4)
233
234		# Set the correct packet length based on header length and data
235		self.length = int(self.ihl * 4 + len(self.data))
236
237		self._version_ihl = self.version << 4 \| self.ihl
238		self._dscp_ecn = self.dscp << 2 \| self.ecn
239		self._flags_offset = self.flags << 13 \| self.offset
240
241		# Set the checksum field before packing
242		self._update_checksum()
243
244		return super().pack()
245
246		def unpack(self, buff, offset=0):
247		"""Unpack a binary struct into this object's attributes.
248
249		Return the values instead of the lib's basic types.
250		"""
251		super().unpack(buff, offset)
252
253		self.version = self._version_ihl.value >> 4
254		self.ihl = self._version_ihl.value & 15
255		self.dscp = self._dscp_ecn.value >> 2
256		self.ecn = self._dscp_ecn.value & 3
257		self.length = self.length.value
258		self.identification = self.identification.value
259		self.flags = self._flags_offset.value >> 13
260		self.offset = self._flags_offset.value & 8191
261		self.ttl = self.ttl.value
262		self.protocol = self.protocol.value
263		self.checksum = self.checksum.value
264		self.source = self.source.value
265		self.destination = self.destination.value
266
267		if self.ihl > 5:
268		options_size = (self.ihl - 5) * 4
269		self.data = self.options.value[options_size:]
270		self.options = self.options.value[:options_size]
271		else:
272		self.data = self.options.value
273		self.options = b''
274
275
276		class TLVWithSubType(GenericTLV):
277		"""Modify the :class:`GenericTLV` to a Organization Specific TLV structure.
278
279		Beyond the standard TLV (type, length, value), we can also have a more
280		specific structure, with the value field being splitted into a sub_type
281		field and a new "sub_value" field.
282		"""
283
284		def __init__(self, tlv_type=1, sub_type=7, sub_value=BinaryData()):
285		"""Create an instance and set its attributes."""
286		super().__init__(tlv_type)
287		self.sub_type = sub_type
288		self.sub_value = sub_value
289
290		@property
291		def value(self):
292		"""Return sub type and sub value as binary data."""
293		binary = UBInt8(self.sub_type).pack() + self.sub_value.pack()
294		return BinaryData(binary)
295
296		def unpack(self, buffer, offset=0):
297		"""Unpack a binary message into this object's attributes.
298
299		Unpack the binary value buff and update this object attributes based
300		on the results.
301
302		Args:
303		buff (bytes): Binary data package to be unpacked.
304		offset (int): Where to begin unpacking.
305
306		Raises:
307		Exception: If there is a struct unpacking error.
308		"""
309		header = UBInt16()
310		header.unpack(buffer[offset:offset+2])
311		self.tlv_type = header.value >> 9
312		length = header.value & 511
313		begin, end = offset + 2, offset + 2 + length
314		sub_type = UBInt8()
315		sub_type.unpack(buffer[begin:begin+1])
316		self.sub_type = sub_type.value
317		self.sub_value = BinaryData(buffer[begin+1:end])
318
319
320		class LLDP(GenericStruct):
321		"""LLDP class.
322
323		Build a LLDP packet with TLVSubtype and Generic Subtypes.
324
325		It contains a chassis_id TLV, a port_id TLV, a TTL (Time to live) and
326		another TVL to represent the end of the LLDP Packet.
327		"""
328
329		chassis_id = TLVWithSubType(tlv_type=1, sub_type=7)
330		port_id = TLVWithSubType(tlv_type=2, sub_type=7)
331		#: TTL time is given in seconds, between 0 and 65535
332		ttl = GenericTLV(tlv_type=3, value=UBInt16(120))
333		# We are not using list of tlvs for now
334		# tlvs = ListOfTLVs()
335		end = GenericTLV(tlv_type=0)
336

kytos / python-openflow

Pull Request — master (#368)

IPv4.pack() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like