fix: Ignore FDv2 objects with unrecognized kinds

ldclient/impl/datasourcev2/polling_common.py

@@ -11,7 +11,8 @@
 from ldclient.impl.datasystem.protocolv2 import (
     DeleteObject,
     EventName,
-    PutObject
+    PutObject,
+    has_unsupported_kind
 )
 from ldclient.impl.util import (
     _LD_ENVID_HEADER,
@@ -68,15 +69,35 @@ def polling_payload_to_changeset(data: dict) -> _Result[ChangeSet, str]:
 
             builder.start(server_intent.payload.code)
         elif event["event"] == EventName.PUT_OBJECT:
+            event_data = event.get("data")
+            if not isinstance(event_data, dict):
+                return _Fail(error="Invalid payload: 'put-object' data must be an object")
+            if not isinstance(event_data.get("kind"), str):
+                return _Fail(error="Invalid payload: 'put-object' kind must be a string")
+
+            if has_unsupported_kind(event_data):
+                log.debug("Ignoring put-object with unsupported kind: %s", event_data["kind"])
+                continue
+
             try:
-                put = PutObject.from_dict(event["data"])
+                put = PutObject.from_dict(event_data)
             except ValueError as err:
                 return _Fail(error="Invalid JSON in put object", exception=err)
 
             builder.add_put(put.kind, put.key, put.version, put.object)
         elif event["event"] == EventName.DELETE_OBJECT:
+            event_data = event.get("data")
+            if not isinstance(event_data, dict):
+                return _Fail(error="Invalid payload: 'delete-object' data must be an object")
+            if not isinstance(event_data.get("kind"), str):
+                return _Fail(error="Invalid payload: 'delete-object' kind must be a string")
+
+            if has_unsupported_kind(event_data):
+                log.debug("Ignoring delete-object with unsupported kind: %s", event_data["kind"])
+                continue
+
             try:
-                delete_object = DeleteObject.from_dict(event["data"])
+                delete_object = DeleteObject.from_dict(event_data)
             except ValueError as err:
                 return _Fail(error="Invalid JSON in delete object", exception=err)
 

ldclient/impl/datasourcev2/streaming_common.py

@@ -16,7 +16,8 @@
     Error,
     EventName,
     Goodbye,
-    PutObject
+    PutObject,
+    has_unsupported_kind
 )
 from ldclient.impl.util import (
     _LD_ENVID_HEADER,
@@ -65,12 +66,32 @@ def process_message(
         return None
 
     if msg.event == EventName.PUT_OBJECT:
-        put = PutObject.from_dict(json.loads(msg.data))
+        data = json.loads(msg.data)
+        if not isinstance(data, dict):
+            raise ValueError("put-object data must be an object")
+        if not isinstance(data.get("kind"), str):
+            raise ValueError("put-object kind must be a string")
+
+        if has_unsupported_kind(data):
+            log.debug("Ignoring put-object with unsupported kind: %s", data["kind"])
+            return None
+
+        put = PutObject.from_dict(data)
         change_set_builder.add_put(put.kind, put.key, put.version, put.object)
         return None
 
     if msg.event == EventName.DELETE_OBJECT:
-        delete = DeleteObject.from_dict(json.loads(msg.data))
+        data = json.loads(msg.data)
+        if not isinstance(data, dict):
+            raise ValueError("delete-object data must be an object")
+        if not isinstance(data.get("kind"), str):
+            raise ValueError("delete-object kind must be a string")
+
+        if has_unsupported_kind(data):
+            log.debug("Ignoring delete-object with unsupported kind: %s", data["kind"])
+            return None
+
+        delete = DeleteObject.from_dict(data)
         change_set_builder.add_delete(delete.kind, delete.key, delete.version)
         return None
 

ldclient/impl/datasystem/protocolv2.py

@@ -7,6 +7,19 @@
 
 from ldclient.interfaces import EventName, ObjectKind
 
+_SUPPORTED_KINDS = tuple(kind.value for kind in ObjectKind)
+
+
+def has_unsupported_kind(data: dict) -> bool:
+    """
+    Returns True if a put-object or delete-object payload names a kind this
+    SDK does not support. Such objects should be ignored so that new kinds can
+    be added to the protocol without breaking older SDKs.
+
+    The caller must first check that the payload's kind is a string.
+    """
+    return data["kind"] not in _SUPPORTED_KINDS
+
 
 @dataclass(frozen=True)
 class DeleteObject:

ldclient/testing/impl/datasourcev2/test_polling_payload_parsing.py

@@ -1,5 +1,7 @@
 import json
 
+import pytest
+
 from ldclient.impl.datasourcev2.polling import (
     fdv1_polling_payload_to_changeset,
     polling_payload_to_changeset
@@ -110,20 +112,69 @@ def test_processes_delete_object():
     assert change_set.selector.version == 461
 
 
+def test_ignores_objects_with_unknown_kinds():
+    data = {
+        "events": [
+            {"event": "server-intent", "data": {"payloads": [{"id": "p", "target": 1, "intentCode": "xfer-full", "reason": "payload-missing"}]}},
+            {"event": "put-object", "data": {"version": 1, "kind": "some-future-kind", "key": "unknown-key", "object": {"foo": 1}}},
+            {"event": "put-object", "data": {"version": 1, "kind": "flag", "key": "flag-key", "object": {"key": "flag-key", "version": 1}}},
+            {"event": "delete-object", "data": {"version": 2, "kind": "some-future-kind", "key": "unknown-key"}},
+            {"event": "delete-object", "data": {"version": 3, "kind": "segment", "key": "segment-key"}},
+            {"event": "payload-transferred", "data": {"state": "(p:p:1)", "version": 1}},
+        ]
+    }
+    # Parse a payload that mixes unknown kinds with known flag and segment changes.
+    result = polling_payload_to_changeset(data)
+
+    # The payload completes with only the known changes.
+    assert isinstance(result, _Success)
+
+    change_set = result.value
+    assert [(c.action, c.kind, c.key) for c in change_set.changes] == [
+        (ChangeType.PUT, ObjectKind.FLAG, "flag-key"),
+        (ChangeType.DELETE, ObjectKind.SEGMENT, "segment-key"),
+    ]
+    assert change_set.selector.state == "(p:p:1)"
+
+
 def test_handles_invalid_put_object():
-    payload_str = '{"events":[ {"event":"server-intent","data":{"payloads":[ {"id":"5A46PZ79FQ9D08YYKT79DECDNV","target":461,"intentCode":"xfer-full","reason":"payload-missing"}]}},{"event": "put-object","data": {}},{"event":"payload-transferred","data":{"state":"(p:5A46PZ79FQ9D08YYKT79DECDNV:461)","id":"5A46PZ79FQ9D08YYKT79DECDNV","version":461}}]}'
+    payload_str = '{"events":[ {"event":"server-intent","data":{"payloads":[ {"id":"5A46PZ79FQ9D08YYKT79DECDNV","target":461,"intentCode":"xfer-full","reason":"payload-missing"}]}},{"event": "put-object","data": {"kind": "flag"}},{"event":"payload-transferred","data":{"state":"(p:5A46PZ79FQ9D08YYKT79DECDNV:461)","id":"5A46PZ79FQ9D08YYKT79DECDNV","version":461}}]}'
     result = polling_payload_to_changeset(json.loads(payload_str))
     assert isinstance(result, _Fail)
     assert result.error == "Invalid JSON in put object"
 
 
 def test_handles_invalid_delete_object():
-    payload_str = '{"events":[ {"event":"server-intent","data":{"payloads":[ {"id":"5A46PZ79FQ9D08YYKT79DECDNV","target":461,"intentCode":"xfer-full","reason":"payload-missing"}]}},{"event": "delete-object","data": {}},{"event":"payload-transferred","data":{"state":"(p:5A46PZ79FQ9D08YYKT79DECDNV:461)","id":"5A46PZ79FQ9D08YYKT79DECDNV","version":461}}]}'
+    payload_str = '{"events":[ {"event":"server-intent","data":{"payloads":[ {"id":"5A46PZ79FQ9D08YYKT79DECDNV","target":461,"intentCode":"xfer-full","reason":"payload-missing"}]}},{"event": "delete-object","data": {"kind": "flag"}},{"event":"payload-transferred","data":{"state":"(p:5A46PZ79FQ9D08YYKT79DECDNV:461)","id":"5A46PZ79FQ9D08YYKT79DECDNV","version":461}}]}'
     result = polling_payload_to_changeset(json.loads(payload_str))
     assert isinstance(result, _Fail)
     assert result.error == "Invalid JSON in delete object"
 
 
+_SERVER_INTENT = {"event": "server-intent", "data": {"payloads": [{"id": "p", "target": 1, "intentCode": "xfer-full", "reason": "payload-missing"}]}}
+_PAYLOAD_TRANSFERRED = {"event": "payload-transferred", "data": {"state": "(p:p:1)", "version": 1}}
+
+
+@pytest.mark.parametrize("event_name", ["put-object", "delete-object"])
+@pytest.mark.parametrize("data_field", [{}, {"data": None}, {"data": "x"}, {"data": [1]}, {"data": 5}])
+def test_handles_objects_whose_data_is_not_an_object(event_name, data_field):
+    data = {"events": [_SERVER_INTENT, {"event": event_name, **data_field}, _PAYLOAD_TRANSFERRED]}
+    result = polling_payload_to_changeset(data)
+    assert isinstance(result, _Fail)
+    assert result.error == f"Invalid payload: '{event_name}' data must be an object"
+
+
+@pytest.mark.parametrize("event_name", ["put-object", "delete-object"])
+@pytest.mark.parametrize("kind_field", [{}, {"kind": None}, {"kind": 5}, {"kind": ["flag"]}])
+def test_handles_objects_whose_kind_is_not_a_string(event_name, kind_field):
+    object_field = {"object": {}} if event_name == "put-object" else {}
+    event_data = {"version": 1, "key": "flag-key", **object_field, **kind_field}
+    data = {"events": [_SERVER_INTENT, {"event": event_name, "data": event_data}, _PAYLOAD_TRANSFERRED]}
+    result = polling_payload_to_changeset(data)
+    assert isinstance(result, _Fail)
+    assert result.error == f"Invalid payload: '{event_name}' kind must be a string"
+
+
 def test_handles_invalid_payload_transferred():
     payload_str = '{"events":[ {"event":"server-intent","data":{"payloads":[ {"id":"5A46PZ79FQ9D08YYKT79DECDNV","target":461,"intentCode":"xfer-full","reason":"payload-missing"}]}},{"event":"payload-transferred","data":{}}]}'
     result = polling_payload_to_changeset(json.loads(payload_str))

ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py

@@ -295,6 +295,78 @@ def test_handles_delete_objects(events):  # pylint: disable=redefined-outer-name
     assert updates[0].change_set.intent_code == IntentCode.TRANSFER_FULL
 
 
+def test_ignores_objects_with_unknown_kinds(events):  # pylint: disable=redefined-outer-name
+    unknown_put_event = Event(
+        event=EventName.PUT_OBJECT,
+        data=json.dumps({"version": 1, "kind": "some-future-kind", "key": "unknown-key", "object": {"foo": 1}}),
+    )
+    unknown_delete_event = Event(
+        event=EventName.DELETE_OBJECT,
+        data=json.dumps({"version": 2, "kind": "some-future-kind", "key": "unknown-key"}),
+    )
+    builder = list_sse_client(
+        [
+            events[EventName.SERVER_INTENT],
+            unknown_put_event,
+            events[EventName.PUT_OBJECT],
+            unknown_delete_event,
+            events[EventName.DELETE_OBJECT],
+            events[EventName.PAYLOAD_TRANSFERRED],
+        ]
+    )
+
+    synchronizer = make_streaming_data_source()
+    synchronizer._sse_client_builder = builder
+    # Stream a payload that mixes unknown kinds with known flag changes.
+    updates = list(synchronizer.sync(MockSelectorStore(Selector.no_selector())))
+
+    # The payload completes with only the known changes and no error.
+    assert len(updates) == 1
+    assert updates[0].state == DataSourceState.VALID
+    assert updates[0].error is None
+
+    assert updates[0].change_set is not None
+    assert [(c.action, c.kind, c.key) for c in updates[0].change_set.changes] == [
+        (ChangeType.PUT, ObjectKind.FLAG, "flag-key"),
+        (ChangeType.DELETE, ObjectKind.FLAG, "flag-key"),
+    ]
+    assert updates[0].change_set.selector.state == "p:SOMETHING:300"
+
+
+@pytest.mark.parametrize("event_name", [EventName.PUT_OBJECT, EventName.DELETE_OBJECT])
+@pytest.mark.parametrize(
+    "data, problem",
+    [
+        ('"x"', "data must be an object"),
+        ("[1]", "data must be an object"),
+        ("null", "data must be an object"),
+        ("5", "data must be an object"),
+        ('{"version": 1, "key": "flag-key", "object": {}}', "kind must be a string"),
+        ('{"version": 1, "kind": null, "key": "flag-key", "object": {}}', "kind must be a string"),
+        ('{"version": 1, "kind": 5, "key": "flag-key", "object": {}}', "kind must be a string"),
+        ('{"version": 1, "kind": ["flag"], "key": "flag-key", "object": {}}', "kind must be a string"),
+    ],
+)
+def test_rejects_malformed_objects(events, event_name, data, problem):  # pylint: disable=redefined-outer-name
+    builder = list_sse_client(
+        [
+            events[EventName.SERVER_INTENT],
+            Event(event=event_name, data=data),
+        ]
+    )
+
+    synchronizer = make_streaming_data_source()
+    synchronizer._sse_client_builder = builder
+    updates = list(synchronizer.sync(MockSelectorStore(Selector.no_selector())))
+
+    assert len(updates) == 1
+    assert updates[0].state == DataSourceState.INTERRUPTED
+    assert updates[0].change_set is None
+    assert updates[0].error is not None
+    assert updates[0].error.kind == DataSourceErrorKind.UNKNOWN
+    assert updates[0].error.message == f"{event_name.value} {problem}"
+
+
 def test_swallows_goodbye(events):  # pylint: disable=redefined-outer-name
     builder = list_sse_client(
         [