ARP.unpack() - Code Metrics - Inspection of "Merge pull request #433 from renanrodrigo/add_arp" - kytos/python-openflow - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( 95cf60...945c40 )

by Carlos Eduardo

created 2017-08-07 17:58 UTC

ARP.unpack() A

↳ Parent: ARP

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	4
CRAP Score	2

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
c	1
b	0
f	0
dl	0
loc	17
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__ = ('ARP', 'Ethernet', 'GenericTLV', 'IPv4', 'VLAN', 'TLVWithSubType',
           'LLDP')


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

    Contains fields for an ARP packet's header and data.
    Designed for Ethernet and IPv4 only: needs to have some attributes changed
    for other HTYPE and PTYPE implementations.
    Must be encapsulated inside an Ethernet frame.
    """

    htype = UBInt16()
    ptype = UBInt16()
    hlen = UBInt8()
    plen = UBInt8()
    oper = UBInt16()
    sha = HWAddress()
    spa = IPAddress()
    tha = HWAddress()
    tpa = IPAddress()

    def __init__(self, htype=1, ptype=0x800, hlen=6, plen=4, oper=1,
                 sha='00:00:00:00:00:00', spa='0.0.0.0',
                 tha="00:00:00:00:00:00", tpa='0.0.0.0'):
        """The constructor receives the parameters below.

        Args:
            htype (int): Hardware protocol type. Defaults to 1 for Ethernet.
            ptype (int): Network protocol type. Defaults to 0x800 for IPv4.
            hlen (int): Length of the hardware address. Defaults to 6 for MAC
                        addresses.
            plen (int): Length of the networking protocol address. Defaults to
                        4 for IPv4 addresses.
            oper (int): Determines the operation for this ARP packet. Must be 1
                        for ARP request or 2 for ARP reply. Defaults to 1.
            sha (str): Sender hardware address. Defaults to
                       '00:00:00:00:00:00'.
            spa (str): Sender protocol address. Defaults to '0.0.0.0'.
            tha (str): Target hardware address. Defaults to
                       '00:00:00:00:00:00'.
            tpa (str): Target protocol address. Defaults to '0.0.0.0'.
        """
        super().__init__()
        self.htype = htype
        self.ptype = ptype
        self.hlen = hlen
        self.plen = plen
        self.oper = oper
        self.sha = sha
        self.spa = spa
        self.tha = tha
        self.tpa = tpa

    def is_valid(self):
        """Assure the ARP contains Ethernet and IPv4 information."""
        return self.htype == 1 and self.ptype == 0x800

    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.
        Check if the protocols involved are Ethernet and IPv4. Other protocols
        are currently not supported.

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

        Raises:
            :exc:`~.exceptions.UnpackException`: If unpack fails.
        """
        super().unpack(buff, offset)
        if not self.is_valid():
            raise UnpackException("Unsupported protocols in ARP packet")


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

kytos / python-openflow

Push — master ( 95cf60...945c40 )

ARP.unpack() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like