noise: Initial version of the noise chat plugin

Based on the WhatSat idea by @joostjager, this plugin implements a simple chat
protocol based on top of `createonion` and `sendonion`.

noise/noise.py

@@ -0,0 +1,157 @@
+#!/usr/bin/env python3
+from pyln.client import Plugin, RpcError
+from pyln.proto.primitives import varint_decode, varint_encode
+from binascii import hexlify, unhexlify
+import struct
+import string
+import random
+from io import BytesIO
+import logging
+from collections import namedtuple
+import shelve
+from pyln.proto.onion import OnionPayload
+
+plugin = Plugin()
+
+
+class Message(object):
+    def __init__(self, sender, body, signature, payment=None, id=None):
+        self.id = id
+        self.sender = sender
+        self.body = body
+        self.signature = signature
+        self.payment = payment
+
+    def to_dict(self):
+        return {
+            "id": self.id,
+            "sender": self.sender,
+            "body": self.body,
+            "signature": hexlify(self.signature).decode('ASCII'),
+            "payment": self.payment,
+        }
+
+
+def serialize_payload(n, blockheight):
+    block, tx, out = n['channel'].split('x')
+    payload = hexlify(struct.pack(
+        "!QQL",
+        int(block) << 40 | int(tx) << 16 | int(out),
+        int(n['amount_msat']),
+        blockheight + n['delay'])).decode('ASCII')
+    payload += "00" * 12
+    return payload
+
+
+def buildpath(plugin, node_id, payload, amt, exclusions):
+    blockheight = plugin.rpc.getinfo()['blockheight']
+    route = plugin.rpc.getroute(node_id, amt, 10, exclude=exclusions)['route']
+    first_hop = route[0]
+    # Need to shift the parameters by one hop
+    hops = []
+    for h, n in zip(route[:-1], route[1:]):
+        # We tell the node h about the parameters to use for n (a.k.a. h + 1)
+        hops.append({
+            "type": "legacy",
+            "pubkey": h['id'],
+            "payload": serialize_payload(n, blockheight)
+        })
+
+    # The last hop has a special payload:
+    hops.append({
+        "type": "tlv",
+        "pubkey": route[-1]['id'],
+        "payload": hexlify(payload).decode('ASCII'),
+    })
+    return first_hop, hops, route
+
+
+def deliver(node_id, payload, amt, max_attempts=5, payment_hash=None):
+    """Do your best to deliver `payload` to `node_id`.
+    """
+    if payment_hash is None:
+        payment_hash = ''.join(random.choice(string.hexdigits) for _ in range(64)).lower()
+
+    exclusions = []
+
+    for attempt in range(max_attempts):
+        logging.debug("Starting attempt {} to deliver message to {}".format(attempt, node_id))
+
+        first_hop, hops, route = buildpath(plugin, node_id, payload, amt, exclusions)
+        onion = plugin.rpc.createonion(hops=hops, assocdata=payment_hash)
+
+        plugin.rpc.sendonion(onion=onion['onion'],
+                             first_hop=first_hop,
+                             payment_hash=payment_hash,
+                             shared_secrets=onion['shared_secrets']
+        )
+        try:
+            plugin.rpc.waitsendpay(payment_hash=payment_hash)
+            return {'route': route, 'payment_hash': payment_hash, 'attempt': attempt}
+        except RpcError as e:
+            print(e)
+            failcode = e.error['data']['failcode']
+            if failcode == 16399:
+                return {'route': route, 'payment_hash': payment_hash, 'attempt': attempt+1}
+
+            plugin.log("Retrying delivery.")
+
+            # TODO Store the failing channel in the exclusions
+    raise ValueError('Could not reach destination {node_id}'.format(node_id=node_id))
+
+
+@plugin.async_method('sendmsg')
+def sendmsg(node_id, msg, plugin, request, amt=1000, **kwargs):
+    payload = BytesIO()
+    varint_encode(34349334, payload)
+    varint_encode(len(msg), payload)
+    payload.write(msg.encode('UTF-8'))
+
+    # Sign the message:
+    sig = plugin.rpc.signmessage(msg)['signature']
+    sig = unhexlify(sig)
+    varint_encode(34349336, payload)
+    varint_encode(len(sig), payload)
+    payload.write(sig)
+
+    res = deliver(node_id, payload.getbuffer(), amt=amt)
+    request.set_result(res)
+
+
+@plugin.async_method('recvmsg')
+def recvmsg(plugin, request, last_id=None, **kwargs):
+    next_id = int(last_id) + 1 if last_id is not None else len(plugin.messages)
+    if next_id < len(plugin.messages):
+        request.set_result(plugin.messages[int(last_id)].to_dict())
+    else:
+        plugin.receive_waiters.append(request)
+
+
+@plugin.hook('htlc_accepted')
+def on_htlc_accepted(onion, htlc, plugin, **kwargs):
+    payload = OnionPayload.from_hex(onion['payload'])
+
+    # TODO verify the signature to extract the sender
+
+    msg = Message(
+        id=len(plugin.messages),
+        sender="AAA",
+        body=payload.get(34349334).value,
+        signature=payload.get(34349336).value,
+        payment=None)
+
+    plugin.messages.append(msg)
+    for r in plugin.receive_waiters:
+        r.set_result(msg.to_dict())
+    plugin.receive_waiters = []
+
+    return {'result': 'continue'}
+
+
+@plugin.init()
+def init(configuration, options, plugin, **kwargs):
+    print("Starting noise chat plugin")
+    plugin.messages = []
+    plugin.receive_waiters = []
+
+plugin.run()

noise/requirements.txt

@@ -0,0 +1,2 @@
+zbase32==1.1.5
+pyln-client>=0.0.7.3

noise/test_chat.py

@@ -0,0 +1,54 @@
+from pyln.testing.fixtures import *
+from pyln.testing.utils import wait_for
+from pprint import pprint
+
+plugin = os.path.join(os.path.dirname(__file__), 'noise.py')
+
+
+def test_sendmsg(node_factory, executor):
+    opts = [{'plugin': plugin}, {}, {'plugin': plugin}]
+    l1, l2, l3 = node_factory.line_graph(3, wait_for_announce=True, opts=opts)
+
+    recv = executor.submit(l3.rpc.recvmsg)
+    l1.rpc.sendmsg(l3.info['id'], "Hello world!")
+
+    # This one is tailing the incoming messages
+    m1 = recv.result(10)
+
+    # This one should get the same result:
+    m2 = l3.rpc.recvmsg(last_id=-1)
+    # They should be the same :-)
+    assert(m1 == m2)
+
+
+def test_sendmsg_retry(node_factory, executor):
+    opts = [{'plugin': plugin}, {}, {'fee-base': 10000}, {'plugin': plugin}]
+    l1, l2, l3, l4 = node_factory.line_graph(4, opts=opts)
+    l5 = node_factory.get_node()
+
+    l2.openchannel(l5, 10**6)
+    l5.openchannel(l4, 10**6)
+
+    def gossip_synced(nodes):
+        for a, b in zip(nodes[:-1], nodes[1:]):
+            if a.rpc.listchannels() != b.rpc.listchannels():
+                return False
+        return True
+
+    wait_for(lambda: gossip_synced([l1, l2, l3, l4, l5]))
+
+    # Now stop l5 so the first attempt will fail.
+    l5.stop()
+
+    recv = executor.submit(l4.rpc.recvmsg)
+
+    send = executor.submit(l1.rpc.sendmsg, l4.info['id'], "Hello world!")
+
+    l1.daemon.wait_for_log(r'Retrying delivery')
+
+    sres = send.result(10)
+    assert(sres['attempt'] == 2)
+    pprint(sres)
+    print(recv.result(10))
+
+    msg = l4.rpc.recvmsg(last_id=-1)