diff --git a/ldclient/impl/datasourcev2/polling_common.py b/ldclient/impl/datasourcev2/polling_common.py index 532efd9b..5abf8cd4 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,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) diff --git a/ldclient/impl/datasourcev2/streaming_common.py b/ldclient/impl/datasourcev2/streaming_common.py index cb0428d8..59b50702 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,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 diff --git a/ldclient/impl/datasystem/protocolv2.py b/ldclient/impl/datasystem/protocolv2.py index 7741a216..d7a4a39c 100644 --- a/ldclient/impl/datasystem/protocolv2.py +++ b/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: diff --git a/ldclient/testing/impl/datasourcev2/test_polling_payload_parsing.py b/ldclient/testing/impl/datasourcev2/test_polling_payload_parsing.py index 8e4c6369..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 @@ -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)) diff --git a/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py b/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py index 216013b4..afea1186 100644 --- a/ldclient/testing/impl/datasourcev2/test_streaming_synchronizer.py +++ b/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( [