Worker executor response signature support

consensus/pbft/execute.go

@@ -1,6 +1,7 @@
 package pbft
 
 import (
+	"encoding/hex"
 	"fmt"
 	"time"
 
@@ -98,6 +99,7 @@ func (r *Replica) execute(view uint, sequence uint, digest string) error {
 			RequestTimestamp: request.Timestamp,
 			Replica:          r.id,
 		},
+		Signature: hex.EncodeToString(res.Signature),
 	}
 
 	// Save this executions in case it's requested again.

executor/config.go

@@ -4,26 +4,32 @@ import (
 	"github.com/spf13/afero"
 
 	"github.com/blocklessnetwork/b7s/models/blockless"
+	"github.com/blocklessnetwork/b7s/models/execute"
 )
 
+type ExecutionSigner interface {
+	Sign(execute.Request, execute.RuntimeOutput) ([]byte, error)
+}
+
 // defaultConfig used to create Executor.
 var defaultConfig = Config{
-	WorkDir:        "workspace",
-	RuntimeDir:     "",
-	ExecutableName: blockless.RuntimeCLI(),
-	FS:             afero.NewOsFs(),
-	Limiter:        &noopLimiter{},
+	WorkDir:         "workspace",
+	RuntimeDir:      "",
+	ExecutableName:  blockless.RuntimeCLI(),
+	FS:              afero.NewOsFs(),
+	Limiter:         &noopLimiter{},
 	DriversRootPath: "",
 }
 
 // Config represents the Executor configuration.
 type Config struct {
-	WorkDir        string   // directory where files needed for the execution are stored
-	RuntimeDir     string   // directory where the executable can be found
-	ExecutableName string   // name for the executable
-	DriversRootPath string // where are cgi drivers stored
-	FS             afero.Fs // FS accessor
-	Limiter        Limiter  // Resource limiter for executed processes
+	WorkDir         string          // directory where files needed for the execution are stored
+	RuntimeDir      string          // directory where the executable can be found
+	ExecutableName  string          // name for the executable
+	DriversRootPath string          // where are cgi drivers stored
+	FS              afero.Fs        // FS accessor
+	Limiter         Limiter         // Resource limiter for executed processes
+	Signer          ExecutionSigner // Signer for the executor
 }
 
 type Option func(*Config)
@@ -62,3 +68,10 @@ func WithLimiter(limiter Limiter) Option {
 		cfg.Limiter = limiter
 	}
 }
+
+// WithSigner sets the signer for the executor.
+func WithSigner(signer ExecutionSigner) Option {
+	return func(cfg *Config) {
+		cfg.Signer = signer
+	}
+}

executor/execute_function.go

@@ -11,16 +11,15 @@ import (
 func (e *Executor) ExecuteFunction(requestID string, req execute.Request) (execute.Result, error) {
 
 	// Execute the function.
-	out, usage, err := e.executeFunction(requestID, req)
+	out, usage, signature, err := e.executeFunction(requestID, req)
 	if err != nil {
-
 		res := execute.Result{
 			Code:      codes.Error,
 			RequestID: requestID,
 			Result:    out,
 			Usage:     usage,
+			Signature: signature,
 		}
-
 		return res, fmt.Errorf("function execution failed: %w", err)
 	}
 
@@ -29,14 +28,15 @@ func (e *Executor) ExecuteFunction(requestID string, req execute.Request) (execu
 		RequestID: requestID,
 		Result:    out,
 		Usage:     usage,
+		Signature: signature,
 	}
 
 	return res, nil
 }
 
 // executeFunction handles the actual execution of the Blockless function. It returns the
 // execution information like standard output, standard error, exit code and resource usage.
-func (e *Executor) executeFunction(requestID string, req execute.Request) (execute.RuntimeOutput, execute.Usage, error) {
+func (e *Executor) executeFunction(requestID string, req execute.Request) (execute.RuntimeOutput, execute.Usage, []byte, error) {
 
 	log := e.log.With().Str("request", requestID).Str("function", req.FunctionID).Logger()
 
@@ -47,7 +47,7 @@ func (e *Executor) executeFunction(requestID string, req execute.Request) (execu
 
 	err := e.cfg.FS.MkdirAll(paths.workdir, defaultPermissions)
 	if err != nil {
-		return execute.RuntimeOutput{}, execute.Usage{}, fmt.Errorf("could not setup working directory for execution (dir: %s): %w", paths.workdir, err)
+		return execute.RuntimeOutput{}, execute.Usage{}, []byte{}, fmt.Errorf("could not setup working directory for execution (dir: %s): %w", paths.workdir, err)
 	}
 	// Remove all temporary files after we're done.
 	defer func() {
@@ -66,10 +66,18 @@ func (e *Executor) executeFunction(requestID string, req execute.Request) (execu
 
 	out, usage, err := e.executeCommand(cmd)
 	if err != nil {
-		return out, execute.Usage{}, fmt.Errorf("command execution failed: %w", err)
+		return out, execute.Usage{}, []byte{}, fmt.Errorf("command execution failed: %w", err)
 	}
 
 	log.Info().Msg("command executed successfully")
 
-	return out, usage, nil
+	var signature []byte
+	if e.cfg.Signer != nil {
+		signature, err = e.cfg.Signer.Sign(req, out)
+		if err != nil {
+			return out, usage, []byte{}, fmt.Errorf("could not sign output: %w", err)
+		}
+	}
+
+	return out, usage, signature, nil
 }

models/execute/response.go

@@ -15,6 +15,7 @@ type Result struct {
 	Result    RuntimeOutput `json:"result"`
 	RequestID string        `json:"request_id"`
 	Usage     Usage         `json:"usage,omitempty"`
+	Signature []byte        `json:"signature,omitempty"`
 }
 
 // Cluster represents the set of peers that executed the request.

node/aggregate/aggregate.go

@@ -17,11 +17,14 @@ type Result struct {
 	Peers []peer.ID `json:"peers,omitempty"`
 	// How frequent was this result, in percentages.
 	Frequency float64 `json:"frequency,omitempty"`
+	// Signature of this result
+	Signature []byte `json:"signature,omitempty"`
 }
 
 type resultStats struct {
-	seen  uint
-	peers []peer.ID
+	seen      uint
+	peers     []peer.ID
+	signature []byte
 }
 
 func Aggregate(results execute.ResultMap) Results {
@@ -40,13 +43,15 @@ func Aggregate(results execute.ResultMap) Results {
 		stat, ok := stats[output]
 		if !ok {
 			stats[output] = resultStats{
-				seen:  0,
-				peers: make([]peer.ID, 0),
+				seen:      0,
+				peers:     make([]peer.ID, 0),
+				signature: res.Signature,
 			}
 		}
 
 		stat.seen++
 		stat.peers = append(stat.peers, executingPeer)
+		stat.signature = res.Signature
 
 		stats[output] = stat
 	}
@@ -59,6 +64,7 @@ func Aggregate(results execute.ResultMap) Results {
 			Result:    res,
 			Peers:     stat.peers,
 			Frequency: 100 * float64(stat.seen) / float64(total),
+			Signature: stat.signature,
 		}
 
 		aggregated = append(aggregated, aggr)

node/worker_execute.go

@@ -2,6 +2,7 @@ package node
 
 import (
 	"context"
+	"encoding/hex"
 	"fmt"
 	"time"
 
@@ -47,6 +48,7 @@ func (n *Node) workerProcessExecute(ctx context.Context, from peer.ID, req reque
 		Results: execute.ResultMap{
 			n.host.ID(): result,
 		},
+		Signature: hex.EncodeToString(result.Signature),
 	}
 
 	// Send the response, whatever it may be (success or failure).