Add in --duplex flag to run HTTP and GRPC servers on the same port

cmd/app/serve.go

@@ -24,13 +24,22 @@ import (
 	"net/http"
 	"os"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"time"
 
+	"chainguard.dev/go-grpc-kit/pkg/duplex"
+	"github.com/goadesign/goa/grpc/middleware"
 	ctclient "github.com/google/certificate-transparency-go/client"
 	"github.com/google/certificate-transparency-go/jsonclient"
+	grpcmw "github.com/grpc-ecosystem/go-grpc-middleware"
+	grpc_zap "github.com/grpc-ecosystem/go-grpc-middleware/logging/zap"
+	grpc_recovery "github.com/grpc-ecosystem/go-grpc-middleware/recovery"
+	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
+	"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"github.com/sigstore/fulcio/pkg/ca"
 	certauth "github.com/sigstore/fulcio/pkg/ca"
 	"github.com/sigstore/fulcio/pkg/ca/ephemeralca"
 	"github.com/sigstore/fulcio/pkg/ca/fileca"
@@ -39,12 +48,18 @@ import (
 	"github.com/sigstore/fulcio/pkg/ca/pkcs11ca"
 	"github.com/sigstore/fulcio/pkg/ca/tinkca"
 	"github.com/sigstore/fulcio/pkg/config"
+	"github.com/sigstore/fulcio/pkg/generated/protobuf"
+	"github.com/sigstore/fulcio/pkg/generated/protobuf/legacy"
 	"github.com/sigstore/fulcio/pkg/log"
+	"github.com/sigstore/fulcio/pkg/server"
 	"github.com/sigstore/sigstore/pkg/cryptoutils"
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
 	"github.com/spf13/viper"
 	"go.uber.org/zap"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/protobuf/proto"
 )
 
 const serveCmdEnvPrefix = "FULCIO_SERVE"
@@ -116,6 +131,7 @@ func (la logAdaptor) Printf(s string, args ...interface{}) {
 }
 
 func runServeCmd(cmd *cobra.Command, args []string) {
+	ctx := cmd.Context()
 	// If a config file is provided, modify the viper config to locate and read it
 	if err := checkServeCmdConfigFile(); err != nil {
 		log.Logger.Fatal(err)
@@ -255,6 +271,22 @@ func runServeCmd(cmd *cobra.Command, args []string) {
 		}
 	}
 
+	if viper.GetString("port") == viper.GetString("grpc-port") {
+		p := viper.GetString("port")
+		port, err := strconv.Atoi(p)
+		if err != nil {
+			log.Logger.Fatal("%s is not a valid port", p)
+		}
+		mp := viper.GetString("metrics-port")
+		metricsPort, err := strconv.Atoi(mp)
+		if err != nil {
+			log.Logger.Fatal("%s is not a valid port", mp)
+		}
+		if err := StartDuplexServer(ctx, cfg, ctClient, baseca, port, metricsPort); err != nil {
+			log.Logger.Fatal(err)
+		}
+	}
+
 	httpServerEndpoint := fmt.Sprintf("%v:%v", viper.GetString("http-host"), viper.GetString("http-port"))
 
 	reg := prometheus.NewRegistry()
@@ -272,7 +304,7 @@ func runServeCmd(cmd *cobra.Command, args []string) {
 	}
 	legacyGRPCServer.startUnixListener()
 
-	httpServer := createHTTPServer(context.Background(), httpServerEndpoint, grpcServer, legacyGRPCServer)
+	httpServer := createHTTPServer(ctx, httpServerEndpoint, grpcServer, legacyGRPCServer)
 	httpServer.startListener()
 
 	readHeaderTimeout := viper.GetDuration("read-header-timeout")
@@ -314,3 +346,79 @@ func checkServeCmdConfigFile() error {
 	}
 	return nil
 }
+
+func StartDuplexServer(ctx context.Context, cfg *config.FulcioConfig, ctClient *ctclient.LogClient, baseca ca.CertificateAuthority, port, metricsPort int) error {
+	logger, opts := log.SetupGRPCLogging()
+
+	d := duplex.New(
+		port,
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		runtime.WithMetadata(extractOIDCTokenFromAuthHeader),
+		grpc.UnaryInterceptor(grpcmw.ChainUnaryServer(
+			grpc_recovery.UnaryServerInterceptor(grpc_recovery.WithRecoveryHandlerContext(panicRecoveryHandler)), // recovers from per-transaction panics elegantly, so put it first
+			middleware.UnaryRequestID(middleware.UseXRequestIDMetadataOption(true), middleware.XRequestMetadataLimitOption(128)),
+			grpc_zap.UnaryServerInterceptor(logger, opts...),
+			PassFulcioConfigThruContext(cfg),
+			grpc_prometheus.UnaryServerInterceptor,
+		)),
+		grpc.MaxRecvMsgSize(int(maxMsgSize)),
+		runtime.WithForwardResponseOption(HTTPResponseModifier),
+	)
+
+	// GRPC server
+	grpcCAServer := server.NewGRPCCAServer(ctClient, baseca)
+	protobuf.RegisterCAServer(d.Server, grpcCAServer)
+	if err := d.RegisterHandler(ctx, protobuf.RegisterCAHandlerFromEndpoint); err != nil {
+		return fmt.Errorf("registering grpc ca handler: %w", err)
+	}
+
+	// Legacy server
+	legacyGRPCCAServer := server.NewLegacyGRPCCAServer(grpcCAServer)
+	legacy.RegisterCAServer(d.Server, legacyGRPCCAServer)
+	if err := d.RegisterHandler(ctx, legacy.RegisterCAHandlerFromEndpoint); err != nil {
+		return fmt.Errorf("registering legacy grpc ca handler: %w", err)
+	}
+
+	// Prometheus
+	reg := prometheus.NewRegistry()
+	grpcMetrics := grpc_prometheus.DefaultServerMetrics
+	grpcMetrics.EnableHandlingTimeHistogram()
+	reg.MustRegister(grpcMetrics, server.MetricLatency, server.RequestsCount)
+	grpc_prometheus.Register(d.Server)
+
+	// Register prometheus handle.
+	d.RegisterListenAndServeMetrics(metricsPort, false)
+
+	logger.Info("Starting duplex server...")
+	if err := d.ListenAndServe(ctx); err != nil {
+		return fmt.Errorf("duplex server: %w", err)
+	}
+	return nil
+}
+
+// The fulcio HTTP legacy client requires a 201 response status code to pass
+// However, even though the legacy client sets the 201 code, GRPC will automatically
+// change it to 200, causing the cert request to fail
+// GRPC recommends controlling HTTP status codes with this response modifier
+// https://grpc-ecosystem.github.io/grpc-gateway/docs/mapping/customizing_your_gateway/#controlling-http-response-status-codes
+// This is required to use fulcio with duplex.
+func HTTPResponseModifier(ctx context.Context, w http.ResponseWriter, p proto.Message) error {
+	md, ok := runtime.ServerMetadataFromContext(ctx)
+	if !ok {
+		return nil
+	}
+
+	// set http status code
+	if vals := md.HeaderMD.Get(server.HTTPResponseCodeMetadataKey); len(vals) > 0 {
+		code, err := strconv.Atoi(vals[0])
+		if err != nil {
+			return err
+		}
+		// delete the headers to not expose any grpc-metadata in http response
+		delete(md.HeaderMD, server.HTTPResponseCodeMetadataKey)
+		delete(w.Header(), "Grpc-Metadata-X-Http-Code")
+		w.WriteHeader(code)
+	}
+
+	return nil
+}

cmd/app/serve_test.go

@@ -0,0 +1,112 @@
+// Copyright 2021 The Sigstore Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+package app
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"net/url"
+	"strings"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/sigstore/fulcio/pkg/api"
+	"github.com/sigstore/fulcio/pkg/ca/ephemeralca"
+	"github.com/sigstore/fulcio/pkg/config"
+	"github.com/sigstore/fulcio/pkg/generated/protobuf"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+func TestDuplex(t *testing.T) {
+	// Start a server with duplex on port 8089
+	ctx := context.Background()
+	ca, err := ephemeralca.NewEphemeralCA()
+	if err != nil {
+		t.Fatal(err)
+	}
+	port := 8089
+	serverURL, err := url.Parse(fmt.Sprintf("http://localhost:%d", port))
+	if err != nil {
+		t.Fatal(err)
+	}
+	metricsPort := 2114
+
+	go func() {
+		if err := StartDuplexServer(ctx, config.DefaultConfig, nil, ca, port, metricsPort); err != nil {
+			log.Fatalf("error starting duplex server: %v", err)
+		}
+	}()
+
+	var rootCert string
+	t.Run("http", func(t *testing.T) {
+		// Make sure we can grab the rootcert with the v1 endpoint
+		legacyClient := api.NewClient(serverURL)
+		resp, err := legacyClient.RootCert()
+		if err != nil {
+			t.Fatal(err)
+		}
+		rootCert = string(resp.ChainPEM)
+	})
+
+	var grpcRootCert string
+	t.Run("grpc", func(t *testing.T) {
+		// Grab the rootcert with the v2 endpoint
+		conn, err := grpc.Dial(fmt.Sprintf("127.0.0.1:%d", port), grpc.WithTransportCredentials(insecure.NewCredentials()))
+		if err != nil {
+			t.Fatal(err)
+		}
+		grpcClient := protobuf.NewCAClient(conn)
+		tb, err := grpcClient.GetTrustBundle(ctx, &protobuf.GetTrustBundleRequest{})
+		if err != nil {
+			t.Fatalf("error getting trust bundle: %v", err)
+		}
+		if len(tb.Chains) != 1 {
+			t.Fatalf("didn't get expected length certificate chain: %v", tb.Chains)
+		}
+		if len(tb.Chains[0].Certificates) != 1 {
+			t.Fatalf("didn't get expected length certs: %v", tb.Chains)
+		}
+		grpcRootCert = strings.TrimSuffix(tb.Chains[0].Certificates[0], "\n")
+	})
+
+	t.Run("compare root certs", func(t *testing.T) {
+		if d := cmp.Diff(rootCert, grpcRootCert); d != "" {
+			t.Fatal(d)
+		}
+	})
+
+	t.Run("prometheus", func(t *testing.T) {
+		// make sure there are metrics on the metrics port
+		url := fmt.Sprintf("http://localhost:%d/metrics", metricsPort)
+		resp, err := http.Get(url)
+		if err != nil {
+			t.Fatal(err)
+		}
+		contents, err := io.ReadAll(resp.Body)
+		if err != nil {
+			t.Fatal(err)
+		}
+		// make sure there's something about hitting the GetTrustBundle in there
+		// this just confirms some metrics are being printed
+		if !strings.Contains(string(contents), "GetTrustBundle") {
+			t.Fatalf("didn't get expected metrics output: %s", string(contents))
+		}
+	})
+}

go.mod

@@ -3,6 +3,7 @@ module github.com/sigstore/fulcio
 go 1.18
 
 require (
+	chainguard.dev/go-grpc-kit v0.11.0
 	cloud.google.com/go/security v1.11.0
 	github.com/PaesslerAG/jsonpath v0.1.1
 	github.com/ThalesIgnite/crypto11 v1.2.5
@@ -15,7 +16,7 @@ require (
 	github.com/google/go-cmp v0.5.9
 	github.com/google/tink/go v1.7.0
 	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0
-	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
+	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.1-0.20210315223345-82c243799c99
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.15.0
 	github.com/hashicorp/golang-lru v0.5.4
 	github.com/magiconair/properties v1.8.7
@@ -76,17 +77,22 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/sts v1.17.5 // indirect
 	github.com/aws/smithy-go v1.13.4 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/blendle/zapdriver v1.3.1 // indirect
 	github.com/cenkalti/backoff/v3 v3.2.2 // indirect
+	github.com/cenkalti/backoff/v4 v4.1.3 // indirect
 	github.com/cespare/xxhash/v2 v2.1.2 // indirect
 	github.com/common-nighthawk/go-figure v0.0.0-20210622060536-734e95fb86be // indirect
 	github.com/dimchansky/utfbom v1.1.1 // indirect
 	github.com/fatih/color v1.13.0 // indirect
 	github.com/go-jose/go-jose/v3 v3.0.0 // indirect
-	github.com/go-logr/logr v1.2.0 // indirect
+	github.com/go-logr/logr v1.2.3 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang-jwt/jwt/v4 v4.4.2 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/google/go-containerregistry v0.12.1 // indirect
+	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/google/uuid v1.3.0 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.2.1 // indirect
 	github.com/googleapis/gax-go/v2 v2.7.0 // indirect
@@ -111,6 +117,7 @@ require (
 	github.com/inconshreveable/mousetrap v1.0.1 // indirect
 	github.com/jellydator/ttlcache/v2 v2.11.1 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/letsencrypt/boulder v0.0.0-20221109233200-85aa52084eaf // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.16 // indirect
@@ -121,6 +128,8 @@ require (
 	github.com/mitchellh/go-testing-interface v1.14.1 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/oklog/run v1.1.0 // indirect
 	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/pelletier/go-toml v1.9.5 // indirect
@@ -139,6 +148,13 @@ require (
 	github.com/theupdateframework/go-tuf v0.5.2-0.20220930112810-3890c1e7ace4 // indirect
 	github.com/titanous/rocacheck v0.0.0-20171023193734-afe73141d399 // indirect
 	go.opencensus.io v0.24.0 // indirect
+	go.opentelemetry.io/otel v1.11.1 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/internal/retry v1.11.1 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.11.1 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.11.1 // indirect
+	go.opentelemetry.io/otel/sdk v1.11.1 // indirect
+	go.opentelemetry.io/otel/trace v1.11.1 // indirect
+	go.opentelemetry.io/proto/otlp v0.19.0 // indirect
 	go.uber.org/atomic v1.10.0 // indirect
 	go.uber.org/multierr v1.8.0 // indirect
 	goa.design/goa v2.2.5+incompatible // indirect
@@ -151,7 +167,14 @@ require (
 	golang.org/x/text v0.6.0 // indirect
 	golang.org/x/time v0.2.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
+	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
+	k8s.io/api v0.25.3 // indirect
+	k8s.io/apimachinery v0.25.3 // indirect
 	k8s.io/klog/v2 v2.80.1 // indirect
+	k8s.io/utils v0.0.0-20220728103510-ee6ede2d64ed // indirect
+	knative.dev/pkg v0.0.0-20221010143036-21d3b47e2efe // indirect
+	sigs.k8s.io/json v0.0.0-20220713155537-f223a00ba0e2 // indirect
+	sigs.k8s.io/structured-merge-diff/v4 v4.2.3 // indirect
 )