Reticulum/RNS/Channel.py

520 lines
18 KiB
Python
Raw Normal View History

2023-02-28 02:05:25 +01:00
# MIT License
#
# Copyright (c) 2016-2023 Mark Qvist / unsigned.io and contributors.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
2023-02-26 01:23:25 +01:00
from __future__ import annotations
import collections
import enum
import threading
import time
from types import TracebackType
from typing import Type, Callable, TypeVar, Generic, NewType
2023-02-26 01:23:25 +01:00
import abc
import contextlib
import struct
import RNS
from abc import ABC, abstractmethod
TPacket = TypeVar("TPacket")
2023-02-26 01:23:25 +01:00
class ChannelOutletBase(ABC, Generic[TPacket]):
2023-02-28 00:36:04 +01:00
"""
An abstract transport layer interface used by Channel.
DEPRECATED: This was created for testing; eventually
Channel will use Link or a LinkBase interface
directly.
"""
2023-02-26 01:23:25 +01:00
@abstractmethod
def send(self, raw: bytes) -> TPacket:
2023-02-26 01:23:25 +01:00
raise NotImplemented()
@abstractmethod
def resend(self, packet: TPacket) -> TPacket:
2023-02-26 01:23:25 +01:00
raise NotImplemented()
@property
@abstractmethod
def mdu(self):
raise NotImplemented()
@property
@abstractmethod
def rtt(self):
raise NotImplemented()
@property
@abstractmethod
def is_usable(self):
raise NotImplemented()
@abstractmethod
def get_packet_state(self, packet: TPacket) -> MessageState:
2023-02-26 01:23:25 +01:00
raise NotImplemented()
@abstractmethod
def timed_out(self):
raise NotImplemented()
@abstractmethod
def __str__(self):
raise NotImplemented()
@abstractmethod
def set_packet_timeout_callback(self, packet: TPacket, callback: Callable[[TPacket], None] | None,
2023-02-26 01:23:25 +01:00
timeout: float | None = None):
raise NotImplemented()
@abstractmethod
def set_packet_delivered_callback(self, packet: TPacket, callback: Callable[[TPacket], None] | None):
2023-02-26 01:23:25 +01:00
raise NotImplemented()
@abstractmethod
def get_packet_id(self, packet: TPacket) -> any:
2023-02-26 01:23:25 +01:00
raise NotImplemented()
class CEType(enum.IntEnum):
2023-02-28 00:36:04 +01:00
"""
ChannelException type codes
"""
2023-02-26 01:23:25 +01:00
ME_NO_MSG_TYPE = 0
ME_INVALID_MSG_TYPE = 1
ME_NOT_REGISTERED = 2
ME_LINK_NOT_READY = 3
ME_ALREADY_SENT = 4
ME_TOO_BIG = 5
class ChannelException(Exception):
2023-02-28 00:36:04 +01:00
"""
An exception thrown by Channel, with a type code.
"""
2023-02-26 01:23:25 +01:00
def __init__(self, ce_type: CEType, *args):
super().__init__(args)
self.type = ce_type
class MessageState(enum.IntEnum):
2023-02-28 00:36:04 +01:00
"""
Set of possible states for a Message
"""
2023-02-26 01:23:25 +01:00
MSGSTATE_NEW = 0
MSGSTATE_SENT = 1
MSGSTATE_DELIVERED = 2
MSGSTATE_FAILED = 3
class MessageBase(abc.ABC):
2023-02-28 00:36:04 +01:00
"""
Base type for any messages sent or received on a Channel.
Subclasses must define the two abstract methods as well as
2023-02-28 02:05:25 +01:00
the ``MSGTYPE`` class variable.
2023-02-28 00:36:04 +01:00
"""
# MSGTYPE must be unique within all classes sent over a
# channel. Additionally, MSGTYPE > 0xf000 are reserved.
2023-02-26 01:23:25 +01:00
MSGTYPE = None
2023-02-28 02:05:25 +01:00
"""
Defines a unique identifier for a message class.
2023-02-28 04:10:28 +01:00
* Must be unique within all classes registered with a ``Channel``
* Must be less than ``0xf000``. Values greater than or equal to ``0xf000`` are reserved.
2023-02-28 02:05:25 +01:00
"""
2023-02-26 01:23:25 +01:00
@abstractmethod
def pack(self) -> bytes:
2023-02-28 00:36:04 +01:00
"""
Create and return the binary representation of the message
2023-02-28 02:05:25 +01:00
:return: binary representation of message
2023-02-28 00:36:04 +01:00
"""
2023-02-26 01:23:25 +01:00
raise NotImplemented()
@abstractmethod
2023-02-28 04:10:28 +01:00
def unpack(self, raw: bytes):
2023-02-28 00:36:04 +01:00
"""
Populate message from binary representation
2023-02-28 02:05:25 +01:00
:param raw: binary representation
2023-02-28 00:36:04 +01:00
"""
2023-02-26 01:23:25 +01:00
raise NotImplemented()
MessageCallbackType = NewType("MessageCallbackType", Callable[[MessageBase], bool])
2023-02-26 01:23:25 +01:00
class Envelope:
2023-02-28 00:36:04 +01:00
"""
Internal wrapper used to transport messages over a channel and
track its state within the channel framework.
"""
2023-02-26 01:23:25 +01:00
def unpack(self, message_factories: dict[int, Type]) -> MessageBase:
msgtype, self.sequence, length = struct.unpack(">HHH", self.raw[:6])
raw = self.raw[6:]
ctor = message_factories.get(msgtype, None)
if ctor is None:
raise ChannelException(CEType.ME_NOT_REGISTERED, f"Unable to find constructor for Channel MSGTYPE {hex(msgtype)}")
message = ctor()
message.unpack(raw)
return message
def pack(self) -> bytes:
if self.message.__class__.MSGTYPE is None:
raise ChannelException(CEType.ME_NO_MSG_TYPE, f"{self.message.__class__} lacks MSGTYPE")
data = self.message.pack()
self.raw = struct.pack(">HHH", self.message.MSGTYPE, self.sequence, len(data)) + data
return self.raw
def __init__(self, outlet: ChannelOutletBase, message: MessageBase = None, raw: bytes = None, sequence: int = None):
self.ts = time.time()
self.id = id(self)
self.message = message
self.raw = raw
self.packet: TPacket = None
2023-02-26 01:23:25 +01:00
self.sequence = sequence
self.outlet = outlet
self.tries = 0
self.tracked = False
class Channel(contextlib.AbstractContextManager):
2023-02-28 00:36:04 +01:00
"""
2023-02-28 04:10:28 +01:00
Provides reliable delivery of messages over
2023-02-28 02:05:25 +01:00
a link.
2023-02-28 04:10:28 +01:00
``Channel`` differs from ``Request`` and
``Resource`` in some important ways:
**Continuous**
Messages can be sent or received as long as
the ``Link`` is open.
**Bi-directional**
Messages can be sent in either direction on
the ``Link``; neither end is the client or
server.
**Size-constrained**
Messages must be encoded into a single packet.
``Channel`` is similar to ``Packet``, except that it
provides reliable delivery (automatic retries) as well
as a structure for exchanging several types of
messages over the ``Link``.
``Channel`` is not instantiated directly, but rather
obtained from a ``Link`` with ``get_channel()``.
2023-02-28 00:36:04 +01:00
"""
2023-02-26 01:23:25 +01:00
def __init__(self, outlet: ChannelOutletBase):
2023-02-28 02:05:25 +01:00
"""
@param outlet:
"""
2023-02-26 01:23:25 +01:00
self._outlet = outlet
self._lock = threading.RLock()
self._tx_ring: collections.deque[Envelope] = collections.deque()
self._rx_ring: collections.deque[Envelope] = collections.deque()
self._message_callbacks: [MessageCallbackType] = []
2023-02-26 01:23:25 +01:00
self._next_sequence = 0
self._message_factories: dict[int, Type[MessageBase]] = {}
2023-02-26 01:23:25 +01:00
self._max_tries = 5
def __enter__(self) -> Channel:
return self
def __exit__(self, __exc_type: Type[BaseException] | None, __exc_value: BaseException | None,
__traceback: TracebackType | None) -> bool | None:
2023-02-28 00:36:04 +01:00
self._shutdown()
2023-02-26 01:23:25 +01:00
return False
2023-02-28 04:10:28 +01:00
def register_message_type(self, message_class: Type[MessageBase]):
2023-02-28 00:36:04 +01:00
"""
2023-02-28 04:10:28 +01:00
Register a message class for reception over a ``Channel``.
Message classes must extend ``MessageBase``.
2023-02-28 02:05:25 +01:00
2023-02-28 04:10:28 +01:00
:param message_class: Class to register
2023-02-28 00:36:04 +01:00
"""
2023-02-28 04:10:28 +01:00
self._register_message_type(message_class, is_system_type=False)
def _register_message_type(self, message_class: Type[MessageBase], *, is_system_type: bool = False):
with self._lock:
if not issubclass(message_class, MessageBase):
raise ChannelException(CEType.ME_INVALID_MSG_TYPE,
f"{message_class} is not a subclass of {MessageBase}.")
if message_class.MSGTYPE is None:
raise ChannelException(CEType.ME_INVALID_MSG_TYPE,
f"{message_class} has invalid MSGTYPE class attribute.")
2023-02-28 00:36:04 +01:00
if message_class.MSGTYPE >= 0xf000 and not is_system_type:
raise ChannelException(CEType.ME_INVALID_MSG_TYPE,
f"{message_class} has system-reserved message type.")
try:
message_class()
except Exception as ex:
raise ChannelException(CEType.ME_INVALID_MSG_TYPE,
f"{message_class} raised an exception when constructed with no arguments: {ex}")
self._message_factories[message_class.MSGTYPE] = message_class
def add_message_handler(self, callback: MessageCallbackType):
2023-02-28 00:36:04 +01:00
"""
Add a handler for incoming messages. A handler
2023-02-28 04:10:28 +01:00
has the following signature:
``(message: MessageBase) -> bool``
2023-02-28 00:36:04 +01:00
Handlers are processed in the order they are
added. If any handler returns True, processing
of the message stops; handlers after the
returning handler will not be called.
2023-02-28 02:05:25 +01:00
:param callback: Function to call
2023-02-28 00:36:04 +01:00
"""
with self._lock:
if callback not in self._message_callbacks:
self._message_callbacks.append(callback)
def remove_message_handler(self, callback: MessageCallbackType):
2023-02-28 00:36:04 +01:00
"""
2023-02-28 04:10:28 +01:00
Remove a handler added with ``add_message_handler``.
2023-02-28 02:05:25 +01:00
:param callback: handler to remove
2023-02-28 00:36:04 +01:00
"""
with self._lock:
self._message_callbacks.remove(callback)
2023-02-26 01:23:25 +01:00
2023-02-28 00:36:04 +01:00
def _shutdown(self):
with self._lock:
self._message_callbacks.clear()
2023-02-28 00:36:04 +01:00
self._clear_rings()
2023-02-26 01:23:25 +01:00
2023-02-28 00:36:04 +01:00
def _clear_rings(self):
2023-02-26 01:23:25 +01:00
with self._lock:
for envelope in self._tx_ring:
if envelope.packet is not None:
self._outlet.set_packet_timeout_callback(envelope.packet, None)
self._outlet.set_packet_delivered_callback(envelope.packet, None)
self._tx_ring.clear()
self._rx_ring.clear()
2023-02-28 00:36:04 +01:00
def _emplace_envelope(self, envelope: Envelope, ring: collections.deque[Envelope]) -> bool:
2023-02-26 01:23:25 +01:00
with self._lock:
i = 0
2023-02-28 00:36:04 +01:00
for existing in ring:
if existing.sequence > envelope.sequence \
and not existing.sequence // 2 > envelope.sequence: # account for overflow
2023-02-26 01:23:25 +01:00
ring.insert(i, envelope)
return True
2023-02-28 00:36:04 +01:00
if existing.sequence == envelope.sequence:
2023-02-26 01:23:25 +01:00
RNS.log(f"Envelope: Emplacement of duplicate envelope sequence.", RNS.LOG_EXTREME)
return False
i += 1
envelope.tracked = True
ring.append(envelope)
return True
2023-02-28 00:36:04 +01:00
def _prune_rx_ring(self):
2023-02-26 01:23:25 +01:00
with self._lock:
# Implementation for fixed window = 1
stale = list(sorted(self._rx_ring, key=lambda env: env.sequence, reverse=True))[1:]
for env in stale:
env.tracked = False
self._rx_ring.remove(env)
def _run_callbacks(self, message: MessageBase):
with self._lock:
cbs = self._message_callbacks.copy()
for cb in cbs:
try:
if cb(message):
return
except Exception as ex:
RNS.log(f"Channel: Error running message callback: {ex}", RNS.LOG_ERROR)
2023-02-28 00:36:04 +01:00
def _receive(self, raw: bytes):
2023-02-26 01:23:25 +01:00
try:
envelope = Envelope(outlet=self._outlet, raw=raw)
with self._lock:
message = envelope.unpack(self._message_factories)
2023-02-28 00:36:04 +01:00
is_new = self._emplace_envelope(envelope, self._rx_ring)
self._prune_rx_ring()
2023-02-26 01:23:25 +01:00
if not is_new:
RNS.log("Channel: Duplicate message received", RNS.LOG_DEBUG)
return
RNS.log(f"Message received: {message}", RNS.LOG_DEBUG)
threading.Thread(target=self._run_callbacks, name="Message Callback", args=[message], daemon=True).start()
2023-02-26 01:23:25 +01:00
except Exception as ex:
RNS.log(f"Channel: Error receiving data: {ex}")
def is_ready_to_send(self) -> bool:
2023-02-28 00:36:04 +01:00
"""
2023-02-28 04:10:28 +01:00
Check if ``Channel`` is ready to send.
2023-02-28 02:05:25 +01:00
:return: True if ready
2023-02-28 00:36:04 +01:00
"""
2023-02-26 01:23:25 +01:00
if not self._outlet.is_usable:
RNS.log("Channel: Link is not usable.", RNS.LOG_EXTREME)
return False
with self._lock:
for envelope in self._tx_ring:
if envelope.outlet == self._outlet and (not envelope.packet
or self._outlet.get_packet_state(envelope.packet) == MessageState.MSGSTATE_SENT):
RNS.log("Channel: Link has a pending message.", RNS.LOG_EXTREME)
return False
return True
def _packet_tx_op(self, packet: TPacket, op: Callable[[TPacket], bool]):
2023-02-26 01:23:25 +01:00
with self._lock:
envelope = next(filter(lambda e: self._outlet.get_packet_id(e.packet) == self._outlet.get_packet_id(packet),
self._tx_ring), None)
if envelope and op(envelope):
envelope.tracked = False
if envelope in self._tx_ring:
self._tx_ring.remove(envelope)
else:
RNS.log("Channel: Envelope not found in TX ring", RNS.LOG_DEBUG)
if not envelope:
RNS.log("Channel: Spurious message received.", RNS.LOG_EXTREME)
def _packet_delivered(self, packet: TPacket):
2023-02-26 01:23:25 +01:00
self._packet_tx_op(packet, lambda env: True)
def _packet_timeout(self, packet: TPacket):
2023-02-26 01:23:25 +01:00
def retry_envelope(envelope: Envelope) -> bool:
if envelope.tries >= self._max_tries:
RNS.log("Channel: Retry count exceeded, tearing down Link.", RNS.LOG_ERROR)
2023-02-28 00:36:04 +01:00
self._shutdown() # start on separate thread?
2023-02-26 01:23:25 +01:00
self._outlet.timed_out()
return True
envelope.tries += 1
self._outlet.resend(envelope.packet)
return False
self._packet_tx_op(packet, retry_envelope)
def send(self, message: MessageBase) -> Envelope:
2023-02-28 00:36:04 +01:00
"""
Send a message. If a message send is attempted and
2023-02-28 04:10:28 +01:00
``Channel`` is not ready, an exception is thrown.
2023-02-28 02:05:25 +01:00
2023-02-28 04:10:28 +01:00
:param message: an instance of a ``MessageBase`` subclass
2023-02-28 00:36:04 +01:00
"""
2023-02-26 01:23:25 +01:00
envelope: Envelope | None = None
with self._lock:
if not self.is_ready_to_send():
raise ChannelException(CEType.ME_LINK_NOT_READY, f"Link is not ready")
envelope = Envelope(self._outlet, message=message, sequence=self._next_sequence)
self._next_sequence = (self._next_sequence + 1) % 0x10000
2023-02-28 00:36:04 +01:00
self._emplace_envelope(envelope, self._tx_ring)
2023-02-26 01:23:25 +01:00
if envelope is None:
raise BlockingIOError()
envelope.pack()
if len(envelope.raw) > self._outlet.mdu:
raise ChannelException(CEType.ME_TOO_BIG, f"Packed message too big for packet: {len(envelope.raw)} > {self._outlet.mdu}")
envelope.packet = self._outlet.send(envelope.raw)
envelope.tries += 1
self._outlet.set_packet_delivered_callback(envelope.packet, self._packet_delivered)
self._outlet.set_packet_timeout_callback(envelope.packet, self._packet_timeout)
return envelope
@property
def MDU(self):
2023-02-28 00:36:04 +01:00
"""
Maximum Data Unit: the number of bytes available
2023-02-28 04:10:28 +01:00
for a message to consume in a single send. This
value is adjusted from the ``Link`` MDU to accommodate
message header information.
2023-02-28 02:05:25 +01:00
:return: number of bytes available
2023-02-28 00:36:04 +01:00
"""
return self._outlet.mdu - 6 # sizeof(msgtype) + sizeof(length) + sizeof(sequence)
2023-02-26 01:23:25 +01:00
class LinkChannelOutlet(ChannelOutletBase):
2023-02-28 00:36:04 +01:00
"""
An implementation of ChannelOutletBase for RNS.Link.
Allows Channel to send packets over an RNS Link with
Packets.
2023-02-28 02:05:25 +01:00
:param link: RNS Link to wrap
2023-02-28 00:36:04 +01:00
"""
2023-02-26 01:23:25 +01:00
def __init__(self, link: RNS.Link):
self.link = link
def send(self, raw: bytes) -> RNS.Packet:
packet = RNS.Packet(self.link, raw, context=RNS.Packet.CHANNEL)
packet.send()
return packet
def resend(self, packet: RNS.Packet) -> RNS.Packet:
if not packet.resend():
RNS.log("Failed to resend packet", RNS.LOG_ERROR)
return packet
@property
def mdu(self):
return self.link.MDU
@property
def rtt(self):
return self.link.rtt
@property
def is_usable(self):
return True # had issues looking at Link.status
def get_packet_state(self, packet: TPacket) -> MessageState:
2023-02-26 01:23:25 +01:00
status = packet.receipt.get_status()
if status == RNS.PacketReceipt.SENT:
return MessageState.MSGSTATE_SENT
if status == RNS.PacketReceipt.DELIVERED:
return MessageState.MSGSTATE_DELIVERED
if status == RNS.PacketReceipt.FAILED:
return MessageState.MSGSTATE_FAILED
else:
raise Exception(f"Unexpected receipt state: {status}")
def timed_out(self):
self.link.teardown()
def __str__(self):
return f"{self.__class__.__name__}({self.link})"
def set_packet_timeout_callback(self, packet: RNS.Packet, callback: Callable[[RNS.Packet], None] | None,
timeout: float | None = None):
if timeout:
packet.receipt.set_timeout(timeout)
def inner(receipt: RNS.PacketReceipt):
callback(packet)
packet.receipt.set_timeout_callback(inner if callback else None)
def set_packet_delivered_callback(self, packet: RNS.Packet, callback: Callable[[RNS.Packet], None] | None):
def inner(receipt: RNS.PacketReceipt):
callback(packet)
packet.receipt.set_delivery_callback(inner if callback else None)
def get_packet_id(self, packet: RNS.Packet) -> any:
return packet.get_hash()