Add support for external modules

cmd/bench/cmd/client.go

@@ -115,7 +115,7 @@ func runClient() error {
 			}
 			return err
 		}
-		rand.Read(txBytes) //nolint:gosec
+		rand.Read(txBytes) //nolint:gosec,staticcheck
 		logger.Log(logging.LevelDebug, fmt.Sprintf("Submitting transaction #%d", i))
 		if err := client.SubmitTransaction(txBytes); err != nil {
 			return err

cmd/externalmodule-test-client/client.go

@@ -0,0 +1,54 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/filecoin-project/mir/pkg/externalmodule"
+	"github.com/filecoin-project/mir/stdevents"
+	"github.com/filecoin-project/mir/stdtypes"
+)
+
+func main() {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	defer cancel()
+
+	beroConn, err := externalmodule.Connect(ctx, "ws://localhost:8080/bero")
+	if err != nil {
+		panic(err)
+	}
+
+	response, err := beroConn.Submit(ctx, stdtypes.ListOf(
+		stdevents.NewTestString("remote", "Ping"),
+		stdevents.NewRaw("remote", []byte{0, 1, 2, 3}),
+	))
+	if err != nil {
+		panic(err)
+	}
+
+	fmt.Printf("Bero received %d events in response.\n", response.Len())
+
+	cecoConn, err := externalmodule.Connect(ctx, "ws://localhost:8080/ceco")
+	if err != nil {
+		panic(err)
+	}
+	response, err = cecoConn.Submit(ctx, stdtypes.ListOf(
+		stdevents.NewTestString("remote", "Ping"),
+		stdevents.NewRaw("remote", []byte{0, 1, 2, 3}),
+	))
+	if err != nil {
+		panic(err)
+	}
+
+	fmt.Printf("Ceco received %d events in response.\n", response.Len())
+
+	err = beroConn.Close(ctx)
+	if err != nil {
+		panic(err)
+	}
+	err = cecoConn.Close(ctx)
+	if err != nil {
+		panic(err)
+	}
+}

cmd/externalmodule-test-server/server.go

@@ -0,0 +1,43 @@
+package main
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/filecoin-project/mir/pkg/externalmodule"
+	"github.com/filecoin-project/mir/stdevents"
+	"github.com/filecoin-project/mir/stdtypes"
+)
+
+type EmptyModule struct {
+	prefix string
+}
+
+func (e EmptyModule) ImplementsModule() {}
+
+func (e EmptyModule) ApplyEvents(events *stdtypes.EventList) (*stdtypes.EventList, error) {
+	fmt.Printf("%s: Received %d event(s).\n", e.prefix, events.Len())
+	return stdtypes.ListOf(stdevents.NewTestString("anonymous-module", "Pong")), nil
+}
+
+func main() {
+	s := externalmodule.NewServer(
+		externalmodule.NewHandler("bero", EmptyModule{"bero"}),
+		externalmodule.NewHandler("ceco", EmptyModule{"ceco"}),
+	)
+
+	time.AfterFunc(10*time.Second, func() {
+		err := s.Stop()
+		if err != nil {
+			fmt.Printf("Error stopping server: %v\n", err)
+		}
+	})
+
+	err := s.Serve("0.0.0.0:8080")
+
+	if err != nil {
+		fmt.Println(err)
+	} else {
+		fmt.Println("Server stopped cleanly.")
+	}
+}

cmd/mircat/debug.go

@@ -111,7 +111,7 @@ func debug(args *arguments) error {
 			for _, event := range entry.Events {
 
 				// Set the index of the event in the event log.
-				metadata.index = uint64(index)
+				metadata.index = uint64(index) //nolint:gosec
 
 				// If the event was selected by the user for inspection, pause before submitting it to the node.
 				// The processing continues after the user's interactive confirmation.
@@ -200,9 +200,6 @@ func debuggerNode(id stdtypes.NodeID, membership *trantorpbtypes.Membership) (*m
 		"iss":   protocol,
 		"timer": timer.New(),
 	}
-	if err != nil {
-		panic(fmt.Errorf("error initializing the Mir modules: %w", err))
-	}
 
 	node, err := mir.NewNode(id, mir.DefaultNodeConfig().WithLogger(logger), nodeModules, nil)
 	if err != nil {

cmd/mircat/display.go

@@ -69,7 +69,7 @@ func displayEvents(args *arguments) error { //nolint:gocognit
 			}
 			// getting events from entry
 			for _, event := range entry.Events {
-				metadata.index = uint64(index)
+				metadata.index = uint64(index) //nolint:gosec
 
 				_, validEvent := args.selectedEventNames[eventName(event)]
 				_, validDest := args.selectedEventDests[event.DestModule]

go.mod

@@ -23,6 +23,7 @@ require github.com/golang/mock v1.6.0
 
 require (
 	filippo.io/keygen v0.0.0-20230306160926-5201437acf8e
+	github.com/coder/websocket v1.8.12
 	github.com/dave/jennifer v1.5.1
 	github.com/drand/kyber v1.2.0
 	github.com/drand/kyber-bls12381 v0.3.0

go.sum

@@ -12,6 +12,8 @@ github.com/alecthomas/units v0.0.0-20211218093645-b94a6e3cc137 h1:s6gZFSlWYmbqAu
 github.com/alecthomas/units v0.0.0-20211218093645-b94a6e3cc137/go.mod h1:OMCwj8VM1Kc9e19TLln2VL61YJF0x1XFtfdL4JdbSyE=
 github.com/cenkalti/backoff/v4 v4.0.0 h1:6VeaLF9aI+MAUQ95106HwWzYZgJJpZ4stumjj6RFYAU=
 github.com/cenkalti/backoff/v4 v4.0.0/go.mod h1:eEew/i+1Q6OrCDZh3WiXYv3+nJwBASZ8Bog/87DQnVg=
+github.com/coder/websocket v1.8.12 h1:5bUXkEPPIbewrnkU8LTCLVaxi4N4J8ahufH2vlo4NAo=
+github.com/coder/websocket v1.8.12/go.mod h1:LNVeNrXQZfe5qhS9ALED3uA+l5pPqvwXg3CKoDBB2gs=
 github.com/cpuguy83/go-md2man/v2 v2.0.2/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
 github.com/dave/jennifer v1.5.1 h1:AI8gaM02nCYRw6/WTH0W+S6UNck9YqPZ05xoIxQtuoE=
 github.com/dave/jennifer v1.5.1/go.mod h1:AxTG893FiZKqxy3FP1kL80VMshSMuz2G+EgvszgGRnk=

node_test.go

@@ -3,6 +3,7 @@ package mir
 import (
 	"context"
 	"fmt"
+	"math"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -140,7 +141,7 @@ func TestNode_Backpressure(t *testing.T) {
 	nodeConfig.Stats.Period = 100 * time.Millisecond
 
 	// Set an input event rate that would fill the node's event buffers in one second in 10 batches.
-	blabberModule := newBlabber(uint64(nodeConfig.PauseInputThreshold/10), 100*time.Millisecond)
+	blabberModule := newBlabber(uint64(nodeConfig.PauseInputThreshold/10), 100*time.Millisecond) //nolint:gosec
 
 	// Set the event consumption rate to 1/2 of the input rate (i.e., draining the buffer in 2 seconds)
 	// and create the consumer module.
@@ -181,8 +182,8 @@ func TestNode_Backpressure(t *testing.T) {
 	fmt.Printf("Total submitted events: %d\n", atomic.LoadUint64(&blabberModule.totalSubmitted))
 	totalSubmitted := atomic.LoadUint64(&blabberModule.totalSubmitted)
 	expectSubmitted := atomic.LoadUint64(&consumerModule.numProcessed) +
-		uint64(nodeConfig.PauseInputThreshold) + // Events left in the buffer
-		uint64(nodeConfig.MaxEventBatchSize) + // Events in the consumer's processing queue
+		uint64(nodeConfig.PauseInputThreshold) + //nolint:gosec // Events left in the buffer
+		uint64(nodeConfig.MaxEventBatchSize) + //nolint:gosec // Events in the consumer's processing queue
 		2*blabberModule.batchSize // one batch of overshooting, one batch waiting in the babbler's output channel.
 	assert.LessOrEqual(t, totalSubmitted, expectSubmitted, "too many events submitted (node event buffer overflow)")
 }
@@ -223,9 +224,12 @@ func (b *blabber) Go() {
 				return
 			default:
 			}
+			if b.batchSize > math.MaxInt {
+				panic("batch size too big for int")
+			}
 			evts := stdtypes.ListOf(sliceutil.Repeat(
 				stdtypes.Event(stdevents.NewTestUint64("consumer", 0)),
-				int(b.batchSize),
+				int(b.batchSize), //nolint:gosec
 			)...)
 			select {
 			case <-b.stop:

pkg/availability/multisigcollector/multisigcollector.go

@@ -1,6 +1,7 @@
 package multisigcollector
 
 import (
+	"fmt"
 	"math"
 
 	"google.golang.org/protobuf/proto"
@@ -69,12 +70,17 @@ func NewReconfigurableModule(mc ModuleConfig, paramsTemplate ModuleParams, logge
 				submc := mc
 				submc.Self = mscID
 
+				// Check for integer overflow
+				if mscParams.MaxRequests > math.MaxInt {
+					return nil, fmt.Errorf("max requests too high for int type: %d", mscParams.MaxRequests)
+				}
+
 				// Fill in instance-specific parameters.
 				moduleParams := paramsTemplate
 				moduleParams.InstanceUID = []byte(mscID)
 				moduleParams.EpochNr = mscParams.Epoch
 				moduleParams.Membership = mscParams.Membership
-				moduleParams.MaxRequests = int(mscParams.MaxRequests)
+				moduleParams.MaxRequests = int(mscParams.MaxRequests) //nolint:gosec
 				// TODO: Use InstanceUIDs properly.
 				//       (E.g., concatenate this with the instantiating protocol's InstanceUID when introduced.)
 

pkg/checkpoint/chkpvalidator/conservativecv.go

@@ -1,15 +1,16 @@
 package chkpvalidator
 
 import (
-	es "github.com/go-errors/errors"
+	"math"
 
-	t "github.com/filecoin-project/mir/stdtypes"
+	es "github.com/go-errors/errors"
 
 	"github.com/filecoin-project/mir/pkg/checkpoint"
 	"github.com/filecoin-project/mir/pkg/crypto"
 	checkpointpbtypes "github.com/filecoin-project/mir/pkg/pb/checkpointpb/types"
 	trantorpbtypes "github.com/filecoin-project/mir/pkg/pb/trantorpb/types"
 	tt "github.com/filecoin-project/mir/pkg/trantor/types"
+	t "github.com/filecoin-project/mir/stdtypes"
 )
 
 type ConservativeCV struct {
@@ -55,14 +56,25 @@ func (ccv *ConservativeCV) Verify(
 		return es.Errorf("nodeID not in membership")
 	}
 
+	// Check if epoch is in integer bounds.
+	if sc.Epoch() > math.MaxInt || epochNr > math.MaxInt {
+		return es.Errorf("epoch number out of integer range")
+	}
+
 	// Check how far the received stable checkpoint is ahead of the local node's state.
-	chkpMembershipOffset := int(sc.Epoch()) - 1 - int(epochNr)
+	// Integer casting required here to prevent underflow.
+	chkpMembershipOffset := int(sc.Epoch()) - 1 - int(epochNr) //nolint:gosec
 	if chkpMembershipOffset <= 0 {
 		// Ignore stable checkpoints that are not far enough
 		// ahead of the current state of the local node.
 		return es.Errorf("checkpoint not far ahead enough")
 	}
 
+	// Make sure ccv.configOffset is non-negative before conversion
+	if ccv.configOffset < 0 {
+		return es.Errorf("configOffset cannot be negative")
+	}
+
 	if chkpMembershipOffset > ccv.configOffset {
 		// cannot verify checkpoint signatures, too far ahead
 		return es.Errorf("checkpoint too far ahead")

pkg/checkpoint/chkpvalidator/permissivecv.go

@@ -1,17 +1,17 @@
 package chkpvalidator
 
 import (
-	es "github.com/go-errors/errors"
-
-	t "github.com/filecoin-project/mir/stdtypes"
+	"math"
 
-	"github.com/filecoin-project/mir/pkg/logging"
+	es "github.com/go-errors/errors"
 
 	"github.com/filecoin-project/mir/pkg/checkpoint"
 	"github.com/filecoin-project/mir/pkg/crypto"
+	"github.com/filecoin-project/mir/pkg/logging"
 	checkpointpbtypes "github.com/filecoin-project/mir/pkg/pb/checkpointpb/types"
 	trantorpbtypes "github.com/filecoin-project/mir/pkg/pb/trantorpb/types"
 	tt "github.com/filecoin-project/mir/pkg/trantor/types"
+	t "github.com/filecoin-project/mir/stdtypes"
 )
 
 type PermissiveCV struct {
@@ -50,13 +50,19 @@ func (pcv *PermissiveCV) Verify(chkp *checkpointpbtypes.StableCheckpoint, epochN
 		return es.Errorf("nodeID not in membership")
 	}
 
+	// Check if epoch is in integer bounds.
+	if sc.Epoch() > math.MaxInt || epochNr > math.MaxInt {
+		return es.Errorf("epoch number out of integer range")
+	}
+
 	// ATTENTION: We are using the membership contained in the checkpoint itself
 	// as the one to verify its certificate against.
 	// This is a vulnerability, since any the state of any node can be corrupted
 	// simply by receiving a maliciously crafted checkpoint.
 	// Thus, the permissive checker is a form of a stub and should not be used in production.
 	chkpMembership := sc.PreviousMembership()
-	chkpMembershipOffset := int(sc.Epoch()) - 1 - int(epochNr)
+	// Integer casting required here to prevent underflow.
+	chkpMembershipOffset := int(sc.Epoch()) - 1 - int(epochNr) //nolint:gosec
 
 	if chkpMembershipOffset > pcv.configOffset {
 		// cannot verify checkpoint signatures, too far ahead

pkg/deploytest/deployment.go

@@ -179,8 +179,8 @@ func (d *Deployment) Run(ctx context.Context) (nodeErrors []error, heapObjects i
 		<-ctx.Done()
 		runtime.GC()
 		runtime.ReadMemStats(&m2)
-		heapObjects = int64(m2.HeapObjects - m1.HeapObjects)
-		heapAlloc = int64(m2.HeapAlloc - m1.HeapAlloc)
+		heapObjects = int64(m2.HeapObjects - m1.HeapObjects) //nolint:gosec
+		heapAlloc = int64(m2.HeapAlloc - m1.HeapAlloc)       //nolint:gosec
 		cancel()
 	}()
 

pkg/deploytest/testreplica.go

@@ -183,7 +183,7 @@ func (tr *TestReplica) submitFakeTransactions(ctx context.Context, node *mir.Nod
 				destModule,
 				[]*trantorpbtypes.Transaction{{
 					ClientId: tt.NewClientIDFromInt(0),
-					TxNo:     tt.TxNo(i),
+					TxNo:     tt.TxNo(i), //nolint:gosec
 					Data:     []byte(fmt.Sprintf("Transaction %d", i)),
 				}},
 			).Pb())

pkg/dsl/dslmodule.go

@@ -162,16 +162,19 @@ func (h Handle) RecoverAndCleanupContext(id ContextID) any {
 // The ImplementsModule method only serves the purpose of indicating that this is a Module and must not be called.
 func (m *dslModuleImpl) ImplementsModule() {}
 
-// EmitEvent adds the event to the queue of output events
-// NB: This function works with the (legacy) protoc-generated types and is likely to be
-// removed in the future, with EmitMirEvent taking its place.
+// EmitEvent adds the event to the queue of output events.
 func EmitEvent(m Module, ev stdtypes.Event) {
 	m.DslHandle().impl.outputEvents.PushBack(ev)
 }
 
-// EmitMirEvent adds the event to the queue of output events
+// EmitEvents adds the events to the queue of output events.
+func EmitEvents(m Module, events *stdtypes.EventList) {
+	m.DslHandle().impl.outputEvents.PushBackList(events)
+}
+
+// EmitMirEvent adds the Mir-generated event to the queue of output events.
 // NB: this function works with the Mir-generated types.
-// For the (legacy) protoc-generated types, EmitEvent can be used.
+// For use with the general event type, see EmitEvent.
 func EmitMirEvent(m Module, ev *eventpbtypes.Event) {
 	m.DslHandle().impl.outputEvents.PushBack(ev.Pb())
 }

pkg/dsl/test/dslmodule_test.go

@@ -278,7 +278,7 @@ func newContextTestingModule(mc *contextTestingModuleModuleConfig) dsl.Module {
 
 			// NB: avoid using primitive types as the context in the actual implementation, prefer named structs,
 			//     remember that the context type is used to match requests with responses.
-			cryptopbdsl.VerifySigs(m, mc.Crypto, sliceutil.Repeat(msg, int(u)), signatures, nodeIDs, &u)
+			cryptopbdsl.VerifySigs(m, mc.Crypto, sliceutil.Repeat(msg, int(u)), signatures, nodeIDs, &u) //nolint:gosec
 		}
 		return nil
 	})