go/worker/compute: Verify I/O root of a proposal

Author: peternose

.changelog/6555.trivial.md

@@ -57,12 +57,14 @@ const (
 	ModeExecutorRunaway           ExecutorMode = 2
 	ModeExecutorStraggler         ExecutorMode = 3
 	ModeExecutorFailureIndicating ExecutorMode = 4
+	ModeExecutorInvalidBatchHash  ExecutorMode = 5
 
 	modeExecutorHonestString            = "executor_honest"
 	modeExecutorDishonestString         = "executor_dishonest"
 	modeExecutorRunawayString           = "executor_runaway"
 	modeExecutorStragglerString         = "executor_straggler"
 	modeExecutorFailureIndicatingString = "executor_failure_indicating"
+	modeExecutorInvalidBatchHashString  = "executor_invalid_batch_hash"
 )
 
 // String returns a string representation of a executor mode.
@@ -78,6 +80,8 @@ func (m ExecutorMode) String() string {
 		return modeExecutorStragglerString
 	case ModeExecutorFailureIndicating:
 		return modeExecutorFailureIndicatingString
+	case ModeExecutorInvalidBatchHash:
+		return modeExecutorInvalidBatchHashString
 	default:
 		return "[unsupported runtime kind]"
 	}
@@ -96,8 +100,10 @@ func (m *ExecutorMode) FromString(str string) error {
 		*m = ModeExecutorStraggler
 	case modeExecutorFailureIndicatingString:
 		*m = ModeExecutorFailureIndicating
+	case modeExecutorInvalidBatchHashString:
+		*m = ModeExecutorInvalidBatchHash
 	default:
-		return fmt.Errorf("invalid executor mode kind: %s", m)
+		return fmt.Errorf("invalid executor mode kind: %s", str)
 	}
 
 	return nil

go/oasis-node/cmd/debug/byzantine/byzantine.go

@@ -9,6 +9,7 @@ import (
 
 	beacon "github.com/oasisprotocol/oasis-core/go/beacon/api"
 	"github.com/oasisprotocol/oasis-core/go/common"
+	"github.com/oasisprotocol/oasis-core/go/common/crypto/hash"
 	"github.com/oasisprotocol/oasis-core/go/common/crypto/signature"
 	"github.com/oasisprotocol/oasis-core/go/common/entity"
 	"github.com/oasisprotocol/oasis-core/go/common/identity"
@@ -76,7 +77,18 @@ func (b *byzantine) receiveAndScheduleTransactions(ctx context.Context, cbc *com
 		panic(fmt.Sprintf("executor proposing batch: %+v", err))
 	}
 
-	if mode == ModeExecutorFailureIndicating {
+	switch mode {
+	case ModeExecutorInvalidBatchHash:
+		validBatchHash := cbc.proposal.Header.BatchHash
+		cbc.proposal.Header.BatchHash = hash.NewFromBytes([]byte("invalid batch hash"))
+		if err := cbc.proposal.Sign(b.identity.NodeSigner, block.Header.Namespace); err != nil {
+			return false, fmt.Errorf("failed to re-sign invalid batch hash proposal: %w", err)
+		}
+		logger.Debug("invalid batch hash: invalidating batch hash",
+			"valid_batch_hash", validBatchHash,
+			"invalid_batch_hash", cbc.proposal.Header.BatchHash,
+		)
+	case ModeExecutorFailureIndicating:
 		// Submit failure indicating commitment and stop.
 		logger.Debug("executor failure indicating: submitting commitment and stopping")
 		schedulerID := b.identity.NodeSigner.Public()

go/oasis-node/cmd/debug/byzantine/node.go

@@ -164,6 +164,35 @@ var (
 			Index: primarySchedulerIndex,
 		},
 	)
+	ByzantineExecutorSchedulerInvalidBatchHash scenario.Scenario = newByzantineImpl(
+		"primary-worker/primary-scheduler/invalid-batch-hash",
+		"executor",
+		[]log.WatcherHandlerFactory{
+			// The Byzantine node will publish a proposal with an invalid batch hash.
+			// Other workers will reject the proposal after determining that the I/O root
+			// does not match. As a result, the second-ranked proposer will prepare a new
+			// proposal but will not have enough commitments to finalize it. This will
+			// trigger the round timeout, leading to discrepancy resolution, which will
+			// succeed once the remaining backup workers submit their commitments.
+			oasis.LogAssertTimeouts(),
+			oasis.LogAssertExecutionDiscrepancyDetected(),
+		},
+		oasis.ByzantineDefaultIdentitySeed,
+		false,
+		// Byzantine node entity should be slashed once for liveness.
+		map[staking.SlashReason]uint64{
+			staking.SlashRuntimeLiveness: 1,
+		},
+		[]oasis.Argument{
+			{Name: byzantine.CfgPrimarySchedulerExpected},
+			{Name: byzantine.CfgExecutorMode, Values: []string{byzantine.ModeExecutorInvalidBatchHash.String()}},
+		},
+		scheduler.ForceElectCommitteeRole{
+			Kind:  scheduler.KindComputeExecutor,
+			Roles: []scheduler.Role{scheduler.RoleWorker},
+			Index: primarySchedulerIndex,
+		},
+	)
 	// ByzantineExecutorStraggler is a scenario in which the Byzantine node acts
 	// as the primary worker, backup scheduler, and a straggler.
 	ByzantineExecutorStraggler scenario.Scenario = newByzantineImpl(

go/oasis-test-runner/scenario/e2e/runtime/byzantine.go

@@ -333,6 +333,7 @@ func RegisterScenarios() error {
 		ByzantineExecutorDishonest,
 		ByzantineExecutorSchedulerRunaway,
 		ByzantineExecutorSchedulerBogus,
+		ByzantineExecutorSchedulerInvalidBatchHash,
 		ByzantineExecutorStraggler,
 		ByzantineExecutorSchedulerStraggler,
 		ByzantineExecutorStragglerAllowed,

go/oasis-test-runner/scenario/e2e/runtime/scenario.go

@@ -588,6 +588,25 @@ func (n *Node) runtimeExecuteTxBatch(
 func (n *Node) startProcessingBatch(ctx context.Context, proposal *commitment.Proposal, rank uint64, batch transaction.RawBatch) {
 	// This method runs within its own goroutine and is always stopped before the runtime
 	// worker finishes. Therefore, it is safe to read local round variables (block info, ...).
+	n.logger.Debug("verifying batch")
+
+	ioRoot, err := n.computeIORoot(ctx, proposal.Header.Round, batch)
+	if err != nil {
+		n.logger.Error("failed to compute I/O root",
+			"err", err,
+		)
+		// Notify the round worker that the execution failed.
+		n.processedBatchCh <- nil
+		return
+	}
+	if !ioRoot.Equal(&proposal.Header.BatchHash) {
+		n.logger.Debug("batch I/O root mismatch")
+		n.proposals.Reject(proposal, rank)
+		// Notify the round worker that the execution failed.
+		n.processedBatchCh <- nil
+		return
+	}
+
 	n.logger.Debug("processing batch",
 		"batch_size", len(batch),
 	)
@@ -624,6 +643,39 @@ func (n *Node) startProcessingBatch(ctx context.Context, proposal *commitment.Pr
 	}
 }
 
+func (n *Node) computeIORoot(ctx context.Context, round uint64, batch transaction.RawBatch) (hash.Hash, error) {
+	txs := make([]*transaction.Transaction, 0, len(batch))
+	for idx, tx := range batch {
+		txs = append(txs, &transaction.Transaction{
+			Input:      tx,
+			BatchOrder: uint32(idx),
+		})
+	}
+
+	emptyRoot := storage.Root{
+		Namespace: n.rt.ID(),
+		Version:   round,
+		Type:      storage.RootTypeIO,
+	}
+	emptyRoot.Hash.Empty()
+
+	ioTree := transaction.NewTree(nil, emptyRoot)
+	defer ioTree.Close()
+
+	for _, tx := range txs {
+		if err := ioTree.AddTransaction(ctx, *tx, nil); err != nil {
+			return hash.Hash{}, fmt.Errorf("failed to add transaction to tree: %w", err)
+		}
+	}
+
+	_, ioRoot, err := ioTree.Commit(ctx)
+	if err != nil {
+		return hash.Hash{}, fmt.Errorf("failed to commit I/O tree: %w", err)
+	}
+
+	return ioRoot, nil
+}
+
 func (n *Node) abortBatch(state *StateProcessingBatch) {
 	n.logger.Warn("aborting processing batch")
 

go/worker/compute/executor/committee/node.go

@@ -91,7 +91,10 @@ func (h *committeeMsgHandler) HandleMessage(_ context.Context, _ signature.Publi
 			"batch_size", len(proposal.Batch),
 		)
 
-		// Add to the queue.
+		if _, ok := h.n.proposals.Get(proposal.Header.Round, rank); ok {
+			return nil
+		}
+
 		if err := h.n.proposals.Add(proposal, rank); err != nil {
 			return err
 		}

go/worker/compute/executor/committee/p2p.go

@@ -16,6 +16,7 @@ const maxPendingProposals = 32
 type proposalInfo struct {
 	proposal *commitment.Proposal
 	rank     uint64
+	invalid  bool
 }
 
 // proposalQueue is a priority queue of pending proposals, ordered by round and rank.
@@ -61,6 +62,8 @@ func (q *proposalQueue) Best(round uint64, minRank uint64, maxRank uint64, exclu
 			return true
 		case pi.rank > maxRank:
 			return false
+		case pi.invalid:
+			return true
 		default:
 			if _, skip := exclude[pi.rank]; skip {
 				return true
@@ -75,6 +78,28 @@ func (q *proposalQueue) Best(round uint64, minRank uint64, maxRank uint64, exclu
 	return proposal, rank, ok
 }
 
+// Get returns a proposal for the given round and rank.
+func (q *proposalQueue) Get(round uint64, rank uint64) (*commitment.Proposal, bool) {
+	q.l.RLock()
+	defer q.l.RUnlock()
+
+	info := proposalInfo{
+		proposal: &commitment.Proposal{
+			Header: commitment.ProposalHeader{
+				Round: round,
+			},
+		},
+		rank: rank,
+	}
+
+	pi, ok := q.q.Get(&info)
+	if !ok {
+		return nil, false
+	}
+
+	return pi.proposal, true
+}
+
 // Add adds a new pending proposal that MUST HAVE already undergone basic validity checks
 // and is therefore considered a valid proposal for the given round, but the node's
 // local consensus view may not yet be ready to process the proposal.
@@ -91,6 +116,11 @@ func (q *proposalQueue) Add(proposal *commitment.Proposal, rank uint64) error {
 		proposal: proposal,
 		rank:     rank,
 	}
+
+	if _, ok := q.q.Get(&info); ok {
+		return fmt.Errorf("proposal already exists")
+	}
+
 	q.q.ReplaceOrInsert(&info)
 
 	// In case of overflows, remove the proposal that is the most in the future.
@@ -105,6 +135,22 @@ func (q *proposalQueue) Add(proposal *commitment.Proposal, rank uint64) error {
 	return nil
 }
 
+// Reject marks a proposal as invalid.
+func (q *proposalQueue) Reject(proposal *commitment.Proposal, rank uint64) {
+	q.l.Lock()
+	defer q.l.Unlock()
+
+	info, ok := q.q.Get(&proposalInfo{
+		proposal: proposal,
+		rank:     rank,
+	})
+	if !ok {
+		return
+	}
+
+	info.invalid = true
+}
+
 // Prune prunes any proposals which are not valid anymore.
 func (q *proposalQueue) Prune(round uint64) {
 	q.l.Lock()