swap CA root watch from Consul Agent API to gRPC

.changelog/443.txt

@@ -0,0 +1,3 @@
+```release-note:enhancement
+consul: swap CA root watch from Consul Agent API to gRPC
+```

.github/workflows/ci.yml

@@ -36,12 +36,10 @@ jobs:
     strategy:
       matrix:
         consul-version:
-        - 1.11.11
-        - 1.11.11+ent
-        - 1.12.6
-        - 1.12.6+ent
         - 1.13.3
         - 1.13.3+ent
+        - 1.14.0-beta1
+        - 1.14.0-beta1+ent
     runs-on: ubuntu-latest
     env:
       TEST_RESULTS_DIR: /tmp/test-results/consul@${{ matrix.consul-version }}
@@ -91,10 +89,6 @@ jobs:
       fail-fast: false
       matrix:
         consul-image:
-        - 'hashicorp/consul:1.11.11'
-        - 'hashicorp/consul-enterprise:1.11.11-ent'
-        - 'hashicorp/consul:1.12.6'
-        - 'hashicorp/consul-enterprise:1.12.6-ent'
         - 'hashicorp/consul:1.13.3'
         - 'hashicorp/consul-enterprise:1.13.3-ent'
         - 'hashicorppreview/consul:1.14-dev'

go.mod

@@ -150,6 +150,7 @@ require (
 	github.com/gostaticanalysis/comment v1.4.2 // indirect
 	github.com/gostaticanalysis/forcetypeassert v0.1.0 // indirect
 	github.com/gostaticanalysis/nilerr v0.1.1 // indirect
+	github.com/hashicorp/consul/proto-public v0.1.1 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-cleanhttp v0.5.2 // indirect
 	github.com/hashicorp/go-immutable-radix v1.3.1 // indirect

go.sum

@@ -680,6 +680,8 @@ github.com/grpc-ecosystem/grpc-gateway v1.9.5/go.mod h1:vNeuVxBJEsws4ogUvrchl83t
 github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw=
 github.com/hashicorp/consul/api v1.15.3 h1:WYONYL2rxTXtlekAqblR2SCdJsizMDIj/uXb5wNy9zU=
 github.com/hashicorp/consul/api v1.15.3/go.mod h1:/g/qgcoBcEXALCNZgRRisyTW0nY86++L0KbeAMXYCeY=
+github.com/hashicorp/consul/proto-public v0.1.1 h1:C28d6xX+DwoD2dSex/kwD0/nJ4XNgoBtQXtZEm4FGEk=
+github.com/hashicorp/consul/proto-public v0.1.1/go.mod h1:vs2KkuWwtjkIgA5ezp4YKPzQp4GitV+q/+PvksrA92k=
 github.com/hashicorp/consul/sdk v0.11.0 h1:HRzj8YSCln2yGgCumN5CL8lYlD3gBurnervJRJAZyC4=
 github.com/hashicorp/consul/sdk v0.11.0/go.mod h1:yPkX5Q6CsxTFMjQQDJwzeNmUUF5NUGGbrDsv9wTb8cw=
 github.com/hashicorp/errwrap v0.0.0-20141028054710-7554cd9344ce/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=

internal/commands/exec/exec.go

@@ -16,6 +16,8 @@ import (
 
 	"github.com/hashicorp/consul-api-gateway/internal/consul"
 	"github.com/hashicorp/consul-api-gateway/internal/envoy"
+
+	"google.golang.org/grpc/metadata"
 )
 
 type AuthConfig struct {
@@ -78,6 +80,7 @@ func RunExec(config ExecConfig) (ret int) {
 
 	// First do the ACL Login, if necessary.
 	var consulClient *api.Client
+	var consulConfig *api.Config
 	var token string
 	var err error
 	if config.AuthConfig.Method != "" {
@@ -89,7 +92,7 @@ func RunExec(config ExecConfig) (ret int) {
 		}
 		config.Logger.Trace("consul login complete")
 	} else {
-		consulConfig := &config.ConsulConfig
+		consulConfig = &config.ConsulConfig
 		consulConfig.Namespace = config.GatewayConfig.Namespace
 		consulClient, err = api.NewClient(consulConfig)
 		if err != nil {
@@ -149,13 +152,28 @@ func RunExec(config ExecConfig) (ret int) {
 		},
 	)
 	options := consul.DefaultCertManagerOptions()
+	options.Addresses = []string{config.EnvoyConfig.XDSAddress}
+	options.GRPCPort = config.EnvoyConfig.XDSPort
 	options.PrimaryDatacenter = config.PrimaryDatacenter
 	options.SDSAddress = config.EnvoyConfig.SDSAddress
 	options.SDSPort = config.EnvoyConfig.SDSPort
 	options.Directory = "/certs"
 	if config.EnvoyConfig.CertificateDirectory != "" {
 		options.Directory = config.EnvoyConfig.CertificateDirectory
 	}
+
+	// This is the same check used in exec/command.go for HTTPS API configuration
+	if config.ConsulConfig.TLSConfig.CAFile != "" {
+		config.Logger.Info("configuring gateway TLS")
+		tlsConfig, err := api.SetupTLSConfig(&config.ConsulConfig.TLSConfig)
+		if err != nil {
+			return 1
+		}
+
+		options.GRPCUseTLS = true
+		options.GRPCTLS = tlsConfig
+	}
+
 	certManager := consul.NewCertManager(
 		config.Logger.Named("cert-manager"),
 		client,
@@ -174,6 +192,11 @@ func RunExec(config ExecConfig) (ret int) {
 	}
 
 	group, groupCtx := errgroup.WithContext(ctx)
+
+	// Append token with service:write permissions to allow certmanager to
+	// watch roots and sign CSRs
+	groupCtx = metadata.AppendToOutgoingContext(groupCtx, "x-consul-token", token)
+
 	group.Go(func() error {
 		return certManager.Manage(groupCtx)
 	})

internal/commands/exec/exec_test.go

@@ -3,22 +3,31 @@ package exec
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
+	"net"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
 	"os"
 	"path"
+	"strconv"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/require"
 
 	"github.com/hashicorp/consul/api"
+	"github.com/hashicorp/consul/proto-public/pbconnectca"
 	"github.com/hashicorp/go-hclog"
 
 	gwTesting "github.com/hashicorp/consul-api-gateway/internal/testing"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 func TestRunExecLoginError(t *testing.T) {
@@ -274,6 +283,8 @@ func TestRunExecEnvoyExecError(t *testing.T) {
 			CertificateDirectory: directory,
 			BootstrapFile:        path.Join(directory, "boostrap.json"),
 			Binary:               "thisisnotabinary",
+			XDSAddress:           consul.address,
+			XDSPort:              consul.grpcPort,
 		},
 		isTest: true,
 	}))
@@ -315,6 +326,8 @@ func TestRunExecShutdown(t *testing.T) {
 			Binary:               "echo",
 			ExtraArgs:            []string{output},
 			Output:               &buffer,
+			XDSAddress:           consul.address,
+			XDSPort:              consul.grpcPort,
 		},
 		isTest: true,
 	}))
@@ -370,6 +383,8 @@ func TestRunExecShutdownACLs(t *testing.T) {
 			Binary:               "echo",
 			ExtraArgs:            []string{output},
 			Output:               &buffer,
+			XDSAddress:           consul.address,
+			XDSPort:              consul.grpcPort,
 		},
 		isTest: true,
 	}))
@@ -391,10 +406,63 @@ type mockConsulServer struct {
 	client *api.Client
 	config *api.Config
 
+	address  string
+	httpPort int
+	grpcPort int
+
 	token            string
 	rootCertPEM      string
 	clientCert       string
 	clientPrivateKey string
+
+	mutex sync.RWMutex
+
+	opts mockConsulOptions
+}
+
+func (c *mockConsulServer) WatchRoots(request *pbconnectca.WatchRootsRequest, stream pbconnectca.ConnectCAService_WatchRootsServer) error {
+	writeCertificate := func() error {
+		fmt.Printf("writing certificate to channel")
+
+		if c.opts.rootCertFail {
+			return status.Error(codes.Internal, "root watch failure")
+		}
+
+		c.mutex.RLock()
+		ca := c.rootCertPEM
+		c.mutex.RUnlock()
+
+		if err := stream.Send(&pbconnectca.WatchRootsResponse{
+			ActiveRootId: "test",
+			Roots: []*pbconnectca.CARoot{{
+				Id:       "test",
+				RootCert: ca,
+			}},
+		}); err != nil {
+			return err
+		}
+		return nil
+	}
+
+	// do initial write
+	if err := writeCertificate(); err != nil {
+		return err
+	}
+
+	for {
+		select {
+		case <-stream.Context().Done():
+			return nil
+			// case <-c.rotate:
+			// 	if err := writeCertificate(); err != nil {
+			// 		return err
+			// 	}
+		}
+	}
+}
+
+func (c *mockConsulServer) Sign(ctx context.Context, request *pbconnectca.SignRequest) (*pbconnectca.SignResponse, error) {
+	return nil, fmt.Errorf("not yet implemented")
 }
 
 func runMockConsulServer(t *testing.T, opts mockConsulOptions) *mockConsulServer {
@@ -406,14 +474,14 @@ func runMockConsulServer(t *testing.T, opts mockConsulOptions) *mockConsulServer
 		rootCertPEM:      string(ca.CertBytes),
 		clientCert:       string(clientCert.CertBytes),
 		clientPrivateKey: string(clientCert.PrivateKeyBytes),
+		opts:             opts,
 	}
 
 	loginPath := "/v1/acl/login"
 	logoutPath := "/v1/acl/logout"
 	registerPath := "/v1/agent/service/register"
 	deregisterPath := "/v1/agent/service/deregister"
 	leafPath := "/v1/agent/connect/ca/leaf"
-	rootPath := "/v1/agent/connect/ca/roots"
 
 	// Start the fake Consul server.
 	consulServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -464,33 +532,50 @@ func runMockConsulServer(t *testing.T, opts mockConsulOptions) *mockConsulServer
 			require.NoError(t, err)
 			return
 		}
-		if r != nil && r.URL.Path == rootPath && r.Method == "GET" {
-			if opts.rootCertFail {
-				w.WriteHeader(http.StatusInternalServerError)
-				return
-			}
-			rootCert, err := json.Marshal(map[string]interface{}{
-				"Roots": []map[string]interface{}{{
-					"RootCert": server.rootCertPEM,
-					"Active":   true,
-				}},
-			})
-			require.NoError(t, err)
-			_, err = w.Write(rootCert)
-			require.NoError(t, err)
-			return
-		}
 		w.WriteHeader(http.StatusInternalServerError)
 	}))
 	t.Cleanup(consulServer.Close)
 
 	serverURL, err := url.Parse(consulServer.URL)
 	require.NoError(t, err)
+	consulHTTPAddress := serverURL.Host
+
 	clientConfig := &api.Config{Address: serverURL.String()}
 	client, err := api.NewClient(clientConfig)
 	require.NoError(t, err)
-
 	server.client = client
 	server.config = clientConfig
+
+	// httptest.NewServer hardcodes 127.0.0.1, so this will be the same as for
+	// the gRPC server, just on a different port
+	consulHTTPAddressParts := strings.Split(consulHTTPAddress, ":")
+	server.address = consulHTTPAddressParts[0]
+	server.httpPort, err = strconv.Atoi(consulHTTPAddressParts[1])
+	require.NoError(t, err)
+	fmt.Printf("running Consul HTTP mock server at %s\n", serverURL.String())
+
+	// Start the fake Consul gRPC server
+	grpcServer := grpc.NewServer()
+	pbconnectca.RegisterConnectCAServiceServer(grpcServer, server)
+	grpcListener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+
+	errEarlyTestTermination := errors.New("early termination")
+	done := make(chan error, 1)
+	go func() {
+		defer func() {
+			// Write an error to the channel, if the server canceled
+			// successfully the err will be nil and the read will get that
+			// first, this will only be read if we have some early exception
+			// that calls runtime.Goexit prior to the server stopping
+			done <- errEarlyTestTermination
+		}()
+		// Start gRPC mock server, send nil error if clean exit
+		done <- grpcServer.Serve(grpcListener)
+	}()
+	server.grpcPort, err = strconv.Atoi(strings.Split(grpcListener.Addr().String(), ":")[1])
+	require.NoError(t, err)
+	fmt.Printf("running Consul gRPC mock server at %s\n", grpcListener.Addr().String())
+
 	return server
 }

internal/commands/server/command.go

@@ -36,7 +36,8 @@ type Command struct {
 	flagCASecret          string // CA Secret for Consul server
 	flagCASecretNamespace string // CA Secret namespace for Consul server
 
-	flagConsulAddress string // Consul server address
+	flagConsulAddress  string // Consul server address
+	flagConsulGRPCPort int    // Consul gRPC port
 
 	flagPrimaryDatacenter string // Primary datacenter, may or may not be the datacenter this controller is running in
 
@@ -71,6 +72,7 @@ func (c *Command) init() {
 	c.flagSet.StringVar(&c.flagCASecret, "ca-secret", "", "CA Secret for Consul server.")
 	c.flagSet.StringVar(&c.flagCASecretNamespace, "ca-secret-namespace", "default", "CA Secret namespace for Consul server.")
 	c.flagSet.StringVar(&c.flagConsulAddress, "consul-address", "", "Consul Address.")
+	c.flagSet.IntVar(&c.flagConsulGRPCPort, "consul-grpc-port", 8502, "Consul gRPC Port.")
 	c.flagSet.StringVar(&c.flagPrimaryDatacenter, "primary-datacenter", "", "Name of the primary Consul datacenter")
 	c.flagSet.StringVar(&c.flagSDSServerHost, "sds-server-host", defaultSDSServerHost, "SDS Server Host.")
 	c.flagSet.StringVar(&c.flagK8sContext, "k8s-context", "", "Kubernetes context to use.")
@@ -170,6 +172,7 @@ func (c *Command) Run(args []string) int {
 		Context:           context.Background(),
 		Logger:            logger,
 		ConsulConfig:      consulCfg,
+		ConsulGRPCPort:    c.flagConsulGRPCPort,
 		K8sConfig:         cfg,
 		ProfilingPort:     c.flagPprofPort,
 		MetricsPort:       c.flagMetricsPort,

internal/commands/server/server.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"os"
 	"os/signal"
+	"strings"
 	"syscall"
 
 	"golang.org/x/sync/errgroup"
@@ -27,6 +28,7 @@ type ServerConfig struct {
 	Context           context.Context
 	Logger            hclog.Logger
 	ConsulConfig      *api.Config
+	ConsulGRPCPort    int
 	K8sConfig         *k8s.Config
 	ProfilingPort     int
 	MetricsPort       int
@@ -73,8 +75,24 @@ func RunServer(config ServerConfig) int {
 	controller.SetStore(store)
 
 	options := consul.DefaultCertManagerOptions()
+	options.Addresses = []string{strings.Split(config.ConsulConfig.Address, ":")[0]}
+	options.GRPCPort = config.ConsulGRPCPort
 	options.PrimaryDatacenter = config.PrimaryDatacenter
 
+	// This is the same check used in server/command.go for HTTPS API configuration
+	if config.ConsulConfig.TLSConfig.CAFile != "" {
+		config.Logger.Info("configuring controller TLS")
+		tlsConfig, err := api.SetupTLSConfig(&config.ConsulConfig.TLSConfig)
+		if err != nil {
+			return 1
+		}
+
+		options.GRPCUseTLS = true
+		options.GRPCTLS = tlsConfig
+	}
+
+	config.Logger.Info(fmt.Sprintf("%+v", options))
+
 	certManager := consul.NewCertManager(
 		config.Logger.Named("cert-manager"),
 		client,