VLAN._validate() - Code Metrics - Inspection of "Merge pull request #424 from erickvermot/less_than..." - cemsbr/python-openflow - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 9eabc4...ddb1e5 )

by Carlos Eduardo

created 2017-07-28 17:36 UTC

VLAN._validate() A

↳ Parent: VLAN

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	4
CRAP Score	2

Importance

Changes

Metric	Value
c	0
b	0
f	0
dl	0
loc	5
ccs	4
cts	4
cp	1
rs	9.4285
cc	2
crap	2

"""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.constants import VLAN_TPID
from pyof.foundation.exceptions import PackException, UnpackException

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


class VLAN(GenericStruct):
    """802.1q VLAN header."""

    #: tpid (:class:`UBInt16`): Tag Protocol Identifier
    tpid = UBInt16(VLAN_TPID)
    #: _tci (:class:`UBInt16`): Tag Control Information - has the
    #: Priority Code Point, DEI/CFI bit and the VLAN ID
    _tci = UBInt16()

    def __init__(self, pcp=None, cfi=None, vid=None):
        """The constructor receives the parameters below.

        If no arguments are set for a particular instance, it is interpreted as
        abscence of VLAN information, and the pack() method will return an
        empty binary string.

        Args:
            tpid (int): Tag Protocol Identifier. Defaults to 0x8100 for 802.1q.
            pcp (int): 802.1p Priority Code Point. Defaults to 0 for Best
                       Effort Queue.
            cfi (int): Canonical Format Indicator. Defaults to 0 for Ethernet.
            vid (int): VLAN ID. If no VLAN is specified, value is 0.
        """
        super().__init__()
        self.tpid = VLAN_TPID
        self.pcp = pcp
        self.cfi = cfi
        self.vid = vid

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

        Merge some fields to ensure correct packing.

        If no arguments are set for a particular instance, it is interpreted as
        abscence of VLAN information, and the pack() method will return an
        empty binary string.

        Returns:
            bytes: Binary representation of this instance.
        """
        if isinstance(value, type(self)):
            return value.pack()

        if self.pcp is None and self.cfi is None and self.vid is None:
            return b''
        self.pcp = self.pcp if self.pcp is not None else 0
        self.cfi = self.cfi if self.cfi is not None else 0
        self.vid = self.vid if self.vid is not None else 0
        self._tci = self.pcp << 13 | self.cfi << 12 | self.vid
        return super().pack()

    def _validate(self):
        """Assure this is a 802.1q VLAN header instance."""
        if self.tpid.value != VLAN_TPID:
            raise UnpackException
        return

    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.

        After unpacking, the abscence of a `tpid` value causes the assignment
        of None to the field values to indicate that there is no VLAN
        information.

        Args:
            buff (bytes): Binary buffer.
            offset (int): Where to begin unpacking.

        Raises:
            :exc:`~.exceptions.UnpackException`: If unpack fails.
        """
        super().unpack(buff, offset)
        if self.tpid.value:
            self._validate()
            self.tpid = self.tpid.value
            self.pcp = self._tci.value >> 13
            self.cfi = (self._tci.value >> 12) & 1
            self.vid = self._tci.value & 4095
        else:
            self.tpid = VLAN_TPID
            self.pcp = None
            self.cfi = None
            self.vid = None


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 = HWAddress()
    source = HWAddress()
    vlan = VLAN()
    ether_type = UBInt16()
    data = BinaryData()

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

        Args:
            destination (:class:`~pyof.foundation.basic_types.HWAddress`):
                The final destination MAC address.
            source (:class:`~pyof.foundation.basic_types.HWAddress`):
                The source Mac address of the packet.
            ether_type (:class:`~pyof.foundation.basic_types.UBInt16`):
                The EtherType of packet.
            data (:class:`~pyof.foundation.basic_types.BinaryData`):
                The content of the packet in binary format.
        """
        super().__init__()
        self.destination = destination
        self.source = source
        self.vlan = vlan
        self.ether_type = ether_type
        self.data = data

    def get_hash(self):
        """Calculate a hash and returns it.

        Returns:
            int: Integer value that identifies this instance.
        """
        return hash(self.pack())

    def unpack(self, buff, 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.

        Ethernet headers may have VLAN tags. If no VLAN tag is found, a
        'wildcard VLAN tag' is inserted to assure correct unpacking.

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

        Raises:
            UnpackException: If there is a struct unpacking error.
        """
        # Checking if the EtherType bytes are actually equal to VLAN_TPID -
        # indicating that the packet is tagged. If it is not, we insert the
        # equivalent to 'NULL VLAN data' (\x00\x00\x00\x00) to enable the
        # correct unpacking process.
        if buff[12:16] != VLAN_TPID.to_bytes(2, 'big'):
            buff = buff[0:12] + b'\x00\x00\x00\x00' + buff[12:]

        super().unpack(buff, offset)


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.

        Args:
            tlv_type (int): Type used by this class. Defaults to 127.
            value (:class:`~pyof.foundation.basic_types.BinaryData`):
                Value stored by GenericTLV.
        """
        super().__init__()
        self.tlv_type = tlv_type
        self._value = value

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

        Returns:
            :class:`~pyof.foundation.basic_types.BinaryData`:
                Value stored by GenericTLV.
        """
        return self._value

    @property
    def length(self):
        """Return the length of value stored by GenericTLV.

        Returns:
            int: Value 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.

        Returns:
            :class:`~pyof.foundation.basic_types.UBInt16`:
                Result after all 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.

        Returns:
            int: Returns the struct size based on inner attributes.
        """
        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''):
        """The constructor receives the parameters below.

        Args:
            version (int): IP protocol version. Defaults to 4.
            ihl (int): Internet Header Length. Default is 5.
            dscp (int): Differentiated Service Code Point. Defaults to 0.
            ecn (int): Explicit Congestion Notification. Defaults to 0.
            length (int): IP packet length in bytes. Defaults to 0.
            identification (int): Packet Id. Defaults to 0.
            flags (int): IPv4 Flags. Defults 0.
            offset (int): IPv4 offset. Defaults to 0.
            ttl (int): Packet time-to-live. Defaults to 255
            protocol (int): Upper layer protocol number. Defaults to 0.
            checksum (int): Header checksum. Defaults to 0.
            source (str): Source IPv4 address. Defaults to "0.0.0.0"
            destination (str): Destination IPv4 address. Defaults to "0.0.0.0"
            options (bytes): IP options. Defaults to empty bytes.
            data (bytes): Packet data. Defaults to empty bytes.
        """
        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.

        Returns:
            bytes: Binary representation of this instance.
        """
        # 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.

        Args:
            buff (bytes): Binary buffer.
            offset (int): Where to begin unpacking.

        Raises:
            :exc:`~.exceptions.UnpackException`: If unpack fails.
        """
        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 :attr:`value` field being splitted into a
    :attr:`sub_type` field and a new :attr:`sub_value` field.
    """

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

        Args:
            tlv_type (int): Type used by this class. Defaults to 1.
            sub_type (int): Sub type value used by this class. Defaults to 7.
            sub_value (:class:`~pyof.foundation.basic_types.BinaryData`):
                Data stored by TLVWithSubType. Defaults to empty BinaryData.
        """
        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.

        Returns:
            :class:`~pyof.foundation.basic_types.BinaryData`:
                BinaryData calculated.
        """
        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 TLV to represent the end of the LLDP Packet.
    """

    #: chassis_id (:class:`~TLVWithSubType`) with tlv_type = 1 and sub_type = 7
    chassis_id = TLVWithSubType(tlv_type=1, sub_type=7)
    #: port_id (:class:`TLVWithSubType`) with tlv = 2 and sub_type = 7
    port_id = TLVWithSubType(tlv_type=2, sub_type=7)
    #: TTL (:class:`GenericTLV`) time is given in seconds, between 0 and 65535,
    #: with tlv_type = 3
    ttl = GenericTLV(tlv_type=3, value=UBInt16(120))
    # We are not using list of tlvs for now
    # tlvs = ListOfTLVs()
    #: end (:class:`GenericTLV`) with tlv_type = 0
    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		BinaryData, HWAddress, IPAddress, UBInt8, UBInt16)
9	1	from pyof.foundation.constants import VLAN_TPID
10	1	from pyof.foundation.exceptions import PackException, UnpackException
11
12	1	__all__ = ('Ethernet', 'GenericTLV', 'IPv4', 'VLAN', 'TLVWithSubType', 'LLDP')
13
14
15	1	class VLAN(GenericStruct):
16		"""802.1q VLAN header."""
17
18		#: tpid (:class:`UBInt16`): Tag Protocol Identifier
19	1	tpid = UBInt16(VLAN_TPID)
20		#: _tci (:class:`UBInt16`): Tag Control Information - has the
21		#: Priority Code Point, DEI/CFI bit and the VLAN ID
22	1	_tci = UBInt16()
23
24	1	def __init__(self, pcp=None, cfi=None, vid=None):
25		"""The constructor receives the parameters below.
26
27		If no arguments are set for a particular instance, it is interpreted as
28		abscence of VLAN information, and the pack() method will return an
29		empty binary string.
30
31		Args:
32		tpid (int): Tag Protocol Identifier. Defaults to 0x8100 for 802.1q.
33		pcp (int): 802.1p Priority Code Point. Defaults to 0 for Best
34		Effort Queue.
35		cfi (int): Canonical Format Indicator. Defaults to 0 for Ethernet.
36		vid (int): VLAN ID. If no VLAN is specified, value is 0.
37		"""
38	1	super().__init__()
39	1	self.tpid = VLAN_TPID
40	1	self.pcp = pcp
41	1	self.cfi = cfi
42	1	self.vid = vid
43
44	1	def pack(self, value=None):
45		"""Pack the struct in a binary representation.
46
47		Merge some fields to ensure correct packing.
48
49		If no arguments are set for a particular instance, it is interpreted as
50		abscence of VLAN information, and the pack() method will return an
51		empty binary string.
52
53		Returns:
54		bytes: Binary representation of this instance.
55		"""
56	1	if isinstance(value, type(self)):
57	1	return value.pack()
58
59	1	if self.pcp is None and self.cfi is None and self.vid is None:
60	1	return b''
61	1	self.pcp = self.pcp if self.pcp is not None else 0
62	1	self.cfi = self.cfi if self.cfi is not None else 0
63	1	self.vid = self.vid if self.vid is not None else 0
64	1	self._tci = self.pcp << 13 \| self.cfi << 12 \| self.vid
65	1	return super().pack()
66
67	1	def _validate(self):
68		"""Assure this is a 802.1q VLAN header instance."""
69	1	if self.tpid.value != VLAN_TPID:
70	1	raise UnpackException
71	1	return
72
73	1	def unpack(self, buff, offset=0):
74		"""Unpack a binary struct into this object's attributes.
75
76		Return the values instead of the lib's basic types.
77
78		After unpacking, the abscence of a `tpid` value causes the assignment
79		of None to the field values to indicate that there is no VLAN
80		information.
81
82		Args:
83		buff (bytes): Binary buffer.
84		offset (int): Where to begin unpacking.
85
86		Raises:
87		:exc:`~.exceptions.UnpackException`: If unpack fails.
88		"""
89	1	super().unpack(buff, offset)
90	1	if self.tpid.value:
91	1	self._validate()
92	1	self.tpid = self.tpid.value
93	1	self.pcp = self._tci.value >> 13
94	1	self.cfi = (self._tci.value >> 12) & 1
95	1	self.vid = self._tci.value & 4095
96		else:
97	1	self.tpid = VLAN_TPID
98	1	self.pcp = None
99	1	self.cfi = None
100	1	self.vid = None
101
102
103	1	class Ethernet(GenericStruct):
104		"""Ethernet "struct".
105
106		Objects of this class represents an ethernet packet. It contains the
107		'Ethernet header', composed by destination (MAC), source (MAC), type
108		(EtherType)[1] and the payload of the packet, as binary data.
109
110		This class does not consider the Ethernet 'Preamble' or the 'CRC'.
111
112		There is also a get_hash method, that hashes the binary representation of
113		the object so we can have a unique representation of the ethernet packet,
114		so we can keep a track of ethernet packets being flooded over the network.
115
116		[1] EtherTypes:
117		http://www.iana.org/assignments/ieee-802-numbers/ieee-802-numbers.xhtml#ieee-802-numbers-1
118		"""
119
120	1	destination = HWAddress()
121	1	source = HWAddress()
122	1	vlan = VLAN()
123	1	ether_type = UBInt16()
124	1	data = BinaryData()
125
126	1	def __init__(self, destination=None, source=None, vlan=VLAN(),
127		ether_type=None, data=b''):
128		"""Create an instance and set its attributes.
129
130		Args:
131		destination (:class:`~pyof.foundation.basic_types.HWAddress`):
132		The final destination MAC address.
133		source (:class:`~pyof.foundation.basic_types.HWAddress`):
134		The source Mac address of the packet.
135		ether_type (:class:`~pyof.foundation.basic_types.UBInt16`):
136		The EtherType of packet.
137		data (:class:`~pyof.foundation.basic_types.BinaryData`):
138		The content of the packet in binary format.
139		"""
140	1	super().__init__()
141	1	self.destination = destination
142	1	self.source = source
143	1	self.vlan = vlan
144	1	self.ether_type = ether_type
145	1	self.data = data
146
147	1	def get_hash(self):
148		"""Calculate a hash and returns it.
149
150		Returns:
151		int: Integer value that identifies this instance.
152		"""
153		return hash(self.pack())
154
155	1	def unpack(self, buff, offset=0):
156		"""Unpack a binary message into this object's attributes.
157
158		Unpack the binary value buff and update this object attributes based
159		on the results.
160
161		Ethernet headers may have VLAN tags. If no VLAN tag is found, a
162		'wildcard VLAN tag' is inserted to assure correct unpacking.
163
164		Args:
165		buff (bytes): Binary data package to be unpacked.
166		offset (int): Where to begin unpacking.
167
168		Raises:
169		UnpackException: If there is a struct unpacking error.
170		"""
171		# Checking if the EtherType bytes are actually equal to VLAN_TPID -
172		# indicating that the packet is tagged. If it is not, we insert the
173		# equivalent to 'NULL VLAN data' (\x00\x00\x00\x00) to enable the
174		# correct unpacking process.
175	1	if buff[12:16] != VLAN_TPID.to_bytes(2, 'big'):
176	1	buff = buff[0:12] + b'\x00\x00\x00\x00' + buff[12:]
177
178	1	super().unpack(buff, offset)
179
180
181	1	class GenericTLV(GenericStruct):
182		"""TLV structure of LLDP packets.
183
184		This is a Type, Length and Value (TLV) struct.
185
186		The LLDP/TLV definition states that the Type field have 7 bits, while
187		the length have 9 bits. The Value must be between 0-511 octets.
188
189		Internally, on the instances of this class, the Type is a integer
190		(0-127) and the Length is dynamically calculated based on the current
191		type and value.
192		"""
193
194	1	def __init__(self, tlv_type=127, value=BinaryData()):
195		"""Create an instance and set its attributes.
196
197		Args:
198		tlv_type (int): Type used by this class. Defaults to 127.
199		value (:class:`~pyof.foundation.basic_types.BinaryData`):
200		Value stored by GenericTLV.
201		"""
202	1	super().__init__()
203	1	self.tlv_type = tlv_type
204	1	self._value = value
205
206	1	@property
207		def value(self):
208		"""Return the value stored by GenericTLV.
209
210		Returns:
211		:class:`~pyof.foundation.basic_types.BinaryData`:
212		Value stored by GenericTLV.
213		"""
214	1	return self._value
215
216	1	@property
217		def length(self):
218		"""Return the length of value stored by GenericTLV.
219
220		Returns:
221		int: Value length in bytes.
222		"""
223	1	return len(self.value.pack())
224
225	1	@property
226		def header(self):
227		"""Header of the TLV Packet.
228
229		The header is composed by the Type (7 bits) and Length (9 bits),
230		summing up 16 bits. To achieve that, we need to do some bitshift
231		operations.
232
233		Returns:
234		:class:`~pyof.foundation.basic_types.UBInt16`:
235		Result after all operations.
236		"""
237	1	return UBInt16(((self.tlv_type & 127) << 9) \| (self.length & 511))
238
239	1	def pack(self, value=None):
240		"""Pack the TLV in a binary representation.
241
242		Returns:
243		bytes: Binary representation of the struct object.
244
245		Raises:
246		:exc:`~.exceptions.ValidationError`: If validation fails.
247		"""
248	1	if value is None:
249	1	output = self.header.pack()
250	1	output += self.value.pack()
251	1	return output
252
253		elif isinstance(value, type(self)):
254		return value.pack()
255		else:
256		msg = "{} is not an instance of {}".format(value,
257		type(self).__name__)
258		raise PackException(msg)
259
260	1	def unpack(self, buffer, offset=0):
261		"""Unpack a binary message into this object's attributes.
262
263		Unpack the binary value buff and update this object attributes based
264		on the results.
265
266		Args:
267		buff (bytes): Binary data package to be unpacked.
268		offset (int): Where to begin unpacking.
269
270		Raises:
271		Exception: If there is a struct unpacking error.
272		"""
273	1	header = UBInt16()
274	1	header.unpack(buffer[offset:offset+2])
275	1	self.tlv_type = header.value >> 9
276	1	length = header.value & 511
277	1	begin, end = offset + 2, offset + 2 + length
278	1	self._value = BinaryData(buffer[begin:end])
279
280	1	def get_size(self, value=None):
281		"""Return struct size.
282
283		Returns:
284		int: Returns the struct size based on inner attributes.
285		"""
286		if isinstance(value, type(self)):
287		return value.get_size()
288
289		return 2 + self.length
290
291
292	1	class IPv4(GenericStruct):
293		"""IPv4 packet "struct".
294
295		Contains all fields of an IP version 4 packet header, plus the upper layer
296		content as binary data.
297		Some of the fields were merged together because of their size being
298		inferior to 8 bits. They are represented as a single class attribute, but
299		pack/unpack methods will take into account the values in individual
300		instance attributes.
301		"""
302
303		#: _version_ihl (:class:`UBInt8`): IP protocol version + Internet Header
304		#: Length (words)
305	1	_version_ihl = UBInt8()
306		#: _dscp_ecn (:class:`UBInt8`): Differentiated Services Code Point
307		#: (ToS - Type of Service) + Explicit Congestion Notification
308	1	_dscp_ecn = UBInt8()
309		#: length (:class:`UBInt16`): IP packet length (bytes)
310	1	length = UBInt16()
311		#: identification (:class:`UBInt16`): Packet ID - common to all fragments
312	1	identification = UBInt16()
313		#: _flags_offset (:class:`UBInt16`): Fragmentation flags + fragmentation
314		#: offset
315	1	_flags_offset = UBInt16()
316		#: ttl (:class:`UBInt8`): Packet time-to-live
317	1	ttl = UBInt8()
318		#: protocol (:class:`UBInt8`): Upper layer protocol number
319	1	protocol = UBInt8()
320		#: checksum (:class:`UBInt16`): Header checksum
321	1	checksum = UBInt16()
322		#: source (:class:`IPAddress`): Source IPv4 address
323	1	source = IPAddress()
324		#: destination (:class:`IPAddress`): Destination IPv4 address
325	1	destination = IPAddress()
326		#: options (:class:`BinaryData`): IP Options - up to 320 bits, always
327		#: padded to 32 bits
328	1	options = BinaryData()
329		#: data (:class:`BinaryData`): Packet data
330	1	data = BinaryData()
331
332	1	def __init__(self, version=4, ihl=5, dscp=0, ecn=0, length=0, # noqa
333		identification=0, flags=0, offset=0, ttl=255, protocol=0,
334		checksum=0, source="0.0.0.0", destination="0.0.0.0",
335		options=b'', data=b''):
336		"""The constructor receives the parameters below.
337
338		Args:
339		version (int): IP protocol version. Defaults to 4.
340		ihl (int): Internet Header Length. Default is 5.
341		dscp (int): Differentiated Service Code Point. Defaults to 0.
342		ecn (int): Explicit Congestion Notification. Defaults to 0.
343		length (int): IP packet length in bytes. Defaults to 0.
344		identification (int): Packet Id. Defaults to 0.
345		flags (int): IPv4 Flags. Defults 0.
346		offset (int): IPv4 offset. Defaults to 0.
347		ttl (int): Packet time-to-live. Defaults to 255
348		protocol (int): Upper layer protocol number. Defaults to 0.
349		checksum (int): Header checksum. Defaults to 0.
350		source (str): Source IPv4 address. Defaults to "0.0.0.0"
351		destination (str): Destination IPv4 address. Defaults to "0.0.0.0"
352		options (bytes): IP options. Defaults to empty bytes.
353		data (bytes): Packet data. Defaults to empty bytes.
354		"""
355	1	super().__init__()
356	1	self.version = version
357	1	self.ihl = ihl
358	1	self.dscp = dscp
359	1	self.ecn = ecn
360	1	self.length = length
361	1	self.identification = identification
362	1	self.flags = flags
363	1	self.offset = offset
364	1	self.ttl = ttl
365	1	self.protocol = protocol
366	1	self.checksum = checksum
367	1	self.source = source
368	1	self.destination = destination
369	1	self.options = options
370	1	self.data = data
371
372	1	def _update_checksum(self):
373		"""Update the packet checksum to enable integrity check."""
374	1	source_list = [int(octet) for octet in self.source.split(".")]
375	1	destination_list = [int(octet) for octet in
376		self.destination.split(".")]
377	1	source_upper = (source_list[0] << 8) + source_list[1]
378	1	source_lower = (source_list[2] << 8) + source_list[3]
379	1	destination_upper = (destination_list[0] << 8) + destination_list[1]
380	1	destination_lower = (destination_list[2] << 8) + destination_list[3]
381
382	1	block_sum = ((self._version_ihl << 8 \| self._dscp_ecn) + self.length +
383		self.identification + self._flags_offset +
384		(self.ttl << 8 \| self.protocol) + source_upper +
385		source_lower + destination_upper + destination_lower)
386
387	1	while block_sum > 65535:
388	1	carry = block_sum >> 16
389	1	block_sum = (block_sum & 65535) + carry
390
391	1	self.checksum = ~block_sum & 65535
392
393	1	def pack(self, value=None):
394		"""Pack the struct in a binary representation.
395
396		Merge some fields to ensure correct packing.
397
398		Returns:
399		bytes: Binary representation of this instance.
400		"""
401		# Set the correct IHL based on options size
402	1	if self.options:
403	1	self.ihl += int(len(self.options) / 4)
404
405		# Set the correct packet length based on header length and data
406	1	self.length = int(self.ihl * 4 + len(self.data))
407
408	1	self._version_ihl = self.version << 4 \| self.ihl
409	1	self._dscp_ecn = self.dscp << 2 \| self.ecn
410	1	self._flags_offset = self.flags << 13 \| self.offset
411
412		# Set the checksum field before packing
413	1	self._update_checksum()
414
415	1	return super().pack()
416
417	1	def unpack(self, buff, offset=0):
418		"""Unpack a binary struct into this object's attributes.
419
420		Return the values instead of the lib's basic types.
421
422		Args:
423		buff (bytes): Binary buffer.
424		offset (int): Where to begin unpacking.
425
426		Raises:
427		:exc:`~.exceptions.UnpackException`: If unpack fails.
428		"""
429	1	super().unpack(buff, offset)
430
431	1	self.version = self._version_ihl.value >> 4
432	1	self.ihl = self._version_ihl.value & 15
433	1	self.dscp = self._dscp_ecn.value >> 2
434	1	self.ecn = self._dscp_ecn.value & 3
435	1	self.length = self.length.value
436	1	self.identification = self.identification.value
437	1	self.flags = self._flags_offset.value >> 13
438	1	self.offset = self._flags_offset.value & 8191
439	1	self.ttl = self.ttl.value
440	1	self.protocol = self.protocol.value
441	1	self.checksum = self.checksum.value
442	1	self.source = self.source.value
443	1	self.destination = self.destination.value
444
445	1	if self.ihl > 5:
446	1	options_size = (self.ihl - 5) * 4
447	1	self.data = self.options.value[options_size:]
448	1	self.options = self.options.value[:options_size]
449		else:
450		self.data = self.options.value
451		self.options = b''
452
453
454	1	class TLVWithSubType(GenericTLV):
455		"""Modify the :class:`GenericTLV` to a Organization Specific TLV structure.
456
457		Beyond the standard TLV (type, length, value), we can also have a more
458		specific structure, with the :attr:`value` field being splitted into a
459		:attr:`sub_type` field and a new :attr:`sub_value` field.
460		"""
461
462	1	def __init__(self, tlv_type=1, sub_type=7, sub_value=BinaryData()):
463		"""Create an instance and set its attributes.
464
465		Args:
466		tlv_type (int): Type used by this class. Defaults to 1.
467		sub_type (int): Sub type value used by this class. Defaults to 7.
468		sub_value (:class:`~pyof.foundation.basic_types.BinaryData`):
469		Data stored by TLVWithSubType. Defaults to empty BinaryData.
470		"""
471	1	super().__init__(tlv_type)
472	1	self.sub_type = sub_type
473	1	self.sub_value = sub_value
474
475	1	@property
476		def value(self):
477		"""Return sub type and sub value as binary data.
478
479		Returns:
480		:class:`~pyof.foundation.basic_types.BinaryData`:
481		BinaryData calculated.
482		"""
483		binary = UBInt8(self.sub_type).pack() + self.sub_value.pack()
484		return BinaryData(binary)
485
486	1	def unpack(self, buffer, offset=0):
487		"""Unpack a binary message into this object's attributes.
488
489		Unpack the binary value buff and update this object attributes based
490		on the results.
491
492		Args:
493		buff (bytes): Binary data package to be unpacked.
494		offset (int): Where to begin unpacking.
495
496		Raises:
497		Exception: If there is a struct unpacking error.
498		"""
499		header = UBInt16()
500		header.unpack(buffer[offset:offset+2])
501		self.tlv_type = header.value >> 9
502		length = header.value & 511
503		begin, end = offset + 2, offset + 2 + length
504		sub_type = UBInt8()
505		sub_type.unpack(buffer[begin:begin+1])
506		self.sub_type = sub_type.value
507		self.sub_value = BinaryData(buffer[begin+1:end])
508
509
510	1	class LLDP(GenericStruct):
511		"""LLDP class.
512
513		Build a LLDP packet with TLVSubtype and Generic Subtypes.
514
515		It contains a chassis_id TLV, a port_id TLV, a TTL (Time to live) and
516		another TLV to represent the end of the LLDP Packet.
517		"""
518
519		#: chassis_id (:class:`~TLVWithSubType`) with tlv_type = 1 and sub_type = 7
520	1	chassis_id = TLVWithSubType(tlv_type=1, sub_type=7)
521		#: port_id (:class:`TLVWithSubType`) with tlv = 2 and sub_type = 7
522	1	port_id = TLVWithSubType(tlv_type=2, sub_type=7)
523		#: TTL (:class:`GenericTLV`) time is given in seconds, between 0 and 65535,
524		#: with tlv_type = 3
525	1	ttl = GenericTLV(tlv_type=3, value=UBInt16(120))
526		# We are not using list of tlvs for now
527		# tlvs = ListOfTLVs()
528		#: end (:class:`GenericTLV`) with tlv_type = 0
529		end = GenericTLV(tlv_type=0)
530

cemsbr / python-openflow

Push — master ( 9eabc4...ddb1e5 )

VLAN._validate() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like