Write received onion messages to the websocket (#2091)

docs/release-notes/eclair-vnext.md

@@ -22,7 +22,7 @@ This lets plugins notify the node operator via external systems (push notificati
 
 Eclair now supports the feature `option_onion_messages`. If this feature is enabled, eclair will relay onion messages.
 It can also send onion messages with the `sendonionmessage` API.
-Messages sent to Eclair will be ignored.
+Messages sent to Eclair can be read with the websocket API.
 
 ### API changes
 

eclair-core/src/main/scala/fr/acinq/eclair/json/JsonSerializers.scala

@@ -22,9 +22,11 @@ import fr.acinq.bitcoin.{Btc, ByteVector32, ByteVector64, OutPoint, Satoshi, Tra
 import fr.acinq.eclair.balance.CheckBalance.GlobalBalance
 import fr.acinq.eclair.blockchain.fee.FeeratePerKw
 import fr.acinq.eclair.channel._
+import fr.acinq.eclair.crypto.Sphinx.RouteBlinding.BlindedRoute
 import fr.acinq.eclair.crypto.{ShaChain, Sphinx}
 import fr.acinq.eclair.db.FailureType.FailureType
 import fr.acinq.eclair.db.{IncomingPaymentStatus, OutgoingPaymentStatus}
+import fr.acinq.eclair.message.OnionMessages
 import fr.acinq.eclair.payment.PaymentFailure.PaymentFailedSummary
 import fr.acinq.eclair.payment._
 import fr.acinq.eclair.router.Router.{ChannelHop, Route}
@@ -402,6 +404,11 @@ object GlobalBalanceSerializer extends MinimalSerializer({
     JObject(JField("total", JDecimal(o.total.toDouble))) merge Extraction.decompose(o)(formats)
 })
 
+private[json] case class MessageReceivedJson(pathId: Option[ByteVector], replyPath: Option[BlindedRoute], unknownTlvs: Map[String, ByteVector])
+object OnionMessageReceivedSerializer extends ConvertClassSerializer[OnionMessages.ReceiveMessage]({ m: OnionMessages.ReceiveMessage =>
+  MessageReceivedJson(m.pathId, m.finalPayload.replyPath.map(_.blindedRoute), m.finalPayload.records.unknown.map(tlv => (tlv.tag.toString -> tlv.value)).toMap)
+})
+
 case class CustomTypeHints(custom: Map[Class[_], String]) extends TypeHints {
   val reverse: Map[String, Class[_]] = custom.map(_.swap)
 
@@ -433,7 +440,11 @@ object CustomTypeHints {
     classOf[TrampolinePaymentRelayed] -> "trampoline-payment-relayed",
     classOf[PaymentReceived] -> "payment-received",
     classOf[PaymentSettlingOnChain] -> "payment-settling-onchain",
-    classOf[PaymentFailed] -> "payment-failed"
+    classOf[PaymentFailed] -> "payment-failed",
+  ))
+
+  val onionMessageEvent: CustomTypeHints = CustomTypeHints(Map(
+    classOf[MessageReceivedJson] -> "onion-message-received"
   ))
 
   val channelStates: ShortTypeHints = ShortTypeHints(
@@ -462,6 +473,7 @@ object JsonSerializers {
     CustomTypeHints.incomingPaymentStatus +
     CustomTypeHints.outgoingPaymentStatus +
     CustomTypeHints.paymentEvent +
+    CustomTypeHints.onionMessageEvent +
     CustomTypeHints.channelStates +
     ByteVectorSerializer +
     ByteVector32Serializer +
@@ -505,6 +517,7 @@ object JsonSerializers {
     JavaUUIDSerializer +
     OriginSerializer +
     GlobalBalanceSerializer +
-    PaymentFailedSummarySerializer
+    PaymentFailedSummarySerializer +
+    OnionMessageReceivedSerializer
 
 }

eclair-node/src/main/scala/fr/acinq/eclair/api/handlers/WebSocket.scala

@@ -24,6 +24,7 @@ import akka.stream.OverflowStrategy
 import akka.stream.scaladsl.{BroadcastHub, Flow, Keep, Source}
 import fr.acinq.eclair.api.Service
 import fr.acinq.eclair.channel.{ChannelClosed, ChannelCreated, ChannelStateChanged, WAIT_FOR_INIT_INTERNAL}
+import fr.acinq.eclair.message.OnionMessages
 import fr.acinq.eclair.payment.PaymentEvent
 
 trait WebSocket {
@@ -53,6 +54,7 @@ trait WebSocket {
         context.system.eventStream.subscribe(self, classOf[ChannelCreated])
         context.system.eventStream.subscribe(self, classOf[ChannelStateChanged])
         context.system.eventStream.subscribe(self, classOf[ChannelClosed])
+        context.system.eventStream.subscribe(self, classOf[OnionMessages.ReceiveMessage])
       }
 
       def receive: Receive = {
@@ -63,6 +65,7 @@ trait WebSocket {
             flowInput.offer(serialization.write(message))
           }
         case message: ChannelClosed => flowInput.offer(serialization.write(message))
+        case message: OnionMessages.ReceiveMessage => flowInput.offer(serialization.write(message))
       }
 
     }))

eclair-node/src/test/scala/fr/acinq/eclair/api/ApiServiceSpec.scala

@@ -24,7 +24,7 @@ import akka.http.scaladsl.server.Route
 import akka.http.scaladsl.testkit.{RouteTestTimeout, ScalatestRouteTest, WSProbe}
 import akka.util.Timeout
 import de.heikoseeberger.akkahttpjson4s.Json4sSupport
-import fr.acinq.bitcoin.Crypto.PublicKey
+import fr.acinq.bitcoin.Crypto.{PrivateKey, PublicKey}
 import fr.acinq.bitcoin.{Block, ByteVector32, ByteVector64, SatoshiLong}
 import fr.acinq.eclair.ApiTypes.ChannelIdentifier
 import fr.acinq.eclair.FeatureSupport.{Mandatory, Optional}
@@ -36,16 +36,18 @@ import fr.acinq.eclair.blockchain.fee.FeeratePerKw
 import fr.acinq.eclair.channel.ChannelOpenResponse.ChannelOpened
 import fr.acinq.eclair.channel.Helpers.Closing
 import fr.acinq.eclair.channel._
+import fr.acinq.eclair.crypto.Sphinx
 import fr.acinq.eclair.db._
 import fr.acinq.eclair.io.NodeURI
 import fr.acinq.eclair.io.Peer.PeerInfo
+import fr.acinq.eclair.message.OnionMessages
 import fr.acinq.eclair.payment._
 import fr.acinq.eclair.payment.relay.Relayer.UsableBalance
 import fr.acinq.eclair.payment.send.MultiPartPaymentLifecycle.PreimageReceived
 import fr.acinq.eclair.payment.send.PaymentInitiator.SendPaymentToRouteResponse
 import fr.acinq.eclair.router.Router.PredefinedNodeRoute
 import fr.acinq.eclair.router.{NetworkStats, Router, Stats}
-import fr.acinq.eclair.wire.protocol.{ChannelUpdate, Color, NodeAddress}
+import fr.acinq.eclair.wire.protocol.{ChannelUpdate, Color, GenericTlv, MessageOnion, NodeAddress, OnionMessagePayloadTlv, TlvStream}
 import org.mockito.scalatest.IdiomaticMockito
 import org.scalatest.funsuite.AnyFunSuite
 import org.scalatest.matchers.should.Matchers
@@ -1168,6 +1170,18 @@ class ApiServiceSpec extends AnyFunSuite with ScalatestRouteTest with IdiomaticM
         assert(serialization.write(chcl) === expectedSerializedChcl)
         system.eventStream.publish(chcl)
         wsClient.expectMessage(expectedSerializedChcl)
+
+        val msgrcv = OnionMessages.ReceiveMessage(MessageOnion.FinalPayload(TlvStream[OnionMessagePayloadTlv](
+          Seq(
+            OnionMessagePayloadTlv.EncryptedData(ByteVector.empty),
+            OnionMessagePayloadTlv.ReplyPath(Sphinx.RouteBlinding.create(PrivateKey(hex"414141414141414141414141414141414141414141414141414141414141414101"), Seq(bobNodeId), Seq(hex"000000")))
+          ), Seq(
+            GenericTlv(UInt64(5), hex"1111")
+          ))), Some(hex"2222"))
+        val expectedSerializedMsgrcv = """{"type":"onion-message-received","pathId":"2222","replyPath":{"introductionNodeId":"039dc0e0b1d25905e44fdf6f8e89755a5e219685840d0bc1d28d3308f9628a3585","blindingKey":"02eec7245d6b7d2ccb30380bfbe2a3648cd7a942653f5aa340edcea1f283686619","blindedNodes":[{"blindedPublicKey":"020303f91e620504cde242df38d04599d8b4d4c555149cc742a5f12de452cbdd40","encryptedPayload":"126a26221759247584d704b382a5789f1d8c5a"}]},"unknownTlvs":{"5":"1111"}}"""
+        assert(serialization.write(msgrcv) === expectedSerializedMsgrcv)
+        system.eventStream.publish(msgrcv)
+        wsClient.expectMessage(expectedSerializedMsgrcv)
       }
   }