from __future__ import annotations from dataclasses import dataclass from functools import reduce from hashlib import sha256 from time import time from nacl.public import Box, SealedBox from nacl.signing import SigningKey, VerifyKey, SignedMessage from queue import SimpleQueue from random import randint from secrets import token_bytes from typing import Protocol import readline import struct """Proof-of-concept gossip protocol implementation. To use in a real application, import the module; create new handlers as necessary for the duck typed protocols SupportsSendAndDeliverMessage, SupportsHandleMessage, and SupportsHandleAction; and register the handlers as shown in main. Note that Node.from_seed is used only for the active node(s) while Node.__init__ is can be used for neighbors for which we know only the public key/address. Most models include a data or metadata property for extensibility. The run_tick, format_address, and action_count functions can be used without modification. Debug message handling can be swapped out from print to a custom function using deregister_debug_handler(print) and register_debug_handler(custom_func), e.g. to write to a log file. """ def license() -> str: """Copyleft (c) 2022 k98kurz Permission to use, copy, modify, and/or distribute this software for any purpose with or without fee is hereby granted, provided that the above copyleft notice and this permission notice appear in all copies. THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. """ return license.__doc__ # global config and toggle/utility functions ENABLE_DEBUG = True DISPLAY_SHORT_ADDRESSES = True SIGN_MESSAGES = True ENCRYPT_MESSAGES = True MESSAGE_TTL = 300 DEBUG_HANDLERS = [print] def format_address(address: bytes) -> str: global DISPLAY_SHORT_ADDRESSES return address.hex()[:8] if DISPLAY_SHORT_ADDRESSES else address.hex() def toggle_short_address() -> bool: global DISPLAY_SHORT_ADDRESSES DISPLAY_SHORT_ADDRESSES = not DISPLAY_SHORT_ADDRESSES return DISPLAY_SHORT_ADDRESSES def debug(msg: str): """Pass debug messages to all debug message handlers.""" global ENABLE_DEBUG, DEBUG_HANDLERS if ENABLE_DEBUG: for d in DEBUG_HANDLERS: d(msg) def register_debug_handler(c: function) -> None: """Register a new function for handling debug messages.""" if not callable(c): raise TypeError('Can only register callables as debug handlers.') global DEBUG_HANDLERS if c not in DEBUG_HANDLERS: DEBUG_HANDLERS.append(c) def unregister_debug_handler(c: function) -> None: """Unregister a function from handling debug messages.""" if not callable(c): raise TypeError('Can only deregister callables as debug handlers.') global DEBUG_HANDLERS if c in DEBUG_HANDLERS: DEBUG_HANDLERS.remove(c) def toggle_debug() -> bool: global ENABLE_DEBUG ENABLE_DEBUG = not ENABLE_DEBUG return ENABLE_DEBUG def toggle_sign_all_messages() -> bool: global SIGN_MESSAGES SIGN_MESSAGES = not SIGN_MESSAGES return SIGN_MESSAGES def toggle_encrypt_messages() -> bool: global ENCRYPT_MESSAGES ENCRYPT_MESSAGES = not ENCRYPT_MESSAGES return ENCRYPT_MESSAGES @dataclass class Connection: """Connection model represent an edge connecting two Nodes together.""" nodes: set[Node] data: dict def __init__(self, nodes: list[Node]) -> None: if type(nodes) is not list or len(nodes) != 2: raise Exception('a Connection must connect exactly 2 nodes') self.nodes = set(nodes) self.data = {} def __hash__(self) -> int: """Enable inclusion in sets.""" node_list = list(self.nodes) node_list.sort() return hash(node_list[0].address + node_list[1].address) @dataclass class Message: """Message model contains the source, destination, content, and optional signature and metadata. """ src: bytes dst: bytes ts: int msg: bytes sig: bytes metadata: dict def __init__(self, src: bytes, dst: bytes, msg: bytes, ts: int = None, sig: bytes = None) -> None: self.src = src self.dst = dst self.ts = int(time()) if ts is None else ts self.msg = msg self.sig = sig self.metadata = {} def __repr__(self) -> str: return f"{format_address(self.src)}->{format_address(self.dst)}: {format_address(sha256(self.msg).digest())}" def __bytes__(self) -> bytes: return self.src + self.dst + self.msg def __hash__(self) -> int: """Enable inclusion in sets.""" return hash(bytes(self)) def pack(self) -> bytes: """Pack the data with struct.""" if self.sig is not None: fstr = '!32s32si64s' + str(len(self.msg)) + 's' return struct.pack(fstr, self.dst, self.src, self.ts, self.sig, self.msg) else: fstr = '!32s32si' + str(len(self.msg)) + 's' return struct.pack(fstr, self.dst, self.src, self.ts, self.msg) @classmethod def unpack(cls, packed: bytes) -> Message: """Unpack the data with struct.""" global SIGN_MESSAGES if SIGN_MESSAGES: fstr = '!32s32si64s' + str(len(packed) - 128) + 's' (dst, src, ts, sig, msg) = struct.unpack(fstr, packed) return Message(src, dst, msg, ts, sig) else: fstr = '!32s32si' + str(len(packed) - 64) + 's' (dst, src, ts, msg) = struct.unpack(fstr, packed) return Message(src, dst, msg, ts) def sign(self, skey: SigningKey) -> SignedMessage: """Generate a signature for the message.""" sig = skey.sign(bytes(self)) self.sig = sig[:64] return sig def verify(self) -> bool: """Verify the message signature""" try: vkey = VerifyKey(self.src) sig = SignedMessage(self.sig + bytes(self)) vkey.verify(sig) return True except: return False def encrypt(self, skey: SigningKey) -> None: """Encrypt the message by the sender.""" if bytes(skey.verify_key) != self.src: raise ValueError('Must use the skey of the sender to encrypt.') privk = skey.to_curve25519_private_key() pubk = VerifyKey(self.dst).to_curve25519_public_key() box = Box(privk, pubk) self.msg = bytes(box.encrypt(self.msg)) def decrypt(self, skey: SigningKey) -> None: """Decrypt the message by the receiver.""" if bytes(skey.verify_key) != self.dst: raise ValueError('Must use the skey of the receiver to decrypt.') privk = skey.to_curve25519_private_key() pubk = VerifyKey(self.src).to_curve25519_public_key() box = Box(privk, pubk) self.msg = box.decrypt(self.msg) def seal(self) -> None: """Encrypt using ephemeral ECDHE.""" sealed_box = SealedBox(VerifyKey(self.dst).to_curve25519_public_key()) self.msg = sealed_box.encrypt(self.msg) def unseal(self, skey: SigningKey) -> None: """Decrypt using ephemeral ECDHE.""" if bytes(skey.verify_key) != self.dst: raise ValueError('Must use the skey of the receiver to decrypt.') privk = skey.to_curve25519_private_key() sealed_box = SealedBox(privk) self.msg = sealed_box.decrypt(self.msg) @dataclass class Action: """Action model contains the name and data for an action a Node will take by passing to the registered action handler. """ name: str data: dict def __init__(self, name: str, data: dict) -> None: self.name = name self.data = data class SupportsSendAndDeliverMessage(Protocol): """Duck type protocol for message sender.""" def send(self, msg: Message) -> None: ... def deliver(self) -> None: ... class SupportsHandleMessage(Protocol): """Duck type protocol for incoming message handler.""" def handle(self, msg: Message) -> None: ... class SupportsHandleAction(Protocol): """Duck type protocol for action handler.""" def handle(self, action: dict) -> None: ... @dataclass class Node: """The core model representing a Node and handling its Connections, Actions, and Messages. Invoke with Node(address) for neighbors and Node.from_seed(seed) for an active node. Optional data property for extensibility. The address is the public key bytes of the node for when SIGN_MESSAGES is set. """ address: bytes msgs_seen: set[bytes] connections: set[Connection] data: dict _seed: bytes _skey: SigningKey _vkey: VerifyKey _inbound: SimpleQueue _outbound: SimpleQueue _actions: SimpleQueue _message_sender: SupportsSendAndDeliverMessage _message_handler: SupportsHandleMessage _action_handler: SupportsHandleAction def __init__(self, address: bytes) -> None: """Create a node from its address (public key bytes).""" self.address = address self.msgs_seen = set() self.connections = set() self.data = {} self._vkey = VerifyKey(address) self._seed = None self._skey = None self._inbound = SimpleQueue() self._outbound = SimpleQueue() self._actions = SimpleQueue() @classmethod def from_seed(cls, seed: bytes): """Create a node from a seed filling out _skey.""" skey = SigningKey(seed) node = cls(bytes(skey.verify_key)) node._skey = skey node._seed = seed return node def __hash__(self) -> int: """Enable inclusion in sets.""" return hash(self.address) def __lt__(self, other: Node) -> bool: return self.address < other.address def __repr__(self) -> str: if self._seed is not None: return "{'address': '" + format_address(self.address) + "','seed':'" + self._seed.hex() + "}" else: return "{'address': '" + format_address(self.address) + "'}" def register_message_sender(self, sndr: SupportsSendAndDeliverMessage) -> None: """Register the message sender.""" if not hasattr(sndr, 'send') or not callable(sndr.send): raise TypeError('sndr must fulfill SupportsSendAndDeliverMessage duck type') self._message_sender = sndr def register_message_handler(self, hndlr: SupportsHandleMessage) -> None: """Register the incoming message handler.""" if not hasattr(hndlr, 'handle') or not callable(hndlr.handle): raise TypeError('hndlr must fulfill SupportsHandleMessage duck type') self._message_handler = hndlr def register_action_handler(self, hndlr: SupportsHandleAction) -> None: """Register the action handler.""" if not hasattr(hndlr, 'handle') or not callable(hndlr.handle): raise TypeError('hndlr must fulfill SupportsHandleAction duck type') self._action_handler = hndlr def add_connection(self, connection: Connection) -> None: """Add the specified connection.""" if not isinstance(connection, Connection): raise TypeError('connection must be a Connection') self.connections.add(connection) def drop_connection(self, connection: Connection) -> None: """Drop the specified connection.""" if not isinstance(connection, Connection): raise TypeError('connection must be a Connection') self.connections.remove(connection) def count_connections(self) -> int: return len(self.connections) def receive_message(self, message: Message): """Queue up an incoming message if its signature is valid or ignored. """ if not isinstance(message, Message): raise TypeError('message must be a Message') global SIGN_MESSAGES, ENCRYPT_MESSAGES, MESSAGE_TTL if int(time()) > (message.ts + MESSAGE_TTL): debug("Node.receive_message: old message discarded") elif message.sig is not None: if message.verify(): if ENCRYPT_MESSAGES: message.unseal(self._skey) self._inbound.put(message) else: debug("Node.receive_message: message signature failed verification") elif SIGN_MESSAGES: debug("Node.receive_message: unsigned message rejected") else: if ENCRYPT_MESSAGES: message.unseal(self._skey) self._inbound.put(message) def send_message(self, dst: bytes, msg: bytes): """Queue up an outgoing message. Sign if necessary and possible.""" if type(dst) is not bytes: raise TypeError("dst must be bytes") if type(msg) is not bytes: raise TypeError("msg must be bytes") message = Message(self.address, dst, msg) global ENCRYPT_MESSAGES if ENCRYPT_MESSAGES: message.seal() if self._skey is not None: message.sign(self._skey) if len(self.connections): if len([c for c in self.connections if dst in [n.address for n in c.nodes]]): self._outbound.put(message) else: debug("cannot deliver message due to lack of connection") else: self._outbound.put(message) def queue_action(self, act: Action) -> None: """Queue an action to be processed by the action handler.""" if not isinstance(act, Action): raise TypeError('act must be an Action') self._actions.put(act) def process(self): """Process actions for this node once.""" if self._outbound.qsize() > 0 and self._message_sender is not None: self._message_sender.send(self._outbound.get()) if self._inbound.qsize() > 0 and self._message_handler is not None: self._message_handler.handle(self._inbound.get()) if self._actions.qsize() > 0 and self._action_handler is not None: self._action_handler.handle(self._actions.get()) def action_count(self): """Count the size of pending messages and actions.""" return self._outbound.qsize() + self._inbound.qsize() + self._actions.qsize() @dataclass class MessageSender: """Example message sender that does not use a network stack.""" nodes: set[Node] message_queue: SimpleQueue dead_letters: list data: dict def __init__(self) -> None: self.nodes = set() self.message_queue = SimpleQueue() self.dead_letters = [] self.data = {} def register_node(self, node: Node) -> None: """Register the specified node for message delivery.""" if not isinstance(node, Node): raise TypeError('node must be a Node') self.nodes.add(node) self.check_dead_letters(node) def register_nodes(self, nodes: list[Node]) -> None: """Register the specified list of nodes for message delivery.""" if not isinstance(nodes, list): raise TypeError('nodes must be a list of Nodes') for n in nodes: self.register_node(n) def send(self, msg: Message) -> None: """Queue up the specified message for delivery.""" if not isinstance(msg, Message): raise TypeError('msg must be a Message') debug(f"MessageSender.send(): {msg}") self.message_queue.put(msg) def check_queue(self) -> int: """Return pending message queue size.""" return self.message_queue.qsize() def deliver(self) -> None: """Deliver all pending messages that can be delivered.""" while self.check_queue() > 0: msg = self.message_queue.get() found = False for n in self.nodes: if n.address == msg.dst: n.receive_message(msg) found = True debug(f'MessageSender.deliver(): delivered {format_address(sha256(msg.msg).digest())} to {format_address(n.address)}') if not found: self.dead_letters.append(msg) debug('MessageSender.deliver(): dead letter') def check_dead_letters(self, node: Node) -> None: """Goes through the dead letters to try to deliver to a newly registered node. """ if not isinstance(node, Node): raise TypeError('node must be a Node') delivered = [] for i, l in enumerate(self.dead_letters): if l.dst == node.address: node.receive_message(l) delivered.append(l) self.dead_letters[:] = [l for l in self.dead_letters if l not in delivered] @dataclass class MessageHandler: """Example message handler. Optional data property for extensibility.""" nodes: set[Node] data: dict def __init__(self) -> None: self.nodes = set() self.data = {} def register_node(self, node: Node) -> None: """Register the specified node for handling incoming messages.""" if not isinstance(node, Node): raise TypeError('node must be a Node') self.nodes.add(node) def register_nodes(self, nodes: list[Node]) -> None: """Register the specified list of nodes for handling incoming messages. """ if not isinstance(nodes, list): raise TypeError('nodes must be a list of Nodes') for n in nodes: self.register_node(n) def handle(self, msg: Message) -> None: """Handle an incoming message.""" if not isinstance(msg, Message): raise TypeError('msg must be a Message') debug(f'MessageHandler.handle(): {msg}') if msg.dst in [n.address for n in self.nodes]: n = [n for n in self.nodes if n.address == msg.dst][0] n.queue_action(Action('store_and_forward', {"msg": msg.msg})) else: debug('MessageHandler.handle(): message dropped') @dataclass class ActionHandler: """Example action handler.""" node: Node other_nodes: set[Node] def __init__(self, node: Node, other_nodes: list[Node]) -> None: self.node = node self.other_nodes = set(other_nodes) def handle(self, act: Action) -> None: """Handle an action. Limited to store_and_forward action.""" if act.name == 'store_and_forward': if sha256(act.data['msg']).digest() not in self.node.msgs_seen: # store self.node.msgs_seen.add(sha256(act.data['msg']).digest()) debug(f"ActionHandler.handle(): store_and_forward [{act.data['msg'].hex()}]") # forward if self.node.count_connections() > 0: # forward to all connected nodes for c in self.node.connections: n = [n for n in c.nodes if n is not self.node][0] self.node.send_message(n.address, act.data['msg']) else: # forward to 2 nodes at random n1 = list(self.other_nodes)[randint(0, len(self.other_nodes)-1)] n2 = list(self.other_nodes)[randint(0, len(self.other_nodes)-1)] self.node.send_message(n1.address, act.data['msg']) self.node.send_message(n2.address, act.data['msg']) else: debug(f"ActionHandler.handle(): store_and_forward skipped for seen message") def run_tick(nodes: list[Node], msg_sender: MessageSender): """Run the process for all nodes, then deliver all pending messages.""" for n in nodes: n.process() msg_sender.deliver() def action_count(nodes: list[Node]): """Returns a count of all pending actions and messages.""" if not isinstance(nodes, list): raise TypeError('nodes must be list of Nodes') return reduce(lambda c, n: c + n.action_count(), nodes, 0) def main(): # create some nodes nodes = [Node.from_seed(token_bytes(32)) for i in range(16)] # create handlers msg_handler = MessageHandler() msg_handler.register_nodes(nodes) msg_sender = MessageSender() msg_sender.register_nodes(nodes) # register handlers for n in nodes: n.register_message_handler(msg_handler) n.register_message_sender(msg_sender) n.register_action_handler(ActionHandler(n, [on for on in nodes if on is not n])) # flag for exit end_signal = False # access global variables global SIGN_MESSAGES, ENCRYPT_MESSAGES while not end_signal: data = input("$: ") command = data.split(' ')[0].strip() data = ' '.join(data.split(' ')[1:]).strip() if command in ('quit', 'q'): end_signal = True elif command in ('list', 'nodes', 'l', 'n', 'ln'): for n in nodes: print(f"{format_address(n.address)}: {[format_address(m) for m in n.msgs_seen]}") elif command in ('listcon', 'connections', 'lc'): connections = set() for n in nodes: connections = connections.union(n.connections) for c in connections: cnodes = list(c.nodes) print(f"{format_address(cnodes[0].address)} - {format_address(cnodes[1].address)}") elif command in ('c', 'connect'): for n in nodes: others = [o for o in nodes if o is not n] for i in range(3): o = others[randint(0, len(others)-1)] n.add_connection(Connection([n, o])) o.add_connection(Connection([n, o])) elif command in ('message', 'm'): src = nodes[randint(0, len(nodes)-1)] message = Message(src.address, src.address, bytes(data, 'utf-8')) if ENCRYPT_MESSAGES: message.seal() if SIGN_MESSAGES: message.sign(src._skey) src.receive_message(message) elif command in ('d', 'debug'): print("debug enabled" if toggle_debug() else "debug disabled") elif command in ('s', 'short'): print("short addresses enabled" if toggle_short_address() else "short addresses disabled") elif command in ('r', 'run'): while action_count(nodes) > 0: run_tick(nodes, msg_sender) elif command in ('h', 'help', '?'): print("options:\t[l|ln|nodes|list] to list nodes and messages seen by each") print("\t\t[m|message] {str} to send a message") print("\t\t[c|connect] to connect nodes together randomly") print("\t\t[lc|listcon|connections] list all connections") print("\t\t[q|quit] to end") print("\t\t[h|help|?] display this text") print("\t\t[d|debug] to toggle debug messages") print("\t\t[s|short] to toggle displaying short address format") print("\t\t[r|run] to run until no pending actions remain") print("\t\tanything else to process a tick") else: run_tick(nodes, msg_sender) if __name__ == '__main__': main()