| #!/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 io |
| import ipaddress |
| import struct |
| |
| from binascii import hexlify |
| |
| import common |
| import network_data |
| |
| from enum import IntEnum |
| |
| |
| class CommandType(IntEnum): |
| LINK_REQUEST = 0 |
| LINK_ACCEPT = 1 |
| LINK_ACCEPT_AND_REQUEST = 2 |
| LINK_REJECT = 3 |
| ADVERTISEMENT = 4 |
| UPDATE = 5 |
| UPDATE_REQUEST = 6 |
| DATA_REQUEST = 7 |
| DATA_RESPONSE = 8 |
| PARENT_REQUEST = 9 |
| PARENT_RESPONSE = 10 |
| CHILD_ID_REQUEST = 11 |
| CHILD_ID_RESPONSE = 12 |
| CHILD_UPDATE_REQUEST = 13 |
| CHILD_UPDATE_RESPONSE = 14 |
| ANNOUNCE = 15 |
| DISCOVERY_REQUEST = 16 |
| DISCOVERY_RESPONSE = 17 |
| |
| |
| class TlvType(IntEnum): |
| SOURCE_ADDRESS = 0 |
| MODE = 1 |
| TIMEOUT = 2 |
| CHALLENGE = 3 |
| RESPONSE = 4 |
| LINK_LAYER_FRAME_COUNTER = 5 |
| MLE_FRAME_COUNTER = 8 |
| ROUTE64 = 9 |
| ADDRESS16 = 10 |
| LEADER_DATA = 11 |
| NETWORK_DATA = 12 |
| TLV_REQUEST = 13 |
| SCAN_MASK = 14 |
| CONNECTIVITY = 15 |
| LINK_MARGIN = 16 |
| STATUS = 17 |
| VERSION = 18 |
| ADDRESS_REGISTRATION = 19 |
| CHANNEL = 20 |
| PANID = 21 |
| ACTIVE_TIMESTAMP = 22 |
| PENDING_TIMESTAMP = 23 |
| ACTIVE_OPERATIONAL_DATASET = 24 |
| PENDING_OPERATIONAL_DATASET = 25 |
| THREAD_DISCOVERY = 26 |
| |
| |
| class SourceAddress(object): |
| |
| def __init__(self, address): |
| self._address = address |
| |
| @property |
| def address(self): |
| return self._address |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.address == other.address |
| |
| def __repr__(self): |
| return "SourceAddress(address={})".format(hex(self._address)) |
| |
| |
| class SourceAddressFactory: |
| |
| def parse(self, data, message_info): |
| address = struct.unpack(">H", data.read(2))[0] |
| return SourceAddress(address) |
| |
| |
| class Mode(object): |
| |
| def __init__(self, receiver, secure, device_type, network_data): |
| self._receiver = receiver |
| self._secure = secure |
| self._device_type = device_type |
| self._network_data = network_data |
| |
| @property |
| def receiver(self): |
| return self._receiver |
| |
| @property |
| def secure(self): |
| return self._secure |
| |
| @property |
| def device_type(self): |
| return self._device_type |
| |
| @property |
| def network_data(self): |
| return self._network_data |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.receiver == other.receiver and \ |
| self.secure == other.secure and \ |
| self.device_type == other.device_type and \ |
| self.network_data == other.network_data |
| |
| def __repr__(self): |
| return "Mode(receiver={}, secure={}, device_type={}, network_data={})".format(self.receiver, |
| self.secure, |
| self.device_type, |
| self.network_data) |
| |
| |
| class ModeFactory: |
| |
| def parse(self, data, message_info): |
| mode = ord(data.read(1)) |
| receiver = (mode >> 3) & 0x01 |
| secure = (mode >> 2) & 0x01 |
| device_type = (mode >> 1) & 0x01 |
| network_data = (mode >> 0) & 0x01 |
| return Mode(receiver, secure, device_type, network_data) |
| |
| |
| class Timeout(object): |
| |
| def __init__(self, timeout): |
| self._timeout = timeout |
| |
| @property |
| def timeout(self): |
| return self._timeout |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.timeout == other.timeout |
| |
| def __repr__(self): |
| return "Timeout(timeout={})".format(self.timeout) |
| |
| |
| class TimeoutFactory: |
| |
| def parse(self, data, message_info): |
| timeout = struct.unpack(">I", data.read(4))[0] |
| return Timeout(timeout) |
| |
| |
| class Challenge(object): |
| |
| def __init__(self, challenge): |
| self._challenge = challenge |
| |
| @property |
| def challenge(self): |
| return self._challenge |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.challenge == other.challenge |
| |
| def __repr__(self): |
| return "Challenge(challenge={})".format(hexlify(self.challenge)) |
| |
| |
| class ChallengeFactory: |
| |
| def parse(self, data, message_info): |
| challenge = data.read() |
| return Challenge(challenge) |
| |
| |
| class Response(object): |
| |
| def __init__(self, response): |
| self._response = response |
| |
| @property |
| def response(self): |
| return self._response |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.response == other.response |
| |
| def __repr__(self): |
| return "Response(response={})".format(hexlify(self.response)) |
| |
| |
| class ResponseFactory: |
| |
| def parse(self, data, message_info): |
| response = data.read() |
| return Response(response) |
| |
| |
| class LinkLayerFrameCounter(object): |
| |
| def __init__(self, frame_counter): |
| self._frame_counter = frame_counter |
| |
| @property |
| def frame_counter(self): |
| return self._frame_counter |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.frame_counter == other.frame_counter |
| |
| def __repr__(self): |
| return "LinkLayerFrameCounter(frame_counter={})".format(self.frame_counter) |
| |
| |
| class LinkLayerFrameCounterFactory: |
| |
| def parse(self, data, message_info): |
| frame_counter = struct.unpack(">I", data.read(4))[0] |
| return LinkLayerFrameCounter(frame_counter) |
| |
| |
| class MleFrameCounter(object): |
| |
| def __init__(self, frame_counter): |
| self._frame_counter = frame_counter |
| |
| @property |
| def frame_counter(self): |
| return self._frame_counter |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.frame_counter == other.frame_counter |
| |
| def __repr__(self): |
| return "MleFrameCounter(frame_counter={})".format(self.frame_counter) |
| |
| |
| class MleFrameCounterFactory: |
| |
| def parse(self, data, message_info): |
| frame_counter = struct.unpack(">I", data.read(4))[0] |
| return MleFrameCounter(frame_counter) |
| |
| |
| class LinkQualityAndRouteData(object): |
| |
| def __init__(self, output, _input, route): |
| self._output = output |
| self._input = _input |
| self._route = route |
| |
| @property |
| def output(self): |
| return self._output |
| |
| @property |
| def input(self): |
| return self._input |
| |
| @property |
| def route(self): |
| return self._route |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.output == other.output and self.input == other.input and self.route == other.route |
| |
| def __repr__(self): |
| return "LinkQualityAndRouteData(ouput={}, input={}, route={})".format(self.output, self.input, self.route) |
| |
| |
| class LinkQualityAndRouteDataFactory: |
| |
| def parse(self, data, message_info): |
| lqrd = ord(data.read(1)) |
| output = (lqrd >> 6) & 0x3 |
| _input = (lqrd >> 4) & 0x3 |
| route = lqrd & 0x0f |
| return LinkQualityAndRouteData(output, _input, route) |
| |
| |
| class Route64(object): |
| |
| def __init__(self, id_sequence, router_id_mask, link_quality_and_route_data): |
| self._id_sequence = id_sequence |
| self._router_id_mask = router_id_mask |
| self._link_quality_and_route_data = link_quality_and_route_data |
| |
| @property |
| def id_sequence(self): |
| return self._id_sequence |
| |
| @property |
| def router_id_mask(self): |
| return self._router_id_mask |
| |
| @property |
| def link_quality_and_route_data(self): |
| return self._link_quality_and_route_data |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.id_sequence == other.id_sequence and \ |
| self.router_id_mask == other.router_id_mask and \ |
| self.link_quality_and_route_data == other.link_quality_and_route_data |
| |
| def __repr__(self): |
| lqrd_str = ", ".join(["{}".format(lqrd) for lqrd in self.link_quality_and_route_data]) |
| return "Route64(id_sequence={}, router_id_mask={}, link_quality_and_route_data=[{}])".format( |
| self.id_sequence, hex(self.router_id_mask), lqrd_str) |
| |
| |
| class Route64Factory: |
| |
| def __init__(self, link_quality_and_route_data_factory): |
| self._lqrd_factory = link_quality_and_route_data_factory |
| |
| def parse(self, data, message_info): |
| id_sequence = ord(data.read(1)) |
| router_id_mask = struct.unpack(">Q", data.read(8))[0] |
| |
| link_quality_and_route_data = [] |
| |
| while data.tell() < len(data.getvalue()): |
| link_quality_and_route_data.append(self._lqrd_factory.parse(data, message_info)) |
| |
| return Route64(id_sequence, router_id_mask, link_quality_and_route_data) |
| |
| |
| class Address16(object): |
| |
| def __init__(self, address): |
| self._address = address |
| |
| @property |
| def address(self): |
| return self._address |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.address == other.address |
| |
| def __repr__(self): |
| return "Address16(address={})".format(hex(self.address)) |
| |
| |
| class Address16Factory: |
| |
| def parse(self, data, message_info): |
| address = struct.unpack(">H", data.read(2))[0] |
| return Address16(address) |
| |
| |
| class LeaderData(object): |
| |
| def __init__(self, partition_id, weighting, data_version, stable_data_version, leader_router_id): |
| self._partition_id = partition_id |
| self._weighting = weighting |
| self._data_version = data_version |
| self._stable_data_version = stable_data_version |
| self._leader_router_id = leader_router_id |
| |
| @property |
| def partition_id(self): |
| return self._partition_id |
| |
| @property |
| def weighting(self): |
| return self._weighting |
| |
| @property |
| def data_version(self): |
| return self._data_version |
| |
| @property |
| def stable_data_version(self): |
| return self._stable_data_version |
| |
| @property |
| def leader_router_id(self): |
| return self._leader_router_id |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.partition_id == other.partition_id and \ |
| self.weighting == other.weighting and \ |
| self.data_version == other.data_version and \ |
| self.stable_data_version == other.stable_data_version and \ |
| self.leader_router_id == other.leader_router_id |
| |
| def __repr__(self): |
| return "LeaderData(partition_id={}, weighting={}, data_version={}, stable_data_version={}, leader_router_id={})".format( |
| self.partition_id, self.weighting, self.data_version, self.stable_data_version, self.leader_router_id) |
| |
| |
| class LeaderDataFactory: |
| |
| def parse(self, data, message_info): |
| partition_id = struct.unpack(">I", data.read(4))[0] |
| weighting = ord(data.read(1)) |
| data_version = ord(data.read(1)) |
| stable_data_version = ord(data.read(1)) |
| leader_router_id = ord(data.read(1)) |
| return LeaderData(partition_id, weighting, data_version, stable_data_version, leader_router_id) |
| |
| |
| class NetworkData(object): |
| |
| def __init__(self, tlvs): |
| self._tlvs = tlvs |
| |
| @property |
| def tlvs(self): |
| return self._tlvs |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.tlvs == other.tlvs |
| |
| def __repr__(self): |
| tlvs_str = ", ".join(["{}".format(tlv) for tlv in self.tlvs]) |
| return "NetworkData(tlvs=[{}])".format(tlvs_str) |
| |
| |
| class NetworkDataFactory: |
| |
| def __init__(self, network_data_tlvs_factory): |
| self._tlvs_factory = network_data_tlvs_factory |
| |
| def parse(self, data, message_info): |
| tlvs = self._tlvs_factory.parse(data, message_info) |
| return NetworkData(tlvs) |
| |
| |
| class TlvRequest(object): |
| |
| def __init__(self, tlvs): |
| self._tlvs = tlvs |
| |
| @property |
| def tlvs(self): |
| return self._tlvs |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.tlvs == other.tlvs |
| |
| def __repr__(self): |
| tlvs_str = ", ".join(["{}".format(tlv) for tlv in self.tlvs]) |
| return "TlvRequest(tlvs=[{}])".format(tlvs_str) |
| |
| |
| class TlvRequestFactory: |
| |
| def parse(self, data, message_info): |
| tlvs = [b for b in bytearray(data.read())] |
| return TlvRequest(tlvs) |
| |
| |
| class ScanMask(object): |
| |
| def __init__(self, router, end_device): |
| self._router = router |
| self._end_device = end_device |
| |
| @property |
| def router(self): |
| return self._router |
| |
| @property |
| def end_device(self): |
| return self._end_device |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.router == other.router and self.end_device == other.end_device |
| |
| def __repr__(self): |
| return "ScanMask(router={}, end_device={})".format(self.router, self.end_device) |
| |
| |
| class ScanMaskFactory: |
| |
| def parse(self, data, message_info): |
| scan_mask = ord(data.read(1)) |
| router = (scan_mask >> 7) & 0x01 |
| end_device = (scan_mask >> 6) & 0x01 |
| return ScanMask(router, end_device) |
| |
| |
| class Connectivity(object): |
| |
| def __init__(self, |
| pp, |
| link_quality_3, |
| link_quality_2, |
| link_quality_1, |
| leader_cost, |
| id_sequence, |
| active_routers, |
| sed_buffer_size=None, |
| sed_datagram_count=None): |
| self._pp = pp |
| self._link_quality_3 = link_quality_3 |
| self._link_quality_2 = link_quality_2 |
| self._link_quality_1 = link_quality_1 |
| self._leader_cost = leader_cost |
| self._id_sequence = id_sequence |
| self._active_routers = active_routers |
| self._sed_buffer_size = sed_buffer_size |
| self._sed_datagram_count = sed_datagram_count |
| |
| @property |
| def pp(self): |
| return self._pp |
| |
| @property |
| def link_quality_3(self): |
| return self._link_quality_3 |
| |
| @property |
| def link_quality_2(self): |
| return self._link_quality_2 |
| |
| @property |
| def link_quality_1(self): |
| return self._link_quality_1 |
| |
| @property |
| def leader_cost(self): |
| return self._leader_cost |
| |
| @property |
| def id_sequence(self): |
| return self._id_sequence |
| |
| @property |
| def active_routers(self): |
| return self._active_routers |
| |
| @property |
| def sed_buffer_size(self): |
| return self._sed_buffer_size |
| |
| @property |
| def sed_datagram_count(self): |
| return self._sed_datagram_count |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.pp == other.pp and \ |
| self.link_quality_3 == other.link_quality_3 and \ |
| self.link_quality_2 == other.link_quality_2 and \ |
| self.link_quality_1 == other.link_quality_1 and \ |
| self.leader_cost == other.leader_cost and \ |
| self.id_sequence == other.id_sequence and \ |
| self.active_routers == other.active_routers and \ |
| self.sed_buffer_size == other.sed_buffer_size and \ |
| self.sed_datagram_count == other.sed_datagram_count |
| |
| def __repr__(self): |
| return r"Connectivity(pp={}, \ |
| link_quality_3={}, \ |
| link_quality_2={}, \ |
| link_quality_1={}, \ |
| leader_cost={}, \ |
| id_sequence={}, \ |
| active_routers={}, \ |
| sed_buffer_size={}, \ |
| sed_datagram_count={})".format( |
| self.pp, |
| self.link_quality_3, |
| self.link_quality_2, |
| self.link_quality_1, |
| self.leader_cost, |
| self.id_sequence, |
| self.active_routers, |
| self.sed_buffer_size, |
| self.sed_datagram_count |
| ) |
| |
| |
| class ConnectivityFactory: |
| |
| def parse(self, data, message_info): |
| pp = ord(data.read(1)) & 0x03 |
| link_quality_3 = ord(data.read(1)) |
| link_quality_2 = ord(data.read(1)) |
| link_quality_1 = ord(data.read(1)) |
| leader_cost = ord(data.read(1)) |
| id_sequence = ord(data.read(1)) |
| active_routers = ord(data.read(1)) |
| |
| sed_data = io.BytesIO(data.read(3)) |
| |
| if len(sed_data.getvalue()) > 0: |
| sed_buffer_size = struct.unpack(">H", sed_data.read(2))[0] |
| sed_datagram_count = ord(sed_data.read(1)) |
| else: |
| sed_buffer_size = None |
| sed_datagram_count = None |
| |
| return Connectivity(pp, |
| link_quality_3, |
| link_quality_2, |
| link_quality_1, |
| leader_cost, |
| id_sequence, |
| active_routers, |
| sed_buffer_size, |
| sed_datagram_count) |
| |
| |
| class LinkMargin(object): |
| |
| def __init__(self, link_margin): |
| self._link_margin = link_margin |
| |
| @property |
| def link_margin(self): |
| return self._link_margin |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.link_margin == other.link_margin |
| |
| def __repr__(self): |
| return "LinkMargin(link_margin={})".format(self.link_margin) |
| |
| |
| class LinkMarginFactory: |
| |
| def parse(self, data, message_info): |
| link_margin = ord(data.read(1)) |
| return LinkMargin(link_margin) |
| |
| |
| class Status(object): |
| |
| def __init__(self, status): |
| self._status = status |
| |
| @property |
| def status(self): |
| return self._status |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.status == other.status |
| |
| def __repr__(self): |
| return "Status(status={})".format(self.status) |
| |
| |
| class StatusFactory: |
| |
| def parse(self, data, message_info): |
| status = ord(data.read(1)) |
| return Status(status) |
| |
| |
| class Version(object): |
| |
| def __init__(self, version): |
| self._version = version |
| |
| @property |
| def version(self): |
| return self._version |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.version == other.version |
| |
| def __repr__(self): |
| return "Version(version={})".format(self.version) |
| |
| |
| class VersionFactory: |
| |
| def parse(self, data, message_info): |
| version = struct.unpack(">H", data.read(2))[0] |
| return Version(version) |
| |
| |
| class AddressFull(object): |
| |
| def __init__(self, ipv6_address): |
| self._ipv6_address = ipv6_address |
| |
| @property |
| def ipv6_address(self): |
| return self._ipv6_address |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.ipv6_address == other.ipv6_address |
| |
| def __repr__(self): |
| return "AddressFull(ipv6_address={}')".format(hexlify(self.ipv6_address)) |
| |
| |
| class AddressFullFactory: |
| |
| def parse(self, data, message_info): |
| data.read(1) # first byte is ignored |
| ipv6_address = data.read(16) |
| return AddressFull(ipv6_address) |
| |
| |
| class AddressCompressed(object): |
| |
| def __init__(self, cid, iid): |
| self._cid = cid |
| self._iid = iid |
| |
| @property |
| def cid(self): |
| return self._cid |
| |
| @property |
| def iid(self): |
| return self._iid |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.cid == other.cid and self.iid == other.iid |
| |
| def __repr__(self): |
| return "AddressCompressed(cid={}, iid={}')".format(self.cid, hexlify(self.iid)) |
| |
| |
| class AddressCompressedFactory: |
| |
| def parse(self, data, message_info): |
| cid = (ord(data.read(1)) & 0x0F) |
| iid = bytearray(data.read(8)) |
| return AddressCompressed(cid, iid) |
| |
| |
| class AddressRegistration(object): |
| |
| def __init__(self, addresses): |
| self._addresses = addresses |
| |
| @property |
| def addresses(self): |
| return self._addresses |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.addresses == other.addresses |
| |
| def __repr__(self): |
| addresses_str = ", ".join(["{}".format(address) for address in self.addresses]) |
| return "AddressRegistration(addresses=[{}])".format(addresses_str) |
| |
| |
| class AddressRegistrationFactory: |
| |
| def __init__(self, addr_compressed_factory, addr_full_factory): |
| self._addr_compressed_factory = addr_compressed_factory |
| self._addr_full_factory = addr_full_factory |
| |
| def parse(self, data, message_info): |
| addresses = [] |
| |
| while data.tell() < len(data.getvalue()): |
| compressed = (ord(data.read(1)) >> 7) & 0x01 |
| data.seek(-1, io.SEEK_CUR) |
| |
| if compressed: |
| addresses.append(self._addr_compressed_factory.parse(data, message_info)) |
| else: |
| addresses.append(self._addr_full_factory.parse(data, message_info)) |
| |
| return AddressRegistration(addresses) |
| |
| |
| class Channel(object): |
| |
| def __init__(self, channel_page, channel): |
| self._channel_page = channel_page |
| self._channel = channel |
| |
| @property |
| def channel_page(self): |
| return self._channel_page |
| |
| @property |
| def channel(self): |
| return self._channel |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.channel_page == other.channel_page and \ |
| self.channel == other.channel |
| |
| def __repr__(self): |
| return "Channel(channel_page={}, channel={})".format(self.channel_page, channel) |
| |
| |
| class ChannelFactory: |
| |
| def parse(self, data, message_info): |
| channel_page = ord(data.read(1)) |
| channel = struct.unpack(">H", data.read(2))[0] |
| return Channel(channel_page, channel) |
| |
| |
| class PanId: |
| |
| def __init__(self, pan_id): |
| self._pan_id = pan_id |
| |
| @property |
| def pan_id(self): |
| return self._pan_id |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.pan_id == other.pan_id |
| |
| def __repr__(self): |
| return "PanId(pan_id={})".format(self.pan_id) |
| |
| |
| class PanIdFactory: |
| |
| def parse(self, data, message_info): |
| pan_id = struct.unpack(">H", data.read(2))[0] |
| return PanId(pan_id) |
| |
| |
| class ActiveTimestamp(object): |
| |
| def __init__(self, timestamp_seconds, timestamp_ticks, u): |
| self._timestamp_seconds = timestamp_seconds |
| self._timestamp_ticks = timestamp_ticks |
| self._u = u |
| |
| @property |
| def timestamp_seconds(self): |
| return self._timestamp_seconds |
| |
| @property |
| def timestamp_ticks(self): |
| return self._timestamp_ticks |
| |
| @property |
| def u(self): |
| return self._u |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.timestamp_seconds == other.timestamp_seconds and \ |
| self.timestamp_ticks == other.timestamp_ticks and \ |
| self.u == other.u |
| |
| def __repr__(self): |
| return "ActiveTimestamp(timestamp_seconds={}, timestamp_ticks={}, u={})".format( |
| self.timestamp_seconds, self.timestamp_ticks, self.u) |
| |
| |
| class ActiveTimestampFactory: |
| |
| def parse(self, data, message_info): |
| seconds = bytearray([0x00, 0x00]) + bytearray(data.read(6)) |
| ticks = struct.unpack(">H", data.read(2))[0] |
| |
| timestamp_seconds = struct.unpack(">Q", bytes(seconds))[0] |
| timestamp_ticks = (ticks >> 1) |
| u = ticks & 0x01 |
| return ActiveTimestamp(timestamp_seconds, timestamp_ticks, u) |
| |
| |
| class PendingTimestamp(object): |
| |
| def __init__(self, timestamp_seconds, timestamp_ticks, u): |
| self._timestamp_seconds = timestamp_seconds |
| self._timestamp_ticks = timestamp_ticks |
| self._u = u |
| |
| @property |
| def timestamp_seconds(self): |
| return self._timestamp_seconds |
| |
| @property |
| def timestamp_ticks(self): |
| return self._timestamp_ticks |
| |
| @property |
| def u(self): |
| return self._u |
| |
| def __eq__(self, other): |
| common.expect_the_same_class(self, other) |
| |
| return self.timestamp_seconds == other.timestamp_seconds and \ |
| self.timestamp_ticks == other.timestamp_ticks and \ |
| self.u == other.u |
| |
| def __repr__(self): |
| return "PendingTimestamp(timestamp_seconds={}, timestamp_ticks={}, u={})".format( |
| self.timestamp_seconds, self.timestamp_ticks, self.u) |
| |
| |
| class PendingTimestampFactory: |
| |
| def parse(self, data, message_info): |
| seconds = bytearray([0x00, 0x00]) + bytearray(data.read(6)) |
| ticks = struct.unpack(">H", data.read(2))[0] |
| |
| timestamp_seconds = struct.unpack(">Q", bytes(seconds))[0] |
| timestamp_ticks = (ticks >> 1) |
| u = ticks & 0x01 |
| return PendingTimestamp(timestamp_seconds, timestamp_ticks, u) |
| |
| |
| class ActiveOperationalDataset: |
| # TODO: Not implemented yet |
| |
| def __init__(self): |
| print("ActiveOperationalDataset is not implemented yet.") |
| |
| |
| class ActiveOperationalDatasetFactory: |
| |
| def parse(self, data, message_info): |
| return ActiveOperationalDataset() |
| |
| |
| class PendingOperationalDataset: |
| # TODO: Not implemented yet |
| |
| def __init__(self): |
| print("PendingOperationalDataset is not implemented yet.") |
| |
| |
| class PendingOperationalDatasetFactory: |
| |
| def parse(self, data, message_info): |
| return PendingOperationalDataset() |
| |
| |
| class ThreadDiscovery: |
| # TODO: Not implemented yet |
| |
| def __init__(self): |
| print("ThreadDiscovery is not implemented yet.") |
| |
| |
| class ThreadDiscoveryFactory: |
| |
| def parse(self, data, message_info): |
| return ThreadDiscovery() |
| |
| |
| class MleCommand(object): |
| |
| def __init__(self, _type, tlvs): |
| self._type = _type |
| self._tlvs = tlvs |
| |
| @property |
| def type(self): |
| return self._type |
| |
| @property |
| def tlvs(self): |
| return self._tlvs |
| |
| def __repr__(self): |
| tlvs_str = ", ".join(["{}".format(tlv) for tlv in self.tlvs]) |
| return "MleCommand(type={}, tlvs=[{}])".format(self.type.name, tlvs_str) |
| |
| |
| class MleCommandFactory: |
| |
| _MARKER_EXTENDED_LENGTH = 0xff |
| |
| def __init__(self, tlvs_factories): |
| self._tlvs_factories = tlvs_factories |
| |
| def _get_length(self, data): |
| length = ord(data.read(1)) |
| |
| if length == self._MARKER_EXTENDED_LENGTH: |
| length = struct.unpack(">H", data.read(2))[0] |
| |
| return length |
| |
| def _get_tlv_factory(self, _type): |
| try: |
| return self._tlvs_factories[_type] |
| except KeyError: |
| raise KeyError("Could not find TLV factory. Unsupported TLV type: {}".format(_type)) |
| |
| def _parse_tlv(self, data, message_info): |
| _type = TlvType(ord(data.read(1))) |
| length = self._get_length(data) |
| value = data.read(length) |
| |
| factory = self._get_tlv_factory(_type) |
| |
| return factory.parse(io.BytesIO(value), message_info) |
| |
| def parse(self, data, message_info): |
| cmd_type = CommandType(ord(data.read(1))) |
| tlvs = [] |
| |
| while data.tell() < len(data.getvalue()): |
| tlv = self._parse_tlv(data, message_info) |
| tlvs.append(tlv) |
| |
| return MleCommand(cmd_type, tlvs) |
| |
| |
| class MleMessage(object): |
| |
| def __init__(self, command): |
| self._command = command |
| |
| @property |
| def command(self): |
| return self._command |
| |
| def __repr__(self): |
| return "MleMessage(command={})".format(self.command) |
| |
| |
| class MleMessageSecured(MleMessage): |
| |
| def __init__(self, aux_sec_hdr, command, mic): |
| super(MleMessageSecured, self).__init__(command) |
| self._aux_sec_hdr = aux_sec_hdr |
| self._mic = mic |
| |
| @property |
| def aux_sec_hdr(self): |
| return self._aux_sec_hdr |
| |
| @property |
| def mic(self): |
| return self._mic |
| |
| def __repr__(self): |
| return "MleMessageSecured(aux_sec_hdr={}, command={}, mic=\"{}\")".format( |
| self.aux_sec_hdr, self.command, hexlify(self.mic)) |
| |
| |
| class MleMessageFactory: |
| |
| def __init__(self, aux_sec_hdr_factory, mle_command_factory, crypto_engine): |
| self._aux_sec_hdr_factory = aux_sec_hdr_factory |
| self._mle_command_factory = mle_command_factory |
| self._crypto_engine = crypto_engine |
| |
| def _create_mle_secured_message(self, data, message_info): |
| aux_sec_hdr = self._aux_sec_hdr_factory.parse(data, message_info) |
| |
| enc_data_length = len(data.getvalue()) |
| |
| enc_data = bytearray(data.read(enc_data_length - data.tell() - self._crypto_engine.mic_length)) |
| mic = bytearray(data.read()) |
| |
| dec_data = self._crypto_engine.decrypt(enc_data, mic, message_info) |
| |
| command = self._mle_command_factory.parse(io.BytesIO(dec_data), message_info) |
| |
| return MleMessageSecured(aux_sec_hdr, command, mic) |
| |
| def _create_mle_message(self, data, message_info): |
| command = self._mle_command_factory.parse(data, message_info) |
| |
| return MleMessage(command) |
| |
| def parse(self, data, message_info): |
| security_indicator = ord(data.read(1)) |
| |
| if security_indicator == 0: |
| return self._create_mle_secured_message(data, message_info) |
| |
| elif security_indicator == 255: |
| return self._create_mle_message(data, message_info) |
| |
| else: |
| raise RuntimeError( |
| "Could not create MLE message. Unknown security indicator value: {}".format(security_indicator)) |
