multi: configure and start hashmail server

With this commit we make it possible to enable the Lightning Node
Connect mailbox server to be enabled and started as a local service
within aperture.

aperture.go

@@ -1,6 +1,7 @@
 package aperture
 
 import (
+	"context"
 	"crypto/tls"
 	"errors"
 	"fmt"
@@ -9,14 +10,17 @@ import (
 	"net/http"
 	"os"
 	"path/filepath"
+	"regexp"
 	"strings"
 	"sync"
 	"time"
 
+	gateway "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
 	flags "github.com/jessevdk/go-flags"
 	"github.com/lightninglabs/aperture/auth"
 	"github.com/lightninglabs/aperture/mint"
 	"github.com/lightninglabs/aperture/proxy"
+	"github.com/lightninglabs/lightning-node-connect/hashmailrpc"
 	"github.com/lightningnetwork/lnd"
 	"github.com/lightningnetwork/lnd/build"
 	"github.com/lightningnetwork/lnd/cert"
@@ -27,6 +31,9 @@ import (
 	"golang.org/x/crypto/acme/autocert"
 	"golang.org/x/net/http2"
 	"golang.org/x/net/http2/h2c"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/protobuf/encoding/protojson"
 	"gopkg.in/yaml.v2"
 )
 
@@ -55,6 +62,14 @@ const (
 	// the certificate validity length to make the chances bigger for it to
 	// be refreshed on a routine server restart.
 	selfSignedCertExpiryMargin = selfSignedCertValidity / 2
+
+	// hashMailGRPCPrefix is the prefix a gRPC request URI has when it is
+	// meant for the hashmailrpc server to be handled.
+	hashMailGRPCPrefix = "/hashmailrpc.HashMail/"
+
+	// hashMailRESTPrefix is the prefix a REST request URI has when it is
+	// meant for the hashmailrpc server to be handled.
+	hashMailRESTPrefix = "/v1/lightning-node-connect/hashmail"
 )
 
 var (
@@ -69,6 +84,12 @@ var (
 		tls.TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305,
 		tls.TLS_ECDHE_ECDSA_WITH_AES_128_CBC_SHA256,
 	}
+
+	// clientStreamingURIs is the list of REST URIs that are
+	// client-streaming and shouldn't be closed after a single message.
+	clientStreamingURIs = []*regexp.Regexp{
+		regexp.MustCompile("^/v1/lightning-node-connect/hashmail/send$"),
+	}
 )
 
 // Main is the true entrypoint of Aperture.
@@ -139,6 +160,7 @@ type Aperture struct {
 	httpsServer   *http.Server
 	torHTTPServer *http.Server
 	proxy         *proxy.Proxy
+	proxyCleanup  func()
 
 	wg   sync.WaitGroup
 	quit chan struct{}
@@ -190,7 +212,9 @@ func (a *Aperture) Start(errChan chan error) error {
 	}
 
 	// Create the proxy and connect it to lnd.
-	a.proxy, err = createProxy(a.cfg, a.challenger, a.etcdClient)
+	a.proxy, a.proxyCleanup, err = createProxy(
+		a.cfg, a.challenger, a.etcdClient,
+	)
 	if err != nil {
 		return err
 	}
@@ -288,6 +312,12 @@ func (a *Aperture) Stop() error {
 		a.challenger.Stop()
 	}
 
+	// Stop everything that was started alongside the proxy, for example the
+	// gRPC and REST servers.
+	if a.proxyCleanup != nil {
+		a.proxyCleanup()
+	}
+
 	// Shut down our client and server connections now. This should cause
 	// the first goroutine to quit.
 	cleanup(a.etcdClient, a.httpsServer, a.proxy)
@@ -585,7 +615,7 @@ func initTorListener(cfg *Config, etcd *clientv3.Client) (*tor.Controller, error
 
 // createProxy creates the proxy with all the services it needs.
 func createProxy(cfg *Config, challenger *LndChallenger,
-	etcdClient *clientv3.Client) (*proxy.Proxy, error) {
+	etcdClient *clientv3.Client) (*proxy.Proxy, func(), error) {
 
 	minter := mint.New(&mint.Config{
 		Challenger:     challenger,
@@ -600,20 +630,112 @@ func createProxy(cfg *Config, challenger *LndChallenger,
 	staticServer := http.NotFoundHandler()
 	if cfg.ServeStatic {
 		if len(strings.TrimSpace(cfg.StaticRoot)) == 0 {
-			return nil, fmt.Errorf("staticroot cannot be empty, " +
+			return nil, nil, fmt.Errorf("staticroot cannot be " +
-				"must contain path to directory that " +
+				"empty, must contain path to directory that " +
 				"contains index.html")
 		}
 		staticServer = http.FileServer(http.Dir(cfg.StaticRoot))
 	}
 
-	localServices := []proxy.LocalService{
+	var (
-		proxy.NewLocalService(staticServer, func(r *http.Request) bool {
+		localServices []proxy.LocalService
-			return true
+		proxyCleanup  = func() {}
-		}),
+	)
+
+	if cfg.HashMail.Enabled {
+		hashMailServices, cleanup, err := createHashMailServer(cfg)
+		if err != nil {
+			return nil, nil, err
+		}
+
+		localServices = append(localServices, hashMailServices...)
+		proxyCleanup = cleanup
 	}
 
-	return proxy.New(authenticator, cfg.Services, localServices...)
+	// The static file server must be last since it will match all calls
+	// that make it to it.
+	localServices = append(localServices, proxy.NewLocalService(
+		staticServer, func(r *http.Request) bool {
+			return true
+		},
+	))
+
+	prxy, err := proxy.New(authenticator, cfg.Services, localServices...)
+	return prxy, proxyCleanup, err
+}
+
+// createHashMailServer creates the gRPC server for the hash mail message
+// gateway and an additional REST and WebSocket capable proxy for that gRPC
+// server.
+func createHashMailServer(cfg *Config) ([]proxy.LocalService, func(), error) {
+	var localServices []proxy.LocalService
+
+	// Create a gRPC server for the hashmail server.
+	hashMailServer := newHashMailServer(hashMailServerConfig{
+		msgRate:           cfg.HashMail.MessageRate,
+		msgBurstAllowance: cfg.HashMail.MessageBurstAllowance,
+	})
+	hashMailGRPC := grpc.NewServer()
+	hashmailrpc.RegisterHashMailServer(hashMailGRPC, hashMailServer)
+	localServices = append(localServices, proxy.NewLocalService(
+		hashMailGRPC, func(r *http.Request) bool {
+			return strings.HasPrefix(r.URL.Path, hashMailGRPCPrefix)
+		}),
+	)
+
+	// And a REST proxy for it as well.
+	// The default JSON marshaler of the REST proxy only sets OrigName to
+	// true, which instructs it to use the same field names as specified in
+	// the proto file and not switch to camel case. What we also want is
+	// that the marshaler prints all values, even if they are falsey.
+	customMarshalerOption := gateway.WithMarshalerOption(
+		gateway.MIMEWildcard, &gateway.JSONPb{
+			MarshalOptions: protojson.MarshalOptions{
+				UseProtoNames:   true,
+				EmitUnpopulated: true,
+			},
+		},
+	)
+
+	// We'll also create and start an accompanying proxy to serve clients
+	// through REST.
+	ctxc, cancel := context.WithCancel(context.Background())
+	proxyCleanup := func() {
+		hashMailServer.Stop()
+		cancel()
+	}
+
+	mux := gateway.NewServeMux(customMarshalerOption)
+	err := hashmailrpc.RegisterHashMailHandlerFromEndpoint(
+		ctxc, mux, cfg.ListenAddr, []grpc.DialOption{
+			grpc.WithTransportCredentials(credentials.NewTLS(
+				&tls.Config{InsecureSkipVerify: true},
+			)),
+		},
+	)
+	if err != nil {
+		proxyCleanup()
+
+		return nil, nil, err
+	}
+
+	// Wrap the default grpc-gateway handler with the WebSocket handler.
+	restHandler := lnrpc.NewWebSocketProxy(
+		mux, log, time.Second*30, time.Second*5,
+		clientStreamingURIs,
+	)
+
+	// Create our proxy chain now. A request will pass
+	// through the following chain:
+	// req ---> CORS handler --> WS proxy ---> REST proxy --> gRPC endpoint
+	corsHandler := allowCORS(restHandler, []string{"*"})
+	localServices = append(localServices, proxy.NewLocalService(
+		corsHandler, func(r *http.Request) bool {
+			return strings.HasPrefix(r.URL.Path, hashMailRESTPrefix)
+		},
+	))
+
+	return localServices, proxyCleanup, nil
 }
 
 // cleanup closes the given server and shuts down the log rotator.
@@ -634,3 +756,55 @@ func cleanup(etcdClient io.Closer, server io.Closer, proxy io.Closer) {
 		log.Errorf("Could not close log rotator: %v", err)
 	}
 }
+
+// allowCORS wraps the given http.Handler with a function that adds the
+// Access-Control-Allow-Origin header to the response.
+func allowCORS(handler http.Handler, origins []string) http.Handler {
+	allowHeaders := "Access-Control-Allow-Headers"
+	allowMethods := "Access-Control-Allow-Methods"
+	allowOrigin := "Access-Control-Allow-Origin"
+
+	// If the user didn't supply any origins that means CORS is disabled
+	// and we should return the original handler.
+	if len(origins) == 0 {
+		return handler
+	}
+
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		origin := r.Header.Get("Origin")
+
+		// Skip everything if the browser doesn't send the Origin field.
+		if origin == "" {
+			handler.ServeHTTP(w, r)
+			return
+		}
+
+		// Set the static header fields first.
+		w.Header().Set(
+			allowHeaders,
+			"Content-Type, Accept, Grpc-Metadata-Macaroon",
+		)
+		w.Header().Set(allowMethods, "GET, POST, DELETE")
+
+		// Either we allow all origins or the incoming request matches
+		// a specific origin in our list of allowed origins.
+		for _, allowedOrigin := range origins {
+			if allowedOrigin == "*" || origin == allowedOrigin {
+				// Only set allowed origin to requested origin.
+				w.Header().Set(allowOrigin, origin)
+
+				break
+			}
+		}
+
+		// For a pre-flight request we only need to send the headers
+		// back. No need to call the rest of the chain.
+		if r.Method == "OPTIONS" {
+			return
+		}
+
+		// Everything's prepared now, we can pass the request along the
+		// chain of handlers.
+		handler.ServeHTTP(w, r)
+	})
+}

config.go

@@ -3,6 +3,7 @@ package aperture
 import (
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/btcsuite/btcutil"
 	"github.com/lightninglabs/aperture/proxy"
@@ -59,6 +60,12 @@ func (a *AuthConfig) validate() error {
 	return nil
 }
 
+type HashMailConfig struct {
+	Enabled               bool          `long:"enabled"`
+	MessageRate           time.Duration `long:"messagerate" description:"The average minimum time that should pass between each message."`
+	MessageBurstAllowance int           `long:"messageburstallowance" description:"The burst rate we allow for messages."`
+}
+
 type TorConfig struct {
 	Control     string `long:"control" description:"The host:port of the Tor instance."`
 	ListenPort  uint16 `long:"listenport" description:"The port we should listen on for client requests over Tor. Note that this port should not be exposed to the outside world, it is only intended to be reached by clients through the onion service."`
@@ -101,6 +108,10 @@ type Config struct {
 	// each backend service to Aperture.
 	Services []*proxy.Service `long:"service" description:"Configurations for each Aperture backend service."`
 
+	// HashMail is the configuration section for configuring the Lightning
+	// Node Connect mailbox server.
+	HashMail *HashMailConfig `long:"hashmail" description:"Configuration for the Lightning Node Connect mailbox server."`
+
 	// DebugLevel is a string defining the log level for the service either
 	// for all subsystems the same or individual level by subsystem.
 	DebugLevel string `long:"debuglevel" description:"Debug level for the Aperture application and its subsystems."`

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/btcsuite/btcwallet/wtxmgr v1.3.1-0.20210706234807-aaf03fee735a
 	github.com/fortytw2/leaktest v1.3.0
 	github.com/golang/protobuf v1.5.2
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.5.0
 	github.com/jessevdk/go-flags v1.4.0
 	github.com/lightninglabs/lightning-node-connect/hashmailrpc v1.0.2
 	github.com/lightninglabs/lndclient v0.12.0-9
@@ -21,6 +22,7 @@ require (
 	golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4
 	golang.org/x/time v0.0.0-20210220033141-f8bda1e9f3ba
 	google.golang.org/grpc v1.39.0
+	google.golang.org/protobuf v1.27.1
 	gopkg.in/macaroon.v2 v2.1.0
 	gopkg.in/yaml.v2 v2.4.0
 )

go.sum

@@ -752,6 +752,7 @@ golang.org/x/sys v0.0.0-20200803210538-64077c9b5642/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20200923182605-d9f96fdee20d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210403161142-5e06dd20ab57/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -760,6 +761,7 @@ golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40 h1:JWgyZ1qgdTaF3N3oxC+MdTV7qvEEgHo3otj+HB5CM7Q=
 golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -965,6 +967,8 @@ gopkg.in/yaml.v2 v2.2.5/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
+gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
+gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b h1:h8qDotaEPuJATrMmW04NCwg7v22aHH28wwpauUhK9Oo=

hashmail_server.go

@@ -9,8 +9,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/btcsuite/btcd/btcec"
-	"github.com/lightninglabs/lndclient"
 	"github.com/lightninglabs/lightning-node-connect/hashmailrpc"
 	"github.com/lightningnetwork/lnd/tlv"
 	"golang.org/x/time/rate"
@@ -76,6 +74,7 @@ func (r *readStream) ReadNextMsg() ([]byte, error) {
 // ReturnStream gives up the read stream by passing it back up through the
 // payment stream.
 func (r *readStream) ReturnStream() {
+	log.Debugf("Returning read stream %x", r.parentStream.id[:])
 	r.parentStream.ReturnReadStream(r)
 }
 
@@ -158,7 +157,7 @@ type stream struct {
 }
 
 // newStream creates a new stream independent of any given stream ID.
-func newStream(id streamID,
+func newStream(id streamID, limiter *rate.Limiter,
 	equivAuth func(auth *hashmailrpc.CipherBoxAuth) error) *stream {
 
 	// Our stream is actually just a plain io.Pipe. This allows us to avoid
@@ -173,10 +172,7 @@ func newStream(id streamID,
 		writeStreamChan: make(chan *writeStream, 1),
 		id:              id,
 		equivAuth:       equivAuth,
-		limiter: rate.NewLimiter(
+		limiter:         limiter,
-			rate.Every(DefaultMsgRate),
-			DefaultMsgBurstAllowance,
-		),
 	}
 
 	// Our tear down function will close the write side of the pipe, which
@@ -187,7 +183,6 @@ func newStream(id streamID,
 		if err != nil {
 			return err
 		}
-
 		s.wg.Wait()
 		return nil
 	}
@@ -267,13 +262,8 @@ func (s *stream) RequestWriteStream() (*writeStream, error) {
 
 // hashMailServerConfig is the main config of the mail server.
 type hashMailServerConfig struct {
-	// IsAccountActive returns true of the passed public key belongs to an
+	msgRate           time.Duration
-	// active non-expired account) within the system.
+	msgBurstAllowance int
-	IsAccountActive func(context.Context, *btcec.PublicKey) bool
-
-	// Signer is a reference to the current lnd signer client which will be
-	// used to verify ECDSA signatures.
-	Signer lndclient.SignerClient
 }
 
 // hashMailServer is an implementation of the HashMailServer gRPC service that
@@ -294,6 +284,13 @@ type hashMailServer struct {
 
 // newHashMailServer returns a new mail server instance given a valid config.
 func newHashMailServer(cfg hashMailServerConfig) *hashMailServer {
+	if cfg.msgRate == 0 {
+		cfg.msgRate = DefaultMsgRate
+	}
+	if cfg.msgBurstAllowance == 0 {
+		cfg.msgBurstAllowance = DefaultMsgBurstAllowance
+	}
+
 	return &hashMailServer{
 		streams: make(map[streamID]*stream),
 		quit:    make(chan struct{}),
@@ -352,9 +349,14 @@ func (h *hashMailServer) InitStream(
 	// TODO(roasbeef): validate that ticket or node doesn't already have
 	// the same stream going
 
-	freshStream := newStream(streamID, func(auth *hashmailrpc.CipherBoxAuth) error {
+	limiter := rate.NewLimiter(
-		return nil
+		rate.Every(h.cfg.msgRate), h.cfg.msgBurstAllowance,
-	})
+	)
+	freshStream := newStream(
+		streamID, limiter, func(auth *hashmailrpc.CipherBoxAuth) error {
+			return nil
+		},
+	)
 
 	h.streams[streamID] = freshStream
 
@@ -598,6 +600,7 @@ func (h *hashMailServer) RecvStream(desc *hashmailrpc.CipherBoxDesc,
 		// exit before shutting down.
 		select {
 		case <-reader.Context().Done():
+			log.Debugf("Read stream context done.")
 			return nil
 		case <-h.quit:
 			return fmt.Errorf("server shutting down")
@@ -607,6 +610,7 @@ func (h *hashMailServer) RecvStream(desc *hashmailrpc.CipherBoxDesc,
 
 		nextMsg, err := readStream.ReadNextMsg()
 		if err != nil {
+			log.Debugf("Got error an read stream read: %v", err)
 			return err
 		}
 
@@ -618,6 +622,8 @@ func (h *hashMailServer) RecvStream(desc *hashmailrpc.CipherBoxDesc,
 			Msg:  nextMsg,
 		})
 		if err != nil {
+			log.Debugf("Got error when sending on read stream: %v",
+				err)
 			return err
 		}
 	}

proxy/proxy_test.go

@@ -115,7 +115,7 @@ func runHTTPTest(t *testing.T, tc *testCase) {
 	}}
 
 	mockAuth := auth.NewMockAuthenticator()
-	p, err := proxy.New(mockAuth, services, true, "static")
+	p, err := proxy.New(mockAuth, services)
 	require.NoError(t, err)
 
 	// Start server that gives requests to the proxy.
@@ -264,7 +264,7 @@ func runGRPCTest(t *testing.T, tc *testCase) {
 
 	// Create the proxy server and start serving on TLS.
 	mockAuth := auth.NewMockAuthenticator()
-	p, err := proxy.New(mockAuth, services, true, "static")
+	p, err := proxy.New(mockAuth, services)
 	require.NoError(t, err)
 	server := &http.Server{
 		Addr:      testProxyAddr,

sample-conf.yaml

@@ -144,3 +144,10 @@ tor:
 
   # Whether a v3 onion service should be created to handle requests.
   v3: false
+
+# Enable the Lightning Node Connect hashmail server, allowing up to 1k messages
+# per burst and a new message every 20 milliseconds.
+hashmail:
+  enabled: true
+  messagerate: 20ms
+  messageburstallowance: 1000