Flag peers that leave liquidity fees unpaid

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

@@ -786,6 +786,9 @@ class Peer(val nodeParams: NodeParams,
       // instead of fulfilled, but that's overkill: if our peer fails one of those payment, they're likely malicious
       // and will fail anyway, even if we try to be clever with fees splitting.
       var remainingFees = feesOwed.max(0 msat)
+      // Set when this payment's fees are already accounted for: either we've been paid (fee credit), or a previous
+      // signature of this purchase recorded them as owed (funded).
+      var feesAccountedFor = false
       pendingOnTheFlyFunding
         .filter { case (paymentHash, _) => paymentHashes.contains(paymentHash) }
         .values.toSeq
@@ -813,13 +816,30 @@ class Peer(val nodeParams: NodeParams,
               // when on-the-fly funding is used, and it's not possible to request a splice while one is already in
               // progress.
               feeCredit = Some(nodeParams.db.liquidity.removeFeeCredit(remoteNodeId, payment.amountOut))
+              feesAccountedFor = true
             case status: OnTheFlyFunding.Status.Funded =>
               log.warning("liquidity purchase was already signed for payment_hash={} (previousTxId={}, currentTxId={})", payment.paymentHash, status.txId, e.txId)
+              feesAccountedFor = true
           }
         })
+      e.purchase.paymentDetails match {
+        case _: LiquidityAds.PaymentDetails.FromFutureHtlc if remainingFees > 0.msat && !feesAccountedFor =>
+          // Our peer withdrew some of the payments that should pay for this liquidity purchase after we validated it,
+          // by failing them, letting them time out or disconnecting. We've signed anyway, so we won't be paid those fees.
+          // They're likely malicious (but not always, they may have simply disconnected while signing): as we do when
+          // they fail funded HTLCs, we disable funding from future HTLCs to limit our exposure to fee siphoning.
+          // We track this purchase separately from payments: fulfilling the remaining payments doesn't pay those fees.
+          log.warning("liquidity purchase signed for txId={} but {} of fees won't be paid: our peer may be malicious", e.txId, remainingFees)
+          nodeParams.onTheFlyFundingConfig.liquidityPurchaseUnderpaid(e.txId, remoteNodeId)
+        case _ => ()
+      }
       stay()
 
-    case Event(e: LiquidityPurchaseAborted, _: ConnectedData) =>
+    case Event(e: LiquidityPurchaseAborted, _) =>
+      // This event is sent by the channel, so it may reach us after we've been disconnected: we must handle it in every
+      // state, otherwise we would keep waiting for a funding transaction that will never be published.
+      // If we had flagged this purchase as underpaid, we didn't lose anything since it was aborted.
+      nodeParams.onTheFlyFundingConfig.liquidityPurchaseAborted(e.txId)
       pendingOnTheFlyFunding.foreach {
         case (paymentHash, pending) => pending.status match {
           case status: OnTheFlyFunding.Status.Funded if status.channelId == e.channelId && status.txId == e.txId && status.fundingTxIndex == e.fundingTxIndex =>

eclair-core/src/main/scala/fr/acinq/eclair/payment/Monitoring.scala

@@ -36,6 +36,7 @@ object Monitoring {
     val ReceivedPaymentDuration = Kamon.timer("payment.duration.received", "Incoming payment duration")
     val RelayedPaymentDuration = Kamon.timer("payment.duration.relayed", "Duration of pending downstream HTLCs during a relay")
     val SuspiciousFromFutureHtlcRelays = Kamon.gauge("payment.on-the-fly-funding.suspicious-htlc-relays", "Number of pending on-the-fly HTLCs that are being rejected by seemingly malicious peers")
+    val UnderpaidLiquidityPurchases = Kamon.gauge("payment.on-the-fly-funding.underpaid-liquidity-purchases", "Number of liquidity purchases funded with from_future_htlc for which some fees won't be paid")
 
     // The goal of this metric is to measure whether retrying MPP payments on failing channels yields useful results.
     // Once enough data has been collected, we will update the MultiPartPaymentLifecycle logic accordingly.

eclair-core/src/main/scala/fr/acinq/eclair/payment/relay/OnTheFlyFunding.scala

@@ -46,38 +46,66 @@ object OnTheFlyFunding {
     // to limit our exposure.
     // Note that this state is flushed when restarting: node operators should explicitly remove the from_future_htlc
     // payment type from their liquidity ads configuration if they want to keep it disabled.
-    private val suspectFromFutureHtlcRelays = scala.collection.concurrent.TrieMap.empty[ByteVector32, PublicKey]
+    private val suspiciousEvents = scala.collection.concurrent.TrieMap.empty[SuspiciousEvent, PublicKey]
 
     /** We allow using from_future_htlc if we don't have too many pending payments that are abusing it. */
     def isFromFutureHtlcAllowed(remoteNodeId: PublicKey): Boolean = {
-      val suspiciousPeers = suspectFromFutureHtlcRelays.values.toSet
+      val suspiciousPeers = suspiciousEvents.values.toSet
       val alreadySuspicious = suspiciousPeers.contains(remoteNodeId)
       val maxSuspiciousReached = suspiciousPeers.size >= maxSuspiciousPeers
       !alreadySuspicious && !maxSuspiciousReached
     }
 
     /** An on-the-fly payment using from_future_htlc was failed by the remote node: they may be malicious. */
     def fromFutureHtlcFailed(paymentHash: ByteVector32, remoteNodeId: PublicKey): Unit = {
-      suspectFromFutureHtlcRelays.addOne(paymentHash, remoteNodeId)
-      Metrics.SuspiciousFromFutureHtlcRelays.withoutTags().update(suspectFromFutureHtlcRelays.size)
+      suspiciousEvents.addOne(SuspiciousEvent.FailedRelay(paymentHash), remoteNodeId)
+      updateMetrics()
     }
 
     /** If a fishy payment is fulfilled, we remove it from the list, which may re-enabled from_future_htlc. */
     def fromFutureHtlcFulfilled(paymentHash: ByteVector32): Unit = {
-      suspectFromFutureHtlcRelays.remove(paymentHash).foreach { _ =>
-        // We only need to update the metric if an entry was actually removed.
-        Metrics.SuspiciousFromFutureHtlcRelays.withoutTags().update(suspectFromFutureHtlcRelays.size)
-      }
+      // We only need to update the metric if an entry was actually removed.
+      suspiciousEvents.remove(SuspiciousEvent.FailedRelay(paymentHash)).foreach(_ => updateMetrics())
+    }
+
+    /** We signed a liquidity purchase using from_future_htlc but our peer withdrew some of the payments that should pay its fees. */
+    def liquidityPurchaseUnderpaid(txId: TxId, remoteNodeId: PublicKey): Unit = {
+      suspiciousEvents.addOne(SuspiciousEvent.UnderpaidPurchase(txId), remoteNodeId)
+      updateMetrics()
     }
 
-    /** Remove suspect payments (except those from the provided list) which potentially re-enables from_future_htlc. */
+    /** If an underpaid liquidity purchase is aborted, we haven't lost anything, which may re-enable from_future_htlc. */
+    def liquidityPurchaseAborted(txId: TxId): Unit = {
+      suspiciousEvents.remove(SuspiciousEvent.UnderpaidPurchase(txId)).foreach(_ => updateMetrics())
+    }
+
+    /** Remove suspicious events (except those from the provided list of nodes) which potentially re-enables from_future_htlc. */
     def enableFromFutureHtlc(suspiciousNodes: Set[PublicKey]): Unit = {
-      val pending = suspectFromFutureHtlcRelays.collect { case (paymentHash, remoteNodeId) if !suspiciousNodes.contains(remoteNodeId) => paymentHash }
-      pending.foreach(paymentHash => suspectFromFutureHtlcRelays.remove(paymentHash))
-      Metrics.SuspiciousFromFutureHtlcRelays.withoutTags().update(suspectFromFutureHtlcRelays.size)
+      suspiciousEvents.filterInPlace { case (_, remoteNodeId) => suspiciousNodes.contains(remoteNodeId) }
+      updateMetrics()
+    }
+
+    private def updateMetrics(): Unit = {
+      Metrics.SuspiciousFromFutureHtlcRelays.withoutTags().update(suspiciousEvents.keys.count(_.isInstanceOf[SuspiciousEvent.FailedRelay]))
+      Metrics.UnderpaidLiquidityPurchases.withoutTags().update(suspiciousEvents.keys.count(_.isInstanceOf[SuspiciousEvent.UnderpaidPurchase]))
     }
   }
 
+  // @formatter:off
+  /** Events that make us suspect a peer of abusing from_future_htlc. */
+  sealed trait SuspiciousEvent
+  object SuspiciousEvent {
+    /** Our peer failed an on-the-fly HTLC while we were still owed fees for it: we forget this if the payment is later fulfilled. */
+    case class FailedRelay(paymentHash: ByteVector32) extends SuspiciousEvent
+    /**
+     * We signed a liquidity purchase after our peer withdrew some of the payments that should pay its fees. Contrary to
+     * failed relays, fulfilling payments later doesn't mean that we've been paid for this purchase: we only forget it if
+     * the purchase is aborted (in which case we haven't lost anything).
+     */
+    case class UnderpaidPurchase(txId: TxId) extends SuspiciousEvent
+  }
+  // @formatter:on
+
   // @formatter:off
   sealed trait Status
   object Status {

eclair-core/src/test/scala/fr/acinq/eclair/payment/relay/OnTheFlyFundingSpec.scala

@@ -1098,8 +1098,8 @@ class OnTheFlyFundingSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike {
 
     val preimage = randomBytes32()
     val paymentHash = Crypto.sha256(preimage)
-    val upstream = upstreamChannel(11_000_000 msat, expiryIn, paymentHash)
-    val willAdd = proposeFunding(10_000_000 msat, expiryOut, paymentHash, upstream)
+    val upstream = upstreamChannel(21_000_000 msat, expiryIn, paymentHash)
+    val willAdd = proposeFunding(20_000_000 msat, expiryOut, paymentHash, upstream)
     val fees = LiquidityAds.Fees(10_000 sat, 5_000 sat)
     val purchase = signLiquidityPurchase(200_000 sat, LiquidityAds.PaymentDetails.FromFutureHtlc(List(paymentHash)), fees = fees)
 
@@ -1119,6 +1119,106 @@ class OnTheFlyFundingSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike {
     awaitCond(nodeParams.onTheFlyFundingConfig.isFromFutureHtlcAllowed(remoteNodeId))
   }
 
+  test("disable from_future_htlc when remote fails HTLCs before liquidity purchase is signed") { f =>
+    import f._
+
+    connect(peer)
+    val channelId = openChannel(500_000 sat)
+
+    val upstream1 = upstreamChannel(60_000_000 msat, expiryIn, paymentHash)
+    val willAdd1 = proposeFunding(50_000_000 msat, expiryOut, paymentHash, upstream1)
+    val requestFunding1 = LiquidityAds.RequestFunding(
+      100_000 sat,
+      LiquidityAds.FundingRate(10_000 sat, 500_000 sat, 0, 250, 1_000 sat, 0 sat),
+      LiquidityAds.PaymentDetails.FromFutureHtlc(paymentHash :: Nil)
+    )
+    val splice1 = createSpliceMessage(channelId, requestFunding1)
+    peerConnection.send(peer, splice1)
+    channel.expectMsg(splice1)
+
+    // Our peer fails the payment that should pay for the liquidity purchase while the splice is being negotiated.
+    peerConnection.send(peer, WillFailHtlc(willAdd1.id, paymentHash, randomBytes(42)))
+    verifyFailedUpstream(upstream1)
+
+    // We sign the splice anyway, so we won't be paid: we disable from_future_htlc for their future payments.
+    assert(nodeParams.onTheFlyFundingConfig.isFromFutureHtlcAllowed(remoteNodeId))
+    signLiquidityPurchase(requestFunding1.requestedAmount, requestFunding1.paymentDetails, channelId, requestFunding1.fees(TestConstants.feeratePerKw, isChannelCreation = false), fundingTxIndex = 1)
+    awaitCond(!nodeParams.onTheFlyFundingConfig.isFromFutureHtlcAllowed(remoteNodeId))
+
+    // Their next liquidity purchase using from_future_htlc is rejected.
+    val paymentHash2 = randomBytes32()
+    val upstream2 = upstreamChannel(60_000_000 msat, expiryIn, paymentHash2)
+    proposeFunding(50_000_000 msat, expiryOut, paymentHash2, upstream2)
+    val requestFunding2 = requestFunding1.copy(paymentDetails = LiquidityAds.PaymentDetails.FromFutureHtlc(paymentHash2 :: Nil))
+    peerConnection.send(peer, createSpliceMessage(channelId, requestFunding2))
+    assert(peerConnection.expectMsgType[CancelOnTheFlyFunding].channelId == channelId)
+    verifyFailedUpstream(upstream2)
+    channel.expectNoMessage(100 millis)
+  }
+
+  test("keep from_future_htlc disabled when the remaining payments of an underpaid liquidity purchase are fulfilled") { f =>
+    import f._
+
+    connect(peer)
+    val channelId = openChannel(500_000 sat)
+
+    // The payment is split in two parts: each of them can only pay part of the liquidity fees.
+    val preimage = randomBytes32()
+    val paymentHash = Crypto.sha256(preimage)
+    val upstream1 = upstreamChannel(7_000_000 msat, expiryIn, paymentHash)
+    val willAdd1 = proposeFunding(6_000_000 msat, expiryOut, paymentHash, upstream1)
+    val upstream2 = upstreamChannel(7_000_000 msat, expiryIn, paymentHash)
+    val willAdd2 = proposeFunding(6_000_000 msat, expiryOut, paymentHash, upstream2)
+
+    // Our peer fails one of the parts while the splice is being negotiated.
+    peerConnection.send(peer, WillFailHtlc(willAdd1.id, paymentHash, randomBytes(42)))
+    verifyFailedUpstream(upstream1)
+
+    // We sign the splice anyway: the remaining part doesn't pay all the fees, so we disable from_future_htlc.
+    val fees = LiquidityAds.Fees(5_000 sat, 5_000 sat)
+    val purchase = signLiquidityPurchase(200_000 sat, LiquidityAds.PaymentDetails.FromFutureHtlc(paymentHash :: Nil), channelId, fees, fundingTxIndex = 1)
+    awaitCond(!nodeParams.onTheFlyFundingConfig.isFromFutureHtlcAllowed(remoteNodeId))
+
+    // Our peer fulfills the remaining part: we've only been paid part of the fees, so from_future_htlc stays disabled.
+    val proposal2 = OnTheFlyFunding.Proposal(willAdd2, upstream2, Nil)
+    peer ! OnTheFlyFunding.PaymentRelayer.RelaySuccess(purchase.channelId, paymentHash, preimage, willAdd2.amount - 1.msat, proposal2 :: Nil)
+    verifyFulfilledUpstream(upstream2, preimage)
+    probe.send(peer, Peer.GetPeerInfo(Some(probe.ref.toTyped)))
+    probe.expectMsgType[Peer.PeerInfo]
+    assert(!nodeParams.onTheFlyFundingConfig.isFromFutureHtlcAllowed(remoteNodeId))
+  }
+
+  test("re-enable from_future_htlc when an underpaid liquidity purchase is aborted") { f =>
+    import f._
+
+    connect(peer)
+    val channelId = openChannel(500_000 sat)
+
+    val upstream = upstreamChannel(25_000_000 msat, expiryIn, paymentHash)
+    val willAdd = proposeFunding(20_000_000 msat, expiryOut, paymentHash, upstream)
+
+    // Our peer fails the payment while the splice is being negotiated (for example because they disconnected).
+    peerConnection.send(peer, WillFailHtlc(willAdd.id, paymentHash, randomBytes(42)))
+    verifyFailedUpstream(upstream)
+
+    // We sign the splice anyway, so we won't be paid: we disable from_future_htlc.
+    val fees = LiquidityAds.Fees(5_000 sat, 5_000 sat)
+    val purchase = signLiquidityPurchase(200_000 sat, LiquidityAds.PaymentDetails.FromFutureHtlc(paymentHash :: Nil), channelId, fees, fundingTxIndex = 1)
+    awaitCond(!nodeParams.onTheFlyFundingConfig.isFromFutureHtlcAllowed(remoteNodeId))
+
+    // An unrelated liquidity purchase is aborted: from_future_htlc stays disabled.
+    peer ! LiquidityPurchaseAborted(channelId, randomTxId(), purchase.fundingTxIndex)
+    probe.send(peer, Peer.GetPeerInfo(Some(probe.ref.toTyped)))
+    probe.expectMsgType[Peer.PeerInfo]
+    assert(!nodeParams.onTheFlyFundingConfig.isFromFutureHtlcAllowed(remoteNodeId))
+
+    // The splice is aborted before being published, so we didn't lose anything: we re-enable from_future_htlc.
+    // Note that the abort may reach us after we've been disconnected, since it is sent asynchronously by the channel.
+    disconnect(channelCount = 1)
+    peer ! LiquidityPurchaseAborted(channelId, purchase.txId, purchase.fundingTxIndex)
+    awaitCond(nodeParams.onTheFlyFundingConfig.isFromFutureHtlcAllowed(remoteNodeId))
+  }
+
   test("don't relay payments if added to fee credit while signing", Tag(withFeeCredit)) { f =>
     import f._
 
@@ -1474,6 +1574,34 @@ class OnTheFlyFundingSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike {
     assert(cfg.isFromFutureHtlcAllowed(remoteNodeId1))
     assert(cfg.isFromFutureHtlcAllowed(remoteNodeId2))
     assert(cfg.isFromFutureHtlcAllowed(remoteNodeId3))
+
+    // We detect liquidity purchases that were signed without being fully paid.
+    val txId1 = randomTxId()
+    val txId2 = randomTxId()
+    cfg.liquidityPurchaseUnderpaid(txId1, remoteNodeId1)
+    assert(!cfg.isFromFutureHtlcAllowed(remoteNodeId1))
+    assert(cfg.isFromFutureHtlcAllowed(remoteNodeId2))
+    cfg.fromFutureHtlcFailed(paymentHash2, remoteNodeId2)
+    assert(!cfg.isFromFutureHtlcAllowed(remoteNodeId3))
+    // Fulfilling a payment with the same payment_hash doesn't pay for an underpaid liquidity purchase.
+    cfg.fromFutureHtlcFulfilled(paymentHash1)
+    cfg.fromFutureHtlcFulfilled(paymentHash2)
+    assert(!cfg.isFromFutureHtlcAllowed(remoteNodeId1))
+    assert(cfg.isFromFutureHtlcAllowed(remoteNodeId2))
+    // If the liquidity purchase is aborted, we didn't lose anything.
+    cfg.liquidityPurchaseAborted(txId2) // noop
+    assert(!cfg.isFromFutureHtlcAllowed(remoteNodeId1))
+    cfg.liquidityPurchaseAborted(txId1)
+    assert(cfg.isFromFutureHtlcAllowed(remoteNodeId1))
+    // We can also manually reactivate from_future_htlc.
+    cfg.liquidityPurchaseUnderpaid(txId1, remoteNodeId1)
+    cfg.liquidityPurchaseUnderpaid(txId2, remoteNodeId2)
+    assert(!cfg.isFromFutureHtlcAllowed(remoteNodeId3))
+    cfg.enableFromFutureHtlc(Set(remoteNodeId2))
+    assert(cfg.isFromFutureHtlcAllowed(remoteNodeId1))
+    assert(!cfg.isFromFutureHtlcAllowed(remoteNodeId2))
+    cfg.enableFromFutureHtlc(Set.empty)
+    assert(cfg.isFromFutureHtlcAllowed(remoteNodeId2))
   }
 
 }