Fix vehicle identity spoofing and bind monitoring servers to localhost

Author: aaronpkahn

metrics/metrics.go

@@ -18,12 +18,18 @@ type MonitoringConfig struct {
 	// PrometheusMetricsPort port to run prometheus on
 	PrometheusMetricsPort int `json:"prometheus_metrics_port,omitempty"`
 
+	// PrometheusMetricsHost is the address to bind the prometheus metrics server to (default "127.0.0.1")
+	PrometheusMetricsHost string `json:"prometheus_metrics_host,omitempty"`
+
 	// Statsd metrics if you are not using prometheus
 	Statsd *StatsdConfig `json:"statsd,omitempty"`
 
 	// ProfilerPort if non-zero enable http profiler on this port
 	ProfilerPort int `json:"profiler_port,omitempty"`
 
+	// ProfilerHost is the address to bind the profiler to (default "127.0.0.1")
+	ProfilerHost string `json:"profiler_host,omitempty"`
+
 	// ProfilingPath is the variable that enable deep profiling is set
 	ProfilingPath string `json:"profiling_path,omitempty"`
 

server/monitoring/metrics_server.go

@@ -6,8 +6,7 @@ import (
 	"sync"
 	"time"
 
-	// This registers the profiler on the default mux which we will use for monitoring port.
-	_ "net/http/pprof"
+	"net/http/pprof"
 
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 
@@ -37,17 +36,32 @@ func StartServerMetrics(config *config.Config, logger *logrus.Logger, registry *
 		promMux := http.NewServeMux()
 		promMux.Handle("/metrics", promhttp.Handler())
 		go func() {
-			if err := http.ListenAndServe(fmt.Sprintf(":%d", config.Monitoring.PrometheusMetricsPort), promMux); err != nil {
+			metricsHost := config.Monitoring.PrometheusMetricsHost
+			if metricsHost == "" {
+				metricsHost = "127.0.0.1"
+			}
+			if err := http.ListenAndServe(fmt.Sprintf("%s:%d", metricsHost, config.Monitoring.PrometheusMetricsPort), promMux); err != nil {
 				logger.ErrorLog("metrics_server_err", err, nil)
 			}
 		}()
 	}
 
 	if config.Monitoring.ProfilerPort > 0 {
 		go func() {
-			StartProfilerServer(config, http.DefaultServeMux, logger)
+			profilerMux := http.NewServeMux()
+			profilerMux.HandleFunc("/debug/pprof/", pprof.Index)
+			profilerMux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
+			profilerMux.HandleFunc("/debug/pprof/profile", pprof.Profile)
+			profilerMux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
+			profilerMux.HandleFunc("/debug/pprof/trace", pprof.Trace)
+
+			StartProfilerServer(config, profilerMux, logger)
 
-			if err := http.ListenAndServe(fmt.Sprintf(":%d", config.Monitoring.ProfilerPort), nil); err != nil {
+			profilerHost := config.Monitoring.ProfilerHost
+			if profilerHost == "" {
+				profilerHost = "127.0.0.1"
+			}
+			if err := http.ListenAndServe(fmt.Sprintf("%s:%d", profilerHost, config.Monitoring.ProfilerPort), profilerMux); err != nil {
 				logger.ErrorLog("profiler_listen_error", err, nil)
 			}
 		}()

server/streaming/server.go

@@ -132,6 +132,8 @@ func (s *Server) ServeBinaryWs(config *config.Config) func(w http.ResponseWriter
 			requestIdentity, err := extractIdentityFromConnection(r)
 			if err != nil {
 				s.logger.ErrorLog("extract_sender_id_err", err, nil)
+				_ = ws.Close()
+				return
 			}
 
 			binarySerializer := telemetry.NewBinarySerializer(requestIdentity, s.DispatchRules, s.logger)
@@ -236,12 +238,13 @@ func extractIdentityFromConnection(r *http.Request) (*telemetry.RequestIdentity,
 }
 
 func extractCertFromHeaders(r *http.Request) (*x509.Certificate, error) {
-	nbCerts := len(r.TLS.PeerCertificates)
-	if nbCerts == 0 {
-		return nil, fmt.Errorf("missing_certificate_error")
+	if r.TLS == nil {
+		return nil, fmt.Errorf("missing_tls_state")
 	}
-
-	return r.TLS.PeerCertificates[nbCerts-1], nil
+	if len(r.TLS.VerifiedChains) == 0 || len(r.TLS.VerifiedChains[0]) == 0 {
+		return nil, fmt.Errorf("missing_verified_client_certificate")
+	}
+	return r.TLS.VerifiedChains[0][0], nil
 }
 
 func registerServerMetricsOnce(metricsCollector metrics.MetricCollector) {

server/streaming/server_test.go

@@ -1,6 +1,9 @@
 package streaming_test
 
 import (
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
@@ -13,12 +16,38 @@ import (
 
 	"github.com/teslamotors/fleet-telemetry/config"
 	logrus "github.com/teslamotors/fleet-telemetry/logger"
+	"github.com/teslamotors/fleet-telemetry/messages"
 	"github.com/teslamotors/fleet-telemetry/metrics/adapter/noop"
 	"github.com/teslamotors/fleet-telemetry/server/airbrake"
 	"github.com/teslamotors/fleet-telemetry/server/streaming"
 	"github.com/teslamotors/fleet-telemetry/telemetry"
 )
 
+// withTLSState wraps a handler to inject a tls.ConnectionState into each request,
+// simulating an mTLS connection for testing without actual TLS transport.
+func withTLSState(handler http.Handler, tlsState *tls.ConnectionState) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		r.TLS = tlsState
+		handler.ServeHTTP(w, r)
+	})
+}
+
+func makeCert(commonName, issuerCommonName string) *x509.Certificate {
+	return &x509.Certificate{
+		Subject: pkix.Name{CommonName: commonName},
+		Issuer:  pkix.Name{CommonName: issuerCommonName},
+	}
+}
+
+type spyProducer struct {
+	captured chan *telemetry.Record
+}
+
+func (p *spyProducer) Close() error                                    { return nil }
+func (p *spyProducer) Produce(entry *telemetry.Record)                 { p.captured <- entry }
+func (p *spyProducer) ProcessReliableAck(_ *telemetry.Record)          {}
+func (p *spyProducer) ReportError(_ string, _ error, _ logrus.LogInfo) {}
+
 var _ = Describe("Socket handler test", func() {
 
 	var producerRules map[string][]telemetry.Producer
@@ -38,8 +67,8 @@ var _ = Describe("Socket handler test", func() {
 		}
 	})
 
-	It("ServeBinaryWs test", func() {
-		logger, hook := logrus.NoOpLogger()
+	It("ServeBinaryWs closes connection when TLS state is missing", func() {
+		logger, _ := logrus.NoOpLogger()
 		conf := &config.Config{
 			RateLimit: &config.RateLimit{
 				MessageLimit:              1,
@@ -49,29 +78,119 @@ var _ = Describe("Socket handler test", func() {
 		}
 
 		registry := streaming.NewSocketRegistry()
-		req := httptest.NewRequest("GET", "http://test-server.example.com", nil)
-
-		producerRules := make(map[string][]telemetry.Producer)
+		producerRules = make(map[string][]telemetry.Producer)
 		_, s, err := streaming.InitServer(conf, airbrake.NewAirbrakeHandler(nil), producerRules, logger, registry)
 		Expect(err).NotTo(HaveOccurred())
 
+		// No TLS state injected, simulates a non-mTLS connection
 		srv := httptest.NewServer(http.HandlerFunc(s.ServeBinaryWs(conf)))
+		defer srv.Close()
 		u, _ := url.Parse(srv.URL)
 		u.Scheme = "ws"
 
 		dialer := &websocket.Dialer{HandshakeTimeout: 1 * time.Second}
-		conn, _, err := dialer.Dial(u.String(), req.Header)
+		conn, _, err := dialer.Dial(u.String(), nil)
 		Expect(err).NotTo(HaveOccurred())
 
-		err = conn.SetReadDeadline(time.Now().Add(100 * time.Millisecond))
+		// Server should close the connection due to missing TLS state
+		err = conn.SetReadDeadline(time.Now().Add(1 * time.Second))
 		Expect(err).NotTo(HaveOccurred())
+		_, _, err = conn.ReadMessage()
+		Expect(err).To(HaveOccurred())
+		_ = conn.Close()
+	})
 
-		err = conn.WriteMessage(websocket.BinaryMessage, []byte(""))
+	It("ServeBinaryWs closes connection when VerifiedChains is empty", func() {
+		logger, _ := logrus.NoOpLogger()
+		conf := &config.Config{
+			RateLimit: &config.RateLimit{
+				MessageLimit:              1,
+				MessageIntervalTimeSecond: 1 * time.Second,
+			},
+			MetricCollector: noop.NewCollector(),
+		}
+
+		registry := streaming.NewSocketRegistry()
+		producerRules = make(map[string][]telemetry.Producer)
+		_, s, err := streaming.InitServer(conf, airbrake.NewAirbrakeHandler(nil), producerRules, logger, registry)
 		Expect(err).NotTo(HaveOccurred())
 
-		_, _, _ = conn.ReadMessage()
+		// TLS state with PeerCertificates but no VerifiedChains
+		tlsState := &tls.ConnectionState{
+			PeerCertificates: []*x509.Certificate{
+				makeCert("device-1", "TeslaMotors"),
+			},
+			VerifiedChains: nil,
+		}
+		srv := httptest.NewServer(withTLSState(http.HandlerFunc(s.ServeBinaryWs(conf)), tlsState))
+		defer srv.Close()
+		u, _ := url.Parse(srv.URL)
+		u.Scheme = "ws"
+
+		dialer := &websocket.Dialer{HandshakeTimeout: 1 * time.Second}
+		conn, _, err := dialer.Dial(u.String(), nil)
+		Expect(err).NotTo(HaveOccurred())
+
+		// Server should close the connection due to empty VerifiedChains
+		err = conn.SetReadDeadline(time.Now().Add(1 * time.Second))
+		Expect(err).NotTo(HaveOccurred())
+		_, _, err = conn.ReadMessage()
+		Expect(err).To(HaveOccurred())
 		_ = conn.Close()
+	})
+
+	It("ServeBinaryWs uses verified leaf cert, not last PeerCertificate", func() {
+		logger, _ := logrus.NoOpLogger()
+
+		spy := &spyProducer{captured: make(chan *telemetry.Record, 1)}
+
+		conf := &config.Config{
+			RateLimit: &config.RateLimit{
+				MessageLimit:              1,
+				MessageIntervalTimeSecond: 1 * time.Second,
+			},
+			MetricCollector: noop.NewCollector(),
+		}
+
+		registry := streaming.NewSocketRegistry()
+		producerRules = map[string][]telemetry.Producer{"V": {spy}}
+		_, s, err := streaming.InitServer(conf, airbrake.NewAirbrakeHandler(nil), producerRules, logger, registry)
+		Expect(err).NotTo(HaveOccurred())
+
+		realCert := makeCert("device-1", "TeslaMotors")
+		spoofedCert := makeCert("SPOOFED-VIN", "TeslaMotors")
+
+		tlsState := &tls.ConnectionState{
+			PeerCertificates: []*x509.Certificate{realCert, spoofedCert},
+			VerifiedChains:   [][]*x509.Certificate{{realCert}},
+		}
+		srv := httptest.NewServer(withTLSState(http.HandlerFunc(s.ServeBinaryWs(conf)), tlsState))
+		defer srv.Close()
+		u, _ := url.Parse(srv.URL)
+		u.Scheme = "ws"
+
+		dialer := &websocket.Dialer{HandshakeTimeout: 1 * time.Second}
+		conn, _, err := dialer.Dial(u.String(), nil)
+		Expect(err).NotTo(HaveOccurred())
+		defer func() { _ = conn.Close() }()
+
+		streamMsg := messages.StreamMessage{
+			TXID:         []byte("test-txid"),
+			SenderID:     []byte("vehicle_device.blah"),
+			DeviceID:     []byte("blah"),
+			DeviceType:   []byte("vehicle_device"),
+			MessageTopic: []byte("V"),
+			Payload:      []byte{},
+		}
+		msgBytes, err := streamMsg.ToBytes()
+		Expect(err).NotTo(HaveOccurred())
+
+		err = conn.WriteMessage(websocket.BinaryMessage, msgBytes)
+		Expect(err).NotTo(HaveOccurred())
 
-		Expect(hook.AllEntries()).To(BeEmpty())
+		// The record VIN must come from the verified cert ("device-1")
+		var record *telemetry.Record
+		Eventually(spy.captured).Should(Receive(&record))
+		Expect(record.Vin).To(Equal("device-1"))
 	})
 })