ARP.__init__() - Code Metrics - Inspection of "Add ARP class" - kytos/python-openflow - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#433)

unknown

created 2017-08-01 20:18 UTC

ARP.init() B

↳ Parent: ARP

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	11
CRAP Score	1

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
c	1
b	0
f	0
dl	0
loc	31
ccs	11
cts	11
cp	1
rs	8.8571
cc	1
crap	1

"""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 (:class:`UBInt16`): The L2 (hardware) protocol type.
    htype = UBInt16()
    #: ptype (:class:`UBInt16`): The L3 protocol type.
    ptype = UBInt16()
    #: hlen (:class:`UBInt8`): Length of hardware address.
    hlen = UBInt8()
    #: plen (:class:`UBInt8`): Length of network protocol address.
    plen = UBInt8()
    #: operation (:class:`UBInt16`): Determines if it is an ARP request
    #: or reply.
    oper = UBInt16()
    #: sha (:class:`HWAddress`): Sender Hardware Address
    sha = HWAddress()
    #: spa (:class:`IPAddress`): Sender Protocol Address
    spa = IPAddress()
    #: tha (:class:`HWAddress`): Target Hardware Address
    tha = HWAddress()
    #: tpa (:class:`IPAddress`): Target Protocol Address
    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."""
        if self.htype != 1 or self.ptype != 0x800:
            return False
        return True

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

kytos / python-openflow

Pull Request — master (#433)

ARP.__init__() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

ARP.init() B