| #!/usr/bin/env python | |
| # | |
| # Copyright (c) 2016, The OpenThread Authors. | |
| # All rights reserved. | |
| # | |
| # Redistribution and use in source and binary forms, with or without | |
| # modification, are permitted provided that the following conditions are met: | |
| # 1. Redistributions of source code must retain the above copyright | |
| # notice, this list of conditions and the following disclaimer. | |
| # 2. Redistributions in binary form must reproduce the above copyright | |
| # notice, this list of conditions and the following disclaimer in the | |
| # documentation and/or other materials provided with the distribution. | |
| # 3. Neither the name of the copyright holder nor the | |
| # names of its contributors may be used to endorse or promote products | |
| # derived from this software without specific prior written permission. | |
| # | |
| # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" | |
| # AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE | |
| # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE | |
| # ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE | |
| # LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR | |
| # CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF | |
| # SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS | |
| # INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN | |
| # CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) | |
| # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE | |
| # POSSIBILITY OF SUCH DAMAGE. | |
| # | |
| import abc | |
| import io | |
| import struct | |
| import sys | |
| from binascii import hexlify | |
| from ipaddress import ip_address | |
| try: | |
| from itertools import izip_longest as zip_longest | |
| except ImportError: | |
| from itertools import zip_longest | |
| # Next headers for IPv6 protocols | |
| IPV6_NEXT_HEADER_HOP_BY_HOP = 0 | |
| IPV6_NEXT_HEADER_TCP = 6 | |
| IPV6_NEXT_HEADER_UDP = 17 | |
| IPV6_NEXT_HEADER_ICMP = 58 | |
| UPPER_LAYER_PROTOCOLS = [ | |
| IPV6_NEXT_HEADER_TCP, | |
| IPV6_NEXT_HEADER_UDP, | |
| IPV6_NEXT_HEADER_ICMP, | |
| ] | |
| # ICMP Protocol codes | |
| ICMP_DESTINATION_UNREACHABLE = 1 | |
| ICMP_ECHO_REQUEST = 128 | |
| ICMP_ECHO_RESPONSE = 129 | |
| # Default hop limit for IPv6 | |
| HOP_LIMIT_DEFAULT = 64 | |
| def calculate_checksum(data): | |
| """ Calculate checksum from data bytes. | |
| How to calculate checksum (RFC 2460): | |
| https://tools.ietf.org/html/rfc2460#page-27 | |
| Args: | |
| data (bytes): input data from which checksum will be calculated | |
| Returns: | |
| int: calculated checksum | |
| """ | |
| # Create halfwords from data bytes. Example: data[0] = 0x01, data[1] = 0xb2 => 0x01b2 | |
| halfwords = [((byte0 << 8) | byte1) for byte0, byte1 in zip_longest(data[::2], data[1::2], fillvalue=0x00)] | |
| checksum = 0 | |
| for halfword in halfwords: | |
| checksum += halfword | |
| checksum = (checksum & 0xFFFF) + (checksum >> 16) | |
| checksum ^= 0xFFFF | |
| if checksum == 0: | |
| return 0xFFFF | |
| else: | |
| return checksum | |
| class PacketFactory(object): | |
| """ Interface for classes that produce objects from data. """ | |
| def parse(self, data, message_info): | |
| """ Convert data to object. | |
| Args: | |
| data (BytesIO) | |
| message_info (MessageInfo) | |
| """ | |
| raise NotImplementedError | |
| class BuildableFromBytes(object): | |
| """ Interface for classes which can be built from bytes. """ | |
| @classmethod | |
| def from_bytes(cls, data): | |
| """ Convert data to object. | |
| Args: | |
| data (bytes) | |
| """ | |
| raise NotImplementedError | |
| class ConvertibleToBytes(object): | |
| """ Interface for classes which can be converted to bytes. """ | |
| def to_bytes(self): | |
| """ Convert object to data. | |
| Returns: | |
| bytes | |
| """ | |
| raise NotImplementedError | |
| def __len__(self): | |
| """ Length of data (in bytes). | |
| Returns: | |
| int | |
| """ | |
| raise NotImplementedError | |
| class Header(object): | |
| """ Interface for header classes. """ | |
| __metaclass__ = abc.ABCMeta | |
| @abc.abstractproperty | |
| def type(self): | |
| """ Number which can be used in the next header field in IPv6 header or next headers. | |
| Returns: | |
| int | |
| """ | |
| class ExtensionHeader(object): | |
| """ Base for classes representing Extension Headers in IPv6 packets. """ | |
| def __init__(self, next_header, hdr_ext_len=0): | |
| self.next_header = next_header | |
| self.hdr_ext_len = hdr_ext_len | |
| class UpperLayerProtocol(Header, ConvertibleToBytes): | |
| """ Base for classes representing upper layer protocol payload in IPv6 packets. """ | |
| def __init__(self, header): | |
| self.header = header | |
| @property | |
| def checksum(self): | |
| """ Return checksum from upper layer protocol header. """ | |
| return self.header.checksum | |
| @checksum.setter | |
| def checksum(self, value): | |
| """ Set checksum value in upper layer protocol header. """ | |
| self.header.checksum = value | |
| def is_valid_checksum(self): | |
| """ Return information if set checksum is valid. | |
| It is not possible to get zero from checksum calculation. | |
| Zero indicates invalid checksum value. | |
| Returns: | |
| bool | |
| """ | |
| return self.checksum != 0 | |
| class IPv6PseudoHeader(ConvertibleToBytes): | |
| """ Class representing IPv6 pseudo header which is required to calculate | |
| upper layer protocol (like e.g. UDP or ICMPv6) checksum. | |
| This class is used only during upper layer protocol checksum calculation. Do not use it outside of this module. | |
| """ | |
| def __init__(self, source_address, destination_address, payload_length, next_header): | |
| self._source_address = self._convert_to_ipaddress(source_address) | |
| self._destination_address = self._convert_to_ipaddress(destination_address) | |
| self.payload_length = payload_length | |
| self.next_header = next_header | |
| def _convert_to_ipaddress(self, value): | |
| if isinstance(value, bytearray): | |
| value = bytes(value) | |
| elif isinstance(value, str) and sys.version_info[0] == 2: | |
| value = value.decode("utf-8") | |
| return ip_address(value) | |
| @property | |
| def source_address(self): | |
| return self._source_address | |
| @source_address.setter | |
| def source_address(self, value): | |
| self._source_address = self._convert_to_ipaddress(value) | |
| @property | |
| def destination_address(self): | |
| return self._destination_address | |
| @destination_address.setter | |
| def destination_address(self, value): | |
| self._source_address = self._convert_to_ipaddress(value) | |
| def to_bytes(self): | |
| data = bytearray() | |
| data += self.source_address.packed | |
| data += self.destination_address.packed | |
| data += struct.pack(">I", self.payload_length) | |
| data += struct.pack(">I", self.next_header) | |
| return data | |
| class IPv6Header(ConvertibleToBytes, BuildableFromBytes): | |
| """ Class representing IPv6 packet header. """ | |
| _version = 6 | |
| _header_length = 40 | |
| def __init__(self, source_address, destination_address, traffic_class=0, flow_label=0, hop_limit=64, | |
| payload_length=0, next_header=0): | |
| self.version = self._version | |
| self._source_address = self._convert_to_ipaddress(source_address) | |
| self._destination_address = self._convert_to_ipaddress(destination_address) | |
| self.traffic_class = traffic_class | |
| self.flow_label = flow_label | |
| self.hop_limit = hop_limit | |
| self.payload_length = payload_length | |
| self.next_header = next_header | |
| def _convert_to_ipaddress(self, value): | |
| if isinstance(value, bytearray): | |
| value = bytes(value) | |
| elif isinstance(value, str) and sys.version_info[0] == 2: | |
| value = value.decode("utf-8") | |
| return ip_address(value) | |
| @property | |
| def source_address(self): | |
| return self._source_address | |
| @source_address.setter | |
| def source_address(self, value): | |
| self._source_address = self._convert_to_ipaddress(value) | |
| @property | |
| def destination_address(self): | |
| return self._destination_address | |
| def to_bytes(self): | |
| data = bytearray([ | |
| ((self.version & 0x0F) << 4) | ((self.traffic_class >> 4) & 0x0F), | |
| ((self.traffic_class & 0x0F) << 4) | ((self.flow_label >> 16) & 0x0F), | |
| ((self.flow_label >> 8) & 0xFF), | |
| ((self.flow_label & 0xFF)) | |
| ]) | |
| data += struct.pack(">H", self.payload_length) | |
| data += bytearray([self.next_header, self.hop_limit]) | |
| data += self.source_address.packed | |
| data += self.destination_address.packed | |
| return data | |
| @classmethod | |
| def from_bytes(cls, data): | |
| b = bytearray(data.read(4)) | |
| version = (b[0] >> 4) & 0x0F | |
| traffic_class = ((b[0] & 0x0F) << 4) | ((b[1] >> 4) & 0x0F) | |
| flow_label = ((b[1] & 0x0F) << 16) | (b[2] << 8) | b[3] | |
| payload_length = struct.unpack(">H", data.read(2))[0] | |
| next_header = ord(data.read(1)) | |
| hop_limit = ord(data.read(1)) | |
| src_addr = bytearray(data.read(16)) | |
| dst_addr = bytearray(data.read(16)) | |
| return cls(src_addr, | |
| dst_addr, | |
| traffic_class, | |
| flow_label, | |
| hop_limit, | |
| payload_length, | |
| next_header) | |
| def __repr__(self): | |
| return "IPv6Header(source_address={}, destination_address={}, next_header={}, payload_length={}, \ | |
| hop_limit={}, traffic_class={}, flow_label={})".format(self.source_address.compressed, | |
| self.destination_address.compressed, | |
| self.next_header, | |
| self.payload_length, | |
| self.hop_limit, | |
| self.traffic_class, | |
| self.flow_label) | |
| def __len__(self): | |
| return self._header_length | |
| class IPv6Packet(ConvertibleToBytes): | |
| """ Class representing IPv6 packet. | |
| IPv6 packet consists of IPv6 header, optional extension header, and upper layer protocol. | |
| IPv6 packet | |
| +-------------+----------------------------------+----------------------------------------------+ | |
| | | | | | |
| | IPv6 header | extension headers (zero or more) | upper layer protocol (e.g. UDP, TCP, ICMPv6) | | |
| | | | | | |
| +-------------+----------------------------------+----------------------------------------------+ | |
| Extension headers: | |
| - HopByHop | |
| - Routing header (not implemented in this module) | |
| Upper layer protocols: | |
| - ICMPv6 | |
| - UDP | |
| - TCP (not implemented in this module) | |
| Example: | |
| IPv6 packet construction without extension headers: | |
| ipv6_packet = IPv6Packet(IPv6Header("fd00:1234:4555::ff:fe00:1800", "ff03::1"), | |
| ICMPv6(ICMPv6Header(128, 0), | |
| ICMPv6EchoBody(0, 2, bytes([0x80, 0x00, 0xc7, 0xbf, 0x00, 0x00, 0x00, 0x01, | |
| 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, | |
| 0x41, 0x41])))) | |
| IPv6 packet construction with extension headers: | |
| ipv6_packet = IPv6Packet(IPv6Header("fd00:1234:4555::ff:fe00:1800", "ff03::1"), | |
| ICMPv6(ICMPv6Header(128, 0), | |
| ICMPv6EchoBody(0, 2, bytes([0x80, 0x00, 0xc7, 0xbf, 0x00, 0x00, 0x00, 0x01, | |
| 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, | |
| 0x41, 0x41])), | |
| [HopByHop(options=[ | |
| HopByHopOption(HopByHopOptionHeader(_type=0x6d), | |
| MPLOption(S=1, M=0, V=0, sequence=2, seed_id=bytes([0x00, 0x18]))) | |
| ])]) | |
| """ | |
| def __init__(self, ipv6_header, upper_layer_protocol, extension_headers=None): | |
| self.ipv6_header = ipv6_header | |
| self.upper_layer_protocol = upper_layer_protocol | |
| self.extension_headers = extension_headers if extension_headers is not None else [] | |
| self._update_next_header_values_in_headers() | |
| if not upper_layer_protocol.is_valid_checksum(): | |
| self.upper_layer_protocol.checksum = self.calculate_checksum() | |
| def _validate_checksum(self): | |
| checksum = self.calculate_checksum() | |
| if self.upper_layer_protocol.checksum != checksum: | |
| raise RuntimeError("Could not create IPv6 packet. " | |
| "Invalid checksum: {}!={}".format(self.upper_layer_protocol.checksum, checksum)) | |
| self.upper_layer_protocol.checksum = checksum | |
| def _update_payload_length_value_in_ipv6_header(self): | |
| self.ipv6_header.payload_length = len(self.upper_layer_protocol) + \ | |
| sum([len(extension_header) for extension_header in self.extension_headers]) | |
| def _update_next_header_values_in_headers(self): | |
| last_header = self.ipv6_header | |
| for extension_header in self.extension_headers: | |
| last_header.next_header = extension_header.type | |
| last_header = extension_header | |
| last_header.next_header = self.upper_layer_protocol.type | |
| def calculate_checksum(self): | |
| saved_checksum = self.upper_layer_protocol.checksum | |
| self.upper_layer_protocol.checksum = 0 | |
| upper_layer_protocol_bytes = self.upper_layer_protocol.to_bytes() | |
| self.upper_layer_protocol.checksum = saved_checksum | |
| pseudo_header = IPv6PseudoHeader(self.ipv6_header.source_address, | |
| self.ipv6_header.destination_address, | |
| len(upper_layer_protocol_bytes), | |
| self.upper_layer_protocol.type) | |
| return calculate_checksum(pseudo_header.to_bytes() + upper_layer_protocol_bytes) | |
| def to_bytes(self): | |
| self._update_payload_length_value_in_ipv6_header() | |
| self._update_next_header_values_in_headers() | |
| self.upper_layer_protocol.checksum = self.calculate_checksum() | |
| ipv6_packet = self.ipv6_header.to_bytes() | |
| for extension_header in self.extension_headers: | |
| ipv6_packet += extension_header.to_bytes() | |
| ipv6_packet += self.upper_layer_protocol.to_bytes() | |
| return ipv6_packet | |
| def __repr__(self): | |
| return "IPv6Packet(header={}, upper_layer_protocol={})".format(self.ipv6_header, self.upper_layer_protocol) | |
| class UDPHeader(ConvertibleToBytes, BuildableFromBytes): | |
| """ Class representing UDP datagram header. | |
| This header is required to construct UDP datagram. | |
| """ | |
| _header_length = 8 | |
| def __init__(self, src_port, dst_port, payload_length=0, checksum=0): | |
| self.src_port = src_port | |
| self.dst_port = dst_port | |
| self._payload_length = payload_length | |
| self.checksum = checksum | |
| @property | |
| def type(self): | |
| return 17 | |
| @property | |
| def payload_length(self): | |
| return self._payload_length | |
| @payload_length.setter | |
| def payload_length(self, value): | |
| self._payload_length = self._header_length + value | |
| def to_bytes(self): | |
| data = struct.pack(">H", self.src_port) | |
| data += struct.pack(">H", self.dst_port) | |
| data += struct.pack(">H", self.payload_length) | |
| data += struct.pack(">H", self.checksum) | |
| return data | |
| @classmethod | |
| def from_bytes(cls, data): | |
| src_port = struct.unpack(">H", data.read(2))[0] | |
| dst_port = struct.unpack(">H", data.read(2))[0] | |
| payload_length = struct.unpack(">H", data.read(2))[0] | |
| checksum = struct.unpack(">H", data.read(2))[0] | |
| return cls(src_port, dst_port, payload_length, checksum) | |
| def __len__(self): | |
| return self._header_length | |
| class UDPDatagram(UpperLayerProtocol): | |
| """ Class representing UDP datagram. | |
| UDP is an upper layer protocol for IPv6 so it can be passed to IPv6 packet as upper_layer_protocol. | |
| This class consists of a UDP header and payload. The example below shows how a UDP datagram can be constructed. | |
| Example: | |
| udp_dgram = UDPDatagram(UDPHeader(src_port=19788, dst_port=19788), | |
| bytes([0x00, 0x15, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, | |
| 0x00, 0x00, 0x01, 0x09, 0x01, 0x01, 0x0b, 0x03, | |
| 0x04, 0xc6, 0x69, 0x73, 0x51, 0x0e, 0x01, 0x80, | |
| 0x12, 0x02, 0x00, 0x01, 0xde, 0xad, 0xbe, 0xef])) | |
| """ | |
| @property | |
| def type(self): | |
| return 17 | |
| def __init__(self, header, payload): | |
| super(UDPDatagram, self).__init__(header) | |
| self.payload = payload | |
| def to_bytes(self): | |
| self.header.payload_length = len(self.payload) | |
| data = bytearray() | |
| data += self.header.to_bytes() | |
| data += self.payload.to_bytes() | |
| return data | |
| def __len__(self): | |
| return len(self.header) + len(self.payload) | |
| class ICMPv6Header(ConvertibleToBytes, BuildableFromBytes): | |
| """ Class representing ICMPv6 message header. | |
| This header is required to construct ICMPv6 message. | |
| """ | |
| _header_length = 4 | |
| def __init__(self, _type, code, checksum=0): | |
| self.type = _type | |
| self.code = code | |
| self.checksum = checksum | |
| def to_bytes(self): | |
| return bytearray([self.type, self.code]) + struct.pack(">H", self.checksum) | |
| @classmethod | |
| def from_bytes(cls, data): | |
| _type = ord(data.read(1)) | |
| code = ord(data.read(1)) | |
| checksum = struct.unpack(">H", data.read(2))[0] | |
| return cls(_type, code, checksum) | |
| def __len__(self): | |
| return self._header_length | |
| class ICMPv6(UpperLayerProtocol): | |
| """ Class representing ICMPv6 message. | |
| ICMPv6 is an upper layer protocol for IPv6 so it can be passed to IPv6 packet as upper_layer_protocol. | |
| This class consists of an ICMPv6 header and body. The example below shows how an ICMPv6 message can be constructed. | |
| Example: | |
| icmpv6_msg = ICMPv6(ICMPv6Header(128, 0), | |
| ICMPv6EchoBody(0, 2, bytes([0x80, 0x00, 0xc7, 0xbf, 0x00, 0x00, 0x00, 0x01, | |
| 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, | |
| 0x41, 0x41]))) | |
| """ | |
| @property | |
| def type(self): | |
| return 58 | |
| def __init__(self, header, body): | |
| super(ICMPv6, self).__init__(header) | |
| self.body = body | |
| def to_bytes(self): | |
| return bytearray(self.header.to_bytes() + self.body.to_bytes()) | |
| def __len__(self): | |
| return len(self.header) + len(self.body) | |
| class HopByHop(ExtensionHeader): | |
| """ Class representing HopByHop extension header. | |
| HopByHop extension header consists of: | |
| - next_header type | |
| - extension header length which is multiple of 8 | |
| - options | |
| """ | |
| _one_byte_padding = 0x00 | |
| _many_bytes_padding = 0x01 | |
| @property | |
| def type(self): | |
| return 0 | |
| def __init__(self, next_header=None, options=None, hdr_ext_len=None): | |
| super(HopByHop, self).__init__(next_header, hdr_ext_len) | |
| self.options = options if options is not None else [] | |
| if hdr_ext_len is not None: | |
| self.hdr_ext_len = hdr_ext_len | |
| else: | |
| payload_length = self._calculate_payload_length() | |
| self.hdr_ext_len = self._calculate_hdr_ext_len(payload_length) | |
| def _calculate_payload_length(self): | |
| payload_length = 2 | |
| for option in self.options: | |
| payload_length += len(option) | |
| return payload_length | |
| def _calculate_hdr_ext_len(self, payload_length): | |
| count = payload_length >> 3 | |
| if (payload_length & 0x7) == 0 and count > 0: | |
| return count - 1 | |
| return count | |
| def to_bytes(self): | |
| data = bytearray([self.next_header, self.hdr_ext_len]) | |
| for option in self.options: | |
| data += option.to_bytes() | |
| # Padding | |
| # | |
| # More details: | |
| # https://tools.ietf.org/html/rfc2460#section-4.2 | |
| # | |
| excess_bytes = len(data) & 0x7 | |
| if excess_bytes > 0: | |
| padding_length = 8 - excess_bytes | |
| if padding_length == 1: | |
| data += bytearray([self._one_byte_padding]) | |
| else: | |
| padding_length -= 2 | |
| data += bytearray([self._many_bytes_padding, padding_length]) | |
| data += bytearray([0x00 for _ in range(padding_length)]) | |
| return data | |
| def __len__(self): | |
| """ HopByHop extension header length | |
| More details: | |
| https://tools.ietf.org/html/rfc2460#section-4.3 | |
| """ | |
| return (self.hdr_ext_len + 1) * 8 | |
| class HopByHopOptionHeader(ConvertibleToBytes, BuildableFromBytes): | |
| """ Class representing HopByHop option header. """ | |
| _header_length = 2 | |
| def __init__(self, _type, length=None): | |
| self.type = _type | |
| self.length = length if length is not None else 0 | |
| def to_bytes(self): | |
| return bytearray([self.type, self.length]) | |
| @classmethod | |
| def from_bytes(cls, data): | |
| _type = ord(data.read(1)) | |
| length = ord(data.read(1)) | |
| return cls(_type, length) | |
| def __len__(self): | |
| return self._header_length | |
| def __repr__(self): | |
| return "HopByHopOptionHeader(type={}, length={})".format(self.type, self.length) | |
| class HopByHopOption(ConvertibleToBytes): | |
| """ Class representing HopByHop option. | |
| Class consists of two elements: HopByHopOptionHeader and value (e.g. for MPLOption). | |
| The following example shows how any HopByHop option can be constructed. | |
| Example: | |
| HopByHop(next_header=0x3a, | |
| options=[HopByHopOption(HopByHopOptionHeader(_type=0x6d), | |
| MPLOption(S=1, M=0, V=0, sequence=2, seed_id=bytes([0x00, 0x18]))) | |
| """ | |
| def __init__(self, header, value): | |
| self.value = value | |
| self.header = header | |
| self.header.length = len(self.value) | |
| def to_bytes(self): | |
| return self.header.to_bytes() + self.value.to_bytes() | |
| def __len__(self): | |
| return len(self.header) + len(self.value) | |
| def __repr__(self): | |
| return "HopByHopOption(header={}, value={})".format(self.header, self.value) | |
| class MPLOption(ConvertibleToBytes): | |
| """ Class representing MPL option. """ | |
| _header_length = 2 | |
| _seed_id_length = { | |
| 0: 0, | |
| 1: 2, | |
| 2: 8, | |
| 3: 16 | |
| } | |
| def __init__(self, S, M, V, sequence, seed_id): | |
| self.S = S | |
| self.M = M | |
| self.V = V | |
| self.sequence = sequence | |
| self.seed_id = seed_id | |
| def to_bytes(self): | |
| smv = ((self.S & 0x03) << 6) | ((self.M & 0x01) << 5) | ((self.V & 0x01) << 4) | |
| return bytearray([smv, self.sequence]) + self.seed_id | |
| @classmethod | |
| def from_bytes(cls, data): | |
| b = ord(data.read(1)) | |
| s = ((b >> 6) & 0x03) | |
| m = ((b >> 5) & 0x01) | |
| v = ((b >> 4) & 0x01) | |
| sequence = ord(data.read(1)) | |
| seed_id = data.read(cls._seed_id_length[s]) | |
| return cls(s, m, v, sequence, seed_id) | |
| def __len__(self): | |
| return self._header_length + self._seed_id_length[self.S] | |
| def __repr__(self): | |
| return "MPLOption(S={}, M={}, V={}, sequence={}, seed_id={})".format(self.S, self.M, self.V, self.sequence, hexlify(self.seed_id)) | |
| class IPv6PacketFactory(PacketFactory): | |
| """ Factory that produces IPv6 packets from data. | |
| This factory must be initialized with factories which allow to parse extension headers and upper layer protocols. | |
| The following example shows preferable setup of IPv6PacketFactory. | |
| Header types: | |
| 0: HopByHop | |
| 17: UDP | |
| 58: ICMPv6 | |
| Option types: | |
| 109: MPL | |
| ICMPv6 body types: | |
| 128: Echo request | |
| 129: Echo response | |
| Example usage: | |
| ipv6_factory = IPv6PacketFactory( | |
| ehf={ | |
| 0: HopByHopFactory(options_factories={ | |
| 109: MPLOptionFactory() | |
| }) | |
| }, | |
| ulpf={ | |
| 17: UDPDatagramFactory(dst_port_factories={ | |
| 19788: MLEMessageFactory(), | |
| 19789: CoAPMessageFactory() | |
| }), | |
| 58: ICMPv6Factory(body_factories={ | |
| 128: ICMPv6EchoBodyFactory(), | |
| 129: ICMPv6EchoBodyFactory() | |
| }) | |
| } | |
| ) | |
| """ | |
| def __init__(self, ehf=None, ulpf=None): | |
| """ | |
| ehf - Extension Header Factory | |
| ulpf - Upper Layer Protocol Factory | |
| Args: | |
| ehf(dict[int: PacketFactory]): Dictionary mapping extension header types on specialized factories. | |
| ulpf(dict[int: PacketFactory]): Dictionary mapping upper layer protocol types on specialized factories. | |
| """ | |
| self._ehf = ehf if ehf is not None else {} | |
| self._ulpf = ulpf if ulpf is not None else {} | |
| def _is_extension_header(self, header_type): | |
| return not header_type in UPPER_LAYER_PROTOCOLS | |
| def _get_extension_header_factory_for(self, next_header): | |
| try: | |
| return self._ehf[next_header] | |
| except KeyError: | |
| raise RuntimeError("Could not get Extension Header factory for next_header={}.".format(next_header)) | |
| def _get_upper_layer_protocol_factory_for(self, next_header): | |
| try: | |
| return self._ulpf[next_header] | |
| except KeyError: | |
| raise RuntimeError("Could not get Upper Layer Protocol factory for next_header={}.".format(next_header)) | |
| def _parse_extension_headers(self, data, next_header, message_info): | |
| extension_headers = [] | |
| while self._is_extension_header(next_header): | |
| factory = self._get_extension_header_factory_for(next_header) | |
| extension_header = factory.parse(data, message_info) | |
| next_header = extension_header.next_header | |
| extension_headers.append(extension_header) | |
| return next_header, extension_headers | |
| def _parse_upper_layer_protocol(self, data, next_header, message_info): | |
| factory = self._get_upper_layer_protocol_factory_for(next_header) | |
| return factory.parse(data, message_info) | |
| def parse(self, data, message_info): | |
| ipv6_header = IPv6Header.from_bytes(data) | |
| message_info.source_ipv6 = ipv6_header.source_address | |
| message_info.destination_ipv6 = ipv6_header.destination_address | |
| next_header, extension_headers = self._parse_extension_headers(data, ipv6_header.next_header, message_info) | |
| upper_layer_protocol = self._parse_upper_layer_protocol(data, next_header, message_info) | |
| return IPv6Packet(ipv6_header, upper_layer_protocol, extension_headers) | |
| class HopByHopOptionsFactory(object): | |
| """ Factory that produces HopByHop options. """ | |
| _one_byte_padding = 0x00 | |
| _many_bytes_padding = 0x01 | |
| def __init__(self, options_factories=None): | |
| self._options_factories = options_factories if options_factories is not None else {} | |
| def _get_HopByHopOption_value_factory(self, _type): | |
| try: | |
| return self._options_factories[_type] | |
| except KeyError: | |
| raise RuntimeError("Could not find HopByHopOption value factory for type={}.".format(_type)) | |
| def parse(self, data, message_info): | |
| options = [] | |
| while data.tell() < len(data.getvalue()): | |
| option_header = HopByHopOptionHeader.from_bytes(data) | |
| if option_header.type == self._one_byte_padding: | |
| # skip one byte padding | |
| data.read(1) | |
| elif option_header.type == self._many_bytes_padding: | |
| # skip n bytes padding | |
| data.read(option_header.length) | |
| else: | |
| factory = self._get_HopByHopOption_value_factory(option_header.type) | |
| option_data = data.read(option_header.length) | |
| option = HopByHopOption(option_header, factory.parse(io.BytesIO(option_data), message_info)) | |
| options.append(option) | |
| return options | |
| class HopByHopFactory(PacketFactory): | |
| """ Factory that produces HopByHop extension headers from data. """ | |
| def __init__(self, hop_by_hop_options_factory): | |
| self._hop_by_hop_options_factory = hop_by_hop_options_factory | |
| def _calculate_extension_header_length(self, hdr_ext_len): | |
| return (hdr_ext_len + 1) * 8 | |
| def parse(self, data, message_info): | |
| next_header = ord(data.read(1)) | |
| hdr_ext_len = ord(data.read(1)) | |
| # Note! Two bytes were read (next_header and hdr_ext_len) so they must be substracted from header length | |
| hop_by_hop_length = self._calculate_extension_header_length(hdr_ext_len) - 2 | |
| hop_by_hop_data = data.read(hop_by_hop_length) | |
| options = self._hop_by_hop_options_factory.parse(io.BytesIO(hop_by_hop_data), message_info) | |
| hop_by_hop = HopByHop(next_header, options, hdr_ext_len) | |
| message_info.payload_length += len(hop_by_hop) | |
| return hop_by_hop | |
| class MPLOptionFactory(PacketFactory): | |
| """ Factory that produces MPL options for HopByHop extension header. """ | |
| def parse(self, data, message_info): | |
| return MPLOption.from_bytes(data) | |
| class UDPHeaderFactory: | |
| """ Factory that produces UDP header. """ | |
| def parse(self, data, message_info): | |
| return UDPHeader.from_bytes(data) | |
| class UdpBasedOnSrcDstPortsPayloadFactory: | |
| # TODO: Unittests | |
| """ Factory that produces UDP payload. """ | |
| def __init__(self, src_dst_port_based_payload_factories): | |
| """ | |
| Args: | |
| src_dst_port_based_payload_factories (PacketFactory): Factories parse UDP payload based on source or destination port. | |
| """ | |
| self._factories = src_dst_port_based_payload_factories | |
| def parse(self, data, message_info): | |
| factory = None | |
| if message_info.dst_port in self._factories: | |
| factory = self._factories[message_info.dst_port] | |
| if message_info.src_port in self._factories: | |
| factory = self._factories[message_info.src_port] | |
| if factory is None: | |
| raise RuntimeError("Could not find factory to build UDP payload.") | |
| return factory.parse(data, message_info) | |
| class UDPDatagramFactory(PacketFactory): | |
| # TODO: Unittests | |
| """ Factory that produces UDP datagrams. """ | |
| def __init__(self, udp_header_factory, udp_payload_factory): | |
| self._udp_header_factory = udp_header_factory | |
| self._udp_payload_factory = udp_payload_factory | |
| def parse(self, data, message_info): | |
| header = self._udp_header_factory.parse(data, message_info) | |
| # Update message payload length: UDP header (8B) + payload length | |
| message_info.payload_length += len(header) + (len(data.getvalue()) - data.tell()) | |
| message_info.src_port = header.src_port | |
| message_info.dst_port = header.dst_port | |
| payload = self._udp_payload_factory.parse(data, message_info) | |
| return UDPDatagram(header, payload) | |
| class ICMPv6Factory(PacketFactory): | |
| """ Factory that produces ICMPv6 messages from data. """ | |
| def __init__(self, body_factories=None): | |
| self._body_factories = body_factories if body_factories is not None else {} | |
| def _get_icmpv6_body_factory(self, _type): | |
| try: | |
| return self._body_factories[_type] | |
| except KeyError: | |
| if "default" not in self._body_factories: | |
| raise RuntimeError("Could not find specialized factory to parse ICMP body. " | |
| "Unsupported ICMP type: {}".format(_type)) | |
| default_factory = self._body_factories["default"] | |
| print("Could not find specialized factory to parse ICMP body. " | |
| "Take the default one: {}".format(type(default_factory))) | |
| return default_factory | |
| def parse(self, data, message_info): | |
| header = ICMPv6Header.from_bytes(data) | |
| factory = self._get_icmpv6_body_factory(header.type) | |
| message_info.payload_length += len(header) + (len(data.getvalue()) - data.tell()) | |
| return ICMPv6(header, factory.parse(data, message_info)) | |
| class ICMPv6EchoBodyFactory(PacketFactory): | |
| """ Factory that produces ICMPv6 echo message body. """ | |
| def parse(self, data, message_info): | |
| return ICMPv6EchoBody.from_bytes(data) | |
| class BytesPayload(ConvertibleToBytes, BuildableFromBytes): | |
| """ Class representing bytes payload. """ | |
| def __init__(self, data): | |
| self.data = data | |
| def to_bytes(self): | |
| return bytearray(self.data) | |
| @classmethod | |
| def from_bytes(cls, data): | |
| return cls(data) | |
| def __len__(self): | |
| return len(self.data) | |
| class BytesPayloadFactory(PacketFactory): | |
| """ Factory that produces bytes payload. """ | |
| def parse(self, data, message_info): | |
| return BytesPayload(data.read()) | |
| class ICMPv6EchoBody(ConvertibleToBytes, BuildableFromBytes): | |
| """ Class representing body of ICMPv6 echo messages. """ | |
| _header_length = 4 | |
| def __init__(self, identifier, sequence_number, data): | |
| self.identifier = identifier | |
| self.sequence_number = sequence_number | |
| self.data = data | |
| def to_bytes(self): | |
| data = struct.pack(">H", self.identifier) | |
| data += struct.pack(">H", self.sequence_number) | |
| data += self.data | |
| return data | |
| @classmethod | |
| def from_bytes(cls, data): | |
| identifier = struct.unpack(">H", data.read(2))[0] | |
| sequence_number = struct.unpack(">H", data.read(2))[0] | |
| return cls(identifier, sequence_number, data.read()) | |
| def __len__(self): | |
| return self._header_length + len(self.data) | |
| class ICMPv6DestinationUnreachableFactory(PacketFactory): | |
| """ Factory that produces ICMPv6 echo message body. """ | |
| def parse(self, data, message_info): | |
| return ICMPv6DestinationUnreachable.from_bytes(data) | |
| class ICMPv6DestinationUnreachable(ConvertibleToBytes, BuildableFromBytes): | |
| """ Class representing body of ICMPv6 Destination Unreachable messages. """ | |
| _header_length = 4 | |
| _unused = 0 | |
| def __init__(self, data): | |
| self.data = data | |
| def to_bytes(self): | |
| data = bytearray(struct.pack(">I", self._unused)) | |
| data += self.data | |
| return data | |
| @classmethod | |
| def from_bytes(cls, data): | |
| unused = struct.unpack(">I", data.read(4))[0] | |
| if unused != 0: | |
| raise RuntimeError( | |
| "Invalid value of unused field in the ICMPv6 Destination Unreachable data. Expected value: 0.") | |
| return cls(bytearray(data.read())) | |
| def __len__(self): | |
| return self._header_length + len(self.data) |