[AppGate] Implement the MVP AppGateServer

pkg/appclient/appclient.go

@@ -0,0 +1,159 @@
+package appclient
+
+import (
+	"context"
+	"log"
+	"net/http"
+	"net/url"
+	"strings"
+
+	sdkclient "github.com/cosmos/cosmos-sdk/client"
+	"github.com/cosmos/cosmos-sdk/crypto/keyring"
+	accounttypes "github.com/cosmos/cosmos-sdk/x/auth/types"
+
+	blocktypes "pocket/pkg/client"
+	"pocket/x/service/types"
+	sessiontypes "pocket/x/session/types"
+)
+
+type appClient struct {
+	// keyName is the name of the key in the keyring that will be used to sign relay requests.
+	keyName string
+	keyring keyring.Keyring
+
+	// clientCtx is the client context for the application.
+	// It is used to query for the application's account to unmarshal the supplier's account
+	// and get the public key to verify the relay response signature.
+	clientCtx sdkclient.Context
+
+	// appAddress is the address of the application that this app client is for.
+	appAddress string
+
+	// sessionQuerier is the querier for the session module.
+	// It used to get the current session for the application given a requested service.
+	sessionQuerier sessiontypes.QueryClient
+
+	// accountQuerier is the querier for the account module.
+	// It is used to get the the supplier's public key to verify the relay response signature.
+	accountQuerier accounttypes.QueryClient
+
+	// blockClient is the client for the block module.
+	// It is used to get the current block height to query for the current session.
+	blockClient blocktypes.BlockClient
+
+	// server is the HTTP server that will be used capture application requests
+	// so that they can be signed and relayed to the supplier.
+	server *http.Server
+}
+
+func NewAppClient(
+	clientCtx sdkclient.Context,
+	keyName string,
+	keyring keyring.Keyring,
+	applicationEndpoint *url.URL,
+	blockClient blocktypes.BlockClient,
+) *appClient {
+	sessionQuerier := sessiontypes.NewQueryClient(clientCtx)
+	accountQuerier := accounttypes.NewQueryClient(clientCtx)
+
+	return &appClient{
+		clientCtx:      clientCtx,
+		keyName:        keyName,
+		keyring:        keyring,
+		sessionQuerier: sessionQuerier,
+		accountQuerier: accountQuerier,
+		blockClient:    blockClient,
+		server:         &http.Server{Addr: applicationEndpoint.Host},
+	}
+}
+
+// Start starts the application server and blocks until the context is done
+// or the server returns an error.
+func (app *appClient) Start(ctx context.Context) error {
+	// Get and populate the application address from the keyring.
+	keyRecord, err := app.keyring.Key(app.keyName)
+	if err != nil {
+		return err
+	}
+
+	accAddress, err := keyRecord.GetAddress()
+	if err != nil {
+		return err
+	}
+
+	app.appAddress = accAddress.String()
+
+	// Shutdown the HTTP server when the context is done.
+	go func() {
+		<-ctx.Done()
+		app.server.Shutdown(ctx)
+	}()
+
+	// Start the HTTP server.
+	return app.server.ListenAndServe()
+}
+
+// Stop stops the application server and returns any error that occurred.
+func (app *appClient) Stop(ctx context.Context) error {
+	return app.server.Shutdown(ctx)
+}
+
+// ServeHTTP is the HTTP handler for the application server.
+// It captures the application request, signs it, and sends it to the supplier.
+// After receiving the response from the supplier, it verifies the response signature
+// before returning the response to the application.
+// The serviceId is extracted from the request path.
+func (app *appClient) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
+	ctx := request.Context()
+
+	// Extract the serviceId from the request path.
+	path := request.URL.Path
+	serviceId := strings.Split(path, "/")[1]
+
+	// Currently only JSON RPC requests are supported.
+	relayReponse, err := app.handleJSONRPCRelays(ctx, serviceId, request)
+	if err != nil {
+		// Reply with an error response if there was an error handling the relay.
+		app.replyWithError(writer, err)
+		log.Printf("ERROR: failed handling relay: %s", err)
+		return
+	}
+
+	// Reply with the RelayResponse payload.
+	if _, err := writer.Write(relayReponse); err != nil {
+		app.replyWithError(writer, err)
+		return
+	}
+	log.Print("INFO: request serviced successfully")
+}
+
+// replyWithError replies to the application with an error response.
+// TODO_TECHDEBT: This method should be aware of the nature of the error to use the appropriate JSONRPC
+// Code, Message and Data. Possibly by augmenting the passed in error with the adequate information.
+func (app *appClient) replyWithError(writer http.ResponseWriter, err error) {
+	relayResponse := &types.RelayResponse{
+		Payload: &types.RelayResponse_JsonRpcPayload{
+			JsonRpcPayload: &types.JSONRPCResponsePayload{
+				Id:      make([]byte, 0),
+				Jsonrpc: "2.0",
+				Error: &types.JSONRPCResponseError{
+					// Using conventional error code indicating internal server error.
+					Code:    -32000,
+					Message: err.Error(),
+					Data:    nil,
+				},
+			},
+		},
+	}
+
+	relayResponseBz, err := relayResponse.Marshal()
+	if err != nil {
+		log.Printf("ERROR: failed marshaling relay response: %s", err)
+		return
+	}
+
+	if _, err = writer.Write(relayResponseBz); err != nil {
+		log.Printf("ERROR: failed writing relay response: %s", err)
+		return
+	}
+}

pkg/appclient/endpoint_selector.go

@@ -0,0 +1,38 @@
+package appclient
+
+import (
+	"context"
+	"net/url"
+
+	sessiontypes "pocket/x/session/types"
+	sharedtypes "pocket/x/shared/types"
+)
+
+// getRelayerUrl gets the URL of the relayer for the given service.
+// It gets the suppliers list from the current session and returns
+// the first relayer URL that matches the JSON RPC RpcType.
+func (app *appClient) getRelayerUrl(
+	ctx context.Context,
+	serviceId string,
+	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.ServiceId.Id != serviceId {
+				continue
+			}
+
+			for _, endpoint := range service.Endpoints {
+				// Return the first endpoint url that matches the JSON RPC RpcType.
+				if endpoint.RpcType == sharedtypes.RPCType_JSON_RPC {
+					supplierUrl, err := url.Parse(endpoint.Url)
+					return supplierUrl, supplier.Address, err
+				}
+			}
+		}
+	}
+
+	// Return an error if no relayer endpoints were found.
+	return nil, "", ErrNoRelayEndpoints
+}

pkg/appclient/errors.go

@@ -0,0 +1,9 @@
+package appclient
+
+import sdkerrors "cosmossdk.io/errors"
+
+var (
+	codespace                        = "appclient"
+	ErrInvalidRelayResponseSignature = sdkerrors.Register(codespace, 1, "invalid relay response signature")
+	ErrNoRelayEndpoints              = sdkerrors.Register(codespace, 2, "no relay endpoints found")
+)

pkg/appclient/jsonrpc.go

@@ -0,0 +1,116 @@
+package appclient
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"net/http"
+
+	"github.com/cometbft/cometbft/crypto"
+
+	"pocket/x/service/types"
+	sessiontypes "pocket/x/session/types"
+	sharedtypes "pocket/x/shared/types"
+)
+
+// handleJSONRPCRelays handles JSON RPC relay requests.
+func (app *appClient) handleJSONRPCRelays(
+	ctx context.Context,
+	serviceId string,
+	request *http.Request,
+) (responseBz []byte, err error) {
+	// Read the request body bytes.
+	payloadBz, err := io.ReadAll(request.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	// Hash and sign the request payload.
+	hash := crypto.Sha256(payloadBz)
+	signature, _, err := app.keyring.Sign(app.keyName, hash)
+	if err != nil {
+		return nil, err
+	}
+
+	// Create the relay request payload.
+	relayRequestPayload := &types.RelayRequest_JsonRpcPayload{}
+	relayRequestPayload.JsonRpcPayload.Unmarshal(payloadBz)
+
+	// Get the current block height to query for the current session.
+	currentBlock := app.blockClient.LatestBlock(ctx)
+
+	// Query for the current session.
+	sessionQueryReq := sessiontypes.QueryGetSessionRequest{
+		ApplicationAddress: app.appAddress,
+		ServiceId:          &sharedtypes.ServiceId{Id: serviceId},
+		BlockHeight:        currentBlock.Height(),
+	}
+	sessionQueryRes, err := app.sessionQuerier.GetSession(ctx, &sessionQueryReq)
+	if err != nil {
+		return nil, err
+	}
+
+	session := sessionQueryRes.Session
+
+	// Get a supplier URL and address for the given service and session.
+	supplierUrl, supplierAddress, err := app.getRelayerUrl(ctx, serviceId, session)
+	if err != nil {
+		return nil, err
+	}
+
+	// Create the relay request.
+	relayRequest := &types.RelayRequest{
+		Meta: &types.RelayRequestMetadata{
+			SessionHeader: session.Header,
+			Signature:     signature,
+		},
+		Payload: relayRequestPayload,
+	}
+
+	// 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 nil, 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.
+	relayHTTPResponse, err := http.DefaultClient.Do(relayHTTPRequest)
+	if err != nil {
+		return nil, err
+	}
+
+	// Read the response body bytes.
+	relayResponseBz, err := io.ReadAll(relayHTTPResponse.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	// Unmarshal the response bytes into a RelayResponse.
+	relayResponse := &types.RelayResponse{}
+	if err := relayResponse.Unmarshal(relayResponseBz); err != nil {
+		return nil, 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.
+	if err := app.verifyResponse(ctx, supplierAddress, relayResponse); err != nil {
+		return nil, err
+	}
+
+	// Marshal the response payload to bytes to be sent back to the application.
+	var responsePayloadBz []byte
+	_, err = relayResponse.Payload.MarshalTo(responsePayloadBz)
+
+	return responsePayloadBz, err
+}

pkg/appclient/relay_verifier.go

@@ -0,0 +1,47 @@
+package appclient
+
+import (
+	"context"
+
+	"github.com/cometbft/cometbft/crypto"
+	accounttypes "github.com/cosmos/cosmos-sdk/x/auth/types"
+
+	"pocket/x/service/types"
+)
+
+// verifyResponse verifies the relay response signature.
+func (app *appClient) verifyResponse(
+	ctx context.Context,
+	supplierAddress string,
+	relayResponse *types.RelayResponse,
+) error {
+	// Query for the supplier account to get the application's public key to verify the relay request signature.
+	accQueryReq := &accounttypes.QueryAccountRequest{Address: supplierAddress}
+	accQueryRes, err := app.accountQuerier.Account(ctx, accQueryReq)
+	if err != nil {
+		return err
+	}
+
+	// Marshal the relay response payload to bytes and get the hash.
+	var payloadBz []byte
+	_, err = relayResponse.Payload.MarshalTo(payloadBz)
+	if err != nil {
+		return err
+	}
+	hash := crypto.Sha256(payloadBz)
+
+	// accQueryRes.Account.Value is a protobuf Any type that should be unmarshaled into an AccountI interface.
+	// TODO_TECHDEBT: Make sure our `AccountI`/`any` unmarshalling is correct.
+	// See https://github.com/pokt-network/poktroll/pull/101/files/edbd628e9146e232ef58c71cfa8f4be2135cdb50..fbba10626df79f6bf6e2218513dfdeb40a629790#r1372464439
+	var account accounttypes.AccountI
+	if err := app.clientCtx.Codec.UnmarshalJSON(accQueryRes.Account.Value, account); err != nil {
+		return err
+	}
+
+	// Verify the relay response signature.
+	if !account.GetPubKey().VerifySignature(hash, relayResponse.Meta.SupplierSignature) {
+		return ErrInvalidRelayResponseSignature
+	}
+
+	return nil
+}