Add integration test for vehicle id spoofing

Makefile

@@ -51,7 +51,6 @@ test-race: test
 integration: generate-certs
 	@echo "** RUNNING INTEGRATION TESTS **"
 	./test/integration/pretest.sh
-	docker compose -p app -f docker-compose.yml build
 	docker compose -p app -f docker-compose.yml up -d --remove-orphans
 	./test/integration/test.sh
 	docker compose -p app -f docker-compose.yml down

docker-compose.yml

@@ -1,14 +1,12 @@
 services:
   app:
-    build:
-      context: .
-      dockerfile: Dockerfile.integration
+    image: fleet-telemetry-integration-tests:latest
     volumes:
       - ./test/integration/config.json:/etc/tesla/config.json:ro
       - ./test/integration/test-certs:/etc/tesla/certs/server:ro
       - /tmp/fleet-telemetry/:/tmp/fleet-telemetry:rw
     command:
-      - "/fleet-telemetry"
+      - "/go/bin/fleet-telemetry"
       - "-config=/etc/tesla/config.json"
     ports:
       - "4443:4443"
@@ -98,6 +96,7 @@ services:
   # Errbit implements airbrake's API, so we can use it to test the airbrake
   errbit:
     image: errbit/errbit:v0.8.0
+    platform: linux/amd64
     ports:
       - '48088:8080'
     command: sh -c "bundle exec rake errbit:bootstrap && bundle exec puma -C config/puma.default.rb"

telemetry/serializer.go

@@ -37,7 +37,7 @@ func NewBinarySerializer(requestIdentity *RequestIdentity, dispatchRules map[str
 func (bs *BinarySerializer) Deserialize(msg []byte, socketID string) (record *Record, err error) {
 	defer func() {
 		if r := recover(); r != nil {
-			err = fmt.Errorf("panic while serializing binary hermes stream: %v", r)
+			err = fmt.Errorf("panic while deserializing binary hermes stream: %v", r)
 		}
 	}()
 

test/integration/integration_test.go

@@ -26,6 +26,7 @@ import (
 )
 
 const (
+	deviceID    = "device-1"
 	vehicleName = "My Test Vehicle"
 	location    = "(37.412374 S, 122.145867 E)"
 	projectID   = "test-project-id"
@@ -73,7 +74,7 @@ var _ = Describe("Test messages", Ordered, func() {
 		Expect(err).NotTo(HaveOccurred())
 		timestamp = timestamppb.Now()
 
-		payload = GenerateVehicleMessage(vehicleName, location, timestamp)
+		payload = GenerateVehicleMessage(deviceID, vehicleName, location, timestamp)
 
 		kinesisConsumer, err = NewTestKinesisConsumer(kinesisHost, []string{kinesisStreamName, kinesisConnectivityStreamName})
 		Expect(err).NotTo(HaveOccurred())
@@ -118,7 +119,32 @@ var _ = Describe("Test messages", Ordered, func() {
 			defer GinkgoRecover()
 			err := kafkaConsumer.Subscribe(vehicleTopic, nil)
 			Expect(err).NotTo(HaveOccurred())
-			err = connection.WriteMessage(websocket.BinaryMessage, GenerateVehicleMessage(vehicleName, location, timestamp))
+			err = connection.WriteMessage(websocket.BinaryMessage, GenerateVehicleMessage(deviceID, vehicleName, location, timestamp))
+			verifyAckMessage(connection, "V")
+			Expect(err).NotTo(HaveOccurred())
+			msg, err := kafkaConsumer.ReadMessage(10 * time.Second)
+			Expect(err).NotTo(HaveOccurred())
+
+			Expect(msg).NotTo(BeNil())
+			Expect(*msg.TopicPartition.Topic).To(Equal(vehicleTopic))
+			Expect(string(msg.Key)).To(Equal(deviceID))
+
+			headers := make(map[string]string)
+			for _, h := range msg.Headers {
+				headers[string(h.Key)] = string(h.Value)
+			}
+			VerifyMessageHeaders(headers)
+			VerifyMessageBody(msg.Value, vehicleName)
+		})
+
+		It("does not allow spoofing", func() {
+			defer GinkgoRecover()
+			err := kafkaConsumer.Subscribe(vehicleTopic, nil)
+			Expect(err).NotTo(HaveOccurred())
+
+			spoofedID := "BAD123"
+			vehicleMessage := GenerateVehicleMessage(spoofedID, vehicleName, location, timestamp)
+			err = connection.WriteMessage(websocket.BinaryMessage, vehicleMessage)
 			verifyAckMessage(connection, "V")
 			Expect(err).NotTo(HaveOccurred())
 			msg, err := kafkaConsumer.ReadMessage(10 * time.Second)
@@ -343,7 +369,7 @@ func VerifyConnectivityMessageBody(body []byte) {
 	Expect(payload.GetStatus()).To(Or(Equal(protos.ConnectivityEvent_CONNECTED), Equal(protos.ConnectivityEvent_DISCONNECTED)))
 }
 
-// VerifyMessageHeaders validates record message returned from dispatchers
+// VerifyMessageBody validates record message returned from dispatchers
 func VerifyMessageBody(body []byte, vehicleName string) {
 	payload := &protos.Payload{}
 	err := proto.Unmarshal(body, payload)

test/integration/server_test.go

@@ -24,7 +24,6 @@ const (
 	messageID           = "integration-test-message-id"
 	senderID            = "vehicle_device.device-1"
 	deviceType          = "vehicle_device"
-	deviceID            = "device-1"
 	deviceClientVersion = "1.0.0"
 
 	serviceURL    = "app:4443"
@@ -37,8 +36,9 @@ const (
 	caClient   = "./test-certs/vehicle_device.CA.cert"
 )
 
-func GenerateVehicleMessage(vehicleName, location string, timestamp *timestamppb.Timestamp) []byte {
-	return tesla.FlatbuffersStreamToBytes([]byte(senderID), []byte("V"), []byte(txid), generatePayload(vehicleName, location, timestamp), 1, []byte(messageID), []byte(deviceType), []byte(deviceID), uint64(time.Now().UnixMilli()))
+func GenerateVehicleMessage(id, vehicleName, location string, timestamp *timestamppb.Timestamp) []byte {
+	senderID := fmt.Sprintf("%s.%s", deviceType, id)
+	return tesla.FlatbuffersStreamToBytes([]byte(senderID), []byte("V"), []byte(txid), generatePayload(vehicleName, location, timestamp), 1, []byte(messageID), []byte(deviceType), []byte(id), uint64(time.Now().UnixMilli()))
 }
 
 func generatePayload(vehicleName, location string, timestamp *timestamppb.Timestamp) []byte {