IPv4.pack() - Code Metrics - Inspection of "Add IPv4 network type class" - kytos/python-openflow - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — master (#371)

unknown

created 2017-06-01 20:38 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	3
Bugs	0	Features	0

Metric	Value
c	3
b	0
f	0
dl	0
loc	16
ccs	0
cts	0
cp	0
rs	9.4285
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.
    """

    # IP protocol version + Internet Header Length (words)
    _version_ihl = UBInt8()
    # Differentiated Services Code Point (ToS - Type of Service) +
    # Explicit Congestion Notification
    _dscp_ecn = UBInt8()
    # IP packet length (bytes)
    length = UBInt16()
    # Packet ID - common to all fragments
    identification = UBInt16()
    # Fragmentation flags + fragmentation offset
    _flags_offset = UBInt16()
    # Packet time-to-live
    ttl = UBInt8()
    # Upper layer protocol number
    protocol = UBInt8()
    # Header checksum
    checksum = UBInt16()
    # Source address
    source = IPAddress()
    # Destination address
    destination = IPAddress()
    # IP Options - up to 320 bits, always padded to 32 bits
    options = 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):
        """Updates 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):
        # 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):
        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 (BinaryData, HWAddress, IPAddress,
8	1	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.
149	1	"""
150
151	1	# IP protocol version + Internet Header Length (words)
152	1	_version_ihl = UBInt8()
153	1	# Differentiated Services Code Point (ToS - Type of Service) +
154		# Explicit Congestion Notification
155	1	_dscp_ecn = UBInt8()
156		# IP packet length (bytes)
157		length = UBInt16()
158		# Packet ID - common to all fragments
159		identification = UBInt16()
160		# Fragmentation flags + fragmentation offset
161	1	_flags_offset = UBInt16()
162		# Packet time-to-live
163		ttl = UBInt8()
164		# Upper layer protocol number
165		protocol = UBInt8()
166		# Header checksum
167		checksum = UBInt16()
168		# Source address
169		source = IPAddress()
170		# Destination address
171		destination = IPAddress()
172		# IP Options - up to 320 bits, always padded to 32 bits
173		options = BinaryData()
174		# Packet data
175		data = BinaryData()
176
177		def __init__(self, version=4, ihl=5, dscp=0, ecn=0, length=0, # noqa
178		identification=0, flags=0, offset=0, ttl=255, protocol=0,
179		checksum=0, source="0.0.0.0", destination="0.0.0.0",
180		options=b'', data=b''):
181		"""Create the Packet and set instance attributes."""
182		super().__init__()
183		self.version = version
184		self.ihl = ihl
185	1	self.dscp = dscp
186		self.ecn = ecn
187		self.length = length
188		self.identification = identification
189		self.flags = flags
190		self.offset = offset
191		self.ttl = ttl
192		self.protocol = protocol
193		self.checksum = checksum
194	1	self.source = source
195	1	self.destination = destination
196		self.options = options
197	1	self.data = data
198
199		def _update_checksum(self):
200	1	"""Updates the packet checksum to enable integrity check."""
201		source_list = [int(octet) for octet in self.source.split(".")]
202		destination_list = [int(octet) for octet in
203		self.destination.split(".")]
204		source_upper = (source_list[0] << 8) + source_list[1]
205		source_lower = (source_list[2] << 8) + source_list[3]
206		destination_upper = (destination_list[0] << 8) + destination_list[1]
207		destination_lower = (destination_list[2] << 8) + destination_list[3]
208
209		block_sum = ((self._version_ihl << 8 \| self._dscp_ecn) + self.length +
210		self.identification + self._flags_offset +
211		(self.ttl << 8 \| self.protocol) + source_upper +
212		source_lower + destination_upper + destination_lower)
213
214		while block_sum > 65535:
215		carry = block_sum >> 16
216		block_sum = (block_sum & 65535) + carry
217
218		self.checksum = ~block_sum & 65535
219
220		def pack(self, value=None):
221		# Set the correct IHL based on options size
222		if self.options:
223		self.ihl += int(len(self.options) / 4)
224
225		# Set the correct packet length based on header length and data
226		self.length = int(self.ihl * 4 + len(self.data))
227
228		self._version_ihl = self.version << 4 \| self.ihl
229		self._dscp_ecn = self.dscp << 2 \| self.ecn
230		self._flags_offset = self.flags << 13 \| self.offset
231
232		# Set the checksum field before packing
233		self._update_checksum()
234
235		return super().pack()
236
237		def unpack(self, buff, offset=0):
238		super().unpack(buff, offset)
239
240		self.version = self._version_ihl.value >> 4
241		self.ihl = self._version_ihl.value & 15
242		self.dscp = self._dscp_ecn.value >> 2
243		self.ecn = self._dscp_ecn.value & 3
244		self.length = self.length.value
245		self.identification = self.identification.value
246		self.flags = self._flags_offset.value >> 13
247		self.offset = self._flags_offset.value & 8191
248		self.ttl = self.ttl.value
249		self.protocol = self.protocol.value
250		self.checksum = self.checksum.value
251		self.source = self.source.value
252		self.destination = self.destination.value
253
254		if self.ihl > 5:
255		options_size = (self.ihl - 5) * 4
256		self.data = self.options.value[options_size:]
257		self.options = self.options.value[:options_size]
258		else:
259		self.data = self.options.value
260		self.options = b''
261
262
263		class TLVWithSubType(GenericTLV):
264		"""Modify the :class:`GenericTLV` to a Organization Specific TLV structure.
265
266		Beyond the standard TLV (type, length, value), we can also have a more
267		specific structure, with the value field being splitted into a sub_type
268		field and a new "sub_value" field.
269		"""
270
271		def __init__(self, tlv_type=1, sub_type=7, sub_value=BinaryData()):
272		"""Create an instance and set its attributes."""
273		super().__init__(tlv_type)
274		self.sub_type = sub_type
275		self.sub_value = sub_value
276
277		@property
278		def value(self):
279		"""Return sub type and sub value as binary data."""
280		binary = UBInt8(self.sub_type).pack() + self.sub_value.pack()
281		return BinaryData(binary)
282
283		def unpack(self, buffer, offset=0):
284		"""Unpack a binary message into this object's attributes.
285
286		Unpack the binary value buff and update this object attributes based
287		on the results.
288
289		Args:
290		buff (bytes): Binary data package to be unpacked.
291		offset (int): Where to begin unpacking.
292
293		Raises:
294		Exception: If there is a struct unpacking error.
295		"""
296		header = UBInt16()
297		header.unpack(buffer[offset:offset+2])
298		self.tlv_type = header.value >> 9
299		length = header.value & 511
300		begin, end = offset + 2, offset + 2 + length
301		sub_type = UBInt8()
302		sub_type.unpack(buffer[begin:begin+1])
303		self.sub_type = sub_type.value
304		self.sub_value = BinaryData(buffer[begin+1:end])
305
306
307		class LLDP(GenericStruct):
308		"""LLDP class.
309
310		Build a LLDP packet with TLVSubtype and Generic Subtypes.
311
312		It contains a chassis_id TLV, a port_id TLV, a TTL (Time to live) and
313		another TVL to represent the end of the LLDP Packet.
314		"""
315
316		chassis_id = TLVWithSubType(tlv_type=1, sub_type=7)
317		port_id = TLVWithSubType(tlv_type=2, sub_type=7)
318		#: TTL time is given in seconds, between 0 and 65535
319		ttl = GenericTLV(tlv_type=3, value=UBInt16(120))
320		# We are not using list of tlvs for now
321		# tlvs = ListOfTLVs()
322		end = GenericTLV(tlv_type=0)
323

kytos / python-openflow

Pull Request — master (#371)

IPv4.pack() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like