Source code for py2p.mesh

from __future__ import print_function
from __future__ import absolute_import

import inspect
import select
import socket
import struct
import sys

from collections import deque
from platform import system
from itertools import chain
from logging import (INFO, DEBUG)
from random import shuffle
from traceback import format_exc

from typing import (cast, Any, MutableSequence, Sequence, Tuple, Union)
# from _collections import deque as DequeType

try:
    from .cbase import protocol as Protocol
except:
    from .base import Protocol
from .base import (
    flags,
    compression,
    to_base_58,
    from_base_58,
    BaseConnection,
    Message,
    MsgPackable,
    BaseDaemon,
    BaseSocket,
    InternalMessage, )
from .utils import (getUTC, get_socket, intersect, inherit_doc, log_entry,
                    awaiting_value)

max_outgoing = 4
default_protocol = Protocol('mesh', "Plaintext")  # SSL")


[docs]class MeshConnection(BaseConnection): """The class for mesh connection abstraction. This inherits from :py:class:`py2p.base.BaseConnection` .. inheritance-diagram:: py2p.mesh.MeshConnection """ @inherit_doc(BaseConnection.send)
[docs] def send(self, msg_type, *args, **kargs): #type: (MeshConnection, MsgPackable, *MsgPackable, **Union[bytes, int]) -> InternalMessage msg = super(MeshConnection, self).send(msg_type, *args, **kargs) if msg and (msg.id, msg.time) not in self.server.waterfalls: self.server.waterfalls.add((msg.id, msg.time)) return msg
@inherit_doc(BaseConnection.found_terminator)
[docs] def found_terminator(self): #type: (MeshConnection) -> InternalMessage try: msg = super(MeshConnection, self).found_terminator() packets = msg.packets self.__print__("Message received: {}".format(packets), level=1) if self.handle_waterfall(msg, packets): return msg elif self.handle_renegotiate(packets): return msg self.server.handle_msg(Message(msg, self.server), self) return msg except (IndexError, struct.error): self.__print__( "Failed to decode message. Expected first compression of: %s." % intersect(compression, self.compression), level=1) self.send(flags.renegotiate, flags.compression, []) self.send(flags.renegotiate, flags.resend)
[docs] def handle_waterfall(self, msg, packets): #type: (MeshConnection, InternalMessage, Tuple[MsgPackable, ...]) -> bool """This method determines whether this message has been previously received or not. If it has been previously received, this method returns ``True``. If it is older than a preset limit, this method returns ``True``. Otherwise this method returns ``False``, and forwards the message appropriately. Args: msg: The message in question packets: The message's packets Returns: Either ``True`` or ``False`` """ if packets[0] == flags.broadcast: if msg.time < getUTC() - 60: self.__print__("Waterfall expired", level=2) return True elif not self.server.waterfall(Message(msg, self.server)): self.__print__("Waterfall already captured", level=2) return True self.__print__( "New waterfall received. Proceeding as normal", level=2) return False
[docs]class MeshDaemon(BaseDaemon): """The class for mesh daemon. This inherits from :py:class:`py2p.base.BaseDaemon` .. inheritance-diagram:: py2p.mesh.MeshDaemon """ @log_entry('py2p.mesh.MeshDaemon', DEBUG) @inherit_doc(BaseDaemon.__init__) def __init__(self, *args, **kwargs): #type: (Any, *Any, **Any) -> None super(MeshDaemon, self).__init__(*args, **kwargs) self.conn_type = MeshConnection if system() != 'Java': def mainloop(self): #type: (MeshDaemon) -> None """Daemon thread which handles all incoming data and connections""" while self.main_thread.is_alive() and self.alive: conns = chain(self.server.routing_table.values(), self.server.awaiting_ids, (self.sock, )) for handler in select.select( cast(Sequence, conns), [], [], 0.01)[0]: if handler == self.sock: self.handle_accept() else: self.process_data(handler) for handler in chain( tuple(self.server.routing_table.values()), self.server.awaiting_ids): self.kill_old_nodes(handler) else:
[docs] def mainloop(self): #type: (MeshDaemon) -> None """Daemon thread which handles all incoming data and connections""" while self.main_thread.is_alive() and self.alive: conns = tuple( chain(self.server.routing_table.values(), self.server.awaiting_ids, (self.sock, ))) for handler in select.select(conns, [], [], 0.01)[0]: if handler == self.sock: self.handle_accept() else: self.process_data(handler) for handler in chain( tuple(self.server.routing_table.values()), self.server.awaiting_ids): self.kill_old_nodes(handler)
[docs] def handle_accept(self): #type: (MeshDaemon) -> Union[None, MeshConnection] """Handle an incoming connection""" if sys.version_info >= (3, 3): exceptions = (socket.error, ConnectionError) else: exceptions = (socket.error, ) try: conn, addr = self.sock.accept() self.__print__('Incoming connection from %s' % repr(addr), level=1) handler = self.conn_type(conn, self.server) self.server._send_handshake(handler) handler.sock.settimeout(1) self.server.awaiting_ids.append(handler) return handler except exceptions: pass
[docs]class MeshSocket(BaseSocket): """The class for mesh socket abstraction. This inherits from :py:class:`py2p.base.BaseSocket` .. inheritance-diagram:: py2p.mesh.MeshSocket Added Events: .. raw:: html <div id="MeshSocket.Event 'connect'"></div> .. py:function:: Event 'connect'(conn) This event is called whenever you have a *new* connection to the service network. In other words, whenever the length of your routing table is increased from zero to one. If you call ``on('connect')``, that will be executed on every connection to the network. So if you are suddenly disconnected, and manage to recover, that function will execute again. To avoid this, call ``once('connect')``. That will usually be more correct. :param py2p.mesh.MeshSocket conn: A reference to this abstract socket .. raw:: html <div id="MeshSocket.Event 'message'"></div> .. py:function:: Event 'message'(conn) This event is called whenever you receive a new message. A reference to the message is *not* passed to you. This is to prevent potential memory leaks. If you want to register a "privileged" handler which *does* get a reference to the message, see :py:func:`~py2p.mesh.MeshSocket.register_handler` :param py2p.mesh.MeshSocket conn: A reference to this abstract socket """ __slots__ = ('requests', 'waterfalls', 'queue', 'daemon') @log_entry('py2p.mesh.MeshSocket', DEBUG) def __init__( self, #type: Any addr, #type: str port, #type: int prot=default_protocol, #type: Protocol out_addr=None, #type: Union[None, Tuple[str, int]] debug_level=0 #type: int ): #type: (...) -> None """Initializes a mesh socket Args: addr: The address you wish to bind to (ie: "192.168.1.1") port: The port you wish to bind to (ie: 44565) prot: The Protocol you wish to operate over, defined by a :py:class:`py2p.base.Protocol` object out_addr: Your outward facing address. Only needed if you're connecting over the internet. If you use '0.0.0.0' for the addr argument, this will automatically be set to your LAN address. debug_level: The verbosity you want this socket to use when printing event data Raises: socket.error: The address you wanted could not be bound, or is otherwise used """ if not hasattr(self, 'daemon'): self.daemon = 'mesh reserved' super(MeshSocket, self).__init__(addr, port, prot, out_addr, debug_level) # Metadata about msg replies where you aren't connected to the sender self.requests = { } #type: Dict[Union[bytes, Tuple[bytes, bytes]], Union[Tuple[MsgPackable, ...], awaiting_value]] # Metadata of messages to waterfall self.waterfalls = set() #type: Set[Tuple[bytes, int]] # Queue of received messages. Access through recv() self.queue = deque() #type: deque if self.daemon == 'mesh reserved': self.daemon = MeshDaemon(addr, port, self) self.register_handler(self.__handle_handshake) self.register_handler(self._handle_peers) self.register_handler(self.__handle_response) self.register_handler(self.__handle_request) @inherit_doc(BaseSocket.handle_msg)
[docs] def handle_msg(self, msg, conn): #type: (MeshSocket, Message, BaseConnection) -> Union[bool, None] if not super(MeshSocket, self).handle_msg(msg, conn): if msg.packets[0] in (flags.whisper, flags.broadcast): self.queue.appendleft(msg) self.emit('message', self) else: self.__print__( "Ignoring message with invalid subflag", level=4) return True
[docs] def _get_peer_list(self): #type: (MeshSocket) -> List[Tuple[Tuple[str, int], bytes]] """This function is used to generate a list-formatted group of your peers. It goes in format ``[ ((addr, port), ID), ...]`` """ peer_list = [(node.addr, key) for key, node in self.routing_table.items() if node.addr] shuffle(peer_list) return peer_list
[docs] def _send_handshake(self, handler): #type: (MeshSocket, MeshConnection) -> None """Shortcut method for sending a handshake to a given handler Args: handler: A :py:class:`~py2p.mesh.MeshConnection` """ tmp_compress = handler.compression handler.compression = [] handler.send(flags.whisper, flags.handshake, self.id, self.protocol.id, self.out_addr, compression) handler.compression = tmp_compress
def __resolve_connection_conflict(self, handler, h_id): #type: (MeshSocket, BaseConnection, bytes) -> None """Sometimes in trying to recover a network a race condition is created. This function applies a heuristic to try and organize the fallout from that race condition. While it isn't perfect, it seems to have increased connection recovery rate from ~20% to ~75%. This statistic is from memory on past tests. Much improvement can be made here, but this statistic can likely never be brought to 100%. In the failure condition, the overall network is unaffected *for large networks*. In small networks this failure condition causes a fork, usually where an individual node is kicked out. Args: handler: The handler with whom you have a connection conflict h_id: The id of this handler """ self.__print__( "Resolving peer conflict on id %s" % repr(h_id), level=1) to_keep, to_kill = None, None #type: Union[None, BaseConnection], Union[None, BaseConnection] if (bool(from_base_58(self.id) > from_base_58(h_id)) ^ bool(handler.outgoing)): # logical xor self.__print__("Closing outgoing connection", level=1) to_keep, to_kill = self.routing_table[h_id], handler self.__print__(to_keep.outgoing, level=1) else: self.__print__("Closing incoming connection", level=1) to_keep, to_kill = handler, self.routing_table[h_id] self.__print__(not to_keep.outgoing, level=1) self.disconnect(cast(MeshConnection, to_kill)) self.routing_table.update({h_id: to_keep})
[docs] def _send_peers(self, handler): #type: (MeshSocket, BaseConnection) -> None """Shortcut method to send a handshake response. This method is extracted from :py:meth:`.__handle_handshake` in order to allow cleaner inheritence from :py:class:`py2p.sync.SyncSocket` """ handler.send(flags.whisper, flags.peers, cast(MsgPackable, self._get_peer_list()))
def __handle_handshake(self, msg, handler): #type: (MeshSocket, Message, BaseConnection) -> Union[bool, None] """This callback is used to deal with handshake signals. Its three primary jobs are: - reject connections seeking a different network - set connection state - deal with connection conflicts Args: msg: A :py:class:`~py2p.base.Message` handler: A :py:class:`~py2p.mesh.MeshConnection` Returns: Either ``True`` or ``None`` """ packets = msg.packets if packets[0] == flags.handshake and len(packets) == 5: if packets[2] != self.protocol.id: self.__print__( "Connected to peer on wrong subnet. ID: %s" % packets[2], level=2) self.disconnect(cast(MeshConnection, handler)) return True elif not handler.addr and len(self.routing_table) == 0: self.emit('connect', self) elif handler is not self.routing_table.get(packets[1], handler): self.__print__( "Connection conflict detected. Trying to resolve", level=2) self.__resolve_connection_conflict(handler, packets[1]) handler.id = packets[1] handler.addr = packets[3] handler.compression = packets[4] self.__print__( "Compression methods changed to %s" % repr(handler.compression), level=4) if handler in self.awaiting_ids: self.awaiting_ids.remove(handler) self.routing_table.update({packets[1]: handler}) self._send_peers(handler) return True
[docs] def _handle_peers(self, msg, handler): #type: (MeshSocket, Message, BaseConnection) -> Union[bool, None] """This callback is used to deal with peer signals. Its primary jobs is to connect to the given peers, if this does not exceed :py:const:`py2p.mesh.max_outgoing` Args: msg: A :py:class:`~py2p.base.Message` handler: A :py:class:`~py2p.mesh.MeshConnection` Returns: Either ``True`` or ``None`` """ packets = msg.packets if packets[0] == flags.peers: new_peers = packets[1] for addr, id in new_peers: if len(tuple(self.outgoing)) < max_outgoing: try: self.connect(addr[0], addr[1], id.encode()) except: # pragma: no cover self.__print__( "Could not connect to %s because\n%s" % (addr, format_exc()), level=1) continue return True
def __handle_response(self, msg, handler): #type: (MeshSocket, Message, BaseConnection) -> Union[bool, None] """This callback is used to deal with response signals. Its two primary jobs are: - if it was your request, send the deferred message - if it was someone else's request, relay the information Args: msg: A :py:class:`~py2p.base.Message` handler: A :py:class:`~py2p.mesh.MeshConnection` Returns: Either ``True`` or ``None`` """ packets = msg.packets if packets[0] == flags.response: self.__print__( "Response received for request id %s" % packets[1], level=1) if self.requests.get(packets[1]): addr = packets[2] if addr: _msg = cast(Tuple[MsgPackable, ...], self.requests.get(packets[1])) self.requests.pop(packets[1]) self.connect(addr[0][0], addr[0][1], addr[1]) self.routing_table[addr[1]].send(*_msg) return True def __handle_request(self, msg, handler): #type: (MeshSocket, Message, BaseConnection) -> Union[bool, None] """This callback is used to deal with request signals. Its three primary jobs are: - respond with a peers signal if packets[1] is ``'*'`` - if you know the ID requested, respond to it - if you don't, make a request with your peers Args: msg: A :py:class:`~py2p.base.Message` handler: A :py:class:`~py2p.mesh.MeshConnection` Returns: Either ``True`` or ``None`` """ packets = msg.packets if packets[0] == flags.request: if packets[1] == b'*': handler.send(flags.whisper, flags.peers, cast(MsgPackable, self._get_peer_list())) elif self.routing_table.get(packets[2]): handler.send( flags.broadcast, flags.response, packets[1], [self.routing_table.get(packets[2]).addr, packets[2]]) return True
[docs] def send(self, *args, **kargs): #type: (MeshSocket, *MsgPackable, **MsgPackable) -> None """This sends a message to all of your peers. If you use default values it will send it to everyone on the network Args: *args: A list of objects you want your peers to receive **kargs: There are two keywords available: flag: A string or bytes-like object which defines your flag. In other words, this defines packet 0. type: A string or bytes-like object which defines your message type. Changing this from default can have adverse effects. Raises: TypeError: If any of the arguments are not serializable. This means your objects must be one of the following: - :py:class:`bool` - :py:class:`float` - :py:class:`int` (if ``2**64 > x > -2**63``) - :py:class:`str` - :py:class:`bytes` - :py:class:`unicode` - :py:class:`tuple` - :py:class:`list` - :py:class:`dict` (if all keys are :py:class:`unicode`) Warning: If you change the type attribute from default values, bad things could happen. It **MUST** be a value from :py:data:`py2p.base.flags`, and more specifically, it **MUST** be either ``broadcast`` or ``whisper``. The only other valid flags are ``waterfall`` and ``renegotiate``, but these are **RESERVED** and must **NOT** be used. """ send_type = kargs.pop('type', flags.broadcast) main_flag = kargs.pop('flag', flags.broadcast) # map(methodcaller('send', 'broadcast', 'broadcast', *args), # self.routing_table.values()) handlers = list(self.routing_table.values()) for handler in handlers: handler.send(main_flag, send_type, *args)
def __clean_waterfalls(self): #type: (MeshSocket) -> None """This function cleans the :py:class:`set` of recently relayed messages based on the following heuristics: * Delete all older than 60 seconds """ self.waterfalls = {i for i in self.waterfalls if i[1] > getUTC() - 60}
[docs] def waterfall(self, msg): #type: (MeshSocket, Message) -> bool """This function handles message relays. Its return value is based on whether it took an action or not. Args: msg: The :py:class:`~py2p.base.Message` in question Returns: ``True`` if the message was then forwarded. ``False`` if not. """ if (msg.id, msg.time) not in self.waterfalls: self.waterfalls.add((msg.id, msg.time)) for handler in tuple(self.routing_table.values()): if handler.id != msg.sender: handler.send_InternalMessage(msg.msg) self.__clean_waterfalls() return True else: self.__print__("Not rebroadcasting", level=3) return False
[docs] def connect(self, addr, port, id=None, conn_type=MeshConnection): #type: (MeshSocket, str, int, bytes, Any) -> Union[None, bool] """This function connects you to a specific node in the overall network. Connecting to one node *should* connect you to the rest of the network, however if you connect to the wrong subnet, the handshake failure involved is silent. You can check this by looking at the truthiness of this objects routing table. Example: .. code:: python >>> conn = mesh.MeshSocket('localhost', 4444) >>> conn.connect('localhost', 5555) >>> # do some other setup for your program >>> if not conn.routing_table: ... conn.connect('localhost', 6666) # any fallback address Args: addr: A string address port: A positive, integral port id: A string-like object which represents the expected ID of this node """ self.__print__( "Attempting connection to %s:%s with id %s" % (addr, port, repr(id)), level=1) if (socket.getaddrinfo( addr, port)[0] == socket.getaddrinfo(*self.out_addr)[0] or id in self.routing_table): self.__print__("Connection already established", level=1) return False conn = get_socket(self.protocol, False) conn.settimeout(1) conn.connect((addr, port)) handler = conn_type(conn, self, outgoing=True) self._send_handshake(handler) if id: self.routing_table.update({id: handler}) else: self.awaiting_ids.append(handler)
[docs] def disconnect(self, handler): #type: (MeshSocket, MeshConnection) -> None """Closes a given connection, and removes it from your routing tables Args: handler: the connection you would like to close """ node_id = handler.id #type: Union[bytes, str] if not node_id: node_id = repr(handler) self.__print__( "Connection to node %s has been closed" % node_id, level=1) if handler in self.awaiting_ids: self.awaiting_ids.remove(handler) elif self.routing_table.get(handler.id) is handler: self.routing_table.pop(handler.id) try: handler.sock.shutdown(socket.SHUT_RDWR) except: pass
[docs] def request_peers(self): #type: (MeshSocket) -> None """Requests your peers' routing tables""" self.send('*', type=flags.request, flag=flags.whisper)
[docs] def recv(self, quantity=1): #type: (MeshSocket, int) -> Union[None, Message, List[Message]] """This function has two behaviors depending on whether quantity is left as default. If quantity is given, it will return a list of :py:class:`~py2p.base.Message` objects up to length quantity. If quantity is left alone, it will return either a single :py:class:`~py2p.base.Message` object, or ``None`` Args: quantity: The maximum number of :py:class:`~py2p.base.Message` s you would like to pull (default: 1) Returns: A list of :py:class:`~py2p.base.Message` s, an empty list, a single :py:class:`~py2p.base.Message` , or ``None`` """ if quantity != 1: ret_list = [] while len(self.queue) and quantity > 0: ret_list.append(self.queue.pop()) quantity -= 1 return ret_list elif len(self.queue): return self.queue.pop() else: return None