From fe487e53a2bfed6ccc177788a7f082f22a4878d9 Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Fri, 16 Oct 2026 21:59:17 +0000 Subject: [PATCH 1/2] fix: Ignore FDv2 objects with unrecognized kinds put-object and delete-object events whose kind the SDK does not support are now dropped before the object is parsed, in both the streaming and polling FDv2 data sources. Previously the strict ObjectKind conversion raised, which interrupted the stream or discarded the whole polling payload. Known objects in the same payload are still applied. --- ldclient/impl/datasourcev2/polling_common.py | 11 +++++- .../impl/datasourcev2/streaming_common.py | 17 +++++++-- ldclient/impl/datasystem/protocolv2.py | 12 ++++++ .../test_polling_payload_parsing.py | 25 ++++++++++++ .../test_streaming_synchronizer.py | 38 +++++++++++++++++++ 5 files changed, 99 insertions(+), 4 deletions(-) diff --git a/ldclient/impl/datasourcev2/polling_common.py b/ldclient/impl/datasourcev2/polling_common.py index 532efd9b..b7f9bc28 100644 --- a/ldclient/impl/datasourcev2/polling_common.py +++ b/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,6 +69,10 @@ def polling_payload_to_changeset(data: dict) -> _Result[ChangeSet, str]: builder.start(server_intent.payload.code) elif event["event"] == EventName.PUT_OBJECT: + if has_unsupported_kind(event["data"]): + log.debug("Ignoring put-object with unsupported kind: %s", event["data"].get("kind")) + continue + try: put = PutObject.from_dict(event["data"]) except ValueError as err: @@ -75,6 +80,10 @@ def polling_payload_to_changeset(data: dict) -> _Result[ChangeSet, str]: builder.add_put(put.kind, put.key, put.version, put.object) elif event["event"] == EventName.DELETE_OBJECT: + if has_unsupported_kind(event["data"]): + log.debug("Ignoring delete-object with unsupported kind: %s", event["data"].get("kind")) + continue + try: delete_object = DeleteObject.from_dict(event["data"]) except ValueError as err: diff --git a/ldclient/impl/datasourcev2/streaming_common.py b/ldclient/impl/datasourcev2/streaming_common.py index cb0428d8..903d5438 100644 --- a/ldclient/impl/datasourcev2/streaming_common.py +++ b/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,22 @@ 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 has_unsupported_kind(data): + log.debug("Ignoring put-object with unsupported kind: %s", data.get("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 has_unsupported_kind(data): + log.debug("Ignoring delete-object with unsupported kind: %s", data.get("kind")) + return None + + delete = DeleteObject.from_dict(data) change_set_builder.add_delete(delete.kind, delete.key, delete.version) return None diff --git a/ldclient/impl/datasystem/protocolv2.py b/ldclient/impl/datasystem/protocolv2.py index 7741a216..ce2e1411 100644 --- a/ldclient/impl/datasystem/protocolv2.py +++ b/ldclient/impl/datasystem/protocolv2.py @@ -7,6 +7,18 @@ 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. + """ + kind = data.get("kind") + return kind is not None and kind not in _SUPPORTED_KINDS + @dataclass(frozen=True) class DeleteObject: diff --git a/ldclient/testing/impl/datasourcev2/test_polling_payload_parsing.py b/ldclient/testing/impl/datasourcev2/test_polling_payload_parsing.py index 8e4c6369..5f9d1676 100644 --- a/ldclient/testing/impl/datasourcev2/test_polling_payload_parsing.py +++ b/ldclient/testing/impl/datasourcev2/test_polling_payload_parsing.py @@ -110,6 +110,31 @@ 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}}]}' result = polling_payload_to_changeset(json.loads(payload_str)) diff --git a/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py b/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py index 216013b4..1c1fab81 100644 --- a/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py +++ b/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py @@ -295,6 +295,44 @@ 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" + + def test_swallows_goodbye(events): # pylint: disable=redefined-outer-name builder = list_sse_client( [ From 31a935ade318f138062e5d8fef798c2def2f4282 Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Thu, 29 Oct 2026 17:26:08 +0000 Subject: [PATCH 2/2] fix: Reject put and delete objects whose data is not an object or whose kind is not a string These are malformed and now fail before the kind is checked, rather than raising from the kind check or being ignored as an unsupported kind. --- ldclient/impl/datasourcev2/polling_common.py | 24 +++++++++---- .../impl/datasourcev2/streaming_common.py | 14 ++++++-- ldclient/impl/datasystem/protocolv2.py | 5 +-- .../test_polling_payload_parsing.py | 30 ++++++++++++++-- .../test_streaming_synchronizer.py | 34 +++++++++++++++++++ 5 files changed, 95 insertions(+), 12 deletions(-) diff --git a/ldclient/impl/datasourcev2/polling_common.py b/ldclient/impl/datasourcev2/polling_common.py index b7f9bc28..5abf8cd4 100644 --- a/ldclient/impl/datasourcev2/polling_common.py +++ b/ldclient/impl/datasourcev2/polling_common.py @@ -69,23 +69,35 @@ def polling_payload_to_changeset(data: dict) -> _Result[ChangeSet, str]: builder.start(server_intent.payload.code) elif event["event"] == EventName.PUT_OBJECT: - if has_unsupported_kind(event["data"]): - log.debug("Ignoring put-object with unsupported kind: %s", event["data"].get("kind")) + 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: - if has_unsupported_kind(event["data"]): - log.debug("Ignoring delete-object with unsupported kind: %s", event["data"].get("kind")) + 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) diff --git a/ldclient/impl/datasourcev2/streaming_common.py b/ldclient/impl/datasourcev2/streaming_common.py index 903d5438..59b50702 100644 --- a/ldclient/impl/datasourcev2/streaming_common.py +++ b/ldclient/impl/datasourcev2/streaming_common.py @@ -67,8 +67,13 @@ def process_message( if msg.event == EventName.PUT_OBJECT: 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.get("kind")) + log.debug("Ignoring put-object with unsupported kind: %s", data["kind"]) return None put = PutObject.from_dict(data) @@ -77,8 +82,13 @@ def process_message( if msg.event == EventName.DELETE_OBJECT: 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.get("kind")) + log.debug("Ignoring delete-object with unsupported kind: %s", data["kind"]) return None delete = DeleteObject.from_dict(data) diff --git a/ldclient/impl/datasystem/protocolv2.py b/ldclient/impl/datasystem/protocolv2.py index ce2e1411..d7a4a39c 100644 --- a/ldclient/impl/datasystem/protocolv2.py +++ b/ldclient/impl/datasystem/protocolv2.py @@ -15,9 +15,10 @@ 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. """ - kind = data.get("kind") - return kind is not None and kind not in _SUPPORTED_KINDS + return data["kind"] not in _SUPPORTED_KINDS @dataclass(frozen=True) diff --git a/ldclient/testing/impl/datasourcev2/test_polling_payload_parsing.py b/ldclient/testing/impl/datasourcev2/test_polling_payload_parsing.py index 5f9d1676..1d21d589 100644 --- a/ldclient/testing/impl/datasourcev2/test_polling_payload_parsing.py +++ b/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 @@ -136,19 +138,43 @@ def test_ignores_objects_with_unknown_kinds(): 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)) diff --git a/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py b/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py index 1c1fab81..afea1186 100644 --- a/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py +++ b/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py @@ -333,6 +333,40 @@ def test_ignores_objects_with_unknown_kinds(events): # pylint: disable=redefine 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( [