Qbft gossip validation (#454)

Author: jframe

app/src/integrationTest/kotlin/maru/app/MaruMultiValidatorTest.kt

@@ -14,7 +14,7 @@ import kotlin.time.Duration.Companion.seconds
 import kotlin.time.toJavaDuration
 import maru.consensus.qbft.ProposerSelectorImpl
 import maru.core.SealedBeaconBlock
-import maru.crypto.Crypto
+import maru.crypto.SecpCrypto
 import maru.database.BeaconChain
 import maru.extensions.encodeHex
 import org.apache.logging.log4j.LogManager
@@ -80,8 +80,8 @@ class MaruMultiValidatorTest {
 
   @Test
   fun `maru with multiple validators is able to produce blocks`() {
-    val validator1Address = Crypto.privateKeyToValidator(Crypto.privateKeyBytesWithoutPrefix(key1))
-    val validator2Address = Crypto.privateKeyToValidator(Crypto.privateKeyBytesWithoutPrefix(key2))
+    val validator1Address = SecpCrypto.privateKeyToValidator(SecpCrypto.privateKeyBytesWithoutPrefix(key1))
+    val validator2Address = SecpCrypto.privateKeyToValidator(SecpCrypto.privateKeyBytesWithoutPrefix(key2))
     log.info("Validator 1 (key1) address: ${validator1Address.address.encodeHex()}")
     log.info("Validator 2 (key2) address: ${validator2Address.address.encodeHex()}")
     val initialValidators = setOf(validator1Address, validator2Address)

app/src/main/kotlin/maru/app/MaruApp.kt

@@ -23,7 +23,7 @@ import maru.consensus.qbft.DifficultyAwareQbftFactory
 import maru.consensus.state.FinalizationProvider
 import maru.core.Protocol
 import maru.core.Validator
-import maru.crypto.Crypto
+import maru.crypto.SecpCrypto
 import maru.database.BeaconChain
 import maru.extensions.encodeHex
 import maru.finalization.LineaFinalizationProvider
@@ -65,13 +65,13 @@ class MaruApp(
 ) : LongRunningCloseable {
   private val log: Logger = LogManager.getLogger(this.javaClass)
 
-  private fun getPrivateKeyWithoutPrefix() = Crypto.privateKeyBytesWithoutPrefix(privateKeyProvider())
+  private fun getPrivateKeyWithoutPrefix() = SecpCrypto.privateKeyBytesWithoutPrefix(privateKeyProvider())
 
   init {
     if (config.qbft == null) {
       log.info("Qbft options are not defined. nodeRole=follower")
     } else {
-      val localValidator = Crypto.privateKeyToValidator(getPrivateKeyWithoutPrefix())
+      val localValidator = SecpCrypto.privateKeyToValidator(getPrivateKeyWithoutPrefix())
       log.info("Qbft options are defined. nodeRole=validator with address={}", localValidator.address.encodeHex())
       // TODO: This may be not needed when we use dynamic validator set from a smart contract
       warnIfValidatorIsNotInTheGenesis(localValidator)

consensus/build.gradle

@@ -15,14 +15,6 @@
 
 plugins {
   id 'maru.kotlin-library-conventions'
-  id 'java-test-fixtures'
-}
-
-sourceSets {
-  testFixtures {
-    compileClasspath += sourceSets.main.compileClasspath
-    runtimeClasspath += sourceSets.main.runtimeClasspath
-  }
 }
 
 dependencies {
@@ -58,6 +50,7 @@ dependencies {
   implementation "tech.pegasys.teku.internal:spec"
 
   testImplementation(testFixtures(project(":core")))
+  testImplementation(testFixtures(project(":crypto")))
   testImplementation(project(":jvm-libs:test-utils"))
   testImplementation "org.jetbrains.kotlin:kotlin-test:2.1.0"
   testImplementation "tech.pegasys.teku.internal:executionclient"

consensus/src/main/kotlin/maru/consensus/qbft/MinimalQbftMessageDecoder.kt

@@ -0,0 +1,101 @@
+/*
+ * Copyright Consensys Software Inc.
+ *
+ * This file is dual-licensed under either the MIT license or Apache License 2.0.
+ * See the LICENSE-MIT and LICENSE-APACHE files in the repository root for details.
+ *
+ * SPDX-License-Identifier: MIT OR Apache-2.0
+ */
+package maru.consensus.qbft
+
+import maru.core.Seal
+import maru.crypto.Crypto
+import org.apache.tuweni.bytes.Bytes
+import org.hyperledger.besu.consensus.qbft.core.messagedata.QbftV1
+import org.hyperledger.besu.consensus.qbft.core.types.QbftMessage
+import org.hyperledger.besu.datatypes.Address
+import org.hyperledger.besu.datatypes.Hash.hash
+import org.hyperledger.besu.ethereum.rlp.BytesValueRLPOutput
+import org.hyperledger.besu.ethereum.rlp.RLP.input
+
+/**
+ * Minimal RLP decoder for QBFT messages, extracting only the roundIdentifier, signature, and author needed for
+ * minimal validation. This prevents more expensive decoding of the full message with the Block or other fields
+ * which can be large.
+ */
+class MinimalQbftMessageDecoder(
+  private val crypto: Crypto,
+) {
+  data class QbftMessageMetadata(
+    val messageCode: Int,
+    val sequenceNumber: Long,
+    val roundNumber: Int,
+    val author: Address,
+  )
+
+  /**
+   * Deserializes a QBFT message to extract metadata.
+   *
+   * The RLP structure of signed data varies by message type:
+   * - **PREPARE/COMMIT**: `[payload, signature]`
+   *   - Schema: [org.hyperledger.besu.consensus.common.bft.payload.SignedData]
+   *   - Message wrappers: [org.hyperledger.besu.consensus.qbft.core.messagewrappers.Prepare],
+   *     [org.hyperledger.besu.consensus.qbft.core.messagewrappers.Commit]
+   * - **PROPOSAL**: `[[payload], signature]` (extra list wrapper around payload)
+   *   - Schema: [org.hyperledger.besu.consensus.qbft.core.messagewrappers.Proposal]
+   * - **ROUND_CHANGE**: `[[payload], signature]` (extra list wrapper around payload)
+   *   - Schema: [org.hyperledger.besu.consensus.qbft.core.messagewrappers.RoundChange]
+   *
+   * All payloads contain: `[sequenceNumber: LONG_SCALAR, roundNumber: INT_SCALAR, ...]`
+   *
+   * @param qbftMessage The QBFT message to decode
+   * @return The decoded metadata
+   */
+  fun deserialize(qbftMessage: QbftMessage): QbftMessageMetadata {
+    val messageCode = qbftMessage.data.code
+    val signedDataBytes = qbftMessage.data.data
+
+    // SignedData list [payload, signature] or [[payload], signature] for PROPOSAL/ROUND_CHANGE
+    val signedDataRlp = input(signedDataBytes)
+    // For PROPOSAL and ROUND_CHANGE, there's an extra list wrapper
+    if (messageCode == QbftV1.PROPOSAL || messageCode == QbftV1.ROUND_CHANGE) {
+      signedDataRlp.enterList()
+    }
+    signedDataRlp.enterList()
+    val payloadRlp = signedDataRlp.readAsRlp()
+
+    // Payload list [sequenceNumber, roundNumber, ...]
+    payloadRlp.enterList()
+    val sequenceNumber = payloadRlp.readLongScalar()
+    val roundNumber = payloadRlp.readIntScalar()
+    val payloadHash = hashForSignature(messageCode, payloadRlp.raw())
+    val signatureBytes = signedDataRlp.readBytes()
+    signedDataRlp.leaveList()
+
+    val author = crypto.signatureToAddress(Seal(signatureBytes.toArray()), payloadHash)
+    return QbftMessageMetadata(
+      messageCode = messageCode,
+      sequenceNumber = sequenceNumber,
+      roundNumber = roundNumber,
+      author = Address.wrap(Bytes.wrap(author)),
+    )
+  }
+
+  /**
+   * Compute the hash used for signature verification, implementing the same algorithm as
+   * [org.hyperledger.besu.consensus.qbft.core.payload.QbftPayload.hashForSignature].
+   *
+   * This is the hash of the RLP encoding of: `LIST [messageType: INT_SCALAR, encodedPayload: RAW_BYTES]`
+   */
+  internal fun hashForSignature(
+    messageType: Int,
+    encodedPayload: Bytes,
+  ): ByteArray {
+    val out = BytesValueRLPOutput()
+    out.startList()
+    out.writeIntScalar(messageType)
+    out.writeRaw(encodedPayload)
+    out.endList()
+    return hash(out.encoded()).toArray()
+  }
+}

consensus/src/main/kotlin/maru/consensus/qbft/QbftMessageProcessor.kt

@@ -0,0 +1,91 @@
+/*
+ * Copyright Consensys Software Inc.
+ *
+ * This file is dual-licensed under either the MIT license or Apache License 2.0.
+ * See the LICENSE-MIT and LICENSE-APACHE files in the repository root for details.
+ *
+ * SPDX-License-Identifier: MIT OR Apache-2.0
+ */
+package maru.consensus.qbft
+
+import maru.consensus.qbft.MinimalQbftMessageDecoder.QbftMessageMetadata
+import maru.consensus.qbft.adapters.QbftBlockchainAdapter
+import maru.consensus.qbft.adapters.QbftValidatorProviderAdapter
+import maru.consensus.qbft.adapters.toQbftReceivedMessageEvent
+import maru.p2p.QbftMessageHandler
+import maru.p2p.ValidationResult
+import maru.p2p.ValidationResult.Companion.Ignore
+import maru.p2p.ValidationResult.Companion.Invalid
+import maru.p2p.ValidationResult.Companion.Valid
+import org.hyperledger.besu.consensus.common.bft.BftEventQueue
+import org.hyperledger.besu.consensus.qbft.core.types.QbftMessage
+import org.hyperledger.besu.datatypes.Address
+import tech.pegasys.teku.infrastructure.async.SafeFuture
+
+/**
+ * Processes QBFT messages received from the P2P network by applying a light validation, adding them to the
+ * event queue if they are valid and returning the validation result to gossip if they are a current message.
+ *
+ * This mirrors the logic in Besu QbftController.processMessage but adapted for LibP2P message handling.
+ *
+ * Validation rules:
+ * - Old messages (sequence < chainHeight): Ignored and not added to the event queue
+ * - Future messages (sequence > chainHeight): Added to the event queue but not gossiped
+ * - Current messages (sequence == chainHeight): Validated for author and local validator status and gossiped if valid
+ */
+class QbftMessageProcessor(
+  private val blockChain: QbftBlockchainAdapter,
+  private val validatorProvider: QbftValidatorProviderAdapter,
+  private val localAddress: Address,
+  private val bftEventQueue: BftEventQueue,
+  private val messageDecoder: MinimalQbftMessageDecoder,
+) : QbftMessageHandler<ValidationResult> {
+  /**
+   * Validates a QBFT message and determines whether it should be gossiped.
+   *
+   * @param qbftMessage The QBFT message to validate
+   * @return A future containing the validation result
+   */
+  override fun handleQbftMessage(qbftMessage: QbftMessage): SafeFuture<ValidationResult> =
+    try {
+      val metadata = messageDecoder.deserialize(qbftMessage)
+      val result = processMessage(qbftMessage, metadata)
+      SafeFuture.completedFuture(result)
+    } catch (e: Exception) {
+      SafeFuture.completedFuture(Invalid("Failed to decode or validate message: ${e.message}"))
+    }
+
+  private fun processMessage(
+    qbftMessage: QbftMessage,
+    metadata: QbftMessageMetadata,
+  ): ValidationResult {
+    if (isMsgForCurrentHeight(metadata.sequenceNumber)) {
+      val validators = validatorProvider.getValidatorsForBlock(blockChain.chainHeadHeader)
+      return if (!isMsgFromKnownValidator(metadata.author, validators)) {
+        Invalid("Message from unknown validator: ${metadata.author}")
+      } else if (!isLocalNodeValidator(validators)) {
+        Ignore("Local node is not a validator")
+      } else {
+        bftEventQueue.add(qbftMessage.toQbftReceivedMessageEvent())
+        Valid
+      }
+    } else if (isMsgForFutureChainHeight(metadata.sequenceNumber)) {
+      bftEventQueue.add(qbftMessage.toQbftReceivedMessageEvent())
+      return Ignore("Future message, will be processed when chain reaches height ${metadata.sequenceNumber}")
+    } else {
+      return Ignore("Old message: sequence ${metadata.sequenceNumber} < height ${blockChain.chainHeadBlockNumber}")
+    }
+  }
+
+  private fun isMsgFromKnownValidator(
+    messageAuthor: Address,
+    validators: Collection<Address>,
+  ): Boolean = validators.contains(messageAuthor)
+
+  private fun isMsgForCurrentHeight(sequenceNumber: Long): Boolean = sequenceNumber == blockChain.chainHeadBlockNumber
+
+  private fun isMsgForFutureChainHeight(sequenceNumber: Long): Boolean =
+    sequenceNumber > blockChain.chainHeadBlockNumber
+
+  private fun isLocalNodeValidator(validators: Collection<Address>): Boolean = validators.contains(localAddress)
+}

consensus/src/main/kotlin/maru/consensus/qbft/QbftValidatorFactory.kt

@@ -34,7 +34,6 @@ import maru.consensus.qbft.adapters.QbftFinalStateAdapter
 import maru.consensus.qbft.adapters.QbftProtocolScheduleAdapter
 import maru.consensus.qbft.adapters.QbftValidatorModeTransitionLoggerAdapter
 import maru.consensus.qbft.adapters.QbftValidatorProviderAdapter
-import maru.consensus.qbft.adapters.toQbftReceivedMessageEvent
 import maru.consensus.qbft.adapters.toSealedBeaconBlock
 import maru.consensus.state.FinalizationProvider
 import maru.consensus.state.StateTransition
@@ -44,6 +43,7 @@ import maru.core.BeaconState
 import maru.core.Protocol
 import maru.core.Validator
 import maru.crypto.Hashing
+import maru.crypto.SecpCrypto
 import maru.crypto.Signing
 import maru.database.BeaconChain
 import maru.executionlayer.manager.ExecutionLayerManager
@@ -66,13 +66,11 @@ import org.hyperledger.besu.consensus.qbft.core.types.QbftMessage
 import org.hyperledger.besu.consensus.qbft.core.types.QbftMinedBlockObserver
 import org.hyperledger.besu.consensus.qbft.core.types.QbftNewChainHead
 import org.hyperledger.besu.consensus.qbft.core.validation.MessageValidatorFactory
-import org.hyperledger.besu.crypto.SignatureAlgorithmFactory
 import org.hyperledger.besu.cryptoservices.KeyPairSecurityModule
 import org.hyperledger.besu.cryptoservices.NodeKey
 import org.hyperledger.besu.ethereum.core.Util
 import org.hyperledger.besu.plugin.services.MetricsSystem
 import org.hyperledger.besu.util.Subscribers
-import tech.pegasys.teku.infrastructure.async.SafeFuture
 
 class QbftValidatorFactory(
   private val beaconChain: BeaconChain,
@@ -91,7 +89,7 @@ class QbftValidatorFactory(
 ) : ProtocolFactory {
   override fun create(forkSpec: ForkSpec): Protocol {
     val protocolConfig = forkSpec.configuration as QbftConsensusConfig
-    val signatureAlgorithm = SignatureAlgorithmFactory.getInstance()
+    val signatureAlgorithm = SecpCrypto.signatureAlgorithm
     val privateKey = signatureAlgorithm.createPrivateKey(Bytes32.wrap(privateKeyBytes))
     val keyPair = signatureAlgorithm.createKeyPair(privateKey)
     val securityModule = KeyPairSecurityModule(keyPair)
@@ -253,12 +251,18 @@ class QbftValidatorFactory(
     val eventProcessor = QbftEventProcessor(bftEventQueue, eventMultiplexer)
     val eventQueueExecutor = Executors.newSingleThreadExecutor(Thread.ofPlatform().daemon().factory())
 
-    // Subscribe to QBFT messages from P2P network and add them to the event queue
-    p2PNetwork.subscribeToQbftMessages { qbftMessage ->
-      bftEventQueue.add(qbftMessage.toQbftReceivedMessageEvent())
-      // TODO: Validate QBFT messages and return the appropriate ValidationResult
-      SafeFuture.completedFuture(ValidationResult.Companion.Valid)
-    }
+    val messageDecoder = MinimalQbftMessageDecoder(SecpCrypto)
+    val qbftMessageProcessor =
+      QbftMessageProcessor(
+        blockChain = blockChain,
+        validatorProvider = besuValidatorProvider,
+        localAddress = localAddress,
+        bftEventQueue = bftEventQueue,
+        messageDecoder = messageDecoder,
+      )
+
+    // Subscribe to QBFT messages from P2P network and validate before adding to event queue
+    p2PNetwork.subscribeToQbftMessages(qbftMessageProcessor)
 
     return QbftConsensusValidator(
       qbftController = qbftController,

consensus/src/main/kotlin/maru/consensus/validation/SealVerifier.kt

@@ -14,10 +14,10 @@ import com.github.michaelbull.result.Result
 import maru.core.BeaconBlockHeader
 import maru.core.Seal
 import maru.core.Validator
+import maru.crypto.SecpCrypto
 import org.apache.tuweni.bytes.Bytes
 import org.apache.tuweni.bytes.Bytes32
 import org.hyperledger.besu.crypto.SignatureAlgorithm
-import org.hyperledger.besu.crypto.SignatureAlgorithmFactory
 import org.hyperledger.besu.ethereum.core.Util
 
 interface SealVerifier {
@@ -32,7 +32,7 @@ interface SealVerifier {
 }
 
 class SCEP256SealVerifier(
-  private val signatureAlgorithm: SignatureAlgorithm = SignatureAlgorithmFactory.getInstance(),
+  private val signatureAlgorithm: SignatureAlgorithm = SecpCrypto.signatureAlgorithm,
 ) : SealVerifier {
   override fun extractValidator(
     seal: Seal,