Add ability to disable mTLS

README.md

@@ -284,3 +284,4 @@ Moreover, the following application-specific considerations apply:
   the frequency they need.
 * Providers agree to take full responsibility for privacy risks, as soon as data
   leave the devices (for more info read our privacy policies).
+* In some cases, fleet telemetry is deployed behind a trusted proxy that handles mTLS and terminates the connection. To set this up, set disable_tls to true in the configuration and ensure that the proxy implements RFC 9440 (https://datatracker.ietf.org/doc/rfc9440/).

cmd/main.go

@@ -72,11 +72,15 @@ func startServer(config *config.Config, airbrakeNotifier *gobrake.Notifier, logg
 		return err
 	}
 
-	if server.TLSConfig, err = config.ExtractServiceTLSConfig(logger); err != nil {
-		return err
+	if config.TLSPassThrough != nil {
+		err = server.ListenAndServe()
+	} else {
+		if server.TLSConfig, err = config.ExtractServiceTLSConfig(logger); err != nil {
+			return err
+		}
+		err = server.ListenAndServeTLS(config.TLS.ServerCert, config.TLS.ServerKey)
 	}
 
-	err = server.ListenAndServeTLS(config.TLS.ServerCert, config.TLS.ServerKey)
 	for dispatcher, producer := range dispatchers {
 		logger.ActivityLog("attempting_to_close", logrus.LogInfo{"dispatcher": dispatcher})
 		// We don't care if this fails. If it does, we'll just continue on.

config/config.go

@@ -5,6 +5,7 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	_ "embed" //Used for default CAs
+	"encoding/json"
 	"errors"
 	"fmt"
 	"net/http"
@@ -47,6 +48,11 @@ type Config struct {
 	// TLS contains certificates & CA info for the webserver
 	TLS *TLS `json:"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"`
 
@@ -154,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,12 +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"},
-			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",
@@ -62,7 +63,7 @@ var _ = Describe("Test full application config", func() {
 	})
 
 	Context("ExtractServiceTLSConfig", func() {
-		It("fails when TLS is nil ", func() {
+		It("fails when TLS is nil", func() {
 			config = &Config{}
 			_, err := config.ExtractServiceTLSConfig(log)
 			Expect(err).To(MatchError("tls config is empty - telemetry server is mTLS only, make sure to provide certificates in the config"))
@@ -121,13 +122,32 @@ var _ = Describe("Test full application config", func() {
 			Expect(config.TransmitDecodedRecords).To(BeFalse())
 		})
 
+		It("tls_pass_through is nil by default", func() {
+			config, err := loadTestApplicationConfig(TestSmallConfig)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(config.TLSPassThrough).To(BeNil())
+		})
+
 		It("transmitrecords enabled", func() {
 			config, err := loadTestApplicationConfig(TestTransmitDecodedRecords)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(config.TransmitDecodedRecords).To(BeTrue())
 		})
 	})
 
+	Context("configure tls", func() {
+		It("read tls_pass_through config", func() {
+			config, err := loadTestApplicationConfig(TestRFC9440TLSConfig)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(*config.TLSPassThrough).To(BeEquivalentTo(RFC9440))
+		})
+
+		It("error on invalid tls_pass_through config", func() {
+			_, err := loadTestApplicationConfig(TestInvalidTLSPassThroughConfig)
+			Expect(err).To(HaveOccurred())
+		})
+	})
+
 	Context("configure kafka", func() {
 		It("converts floats to int", func() {
 			config, err := loadTestApplicationConfig(TestSmallConfig)
@@ -296,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,6 +62,46 @@ const TestSmallConfig = `
 }
 `
 
+const TestRFC9440TLSConfig = `
+{
+	"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": "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"
+}
+`
+
 const TestBadReliableAckConfig = `
 {
 	"host": "127.0.0.1",

server/streaming/server.go

@@ -3,8 +3,12 @@ package streaming
 import (
 	"context"
 	"crypto/x509"
+	"encoding/base64"
+	"encoding/pem"
 	"fmt"
+	"github.com/pkg/errors"
 	"net/http"
+	"net/url"
 	"sync"
 	"time"
 
@@ -129,7 +133,7 @@ func (s *Server) ServeBinaryWs(config *config.Config) func(w http.ResponseWriter
 	return func(w http.ResponseWriter, r *http.Request) {
 		if ws := s.promoteToWebsocket(w, r); ws != nil {
 			ctx := context.WithValue(context.Background(), SocketContext, map[string]interface{}{"request": r})
-			requestIdentity, err := extractIdentityFromConnection(r)
+			requestIdentity, err := extractIdentity(r, config)
 			if err != nil {
 				s.logger.ErrorLog("extract_sender_id_err", err, nil)
 			}
@@ -218,8 +222,21 @@ func (s *Server) promoteToWebsocket(w http.ResponseWriter, r *http.Request) *web
 	return ws
 }
 
-func extractIdentityFromConnection(r *http.Request) (*telemetry.RequestIdentity, error) {
-	cert, err := extractCertFromHeaders(r)
+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.TLSPassThrough != nil {
+		cert, err = headerExtractConfigMap[*config.TLSPassThrough](r)
+	} else {
+		cert, err = extractCertFromTLS(r)
+	}
 	if err != nil {
 		return nil, err
 	}
@@ -234,7 +251,49 @@ func extractIdentityFromConnection(r *http.Request) (*telemetry.RequestIdentity,
 	}, nil
 }
 
-func extractCertFromHeaders(r *http.Request) (*x509.Certificate, error) {
+// 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)
+	}
+	block, _ := pem.Decode(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
+}
+
+// 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")
+	}
+	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) {
 	nbCerts := len(r.TLS.PeerCertificates)
 	if nbCerts == 0 {
 		return nil, fmt.Errorf("missing_certificate_error")