Close connection when receiving malformed messages (#3273)
What the developer wrote
Authored by Bastien Teinturier
Close connection when receiving malformed messages (#3273)
When receiving a message that we're supposed to be able to decode, for
which decoding fails, we must send a warning and close the connection.
This ensures that the connection state cannot become corrupted, which
may otherwise happen if we fail to deserialize a `commit_sig` message
(for example) and simply ignored it while continuing to process the
next messages.
Fixes #3271
What changed, and why it matters
This patch fixes a bug in Eclair, a Bitcoin Lightning Network node implementation. Previously, if a peer sent a message that Eclair couldn't properly decode (for example, a corrupted or unexpected 'commit_sig' message), Eclair would log a warning but otherwise ignore it and keep processing later messages. This could cause the two peers' views of the channel state to drift out of sync, potentially leading to incorrect behavior or security issues. The fix makes Eclair send a warning back to the peer and immediately close the connection when it fails to decode a message it expected to understand.
Reviewers should verify that the Warning message is correctly encoded and sent before stop(FSM.Normal) terminates the actor, confirm that no other code paths silently ignore codec failures, and consider whether an authenticated peer can intentionally trigger a close-of-connection as a denial-of-service vector. The fix should be backported to supported release branches.
Security signals we found
state desynchronization risk from ignored deserialization failures
missing error handling on Attempt.Failure in message codec
connection now closed on malformed messages with a warning
test renamed and rewritten to assert termination on malformed decode
Evidence from the diff
TransportHandler.scala’s decodeAndSendToListener previously returned Map[LightningMessage, Int] and silently swallowed Attempt.Failure deserialization errors. The patch changes the return type to Either[ByteVector, Map[LightningMessage, Int]], returning Left(plaintext) on the first decode failure. Both call sites (WaitingForListenerData and NormalData handling of Tcp.Received) now pattern-match on the result: on Left(msg) they log a warning, send a Warning message containing the offending hex to the peer, and stop the FSM. The test is updated from ‘handle unknown messages’ to ‘close connection on malformed messages’ and now asserts that a Warning is sent and both TransportHandler actors terminate.
Changed components
eclair-core/src/main/scala/fr/acinq/eclair/crypto/TransportHandler.scalaeclair-core/src/test/scala/fr/acinq/eclair/crypto/TransportHandlerSpec.scalaInspect captured patch +43 / −35
diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/crypto/TransportHandler.scala b/eclair-core/src/main/scala/fr/acinq/eclair/crypto/TransportHandler.scala
index 74d9b2f..25e9277 100644
--- a/eclair-core/src/main/scala/fr/acinq/eclair/crypto/TransportHandler.scala
+++ b/eclair-core/src/main/scala/fr/acinq/eclair/crypto/TransportHandler.scala
@@ -95,7 +95,7 @@ class TransportHandler(keyPair: KeyPair, rs: Option[ByteVector], connection: Act
/** We keep track of pending pings to defend against ping flooding. */
private var pendingPings = 0
- private def decodeAndSendToListener(listener: ActorRef, plaintextMessages: Seq[ByteVector]): Map[LightningMessage, Int] = {
+ private def decodeAndSendToListener(listener: ActorRef, plaintextMessages: Seq[ByteVector]): Either[ByteVector, Map[LightningMessage, Int]] = {
log.debug("decoding {} plaintext messages", plaintextMessages.size)
var m = Map.empty[LightningMessage, Int]
plaintextMessages.foreach(plaintext => codec.decode(plaintext.bits) match {
@@ -106,16 +106,17 @@ class TransportHandler(keyPair: KeyPair, rs: Option[ByteVector], connection: Act
pendingPings += 1
if (pendingPings > 1) {
// We will kill the connection anyway, no need to process remaining messages
- return m
+ return Right(m)
}
}
listener ! message
m += (message -> (m.getOrElse(message, 0) + 1))
case Attempt.Failure(err) =>
log.warning("cannot deserialize {}: {}", plaintext.toHex, err.message)
+ return Left(plaintext)
})
log.debug("decoded {} messages", m.values.sum)
- m
+ Right(m)
}
private def encodeAndSendToPeer(encryptor: Encryptor, t: LightningMessage): Encryptor = {
@@ -182,16 +183,20 @@ class TransportHandler(keyPair: KeyPair, rs: Option[ByteVector], connection: Act
case Event(Listener(listener), d@WaitingForListenerData(_, dec)) =>
context.watch(listener)
val (dec1, plaintextMessages) = dec.decrypt()
- val unackedReceived1 = decodeAndSendToListener(listener, plaintextMessages)
- if (pendingPings > 1) {
- log.warning("ping flood detected (pendingPings={}): closing connection", pendingPings)
- stop(FSM.Normal)
- } else {
- if (unackedReceived1.isEmpty) {
- log.debug("no decoded messages, resuming reading")
- connection ! Tcp.ResumeReading
- }
- goto(Normal) using NormalData(d.encryptor, dec1, listener, sendBuffer = SendBuffer(Queue.empty[LightningMessage], Queue.empty[LightningMessage]), unackedReceived = unackedReceived1, unackedSent = None)
+ decodeAndSendToListener(listener, plaintextMessages) match {
+ case Left(msg) =>
+ log.warning("malformed message detected: closing connection")
+ encodeAndSendToPeer(d.encryptor, Warning(s"could not decode message: ${msg.toHex}"))
+ stop(FSM.Normal)
+ case _ if pendingPings > 1 =>
+ log.warning("ping flood detected (pendingPings={}): closing connection", pendingPings)
+ stop(FSM.Normal)
+ case Right(unackedReceived1) =>
+ if (unackedReceived1.isEmpty) {
+ log.debug("no decoded messages, resuming reading")
+ connection ! Tcp.ResumeReading
+ }
+ goto(Normal) using NormalData(d.encryptor, dec1, listener, sendBuffer = SendBuffer(Queue.empty[LightningMessage], Queue.empty[LightningMessage]), unackedReceived = unackedReceived1, unackedSent = None)
}
}
}
@@ -201,16 +206,20 @@ class TransportHandler(keyPair: KeyPair, rs: Option[ByteVector], connection: Act
case Event(Tcp.Received(data), d: NormalData) =>
log.debug("received chunk of size={}", data.size)
val (dec1, plaintextMessages) = d.decryptor.copy(buffer = d.decryptor.buffer ++ data).decrypt()
- val unackedReceived1 = decodeAndSendToListener(d.listener, plaintextMessages)
- if (pendingPings > 1) {
- log.warning("ping flood detected (pendingPings={}): closing connection", pendingPings)
- stop(FSM.Normal)
- } else {
- if (unackedReceived1.isEmpty) {
- log.debug("no decoded messages, resuming reading")
- connection ! Tcp.ResumeReading
- }
- stay() using d.copy(decryptor = dec1, unackedReceived = unackedReceived1)
+ decodeAndSendToListener(d.listener, plaintextMessages) match {
+ case Left(msg) =>
+ log.warning("malformed message detected: closing connection")
+ encodeAndSendToPeer(d.encryptor, Warning(s"could not decode message: ${msg.toHex}"))
+ stop(FSM.Normal)
+ case _ if pendingPings > 1 =>
+ log.warning("ping flood detected (pendingPings={}): closing connection", pendingPings)
+ stop(FSM.Normal)
+ case Right(unackedReceived1) =>
+ if (unackedReceived1.isEmpty) {
+ log.debug("no decoded messages, resuming reading")
+ connection ! Tcp.ResumeReading
+ }
+ stay() using d.copy(decryptor = dec1, unackedReceived = unackedReceived1)
}
case Event(ReadAck(msg: LightningMessage), d: NormalData) =>
diff --git a/eclair-core/src/test/scala/fr/acinq/eclair/crypto/TransportHandlerSpec.scala b/eclair-core/src/test/scala/fr/acinq/eclair/crypto/TransportHandlerSpec.scala
index e0d8b72..8f9adce 100644
--- a/eclair-core/src/test/scala/fr/acinq/eclair/crypto/TransportHandlerSpec.scala
+++ b/eclair-core/src/test/scala/fr/acinq/eclair/crypto/TransportHandlerSpec.scala
@@ -19,7 +19,7 @@ package fr.acinq.eclair.crypto
import akka.actor.{Actor, ActorLogging, ActorRef, OneForOneStrategy, Props, Stash, SupervisorStrategy, Terminated}
import akka.io.Tcp
import akka.testkit.{TestActorRef, TestFSMRef, TestProbe}
-import fr.acinq.eclair.{TestKitBaseClass, randomBytes32}
+import fr.acinq.eclair.TestKitBaseClass
import fr.acinq.eclair.crypto.Noise.{Chacha20Poly1305CipherFunctions, CipherState}
import fr.acinq.eclair.crypto.TransportHandler.{Encryptor, ExtendedCipherState, Listener}
import fr.acinq.eclair.wire.protocol.LightningMessageCodecs.{lightningMessageCodec, pingCodec, warningCodec}
@@ -75,12 +75,12 @@ class TransportHandlerSpec extends TestKitBaseClass with AnyFunSuiteLike with Be
probe1.expectTerminated(pipe)
}
- test("handle unknown messages") {
+ test("close connection on malformed messages") {
val incompleteCodec: Codec[LightningMessage] = discriminated[LightningMessage].by(uint16)
.typecase(1, warningCodec)
.typecase(18, pingCodec)
- val pipe = system.actorOf(Props[MyPipePull]())
+ val pipe = system.actorOf(Props[MyPipe]())
val probe1 = TestProbe()
val probe2 = TestProbe()
val initiator = TestFSMRef(new TransportHandler(Initiator.s, Some(Responder.s.pub), pipe, incompleteCodec))
@@ -96,23 +96,22 @@ class TransportHandlerSpec extends TestKitBaseClass with AnyFunSuiteLike with Be
awaitCond(initiator.stateName == TransportHandler.Normal)
awaitCond(responder.stateName == TransportHandler.Normal)
+ // We receive a ping message that we know how to decode.
val msg1 = Ping(130, hex"deadbeef")
responder ! msg1
probe1.expectMsg(msg1)
probe1.reply(TransportHandler.ReadAck(msg1))
+ // We receive a pong message which we don't know how to decode.
responder ! Pong(hex"deadbeef")
- probe1.expectNoMessage(2 seconds) // unknown message
-
- val msg2 = Warning(randomBytes32(), hex"beefdead")
- responder ! msg2
- probe1.expectMsg(msg2)
- probe1.reply(TransportHandler.ReadAck(msg2))
+ // We send a warning and close the connection.
+ probe2.expectMsgType[Warning]
+ probe1.watch(initiator)
+ probe1.expectTerminated(initiator)
+ probe1.watch(responder)
+ probe1.expectTerminated(responder)
probe1.watch(pipe)
- initiator.stop()
- responder.stop()
- system.stop(pipe)
probe1.expectTerminated(pipe)
}
Why this scored 70/100
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.