[AppGate] Implement the MVP AppGateServer

cmd/pocketd/cmd/root.go

@@ -43,6 +43,7 @@ import (
 
 	"github.com/pokt-network/poktroll/app"
 	appparams "github.com/pokt-network/poktroll/app/params"
+	appgateservercmd "github.com/pokt-network/poktroll/pkg/appgateserver/cmd"
 )
 
 // NewRootCmd creates a new root command for a Cosmos SDK application
@@ -148,6 +149,11 @@ func initRootCmd(
 		txCommand(),
 		keys.Commands(app.DefaultNodeHome),
 	)
+
+	// add the appgate server command
+	rootCmd.AddCommand(
+		appgateservercmd.AppGateServerCmd(),
+	)
 }
 
 // queryCommand returns the sub-command to send queries to the app

go.mod

@@ -7,6 +7,7 @@ require (
 	cosmossdk.io/depinject v1.0.0-alpha.3
 	cosmossdk.io/errors v1.0.0-beta.7
 	cosmossdk.io/math v1.0.1
+	github.com/athanorlabs/go-dleq v0.1.0
 	github.com/cometbft/cometbft v0.37.2
 	github.com/cometbft/cometbft-db v0.8.0
 	github.com/cosmos/cosmos-proto v1.0.0-beta.2
@@ -20,6 +21,7 @@ require (
 	github.com/gorilla/websocket v1.5.0
 	github.com/grpc-ecosystem/grpc-gateway v1.16.0
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.15.2
+	github.com/noot/ring-go v0.0.0-20231019173746-6c4b33bcf03f
 	github.com/pokt-network/smt v0.7.1
 	github.com/regen-network/gocuke v0.6.2
 	github.com/spf13/cast v1.5.1

go.sum

@@ -300,6 +300,8 @@ github.com/armon/go-radix v1.0.0/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgI
 github.com/aryann/difflib v0.0.0-20170710044230-e206f873d14a/go.mod h1:DAHtR1m6lCRdSC2Tm3DSWRPvIPr6xNKyeHdqDQSQT+A=
 github.com/ashanbrown/forbidigo v1.3.0/go.mod h1:vVW7PEdqEFqapJe95xHkTfB1+XvZXBFg8t0sG2FIxmI=
 github.com/ashanbrown/makezero v1.1.1/go.mod h1:i1bJLCRSCHOcOa9Y6MyF2FTfMZMFdHvxKHxgO5Z1axI=
+github.com/athanorlabs/go-dleq v0.1.0 h1:0/llWZG8fz2uintMBKOiBC502zCsDA8nt8vxI73W9Qc=
+github.com/athanorlabs/go-dleq v0.1.0/go.mod h1:DWry6jSD7A13MKmeZA0AX3/xBeQCXDoygX99VPwL3yU=
 github.com/aws/aws-lambda-go v1.13.3/go.mod h1:4UKl9IzQMoD+QF79YdCuzCwp8VbmG4VAQwij/eHl5CU=
 github.com/aws/aws-sdk-go v1.23.20/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo=
 github.com/aws/aws-sdk-go v1.25.37/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo=
@@ -1481,6 +1483,8 @@ github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLA
 github.com/nishanths/exhaustive v0.8.1/go.mod h1:qj+zJJUgJ76tR92+25+03oYUhzF4R7/2Wk7fGTfCHmg=
 github.com/nishanths/predeclared v0.0.0-20190419143655-18a43bb90ffc/go.mod h1:62PewwiQTlm/7Rj+cxVYqZvDIUc+JjZq6GHAC1fsObQ=
 github.com/nishanths/predeclared v0.2.2/go.mod h1:RROzoN6TnGQupbC+lqggsOlcgysk3LMK/HI84Mp280c=
+github.com/noot/ring-go v0.0.0-20231019173746-6c4b33bcf03f h1:1+NP/H13eFAqBYrGpRkbJUWVWIO2Zr2eP7a/q0UtZVQ=
+github.com/noot/ring-go v0.0.0-20231019173746-6c4b33bcf03f/go.mod h1:0t3gzoSfW2bkTce1E/Jis3MQpjiKGhAgqieFK+nkQsI=
 github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
 github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
 github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU=

pkg/appgateserver/cmd/cmd.go

@@ -0,0 +1,147 @@
+package cmd
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"net/http"
+	"net/url"
+	"os"
+	"os/signal"
+
+	"cosmossdk.io/depinject"
+	cosmosclient "github.com/cosmos/cosmos-sdk/client"
+	"github.com/cosmos/cosmos-sdk/client/flags"
+	"github.com/spf13/cobra"
+
+	"github.com/pokt-network/poktroll/pkg/appgateserver"
+	blockclient "github.com/pokt-network/poktroll/pkg/client/block"
+	eventsquery "github.com/pokt-network/poktroll/pkg/client/events_query"
+)
+
+var (
+	flagSigningKey        string
+	flagSelfSigning       bool
+	flagListeningEndpoint string
+	flagCometWebsocketUrl string
+)
+
+func AppGateServerCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "appgate-server",
+		Short: "Starts the AppGate server",
+		Long: `Starts the AppGate server that listens for incoming relay requests and handles
+the necessary on-chain interactions (sessions, suppliers, etc) to receive the
+respective relay response.
+
+-- App Mode (Flag)- -
+If the server is started with a defined '--self-signing' flag, it will behave
+as an Application. Any incoming requests will be signed by using the private
+key and ring associated with the '--signing-key' flag.
+
+-- Gateway Mode (Flag)--
+If the '--self-signing' flag is not provided, the server will behave as a Gateway.
+It will sign relays on behalf of any Application sending it relays, provided
+that the address associated with '--signing-key' has been delegated to. This is
+necessary for the application<->gateway ring signature to function.
+
+-- App Mode (HTTP) --
+If an application doesn't provide the '--self-signing' flag, it can still send
+relays to the AppGate server and function as an Application, provided that:
+1. Each request contains the '?senderAddress=[address]' query parameter
+2. The key associated with the '--signing-key' flag belongs to the address
+   provided in the request, otherwise the ring signature will not be valid.`,
+		Args: cobra.NoArgs,
+		RunE: runAppGateServer,
+	}
+
+	cmd.Flags().StringVar(&flagSigningKey, "signing-key", "", "The name of the key that will be used to sign relays")
+	cmd.Flags().StringVar(&flagListeningEndpoint, "listening-endpoint", "http://localhost:42069", "The host and port that the appgate server will listen on")
+	cmd.Flags().StringVar(&flagCometWebsocketUrl, "comet-websocket-url", "ws://localhost:36657/websocket", "The URL of the comet websocket endpoint to communicate with the pocket blockchain")
+	cmd.Flags().BoolVar(&flagSelfSigning, "self-signing", false, "Whether the server should sign all incoming requests with its own ring (for applications)")
+
+	cmd.Flags().String(flags.FlagKeyringBackend, "", "Select keyring's backend (os|file|kwallet|pass|test)")
+	cmd.Flags().String(flags.FlagNode, "tcp://localhost:36657", "The URL of the comet tcp endpoint to communicate with the pocket blockchain")
+
+	return cmd
+}
+
+func runAppGateServer(cmd *cobra.Command, _ []string) error {
+	// Create a context that is canceled when the command is interrupted
+	ctx, cancelCtx := context.WithCancel(cmd.Context())
+	defer cancelCtx()
+
+	// Handle interrupts in a goroutine.
+	go func() {
+		sigCh := make(chan os.Signal, 1)
+		signal.Notify(sigCh, os.Interrupt)
+
+		// Block until we receive an interrupt or kill signal (OS-agnostic)
+		<-sigCh
+		log.Println("INFO: Interrupt signal received, shutting down...")
+
+		// Signal goroutines to stop
+		cancelCtx()
+	}()
+
+	// Parse the listening endpoint.
+	listeningUrl, err := url.Parse(flagListeningEndpoint)
+	if err != nil {
+		return fmt.Errorf("failed to parse listening endpoint: %w", err)
+	}
+
+	// Setup the AppGate server dependencies.
+	appGateServerDeps, err := setupAppGateServerDependencies(cmd, ctx, flagCometWebsocketUrl)
+	if err != nil {
+		return fmt.Errorf("failed to setup AppGate server dependencies: %w", err)
+	}
+
+	log.Println("INFO: Creating AppGate server...")
+
+	// Create the AppGate server.
+	appGateServer, err := appgateserver.NewAppGateServer(
+		appGateServerDeps,
+		appgateserver.WithSigningInformation(&appgateserver.SigningInformation{
+			// provide the name of the key to use for signing all incoming requests
+			SigningKeyName: flagSigningKey,
+			// provide whether the appgate server should sign all incoming requests
+			// with its own ring (for applications) or not (for gateways)
+			SelfSigning: flagSelfSigning,
+		}),
+		appgateserver.WithListeningUrl(listeningUrl),
+	)
+	if err != nil {
+		return fmt.Errorf("failed to create AppGate server: %w", err)
+	}
+
+	log.Printf("INFO: Starting AppGate server, listening on %s...", listeningUrl.String())
+
+	// Start the AppGate server.
+	if err := appGateServer.Start(ctx); err != nil && !errors.Is(err, http.ErrServerClosed) {
+		return fmt.Errorf("failed to start app gate server: %w", err)
+	} else if errors.Is(err, http.ErrServerClosed) {
+		log.Println("INFO: AppGate server stopped")
+	}
+
+	return nil
+}
+
+func setupAppGateServerDependencies(cmd *cobra.Command, ctx context.Context, cometWebsocketUrl string) (depinject.Config, error) {
+	// Retrieve the client context for the chain interactions.
+	clientCtx := cosmosclient.GetClientContextFromCmd(cmd)
+
+	// Create the events client.
+	eventsQueryClient := eventsquery.NewEventsQueryClient(flagCometWebsocketUrl)
+
+	// Create the block client.
+	log.Printf("INFO: Creating block client, using comet websocket URL: %s...", flagCometWebsocketUrl)
+	deps := depinject.Supply(eventsQueryClient)
+	blockClient, err := blockclient.NewBlockClient(ctx, deps, flagCometWebsocketUrl)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create block client: %w", err)
+	}
+
+	// Return the dependencie config.
+	return depinject.Supply(clientCtx, blockClient), nil
+}

pkg/appgateserver/endpoint_selector.go

@@ -0,0 +1,45 @@
+package appgateserver
+
+import (
+	"context"
+	"log"
+	"net/url"
+
+	sessiontypes "github.com/pokt-network/poktroll/x/session/types"
+	sharedtypes "github.com/pokt-network/poktroll/x/shared/types"
+)
+
+// TODO_IMPROVE: This implements a naive greedy approach that defaults to the
+// first available supplier. Future optimizations (e.g. Quality-of-Service) can be introduced here.
+// TODO(@h5law): Look into different endpoint selection depending on their suitability.
+// getRelayerUrl gets the URL of the relayer for the given service.
+func (app *appGateServer) getRelayerUrl(
+	ctx context.Context,
+	serviceId string,
+	rpcType sharedtypes.RPCType,
+	session *sessiontypes.Session,
+) (supplierUrl *url.URL, supplierAddress string, err error) {
+	for _, supplier := range session.Suppliers {
+		for _, service := range supplier.Services {
+			// Skip services that don't match the requested serviceId.
+			if service.Service.Id != serviceId {
+				continue
+			}
+
+			for _, endpoint := range service.Endpoints {
+				// Return the first endpoint url that matches the JSON RPC RpcType.
+				if endpoint.RpcType == rpcType {
+					supplierUrl, err := url.Parse(endpoint.Url)
+					if err != nil {
+						log.Printf("error parsing url: %s", err)
+						continue
+					}
+					return supplierUrl, supplier.Address, nil
+				}
+			}
+		}
+	}
+
+	// Return an error if no relayer endpoints were found.
+	return nil, "", ErrAppGateNoRelayEndpoints
+}

pkg/appgateserver/errors.go

@@ -0,0 +1,13 @@
+package appgateserver
+
+import sdkerrors "cosmossdk.io/errors"
+
+var (
+	codespace                               = "appgateserver"
+	ErrAppGateInvalidRelayResponseSignature = sdkerrors.Register(codespace, 1, "invalid relay response signature")
+	ErrAppGateNoRelayEndpoints              = sdkerrors.Register(codespace, 2, "no relay endpoints found")
+	ErrAppGateInvalidRequestURL             = sdkerrors.Register(codespace, 3, "invalid request URL")
+	ErrAppGateMissingAppAddress             = sdkerrors.Register(codespace, 4, "missing application address")
+	ErrAppGateMissingSigningInformation     = sdkerrors.Register(codespace, 5, "missing app client signing information")
+	ErrAppGateMissingListeningEndpoint      = sdkerrors.Register(codespace, 6, "missing app client listening endpoint")
+)

pkg/appgateserver/jsonrpc.go

@@ -0,0 +1,132 @@
+package appgateserver
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"log"
+	"net/http"
+
+	"github.com/cometbft/cometbft/crypto"
+
+	"github.com/pokt-network/poktroll/x/service/types"
+	sharedtypes "github.com/pokt-network/poktroll/x/shared/types"
+)
+
+// handleJSONRPCRelay handles JSON RPC relay requests.
+// It does everything from preparing, signing and sending the request.
+// It then blocks on the response to come back and forward it to the provided writer.
+func (app *appGateServer) handleJSONRPCRelay(
+	ctx context.Context,
+	appAddress, serviceId string,
+	request *http.Request,
+	writer http.ResponseWriter,
+) error {
+	// Read the request body bytes.
+	payloadBz, err := io.ReadAll(request.Body)
+	if err != nil {
+		return err
+	}
+
+	// Create the relay request payload.
+	relayRequestPayload := &types.RelayRequest_JsonRpcPayload{}
+	relayRequestPayload.JsonRpcPayload.Unmarshal(payloadBz)
+
+	session, err := app.getCurrentSession(ctx, appAddress, serviceId)
+	if err != nil {
+		return err
+	}
+	log.Printf("DEBUG: Current session ID: %s", session.SessionId)
+
+	// Get a supplier URL and address for the given service and session.
+	supplierUrl, supplierAddress, err := app.getRelayerUrl(ctx, serviceId, sharedtypes.RPCType_JSON_RPC, session)
+	if err != nil {
+		return err
+	}
+
+	// Create the relay request.
+	relayRequest := &types.RelayRequest{
+		Meta: &types.RelayRequestMetadata{
+			SessionHeader: session.Header,
+			Signature:     nil, // signature added below
+		},
+		Payload: relayRequestPayload,
+	}
+
+	// Get the application's signer.
+	signer, err := app.getRingSingerForAppAddress(ctx, appAddress)
+	if err != nil {
+		return err
+	}
+
+	// Hash and sign the request's signable bytes.
+	signableBz, err := relayRequest.GetSignableBytes()
+	if err != nil {
+		return err
+	}
+
+	hash := crypto.Sha256(signableBz)
+	signature, err := signer.Sign(hash)
+	if err != nil {
+		return err
+	}
+	relayRequest.Meta.Signature = signature
+
+	// Marshal the relay request to bytes and create a reader to be used as an HTTP request body.
+	relayRequestBz, err := relayRequest.Marshal()
+	if err != nil {
+		return err
+	}
+	relayRequestReader := io.NopCloser(bytes.NewReader(relayRequestBz))
+
+	// Create the HTTP request to send the request to the relayer.
+	relayHTTPRequest := &http.Request{
+		Method: request.Method,
+		Header: request.Header,
+		URL:    supplierUrl,
+		Body:   relayRequestReader,
+	}
+
+	// Perform the HTTP request to the relayer.
+	log.Printf("DEBUG: Sending signed relay request to %s", supplierUrl)
+	relayHTTPResponse, err := http.DefaultClient.Do(relayHTTPRequest)
+	if err != nil {
+		return err
+	}
+
+	// Read the response body bytes.
+	relayResponseBz, err := io.ReadAll(relayHTTPResponse.Body)
+	if err != nil {
+		return err
+	}
+
+	// Unmarshal the response bytes into a RelayResponse.
+	relayResponse := &types.RelayResponse{}
+	if err := relayResponse.Unmarshal(relayResponseBz); err != nil {
+		return err
+	}
+
+	// Verify the response signature. We use the supplier address that we got from
+	// the getRelayerUrl function since this is the address we are expecting to sign the response.
+	// TODO_TECHDEBT: if the RelayResponse is an internal error response, we should not verify the signature
+	// as in some relayer early failures, it may not be signed by the supplier.
+	// TODO_IMPROVE: Add more logging & telemetry so we can get visibility and signal into
+	// failed responses.
+	log.Println("DEBUG: Verifying signed relay response from...")
+	if err := app.verifyResponse(ctx, supplierAddress, relayResponse); err != nil {
+		return err
+	}
+
+	// Marshal the response payload to bytes to be sent back to the application.
+	var responsePayloadBz []byte
+	if _, err = relayResponse.Payload.MarshalTo(responsePayloadBz); err != nil {
+		return err
+	}
+
+	// Reply with the RelayResponse payload.
+	if _, err := writer.Write(relayRequestBz); err != nil {
+		return err
+	}
+
+	return nil
+}