Reuse validated immutable FLM input metadata

openpilot/starpilot/flm/native_refinement_wire.py

@@ -11,6 +11,10 @@ FORMAT = '!BQQBd9f'
 ORDINARY_INPUTS = 32
 EXTENDED_INPUTS = 33
 
+# Only deep-immutable metadata tuples already checked in this process. Live
+# source/native permission and receiver clocks remain outside this proof.
+_INPUT_PROOFS: tuple[object, ...] = ()
+
 
 @dataclass(frozen=True)
 class NativeProof:
@@ -63,6 +67,9 @@ def decode(value):
 
 def valid_inputs(values):
   """Legacy32 ordinary inputs, plus ONLY one recognized bounded extension."""
+  global _INPUT_PROOFS
+  if any(values is previous for previous in _INPUT_PROOFS):
+    return True
   try:
     if (type(values) is not tuple or len(values) > EXTENDED_INPUTS or
         any(type(row) is not tuple or len(row) != 2 for row in values) or
@@ -73,9 +80,11 @@ def valid_inputs(values):
         not (type(value) is bool or type(value) in (int,float) and math.isfinite(value) or
              type(value) is str and len(value) <= 96) for name,value in ordinary):
       return False
-    proof = tuple(value for name,value in values if name == FIELD)
+    proof = tuple((name,value) for name,value in values if name == FIELD)
     if proof:
-      decode(proof[0])
+      decode(proof[0][1])
+    if not proof or type(proof[0][0]) is str:
+      _INPUT_PROOFS = (*_INPUT_PROOFS[-3:], values)
     return True
   except (TypeError,ValueError,OverflowError):
     return False

openpilot/starpilot/flm/tests/test_native_metadata_proof.py

@@ -0,0 +1,70 @@
+import base64
+import struct
+from unittest import TestCase
+from unittest.mock import patch
+
+from openpilot.starpilot.flm import native_refinement_wire as wire
+
+
+def packet():
+  raw = struct.pack(wire.FORMAT, 1, 10000, 10001, 1, .01, *([0.] * 9))
+  return (('value', 1.), (wire.FIELD, wire.PREFIX + base64.b64encode(raw).decode()))
+
+
+class NativeMetadataProofTest(TestCase):
+  def setUp(self):
+    replacement = patch.object(wire, '_INPUT_PROOFS', ())
+    replacement.start()
+    self.addCleanup(replacement.stop)
+
+  def test_reuses_only_the_same_validated_immutable_tuple(self):
+    calls = []
+    original = wire.decode
+
+    def counted(value):
+      calls.append(value)
+      return original(value)
+
+    replacement = patch.object(wire, 'decode', counted)
+    replacement.start()
+    self.addCleanup(replacement.stop)
+    value = packet()
+    for _ in range(3):
+      assert wire.valid_inputs(value)
+    assert len(calls) == 1
+    equal = (value[0], value[1])
+    assert equal == value and equal is not value
+    assert wire.valid_inputs(equal)
+    assert len(calls) == 2
+    assert not wire.valid_inputs(value[:-1] + ((wire.FIELD, 'changed'),))
+
+
+  def test_proof_retention_is_bounded_and_evicted_inputs_are_checked(self):
+    first = packet()
+    assert wire.valid_inputs(first)
+    for i in range(12):
+      assert wire.valid_inputs((('value', i),))
+    assert len(wire._INPUT_PROOFS) == 4
+    assert all(value is not first for value in wire._INPUT_PROOFS)
+    assert wire.valid_inputs(first)
+
+
+  def test_mutable_string_subclass_cannot_reuse_an_immutable_proof(self):
+
+    class Name(str):
+      matches = True
+
+      def __eq__(self, other):
+        return self.matches and super().__eq__(other)
+
+      def __ne__(self, other):
+        return not self.__eq__(other)
+
+      __hash__ = str.__hash__
+
+    name = Name(wire.FIELD)
+    value = ((name, packet()[-1][1]),)
+    assert wire.valid_inputs(value)
+    assert not wire._INPUT_PROOFS
+    name.matches = False
+    assert not wire.valid_inputs(value)

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "47553ce6423062614b986934423f5ae1d3d4acb3b9002ec4de081e9a6db28ea8", "schema_version": 1}
+{"protected_inventory_sha256": "2a827534886fd3db0e2ddbed1fec22c9b498b69a80d08e3c01a08947325d794f", "schema_version": 1}