Relay onion messages

eclair-core/src/main/resources/reference.conf

@@ -54,6 +54,7 @@ eclair {
     option_anchor_outputs = disabled
     option_anchors_zero_fee_htlc_tx = disabled
     option_shutdown_anysegwit = optional
+    option_onion_messages = optional
     trampoline_payment = disabled
     keysend = disabled
   }
@@ -333,6 +334,10 @@ eclair {
       "mempool.space"
     ]
   }
+
+  onion-messages {
+    rate-limit-per-second = 10
+  }
 }
 
 akka {

eclair-core/src/main/scala/fr/acinq/eclair/Features.scala

@@ -203,6 +203,11 @@ object Features {
     val mandatory = 26
   }
 
+  case object OnionMessages extends Feature {
+    val rfcName = "option_onion_messages"
+    val mandatory = 38
+  }
+
   // TODO: @t-bast: update feature bits once spec-ed (currently reserved here: https://github.com/lightningnetwork/lightning-rfc/issues/605)
   // We're not advertising these bits yet in our announcements, clients have to assume support.
   // This is why we haven't added them yet to `areSupported`.
@@ -231,6 +236,7 @@ object Features {
     AnchorOutputs,
     AnchorOutputsZeroFeeHtlcTx,
     ShutdownAnySegwit,
+    OnionMessages,
     KeySend
   )
 

eclair-core/src/main/scala/fr/acinq/eclair/NodeParams.scala

@@ -94,7 +94,8 @@ case class NodeParams(nodeKeyManager: NodeKeyManager,
                       maxPaymentAttempts: Int,
                       enableTrampolinePayment: Boolean,
                       balanceCheckInterval: FiniteDuration,
-                      blockchainWatchdogSources: Seq[String]) {
+                      blockchainWatchdogSources: Seq[String],
+                      onionMessageRateLimitPerSecond: Double) {
   val privateKey: Crypto.PrivateKey = nodeKeyManager.nodeKey.privateKey
 
   val nodeId: PublicKey = nodeKeyManager.nodeId
@@ -457,7 +458,8 @@ object NodeParams extends Logging {
       maxPaymentAttempts = config.getInt("max-payment-attempts"),
       enableTrampolinePayment = config.getBoolean("trampoline-payments-enable"),
       balanceCheckInterval = FiniteDuration(config.getDuration("balance-check-interval").getSeconds, TimeUnit.SECONDS),
-      blockchainWatchdogSources = config.getStringList("blockchain-watchdog.sources").asScala.toSeq
+      blockchainWatchdogSources = config.getStringList("blockchain-watchdog.sources").asScala.toSeq,
+      onionMessageRateLimitPerSecond = config.getDouble("onion-messages.rate-limit-per-second")
     )
   }
 }

eclair-core/src/main/scala/fr/acinq/eclair/crypto/Sphinx.scala

@@ -24,6 +24,7 @@ import fr.acinq.eclair.wire.protocol._
 import grizzled.slf4j.Logging
 import scodec.Attempt
 import scodec.bits.ByteVector
+import scodec.codecs.provide
 
 import scala.annotation.tailrec
 import scala.util.{Failure, Success, Try}
@@ -272,7 +273,7 @@ object Sphinx extends Logging {
      * When an invalid onion is received, its hash should be included in the failure message.
      */
     def hash(onion: protocol.OnionRoutingPacket): ByteVector32 =
-      Crypto.sha256(OnionCodecs.onionRoutingPacketCodec(onion.payload.length.toInt).encode(onion).require.toByteVector)
+      Crypto.sha256(OnionCodecs.onionRoutingPacketCodec(provide(onion.payload.length.toInt)).encode(onion).require.toByteVector)
 
   }
 
@@ -291,6 +292,12 @@ object Sphinx extends Logging {
     override val PayloadLength = 400
   }
 
+  /**
+   * A message onion packet is used when requesting/sending an invoice from/to a remote node when using offers (BOLT12).
+   * @param PayloadLength SHOULD be 1300 or 32768
+   */
+  case class MessagePacket(PayloadLength: Int) extends OnionRoutingPacket[Onion.MessagePacket]
+
   /**
    * A properly decrypted failure from a node in the route.
    *
@@ -378,15 +385,6 @@ object Sphinx extends Logging {
    */
   object RouteBlinding {
 
-    /**
-     * @param publicKey            introduction node's public key (which cannot be blinded since the sender need to find a route to it).
-     * @param blindingEphemeralKey blinding tweak that can be used by the introduction node to derive the private key that
-     *                             lets it decrypt the encrypted payload.
-     * @param encryptedPayload     encrypted payload that can be decrypted with the introduction node's private key and the
-     *                             blinding ephemeral key.
-     */
-    case class IntroductionNode(publicKey: PublicKey, blindingEphemeralKey: PublicKey, encryptedPayload: ByteVector)
-
     /**
      * @param blindedPublicKey     blinded public key, which hides the real public key.
      * @param blindingEphemeralKey blinding tweak that can be used by the receiving node to derive the private key that
@@ -397,13 +395,13 @@ object Sphinx extends Logging {
     case class BlindedNode(blindedPublicKey: PublicKey, blindingEphemeralKey: PublicKey, encryptedPayload: ByteVector)
 
     /**
-     * @param introductionNode the first node should not be blinded, otherwise the sender cannot locate it.
-     * @param blindedNodes     blinded nodes (not including the introduction node).
+     * @param introductionNodeId the first node, not be blinded so that the sender can locate it.
+     * @param blindedNodes       blinded nodes (including the introduction node).
      */
-    case class BlindedRoute(introductionNode: IntroductionNode, blindedNodes: Seq[BlindedNode]) {
-      val nodeIds: Seq[PublicKey] = introductionNode.publicKey +: blindedNodes.map(_.blindedPublicKey)
-      val blindingEphemeralKeys: Seq[PublicKey] = introductionNode.blindingEphemeralKey +: blindedNodes.map(_.blindingEphemeralKey)
-      val encryptedPayloads: Seq[ByteVector] = introductionNode.encryptedPayload +: blindedNodes.map(_.encryptedPayload)
+    case class BlindedRoute(introductionNodeId: PublicKey, blindedNodes: Seq[BlindedNode]) {
+      val nodeIds: Seq[PublicKey] = introductionNodeId +: blindedNodes.tail.map(_.blindedPublicKey)
+      val blindingEphemeralKeys: Seq[PublicKey] = blindedNodes.map(_.blindingEphemeralKey)
+      val encryptedPayloads: Seq[ByteVector] = blindedNodes.map(_.encryptedPayload)
     }
 
     /**
@@ -426,8 +424,7 @@ object Sphinx extends Logging {
         e = e.multiply(PrivateKey(Crypto.sha256(blindingKey.value ++ sharedSecret.bytes)))
         BlindedNode(blindedPublicKey, blindingKey, encryptedPayload ++ mac)
       }
-      val introductionNode = IntroductionNode(publicKeys.head, blindedHops.head.blindingEphemeralKey, blindedHops.head.encryptedPayload)
-      BlindedRoute(introductionNode, blindedHops.tail)
+      BlindedRoute(publicKeys.head, blindedHops)
     }
 
     /**

eclair-core/src/main/scala/fr/acinq/eclair/io/Peer.scala

@@ -21,6 +21,7 @@ import akka.event.Logging.MDC
 import akka.event.{BusLogging, DiagnosticLoggingAdapter}
 import akka.util.Timeout
 import com.google.common.net.HostAndPort
+import com.google.common.util.concurrent.RateLimiter
 import fr.acinq.bitcoin.Crypto.PublicKey
 import fr.acinq.bitcoin.{ByteVector32, Satoshi, SatoshiLong, Script}
 import fr.acinq.eclair.Features.Wumbo
@@ -34,7 +35,7 @@ import fr.acinq.eclair.io.Monitoring.Metrics
 import fr.acinq.eclair.io.PeerConnection.KillReason
 import fr.acinq.eclair.remote.EclairInternalsSerializer.RemoteTypes
 import fr.acinq.eclair.wire.protocol
-import fr.acinq.eclair.wire.protocol.{Error, HasChannelId, HasTemporaryChannelId, LightningMessage, NodeAddress, RoutingMessage, UnknownMessage, Warning}
+import fr.acinq.eclair.wire.protocol.{Error, HasChannelId, HasTemporaryChannelId, LightningMessage, NodeAddress, OnionMessage, OnionMessages, RoutingMessage, UnknownMessage, Warning}
 import scodec.bits.ByteVector
 
 import java.net.InetSocketAddress
@@ -54,6 +55,8 @@ class Peer(val nodeParams: NodeParams, remoteNodeId: PublicKey, wallet: OnChainA
 
   import Peer._
 
+  private val messageRelayRateLimiter = RateLimiter.create(nodeParams.onionMessageRateLimitPerSecond)
+
   startWith(INSTANTIATING, Nothing)
 
   when(INSTANTIATING) {
@@ -241,6 +244,16 @@ class Peer(val nodeParams: NodeParams, remoteNodeId: PublicKey, wallet: OnChainA
         d.channels.values.toSet[ActorRef].foreach(_ ! INPUT_DISCONNECTED) // we deduplicate with toSet because there might be two entries per channel (tmp id and final id)
         gotoConnected(connectionReady, d.channels)
 
+      case Event(msg: OnionMessage, d: ConnectedData) =>
+        if (nodeParams.features.hasFeature(Features.OnionMessages) && messageRelayRateLimiter.tryAcquire()) {
+          relayOnionMessage(msg)
+        }
+        stay()
+
+      case Event(Peer.SendOnionMessage(toNodeId, msg), d: ConnectedData) if toNodeId == remoteNodeId =>
+        d.peerConnection ! msg
+        stay()
+
       case Event(unknownMsg: UnknownMessage, d: ConnectedData) if nodeParams.pluginMessageTags.contains(unknownMsg.tag) =>
         context.system.eventStream.publish(UnknownMessageReceived(self, remoteNodeId, unknownMsg, d.connectionInfo))
         stay()
@@ -251,6 +264,14 @@ class Peer(val nodeParams: NodeParams, remoteNodeId: PublicKey, wallet: OnChainA
     }
   }
 
+  private def relayOnionMessage(msg: OnionMessage): Unit = {
+    OnionMessages.process(nodeParams.privateKey, msg) match {
+      case OnionMessages.DropMessage(_) => () // We ignore bad messages
+      case OnionMessages.RelayMessage(nextNodeId, dataToRelay) => context.parent ! Peer.SendOnionMessage(nextNodeId, dataToRelay)
+      case OnionMessages.ReceiveMessage(_) => () // We only relay messages
+    }
+  }
+
   whenUnhandled {
     case Event(_: Peer.OpenChannel, _) =>
       sender() ! Status.Failure(new RuntimeException("not connected"))
@@ -425,6 +446,8 @@ object Peer {
     def apply(uri: NodeURI): Connect = new Connect(uri.nodeId, Some(uri.address))
   }
 
+  case class SendOnionMessage(nodeId: PublicKey, message: OnionMessage) extends PossiblyHarmful
+
   case class Disconnect(nodeId: PublicKey) extends PossiblyHarmful
   case class OpenChannel(remoteNodeId: PublicKey, fundingSatoshis: Satoshi, pushMsat: MilliSatoshi, channelType_opt: Option[SupportedChannelType], fundingTxFeeratePerKw_opt: Option[FeeratePerKw], channelFlags: Option[Byte], timeout_opt: Option[Timeout]) extends PossiblyHarmful {
     require(pushMsat <= fundingSatoshis, s"pushMsat must be less or equal to fundingSatoshis")

eclair-core/src/main/scala/fr/acinq/eclair/io/Switchboard.scala

@@ -71,6 +71,10 @@ class Switchboard(nodeParams: NodeParams, peerFactory: Switchboard.PeerFactory)
         case None => sender() ! Status.Failure(new RuntimeException("peer not found"))
       }
 
+    case s@Peer.SendOnionMessage(nodeId, _) =>
+      val peer = createOrGetPeer(nodeId, offlineChannels = Set.empty)
+      peer forward s
+
     case o: Peer.OpenChannel =>
       getPeer(o.remoteNodeId) match {
         case Some(peer) => peer forward o

eclair-core/src/main/scala/fr/acinq/eclair/wire/protocol/LightningMessageCodecs.scala

@@ -18,6 +18,7 @@ package fr.acinq.eclair.wire.protocol
 
 import fr.acinq.eclair.wire.Monitoring.{Metrics, Tags}
 import fr.acinq.eclair.wire.protocol.CommonCodecs._
+import fr.acinq.eclair.wire.protocol.OnionCodecs.onionRoutingPacketCodec
 import fr.acinq.eclair.{Features, KamonExt}
 import scodec.bits.{BitVector, ByteVector}
 import scodec.codecs._
@@ -308,6 +309,11 @@ object LightningMessageCodecs {
       ("timestampRange" | uint32) ::
       ("tlvStream" | GossipTimestampFilterTlv.gossipTimestampFilterTlvCodec)).as[GossipTimestampFilter]
 
+  val onionMessageCodec: Codec[OnionMessage] = (
+    ("blindingKey" | publicKey) ::
+      ("onionPacket" | OnionCodecs.messageOnionPacketCodec) ::
+      ("tlvStream" | OnionMessageTlv.onionMessageTlvCodec)).as[OnionMessage]
+
   // NB: blank lines to minimize merge conflicts
 
   //
@@ -361,6 +367,7 @@ object LightningMessageCodecs {
     .typecase(263, queryChannelRangeCodec)
     .typecase(264, replyChannelRangeCodec)
     .typecase(265, gossipTimestampFilterCodec)
+    .typecase(387, onionMessageCodec)
   // NB: blank lines to minimize merge conflicts
 
   //

eclair-core/src/main/scala/fr/acinq/eclair/wire/protocol/LightningMessageTypes.scala

@@ -320,6 +320,8 @@ object ReplyChannelRange {
 
 case class GossipTimestampFilter(chainHash: ByteVector32, firstTimestamp: TimestampSecond, timestampRange: Long, tlvStream: TlvStream[GossipTimestampFilterTlv] = TlvStream.empty) extends RoutingMessage with HasChainHash
 
+case class OnionMessage(blindingKey: PublicKey, onionRoutingPacket: OnionRoutingPacket, tlvStream: TlvStream[OnionMessageTlv] = TlvStream.empty) extends LightningMessage
+
 // NB: blank lines to minimize merge conflicts
 
 //