fix(microservices): validate pre-identified telemetry and keep stored data out of CVT

docs.openc3.com/docs/configuration/protocols.md

@@ -323,6 +323,8 @@ For a full example, please see [openc3-cosmos-gems-interface](https://repos.open
 
 The Preidentified Protocol delineates packets using the COSMOS header. This Protocol was created to allow tools to connect and receive the entire packet stream. It can be used with the [FileInterface](/docs/configuration/interfaces#file-interface) to process COSMOS 4 log files. It can also be used to chain COSMOS instances together although that should rarely be needed with the new web native implementation.
 
+Every target named in the incoming packet headers must be mapped to the interface with [MAP_TARGET](plugins.md#map_target) or [MAP_TLM_TARGET](plugins.md#map_tlm_target). Packets for unmapped targets are dropped, and the interface logs a warning the first time it sees each unmapped packet.
+
 | Parameter    | Description                                                                                                                                                                                                                    | Required | Default                 |
 | ------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | -------- | ----------------------- |
 | Sync Pattern | Hex string representing a byte pattern that will be searched for in the raw data. This pattern represents a packet delimiter and all data found AFTER the sync pattern will be returned. The sync pattern itself is discarded. | No       | nil (no sync pattern)   |

openc3/lib/openc3/microservices/interface_microservice.rb

@@ -559,6 +559,8 @@ def run
   class InterfaceMicroservice < Microservice
     UNKNOWN_BYTES_TO_PRINT = 16
     DEFAULT_UPDATE_INTERVAL = 0.1 # seconds
+    # Most distinct unmapped target / packet names to remember warning about
+    UNMAPPED_TLM_WARNED_MAX = 1000
 
     # Interfaces never use descriptions so don't keep them in memory
     def load_descriptions?
@@ -600,6 +602,8 @@ def initialize(name)
         RouterStatusModel.set(@interface.as_json(), scope: @scope)
       end
 
+      # Insertion ordered so the oldest key can be evicted (see handle_packet)
+      @unmapped_tlm_warned = {}
       # Queue Redis writes by default and flush them every update_interval seconds
       update_interval = DEFAULT_UPDATE_INTERVAL
       @interface.options.each do |option_name, option_values|
@@ -794,26 +798,38 @@ def handle_packet(packet)
       InterfaceStatusModel.set(@interface.as_json(), queued: true, scope: @scope) unless @cancel_thread
       packet.received_time = Time.now.sys unless packet.received_time
 
+      if packet.identified?
+        unless @interface.tlm_target_names.include?(packet.target_name)
+          # Trust the identity but drop it because the target isn't mapped to this interface
+          key = "#{packet.target_name} #{packet.packet_name}"
+          unless @unmapped_tlm_warned.key?(key)
+            # The names come from the wire so bound how many are remembered.
+            # Evict the oldest rather than stop adding so every new name still warns.
+            @unmapped_tlm_warned.delete(@unmapped_tlm_warned.first[0]) if @unmapped_tlm_warned.length >= UNMAPPED_TLM_WARNED_MAX
+            @unmapped_tlm_warned[key] = true
+            @logger.warn "#{@interface.name}: Dropping telemetry for unmapped target: #{key}"
+          end
+          return
+        end
+        unless known_tlm_packet?(packet)
+          # Packet identified but we don't know about it
+          # Clear packet_name and target_name and try to identify
+          @logger.warn "#{@interface.name}: Received unknown identified telemetry: #{packet.target_name} #{packet.packet_name}"
+          packet.target_name = nil
+          packet.packet_name = nil
+        end
+      end
+
       if packet.stored
         # Stored telemetry does not update the current value table
         identified_packet = System.telemetry.identify_and_define_packet(packet, @interface.tlm_target_names)
       else
         # Identify and update packet
         if packet.identified?
-          begin
-            # Preidentifed packet - place it into the current value table
-            identified_packet = System.telemetry.update!(packet.target_name,
-                                                         packet.packet_name,
-                                                         packet.buffer)
-          rescue RuntimeError
-            # Packet identified but we don't know about it
-            # Clear packet_name and target_name and try to identify
-            @logger.warn "#{@interface.name}: Received unknown identified telemetry: #{packet.target_name} #{packet.packet_name}"
-            packet.target_name = nil
-            packet.packet_name = nil
-            identified_packet = System.telemetry.identify!(packet.buffer,
-                                                           @interface.tlm_target_names)
-          end
+          # Preidentifed packet - place it into the current value table
+          identified_packet = System.telemetry.update!(packet.target_name,
+                                                       packet.packet_name,
+                                                       packet.buffer)
         else
           # Packet needs to be identified
           identified_packet = System.telemetry.identify!(packet.buffer,
@@ -827,13 +843,21 @@ def handle_packet(packet)
         identified_packet.extra = packet.extra
         packet = identified_packet
       else
-        unknown_packet = System.telemetry.update!('UNKNOWN', 'UNKNOWN', packet.buffer)
+        if packet.stored
+          # Stored telemetry does not update the current value table
+          unknown_packet = System.telemetry.packet('UNKNOWN', 'UNKNOWN').clone
+          unknown_packet.buffer = packet.buffer
+        else
+          unknown_packet = System.telemetry.update!('UNKNOWN', 'UNKNOWN', packet.buffer)
+        end
         unknown_packet.received_time = packet.received_time
         unknown_packet.stored = packet.stored
         unknown_packet.extra = packet.extra
         packet = unknown_packet
-        json_hash = CvtModel.build_json_from_packet(packet)
-        CvtModel.set(json_hash, target_name: packet.target_name, packet_name: packet.packet_name, queued: @queued, scope: @scope)
+        unless packet.stored
+          json_hash = CvtModel.build_json_from_packet(packet)
+          CvtModel.set(json_hash, target_name: packet.target_name, packet_name: packet.packet_name, queued: @queued, scope: @scope)
+        end
         num_bytes_to_print = [UNKNOWN_BYTES_TO_PRINT, packet.length].min
         data = packet.buffer(false)[0..(num_bytes_to_print - 1)]
         prefix = data.each_byte.map { | byte | sprintf("%02X", byte) }.join()
@@ -847,6 +871,14 @@ def handle_packet(packet)
       end
     end
 
+    # @return [Boolean] Whether a pre-identified packet names a defined packet
+    def known_tlm_packet?(packet)
+      System.telemetry.packet(packet.target_name, packet.packet_name)
+      true
+    rescue RuntimeError
+      false
+    end
+
     def handle_connection_failed(connection, connect_error)
       @error = connect_error
       @logger.error "#{@interface.name}: Connection #{connection} failed due to #{connect_error.formatted(false, false)}"

openc3/python/openc3/microservices/interface_microservice.py

@@ -657,6 +657,8 @@ class InterfaceMicroservice(Microservice):
     UNKNOWN_BYTES_TO_PRINT = 16
     DISCONNECT_WAIT_TIME = 1
     DEFAULT_UPDATE_INTERVAL = 0.1  # seconds
+    # Most distinct unmapped target / packet names to remember warning about
+    UNMAPPED_TLM_WARNED_MAX = 1000
 
     # Interfaces never use descriptions so don't keep them in memory
     def load_descriptions(self):
@@ -682,6 +684,7 @@ def __init__(self, name):
 
         self.scope = name.split("__")[0]
         interface_name = name.split("__")[2]
+        self.interface: Interface
         if self.interface_or_router == "INTERFACE":
             self.interface = InterfaceModel.get_model(name=interface_name, scope=self.scope).build()
         else:
@@ -701,6 +704,8 @@ def __init__(self, name):
         else:
             RouterStatusModel.set(self.interface.as_json(), scope=self.scope)
 
+        # Insertion ordered so the oldest key can be evicted (see handle_packet)
+        self.unmapped_tlm_warned: dict[str, bool] = {}
         # Queue Redis writes by default and flush them every update_interval seconds
         update_interval = InterfaceMicroservice.DEFAULT_UPDATE_INTERVAL
         for option_name, option_values in self.interface.options.items():
@@ -798,7 +803,7 @@ def attempt_connection(self, *params):
         # Catch Exception so any rebuild failure (bad parameters, import errors) keeps the original interface
         except Exception:
             self.logger.error(
-                f"Attempting connection #{self.interface.connection_string} failed due to {traceback.format_exc()}"
+                f"Attempting connection #{self.interface.connection_string()} failed due to {traceback.format_exc()}"
             )
             # if SignalException === error:
             #   self.logger.info(f"{self.interface.name}: Closing from signal")
@@ -905,26 +910,35 @@ def handle_packet(self, packet):
         if packet.received_time is None:
             packet.received_time = datetime.now(timezone.utc)
 
+        if packet.identified():
+            if packet.target_name not in self.interface.tlm_target_names:
+                # Trust the identity but drop it because the target isn't mapped to this interface
+                key = f"{packet.target_name} {packet.packet_name}"
+                if key not in self.unmapped_tlm_warned:
+                    # The names come from the wire so bound how many are remembered.
+                    # Evict the oldest rather than stop adding so every new name still warns.
+                    if len(self.unmapped_tlm_warned) >= self.UNMAPPED_TLM_WARNED_MAX:
+                        del self.unmapped_tlm_warned[next(iter(self.unmapped_tlm_warned))]
+                    self.unmapped_tlm_warned[key] = True
+                    self.logger.warn(f"{self.interface.name}: Dropping telemetry for unmapped target: {key}")
+                return
+            if not self._known_tlm_packet(packet):
+                # Packet identified but we don't know about it
+                # Clear packet_name and target_name and try to identify
+                self.logger.warn(
+                    f"{self.interface.name}: Received unknown identified telemetry: {packet.target_name} {packet.packet_name}"
+                )
+                packet.target_name = None
+                packet.packet_name = None
+
         if packet.stored:
             # Stored telemetry does not update the current value table
             identified_packet = System.telemetry.identify_and_define_packet(packet, self.interface.tlm_target_names)
         else:
             # Identify and update packet
             if packet.identified():
-                try:
-                    # Preidentifed packet - place it into the current value table
-                    identified_packet = System.telemetry.update(packet.target_name, packet.packet_name, packet.buffer)
-                except Exception:
-                    # Packet identified but we don't know about it
-                    # Clear packet_name and target_name and try to identify
-                    self.logger.warn(
-                        f"{self.interface.name}: Received unknown identified telemetry: {packet.target_name} {packet.packet_name}"
-                    )
-                    packet.target_name = None
-                    packet.packet_name = None
-                    identified_packet = System.telemetry.identify_and_set_buffer(
-                        packet.buffer, self.interface.tlm_target_names
-                    )
+                # Preidentifed packet - place it into the current value table
+                identified_packet = System.telemetry.update(packet.target_name, packet.packet_name, packet.buffer)
             else:
                 # Packet needs to be identified
                 identified_packet = System.telemetry.identify_and_set_buffer(
@@ -937,22 +951,28 @@ def handle_packet(self, packet):
             identified_packet.extra = packet.extra
             packet = identified_packet
         else:
-            unknown_packet = System.telemetry.update("UNKNOWN", "UNKNOWN", packet.buffer)
+            if packet.stored:
+                # Stored telemetry does not update the current value table
+                unknown_packet = System.telemetry.packet("UNKNOWN", "UNKNOWN").clone()
+                unknown_packet.buffer = packet.buffer
+            else:
+                unknown_packet = System.telemetry.update("UNKNOWN", "UNKNOWN", packet.buffer)
             unknown_packet.received_time = packet.received_time
             unknown_packet.stored = packet.stored
             unknown_packet.extra = packet.extra
             packet = unknown_packet
-            json_hash = CvtModel.build_json_from_packet(packet)
-            CvtModel.set(
-                json_hash,
-                packet.target_name,
-                packet.packet_name,
-                queued=self.queued,
-                scope=self.scope,
-            )
+            if not packet.stored:
+                json_hash = CvtModel.build_json_from_packet(packet)
+                CvtModel.set(
+                    json_hash,
+                    packet.target_name,
+                    packet.packet_name,
+                    queued=self.queued,
+                    scope=self.scope,
+                )
             num_bytes_to_print = min(InterfaceMicroservice.UNKNOWN_BYTES_TO_PRINT, len(packet.buffer))
             data = packet.buffer_no_copy()[0:(num_bytes_to_print)]
-            prefix = "".join([format(x, "02x") for x in data])
+            prefix = "".join([format(x, "02X") for x in data])
             self.logger.warn(
                 f"{self.interface.name} {packet.target_name} packet length: {len(packet.buffer)} starting with: {prefix}"
             )
@@ -962,6 +982,14 @@ def handle_packet(self, packet):
             TargetModel.sync_tlm_packet_counts(packet, self.interface.tlm_target_names, scope=self.scope)
             TelemetryTopic.write_packet(packet, queued=self.queued, scope=self.scope)
 
+    # Whether a pre-identified packet names a defined packet
+    def _known_tlm_packet(self, packet):
+        try:
+            System.telemetry.packet(packet.target_name, packet.packet_name)
+            return True
+        except Exception:
+            return False
+
     # Equivalent of the Ruby Errno lists. Use the specific ConnectionError subclasses
     # rather than ConnectionError itself, which also covers BrokenPipeError (EPIPE)
     CONNECTION_FAILED_ERRORS = (ConnectionRefusedError, ConnectionResetError, TimeoutError)

openc3/python/openc3/topics/telemetry_topic.py

@@ -23,7 +23,7 @@ def write_packet(cls, packet, scope, queued=False):
         msg_hash = {
             "time": to_nsec_from_epoch(packet.packet_time),
             "received_time": to_nsec_from_epoch(packet.received_time),
-            "stored": str(packet.stored),
+            "stored": str(packet.stored).lower(),
             "target_name": packet.target_name,
             "packet_name": packet.packet_name,
             "received_count": packet.received_count,

openc3/python/test/microservices/test_interface_microservice.py

@@ -27,6 +27,7 @@
 from openc3.models.interface_status_model import InterfaceStatusModel
 from openc3.models.microservice_model import MicroserviceModel
 from openc3.models.target_model import TargetModel
+from openc3.packets.packet import Packet
 from openc3.system.system import System
 from openc3.topics.interface_topic import InterfaceTopic
 from openc3.topics.telemetry_decom_topic import TelemetryDecomTopic
@@ -1053,6 +1054,83 @@ def test_handle_packet_does_not_write_status_after_cancel_thread_set(self):
         im.shutdown()
         time.sleep(0.1)  # Allow threads to exit
 
+    def setup_handle_packet(self):
+        im = InterfaceMicroservice("DEFAULT__INTERFACE__INST_INT")
+        self.addCleanup(im.shutdown)
+        warnings = []
+        im.logger = Mock()
+        im.logger.warn.side_effect = lambda msg, *args, **kwargs: warnings.append(msg)
+        written = []
+        patcher = patch(
+            "openc3.microservices.interface_microservice.TelemetryTopic.write_packet",
+            side_effect=lambda packet, *args, **kwargs: written.append(packet),
+        )
+        patcher.start()
+        self.addCleanup(patcher.stop)
+        return im, warnings, written
+
+    def health_status_buffer(self):
+        packet = System.telemetry.packet("INST", "HEALTH_STATUS").clone()
+        for item in packet.id_items:
+            packet.write_item(item, item.id_value, "RAW")
+        return packet.buffer
+
+    def handle(self, im, buffer, target_name=None, packet_name=None, stored=False):
+        packet = Packet(target_name, packet_name)
+        packet.buffer = buffer
+        packet.stored = stored
+        im.handle_packet(packet)
+
+    def test_handle_packet_drops_a_target_not_mapped_to_the_interface_and_warns_once(self):
+        im, warnings, written = self.setup_handle_packet()
+        with (
+            patch.object(System.telemetry, "identify_and_set_buffer") as mock_identify,
+            patch.object(System.telemetry, "identify_and_define_packet") as mock_define,
+        ):
+            for _ in range(2):
+                self.handle(im, self.health_status_buffer(), "SYSTEM", "LIMITS_CHANGE")
+            self.handle(im, self.health_status_buffer(), "SYSTEM", "LIMITS_CHANGE", stored=True)
+            mock_identify.assert_not_called()
+            mock_define.assert_not_called()
+        self.assertEqual(
+            [w for w in warnings if "unmapped target" in w],
+            ["INST_INT: Dropping telemetry for unmapped target: SYSTEM LIMITS_CHANGE"],
+        )
+        self.assertEqual(written, [])
+
+    def test_handle_packet_bounds_unmapped_warnings_by_evicting_the_oldest(self):
+        im, warnings, _ = self.setup_handle_packet()
+        with patch.object(InterfaceMicroservice, "UNMAPPED_TLM_WARNED_MAX", 2):
+            for name in ["A", "B", "C", "A"]:
+                self.handle(im, self.health_status_buffer(), "SYSTEM", name)
+        # C evicted A so A warns again, and every new name warned
+        self.assertEqual([w.split(" ")[-1] for w in warnings if "unmapped target" in w], ["A", "B", "C", "A"])
+        self.assertEqual(len(im.unmapped_tlm_warned), 2)
+
+    def test_handle_packet_falls_back_for_a_stored_packet_with_an_unknown_packet(self):
+        im, warnings, written = self.setup_handle_packet()
+        self.handle(im, self.health_status_buffer(), "INST", "NOPE", stored=True)
+        self.assertIn("INST_INT: Received unknown identified telemetry: INST NOPE", warnings)
+        self.assertEqual(len(written), 1)
+        self.assertEqual(written[0].target_name, "INST")
+        self.assertEqual(written[0].packet_name, "HEALTH_STATUS")
+        self.assertTrue(written[0].stored)
+
+    def test_handle_packet_publishes_an_unknown_stored_packet_without_updating_the_cvt(self):
+        im, warnings, written = self.setup_handle_packet()
+        with (
+            patch.object(CvtModel, "set") as mock_set,
+            patch.object(System.telemetry, "update", wraps=System.telemetry.update) as mock_update,
+        ):
+            self.handle(im, b"\xab\xcd", stored=True)
+            mock_set.assert_not_called()
+            mock_update.assert_not_called()
+        self.assertEqual(len(written), 1)
+        self.assertEqual(written[0].target_name, "UNKNOWN")
+        self.assertEqual(written[0].packet_name, "UNKNOWN")
+        self.assertTrue(written[0].stored)
+        self.assertIn("INST_INT UNKNOWN packet length: 2 starting with: ABCD", warnings)
+
     def test_supports_optimize_throughput_option_for_backward_compatibility(self):
         # Update the model to use OPTIMIZE_THROUGHPUT option (legacy name)
         model = InterfaceModel(