make it support various type of ssl certificate header

cmd/main.go

@@ -72,7 +72,7 @@ func startServer(config *config.Config, airbrakeNotifier *gobrake.Notifier, logg
 		return err
 	}
 
-	if config.DisableTLS {
+	if config.TLSPassThrough != nil {
 		err = server.ListenAndServe()
 	} else {
 		if server.TLSConfig, err = config.ExtractServiceTLSConfig(logger); err != nil {

config/config.go

@@ -5,6 +5,7 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	_ "embed" //Used for default CAs
+	"encoding/json"
 	"errors"
 	"fmt"
 	"net/http"
@@ -47,12 +48,10 @@ type Config struct {
 	// TLS contains certificates & CA info for the webserver
 	TLS *TLS `json:"tls,omitempty"`
 
-	// DisableTLS indicates whether to disable mutual TLS (mTLS) for incoming connections.
-	// This should only be set to true if there is a reverse proxy that is already handling
-	// mTLS on behalf of this service and can convey the client certificate following
-	// the guidelines specified in RFC 9440 (https://datatracker.ietf.org/doc/rfc9440/).
-	// TLS configuration will be ignored when this option is set to true.
-	DisableTLS bool `json:"disable_tls,omitempty"`
+	// TLSPassThrough when set will disable mutual TLS (mTLS) for incoming connections
+	// and ignore TLS config. This should only be set to true if there is a reverse proxy that
+	// is already handling mTLS on behalf of this service.
+	TLSPassThrough *TLSPassThrough `json:"tls_pass_through,omitempty"`
 
 	// UseDefaultEngCA overrides default CA to eng
 	UseDefaultEngCA bool `json:"use_default_eng_ca"`
@@ -161,6 +160,34 @@ type TLS struct {
 	ServerKey  string `json:"server_key"`
 }
 
+type TLSPassThrough string
+
+const (
+	RFC9440                    TLSPassThrough = "rfc9440"
+	AWSApplicationLoadBalancer TLSPassThrough = "aws_alb"
+)
+
+func (t *TLSPassThrough) IsValid() bool {
+	switch *t {
+	case RFC9440, AWSApplicationLoadBalancer:
+		return true
+	default:
+		return false
+	}
+}
+
+func (t *TLSPassThrough) UnmarshalJSON(data []byte) error {
+	var temp string
+	if err := json.Unmarshal(data, &temp); err != nil {
+		return err
+	}
+	*t = TLSPassThrough(temp)
+	if !t.IsValid() {
+		return errors.New("invalid value for TLSPassThrough")
+	}
+	return nil
+}
+
 // AirbrakeTLSConfig return the TLS config needed for connecting with airbrake server
 func (c *Config) AirbrakeTLSConfig() (*tls.Config, error) {
 	if c.Airbrake.TLS == nil {

config/config_test.go

@@ -27,13 +27,13 @@ var _ = Describe("Test full application config", func() {
 	BeforeEach(func() {
 		log, _ = logrus.NoOpLogger()
 		config = &Config{
-			Host:       "127.0.0.1",
-			Port:       443,
-			StatusPort: 8080,
-			Namespace:  "tesla_telemetry",
-			TLS:        &TLS{CAFile: "tesla.ca", ServerCert: "your_own_cert.crt", ServerKey: "your_own_key.key"},
-			DisableTLS: false,
-			RateLimit:  &RateLimit{Enabled: true, MessageLimit: 1000, MessageInterval: 30},
+			Host:           "127.0.0.1",
+			Port:           443,
+			StatusPort:     8080,
+			Namespace:      "tesla_telemetry",
+			TLS:            &TLS{CAFile: "tesla.ca", ServerCert: "your_own_cert.crt", ServerKey: "your_own_key.key"},
+			TLSPassThrough: ptr(RFC9440),
+			RateLimit:      &RateLimit{Enabled: true, MessageLimit: 1000, MessageInterval: 30},
 			Kafka: &confluent.ConfigMap{
 				"bootstrap.servers":        "some.broker:9093",
 				"ssl.ca.location":          "kafka.ca",
@@ -122,10 +122,10 @@ var _ = Describe("Test full application config", func() {
 			Expect(config.TransmitDecodedRecords).To(BeFalse())
 		})
 
-		It("disable_tls false by default", func() {
+		It("tls_pass_through is nil by default", func() {
 			config, err := loadTestApplicationConfig(TestSmallConfig)
 			Expect(err).NotTo(HaveOccurred())
-			Expect(config.DisableTLS).To(BeFalse())
+			Expect(config.TLSPassThrough).To(BeNil())
 		})
 
 		It("transmitrecords enabled", func() {
@@ -136,10 +136,15 @@ var _ = Describe("Test full application config", func() {
 	})
 
 	Context("configure tls", func() {
-		It("read disable tls config", func() {
-			config, err := loadTestApplicationConfig(TestDisableTLSConfig)
+		It("read tls_pass_through config", func() {
+			config, err := loadTestApplicationConfig(TestRFC9440TLSConfig)
 			Expect(err).NotTo(HaveOccurred())
-			Expect(config.DisableTLS).To(BeEquivalentTo(true))
+			Expect(*config.TLSPassThrough).To(BeEquivalentTo(RFC9440))
+		})
+
+		It("error on invalid tls_pass_through config", func() {
+			_, err := loadTestApplicationConfig(TestInvalidTLSPassThroughConfig)
+			Expect(err).To(HaveOccurred())
 		})
 	})
 
@@ -311,3 +316,7 @@ var _ = Describe("Test full application config", func() {
 		})
 	})
 })
+
+func ptr[T any](x T) *T {
+	return &x
+}

config/test_configs_test.go

@@ -62,7 +62,7 @@ const TestSmallConfig = `
 }
 `
 
-const TestDisableTLSConfig = `
+const TestRFC9440TLSConfig = `
 {
 	"host": "127.0.0.1",
 	"port": 443,
@@ -78,12 +78,27 @@ const TestDisableTLSConfig = `
 	"records": {
 		"V": ["kafka"]
 	},
-	"disable_tls": true,
-	"tls": {
-		"ca_file": "tesla.ca",
-		"server_cert": "your_own_cert.crt",
-		"server_key": "your_own_key.key"
-	}
+	"tls_pass_through": "rfc9440"
+}
+`
+
+const TestInvalidTLSPassThroughConfig = `
+{
+	"host": "127.0.0.1",
+	"port": 443,
+	"status_port": 8080,
+	"namespace": "tesla_telemetry",
+	"kafka": {
+		"bootstrap.servers": "some.broker1:9093,some.broker1:9093",
+		"ssl.ca.location": "kafka.ca",
+		"ssl.certificate.location": "kafka.crt",
+		"ssl.key.location": "kafka.key",
+		"queue.buffering.max.messages": 1000000
+	},
+	"records": {
+		"V": ["kafka"]
+	},
+	"tls_pass_through": "abcde"
 }
 `
 

server/streaming/server.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"github.com/pkg/errors"
 	"net/http"
+	"net/url"
 	"sync"
 	"time"
 
@@ -221,11 +222,18 @@ func (s *Server) promoteToWebsocket(w http.ResponseWriter, r *http.Request) *web
 	return ws
 }
 
+type extractCertFunc func(r *http.Request) (*x509.Certificate, error)
+
+var headerExtractConfigMap = map[config.TLSPassThrough]extractCertFunc{
+	config.RFC9440:                    extractCertRFC2440,
+	config.AWSApplicationLoadBalancer: extractCertAWSALB,
+}
+
 func extractIdentity(r *http.Request, config *config.Config) (*telemetry.RequestIdentity, error) {
 	var cert *x509.Certificate
 	var err error
-	if config.DisableTLS {
-		cert, err = extractCertFromHeaders(r)
+	if config.TLSPassThrough != nil {
+		cert, err = headerExtractConfigMap[*config.TLSPassThrough](r)
 	} else {
 		cert, err = extractCertFromTLS(r)
 	}
@@ -243,33 +251,46 @@ func extractIdentity(r *http.Request, config *config.Config) (*telemetry.Request
 	}, nil
 }
 
-func extractCertFromHeaders(r *http.Request) (*x509.Certificate, error) {
-	raw := r.Header.Get("Client-Cert-Chain") // Client-Cert and Client-Cert-Chain HTTP Header Field  https://datatracker.ietf.org/doc/rfc9440/
+// extractCertRFC2440 implements https://datatracker.ietf.org/doc/rfc9440/
+func extractCertRFC2440(r *http.Request) (*x509.Certificate, error) {
+	raw := r.Header.Get("Client-Cert-Chain")
 	if raw == "" {
 		return nil, errors.New("missing_certificate_error")
 	}
 	rest, err := base64.StdEncoding.DecodeString(raw)
 	if err != nil {
 		return nil, fmt.Errorf("failed to parse certificates: %w", err)
 	}
-	var allCerts []*x509.Certificate
-	for {
-		block, remaining := pem.Decode(rest)
-		if block == nil {
-			break
-		}
-		certs, err1 := x509.ParseCertificates(block.Bytes)
-		if err1 != nil {
-			fmt.Printf(err1.Error())
-			return nil, fmt.Errorf("failed to parse certificates: %w", err)
-		}
-		allCerts = append(allCerts, certs...)
-		rest = remaining
+	block, _ := pem.Decode(rest)
+	if block == nil {
+		return nil, fmt.Errorf("failed to parse certificates: %w", err)
 	}
-	if len(allCerts) == 0 {
+	certs, err := x509.ParseCertificates(block.Bytes)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse certificates: %w", err)
+	}
+	return certs[0], nil
+}
+
+// extractCertAWSALB implements https://docs.aws.amazon.com/elasticloadbalancing/latest/application/mutual-authentication.html#mtls-http-headers
+func extractCertAWSALB(r *http.Request) (*x509.Certificate, error) {
+	raw := r.Header.Get("X-Amzn-Mtls-Clientcert")
+	if raw == "" {
 		return nil, errors.New("missing_certificate_error")
 	}
-	return allCerts[0], nil
+	rest, err := url.QueryUnescape(raw)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse certificates: %w", err)
+	}
+	block, _ := pem.Decode([]byte(rest))
+	if block == nil {
+		return nil, fmt.Errorf("failed to parse certificates: %w", err)
+	}
+	certs, err := x509.ParseCertificates(block.Bytes)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse certificates: %w", err)
+	}
+	return certs[0], nil
 }
 
 func extractCertFromTLS(r *http.Request) (*x509.Certificate, error) {

server/streaming/server_test.go

@@ -33,7 +33,7 @@ var _ = Describe("Extract certificate from header test", func() {
 		producerRules := make(map[string][]telemetry.Producer)
 		registry := streaming.NewSocketRegistry()
 		conf := &config.Config{
-			DisableTLS:      true,
+			TLSPassThrough:  ptr(config.RFC9440),
 			TLS:             nil,
 			MetricCollector: noop.NewCollector(),
 		}
@@ -168,3 +168,7 @@ var _ = Describe("Socket handler test", func() {
 		Expect(hook.AllEntries()).To(BeEmpty())
 	})
 })
+
+func ptr[T any](x T) *T {
+	return &x
+}