Fix a batch of low-severity issues (#3355)

### docs/Cluster.md
@@ -29,6 +29,7 @@ Front servers are stateless, they can be stopped/killed at will. The node will r
 ```
 
 The goal is to offload your node from connection and routing table management:
+
 - incoming connections
 - outgoing connections
 - gossip queries + pings
@@ -42,6 +43,7 @@ You already have a lightning node up and running in a standalone setup (with Bit
 You know what your `node id` is.
 
 Conventions used in this document:
+
 - what we previously called `eclair-node` will be called `backend`. It is to be launched, configured and backed-up exactly like in a standalone setup.
 - `node` refer to *akka cluster nodes*, not to be confused with lightning nodes. Together, all *cluster nodes* form a single logical *lighting node*.
 
@@ -50,6 +52,7 @@ Conventions used in this document:
 Use this if you want to experiment with the cluster mode on a single local server.
 
 Set the following values in `.eclair/eclair.conf`:
+
 ```
 akka.actor.provider = cluster
 akka.extensions = ["akka.cluster.pubsub.DistributedPubSub"]
@@ -60,11 +63,13 @@ eclair.front.pub = 03...............
 ```
 
 Start the `backend`:
+
 ```shell
 $ ./eclair-node.sh
 ```
 
 Then run an instance of `frontend`:
+
 ```shell
 $ ./eclair-front.sh -Dakka.remote.artery.canonical.port=25521 -Declair.server.port=9736
 ```
@@ -74,9 +79,10 @@ NB: we override the ports, otherwise they would conflict since in this example e
 ### Production setup
 
 In production you should:
+
 - run multiple `frontend` servers
 - run one app per server
-- enable `tcp-tls` to encrypt communications between members of the cluster with your own generated certificate (see below)
+- you MUST enable `tls-tcp` to encrypt communications between members of the cluster with your own generated certificate (see below)
 - use a load balancer to hide all your `frontend` servers under the same ip address
 - set firewall rules to disable lightning connections (port 9735) on your `backend` server, so all connections go through the `frontend`
 - enable [monitoring](Monitoring.md)
@@ -85,12 +91,14 @@ In production you should:
 #### Enable encrypted communication for the cluster
 
 We use a self-signed certificate, which offers a good compromise. More advanced options are available, see [akka doc](https://doc.akka.io/docs/akka/current/remoting-artery.html#remote-security).
+
 > Have a single set of keys and a single certificate for all nodes and disable hostname checking
 > - The single set of keys and the single certificate is distributed to all nodes. The certificate can be self-signed as it is distributed both as a certificate for authentication but also as the trusted certificate.
 > - If the keys/certificate are lost, someone else can connect to your cluster.
 > - Adding nodes to the cluster is simple as the key material can be deployed / distributed to the new cluster node.
 
 Generate a self-signed certificate (set a strong password):
+
 ```shell
 $ keytool -genkeypair -v \
           -keystore akka-cluster-tls.jks \
@@ -103,30 +111,38 @@ $ keytool -genkeypair -v \
 ```
 
 Copy the resulting certificate to the `.eclair` directory on your backend node and all your frontend nodes:
+
 ```shell
 $ cp akka-cluster-tls.jks ~/.eclair
 ```
+
 Add this to `eclair.conf` on all your frontend nodes:
+
 ```
 akka.remote.artery.transport = "tls-tcp"
 ```
 
+Note that `tls-tcp` MUST be used when using the cluster mode.
+
 #### Run cluster nodes on separate servers
 
 Start all your frontend nodes with the following environment variables:
+
 * BACKEND_IP set to the IP address of your backend node
 * LOCAL_IP set to the IP address of this frontend node (this is typically a private IP address, reachable from your backend node)
 * NODE_PUB_KEY set to your node public key
 * AKKA_TLS_PASSWORD set to the password of your Akka certificate
 
 Add this to `eclair.conf` on your backend node:
+
 ```
 akka.remote.artery.transport = "tls-tcp"
 akka.remote.artery.canonical.hostname="ip-of-this-backend-node"
 akka.cluster.seed-nodes=["akka://eclair-node@ip-of-this-backend-node:25520"]
 ```
 
 Start your backend node with the following environment variables:
+
 * AKKA_TLS_PASSWORD set to the password of your Akka certificate
 
 ### AWS Deployment
@@ -138,17 +154,19 @@ You can run it as-is for testing.
 #### TLS encryption
 
 If you intend to use it in production, you need to enable encryption with your own certificate:
-1. Follow the procedure above to generate your `akka-tls.jks`
 
+1. Follow the procedure above to generate your `akka-tls.jks`
 2. We recommend forking the project and building your own bundle:
+
 ```shell
 $ git clone git@github.com:ACINQ/eclair.git
 $ vi eclair-core/src/main/reference.conf # set akka.remote.artery.transport = "tls-tcp"
 $ cp akka-cluster-tls.jks eclair-front/modules/awseb/ # copy the file you generated
 $ vi eclair-front/modules/awseb.xml # uncomment the relevant parts
 $ ./mvnw package -DskipTests
 ```
-Alternatively, you can also edit the existing bundle and manually add the `akka-cluster-tls.jks` file to the root of the zip archive. You will also need to set `akka.remote.artery.transport=tls-tcp` at runtime. 
+
+Alternatively, you can also edit the existing bundle and manually add the `akka-cluster-tls.jks` file to the root of the zip archive. You will also need to set `akka.remote.artery.transport=tls-tcp` at runtime.
 
 #### Private key
 
@@ -157,6 +175,7 @@ In production, we highly recommend using AWS Secrets manager to provide the node
 #### Configuration
 
 We recommend using Beanstalk environment variables for `AKKA_TLS_PASSWORD`, `BACKEND_IP`, and `NODE_PUB_KEY`. Other configuration keys should be set in the `AKKA_CONF` environment variable, semicolon separated. Example:
+
 - `AKKA_CONF`: `eclair.enable-kamon=true; akka.remote.artery.transport=tls-tcp; eclair.front.priv-key-provider=aws-sm...`
 - `AKKA_TLS_PASSWORD`: `xxxxxxxx`
 - `BACKEND_IP`: `1.2.3.4`

### eclair-core/src/main/scala/fr/acinq/eclair/io/MessageRelay.scala
@@ -29,7 +29,7 @@ import fr.acinq.eclair.io.Monitoring.{Metrics, Tags}
 import fr.acinq.eclair.io.Peer.{PeerInfo, PeerInfoResponse}
 import fr.acinq.eclair.io.Switchboard.GetPeerInfo
 import fr.acinq.eclair.message.OnionMessages
-import fr.acinq.eclair.message.OnionMessages.DropReason
+import fr.acinq.eclair.message.OnionMessages.{DropReason, TooManyDummyHops}
 import fr.acinq.eclair.router.Router
 import fr.acinq.eclair.wire.protocol.OnionMessage
 import fr.acinq.eclair.{EncodedNodeId, Logs, NodeParams, ShortChannelId}
@@ -90,6 +90,7 @@ private class MessageRelay(nodeParams: NodeParams,
 
   import MessageRelay._
 
+  private var selfHopsCount = 0
   private val log = context.log
 
   def queryNextNodeId(msg: OnionMessage, nextNode: Either[ShortChannelId, EncodedNodeId]): Behavior[Command] = {
@@ -123,18 +124,29 @@ private class MessageRelay(nodeParams: NodeParams,
   private def withNextNodeId(msg: OnionMessage, nextNodeId: EncodedNodeId.WithPublicKey): Behavior[Command] = {
     nextNodeId match {
       case EncodedNodeId.WithPublicKey.Plain(nodeId) if nodeId == nodeParams.nodeId =>
-        OnionMessages.process(nodeParams.privateKey, msg) match {
-          case OnionMessages.DropMessage(reason) =>
-            Metrics.OnionMessagesNotRelayed.withTag(Tags.Reason, reason.getClass.getSimpleName).increment()
-            replyTo_opt.foreach(_ ! DroppedMessage(messageId, reason))
-            Behaviors.stopped
-          case OnionMessages.SendMessage(nextNode, nextMessage) =>
-            // We need to repeat the process until we identify the (real) next node, or find out that we're the recipient.
-            queryNextNodeId(nextMessage, nextNode)
-          case received: OnionMessages.ReceiveMessage =>
-            context.system.eventStream ! EventStream.Publish(received)
-            replyTo_opt.foreach(_ ! Sent(messageId))
-            Behaviors.stopped
+        if (selfHopsCount > 2 * nodeParams.offersConfig.messagePathMinLength + 1) {
+          // We only include ourselves multiple times in a path when using dummy hops.
+          // If we're included too many times in a path, that's most likely a remote node messing with us.
+          log.debug("too many self hops (max={})", 2 * nodeParams.offersConfig.messagePathMinLength + 1)
+          val reason = TooManyDummyHops(nodeParams.offersConfig.messagePathMinLength)
+          Metrics.OnionMessagesNotRelayed.withTag(Tags.Reason, reason.getClass.getSimpleName).increment()
+          replyTo_opt.foreach(_ ! DroppedMessage(messageId, reason))
+          Behaviors.stopped
+        } else {
+          selfHopsCount += 1
+          OnionMessages.process(nodeParams.privateKey, msg) match {
+            case OnionMessages.DropMessage(reason) =>
+              Metrics.OnionMessagesNotRelayed.withTag(Tags.Reason, reason.getClass.getSimpleName).increment()
+              replyTo_opt.foreach(_ ! DroppedMessage(messageId, reason))
+              Behaviors.stopped
+            case OnionMessages.SendMessage(nextNode, nextMessage) =>
+              // We need to repeat the process until we identify the (real) next node, or find out that we're the recipient.
+              queryNextNodeId(nextMessage, nextNode)
+            case received: OnionMessages.ReceiveMessage =>
+              context.system.eventStream ! EventStream.Publish(received)
+              replyTo_opt.foreach(_ ! Sent(messageId))
+              Behaviors.stopped
+          }
         }
       case EncodedNodeId.WithPublicKey.Plain(nodeId) =>
         policy match {

### eclair-core/src/main/scala/fr/acinq/eclair/message/OnionMessages.scala
@@ -159,6 +159,7 @@ object OnionMessages {
   case class CannotDecodeOnion(message: String) extends DropReason { override def toString = s"can't decode onion: $message" }
   case class CannotDecryptBlindedPayload(message: String) extends DropReason { override def toString = s"can't decrypt blinded payload: $message" }
   case class CannotDecodeBlindedPayload(message: String) extends DropReason { override def toString = s"can't decode blinded payload: $message" }
+  case class TooManyDummyHops(max: Int) extends DropReason { override def toString = s"onion message contains too many dummy hops (max=$max)" }
   // @formatter:on
 
   case class DecodedOnionPacket(payload: TlvStream[OnionMessagePayloadTlv], next_opt: Option[OnionRoutingPacket])

### eclair-core/src/main/scala/fr/acinq/eclair/wire/internal/channel/version5/ChannelCodecs5.scala
@@ -319,7 +319,10 @@ private[channel] object ChannelCodecs5 {
 
     private val waitForRevCodec: Codec[WaitForRev] = ("sentAfterLocalCommitIndex" | uint64overflow).as[WaitForRev]
 
-    private val updateMessageCodec: Codec[UpdateMessage] = lengthDelimited(lightningMessageCodec.narrow[UpdateMessage](f => Attempt.successful(f.asInstanceOf[UpdateMessage]), g => g))
+    private val updateMessageCodec: Codec[UpdateMessage] = lengthDelimited(lightningMessageCodec.narrow[UpdateMessage]({
+      case f: UpdateMessage => Attempt.successful(f)
+      case _ => Attempt.failure(Err("not an UpdateMessage"))
+    }, g => g))
 
     private val localChangesCodec: Codec[LocalChanges] = (
       ("proposed" | listOfN(uint16, updateMessageCodec)) ::

### eclair-core/src/main/scala/fr/acinq/eclair/wire/protocol/OnionMessageTlv.scala
@@ -16,10 +16,8 @@
 
 package fr.acinq.eclair.wire.protocol
 
-import fr.acinq.eclair.wire.protocol.CommonCodecs.varint
-import fr.acinq.eclair.wire.protocol.TlvCodecs.tlvStream
 import scodec.Codec
-import scodec.codecs.discriminated
+import scodec.codecs.provide
 
 /**
  * Created by thomash on 10/09/2021.
@@ -28,5 +26,7 @@ import scodec.codecs.discriminated
 sealed trait OnionMessageTlv extends Tlv
 
 object OnionMessageTlv {
-  val onionMessageTlvCodec: Codec[TlvStream[OnionMessageTlv]] = tlvStream(discriminated[OnionMessageTlv].by(varint))
+  // We don't support any TLV for onion messages yet. Since onion messages can be spammy, we don't need to waste any
+  // resources trying to decode unknown TLVs that we'll throw away anyway.
+  val onionMessageTlvCodec: Codec[TlvStream[OnionMessageTlv]] = provide(TlvStream.empty[OnionMessageTlv])
 }

### eclair-front/src/main/scala/fr/acinq/eclair/Boot.scala
@@ -16,25 +16,26 @@
 
 package fr.acinq.eclair
 
-import java.io.File
-
 import akka.actor.ActorSystem
 import com.typesafe.config.{ConfigFactory, ConfigParseOptions, ConfigSyntax}
 import grizzled.slf4j.Logging
 import kamon.Kamon
 
+import java.io.File
 import scala.concurrent.ExecutionContext
 import scala.util.{Failure, Success}
 
 object Boot extends App with Logging {
   try {
     val datadir = new File(sys.props.getOrElse("eclair.datadir", sys.props("user.home") + "/.eclair"))
     val config = ConfigFactory.parseString(
-      sys.env.getOrElse("AKKA_CONF", "").replace(";", "\n"),
-      ConfigParseOptions.defaults().setSyntax(ConfigSyntax.PROPERTIES))
+        sys.env.getOrElse("AKKA_CONF", "").replace(";", "\n"),
+        ConfigParseOptions.defaults().setSyntax(ConfigSyntax.PROPERTIES))
       .withFallback(ConfigFactory.parseProperties(System.getProperties))
       .withFallback(ConfigFactory.parseFile(new File(datadir, "eclair.conf")))
       .withFallback(ConfigFactory.load())
+    // It's unsafe to use the cluster mode with plain tcp: communications MUST be encrypted.
+    require(config.getString("akka.remote.artery.transport") == "tls-tcp", "frontend cluster communication must use tls-tcp")
 
     // the actor system name needs to be the same for all members of the cluster
     implicit val system: ActorSystem = ActorSystem("eclair-node", config)
@@ -54,7 +55,7 @@ object Boot extends App with Logging {
     case t: Throwable => onError(t)
   }
 
-  def onError(t: Throwable): Unit = {
+  private def onError(t: Throwable): Unit = {
     val errorMsg = if (t.getMessage != null) t.getMessage else t.getClass.getSimpleName
     System.err.println(s"fatal error: $errorMsg")
     logger.error(s"fatal error: $errorMsg", t)

### eclair-node/src/main/scala/fr/acinq/eclair/Boot.scala
@@ -39,6 +39,9 @@ object Boot extends App with Logging {
 
     val datadir = new File(System.getProperty("eclair.datadir", System.getProperty("user.home") + "/.eclair"))
     val config = NodeParams.loadConfiguration(datadir)
+    if (config.getString("akka.actor.provider") == "cluster") {
+      require(config.getString("akka.remote.artery.transport") == "tls-tcp", "backend cluster communication must use tls-tcp")
+    }
 
     val plugins = Plugin.loadPlugins(args.toIndexedSeq.map(new File(_)))
     plugins.foreach(plugin => logger.info(s"loaded plugin ${plugin.getClass.getSimpleName}"))

### eclair-node/src/main/scala/fr/acinq/eclair/api/directives/ErrorDirective.scala
@@ -33,10 +33,10 @@ trait ErrorDirective {
   private val apiExceptionHandler = ExceptionHandler {
     case t: IllegalArgumentException =>
       logger.error(s"API call failed with cause=${t.getMessage}")
-      complete(StatusCodes.BadRequest, ErrorResponse(t.getMessage))
+      complete(StatusCodes.BadRequest, ErrorResponse("API call failed: check logs for more details"))
     case t: Throwable =>
       logger.error(s"API call failed with cause=${t.getMessage}")
-      complete(StatusCodes.InternalServerError, ErrorResponse(t.getMessage))
+      complete(StatusCodes.InternalServerError, ErrorResponse("API call failed: check logs for more details"))
   }
 
   // map all the rejections to a JSON error object ErrorResponse

### eclair-node/src/test/scala/fr/acinq/eclair/api/ApiServiceSpec.scala
@@ -612,7 +612,7 @@ class ApiServiceSpec extends AnyFunSuite with ScalatestRouteTest with IdiomaticM
         assert(handled)
         assert(status == BadRequest)
         val resp = entityAs[ErrorResponse](Json4sSupport.unmarshaller, ClassTag(classOf[ErrorResponse]))
-        assert(resp.error == "invoice has expired")
+        assert(resp.error == "API call failed: check logs for more details")
         eclair.send(None, 1258000 msat, any, any, any, any, any)(any[Timeout]).wasCalled(once)
       }
   }