From 74aab5611e3f2de8ff4b939bab51ec444bc6858f Mon Sep 17 00:00:00 2001 From: Sylvester Kaczmarek <16242628+sylvesterkaczmarek@users.noreply.github.com> Date: Sun, 16 Aug 2026 10:00:28 +0000 Subject: [PATCH] fix(agent): reject invalid telemetry injection payloads --- backend/routers/ai_intel.py | 42 ++------ backend/services/ai_intel_store.py | 7 ++ backend/services/openclaw_channel.py | 4 +- .../test_ai_intel_store_copy_on_write.py | 54 +++++----- backend/tests/test_injection_entrypoints.py | 102 ++++++++++++++++++ 5 files changed, 150 insertions(+), 59 deletions(-) create mode 100644 backend/tests/test_injection_entrypoints.py diff --git a/backend/routers/ai_intel.py b/backend/routers/ai_intel.py index 0220249..1a56b42 100644 --- a/backend/routers/ai_intel.py +++ b/backend/routers/ai_intel.py @@ -125,8 +125,8 @@ class LayerUpdate(BaseModel): class InjectRequest(BaseModel): layer: str - items: list[dict[str, Any]] = Field(..., max_length=200) - mode: str = "append" # "append" or "replace" + items: list[Any] = Field(..., max_length=200) + mode: str = "append" # validated by inject_layer_data # --------------------------------------------------------------------------- @@ -641,39 +641,15 @@ INJECTABLE_LAYERS = { @router.post("/api/ai/inject", dependencies=[Depends(require_openclaw_or_local)]) @limiter.limit("30/minute") async def inject_data(request: Request, body: InjectRequest): - """Inject custom data into ANY native ShadowBroker layer. - Items appear as real telemetry alongside automated feeds. - Tagged with _source='user:openclaw' so they can be filtered/removed.""" - from services.fetchers._store import latest_data, _data_lock, bump_data_version + """Inject custom data through the shared validated layer helper.""" + from services.ai_intel_store import inject_layer_data - if body.layer not in INJECTABLE_LAYERS: - raise HTTPException(400, f"Layer '{body.layer}' is not injectable. " - f"Valid layers: {sorted(INJECTABLE_LAYERS)}") + result = inject_layer_data(body.layer, body.items, mode=body.mode) + if not result.get("ok"): + raise HTTPException(400, result.get("detail", "invalid injection payload")) - now = time.time() - items = body.items[:200] # cap at 200 - - # Tag every injected item - for item in items: - item["_injected"] = True - item["_source"] = "user:openclaw" - item["_injected_at"] = now - - with _data_lock: - existing = list(latest_data.get(body.layer) or []) - if body.mode == "replace": - existing = [x for x in existing if not x.get("_injected")] - existing.extend(items) - latest_data[body.layer] = existing - bump_data_version() - - total = len(latest_data.get(body.layer, [])) - return { - "ok": True, - "layer": body.layer, - "injected": len(items), - "total": total, - } + total = len(_latest_data.get(body.layer, [])) + return {**result, "total": total} @router.delete("/api/ai/inject", dependencies=[Depends(require_openclaw_or_local)]) diff --git a/backend/services/ai_intel_store.py b/backend/services/ai_intel_store.py index 1006c05..a479bc3 100644 --- a/backend/services/ai_intel_store.py +++ b/backend/services/ai_intel_store.py @@ -121,6 +121,10 @@ def inject_layer_data( if layer not in _INJECTABLE_LAYERS: return {"ok": False, "detail": f"layer '{layer}' not injectable"} + mode = str(mode or "").strip().lower() + if mode not in {"append", "replace"}: + return {"ok": False, "detail": "mode must be 'append' or 'replace'"} + items = list(items or [])[:200] if not items: return {"ok": False, "detail": "no items provided"} @@ -136,6 +140,9 @@ def inject_layer_data( entry["_injected_at"] = now tagged.append(entry) + if not tagged: + return {"ok": False, "detail": "no valid items provided"} + with _data_lock: current = latest_data.get(layer) existing = list(current) if isinstance(current, list) else [] diff --git a/backend/services/openclaw_channel.py b/backend/services/openclaw_channel.py index e723664..6e8ca1c 100644 --- a/backend/services/openclaw_channel.py +++ b/backend/services/openclaw_channel.py @@ -1362,7 +1362,9 @@ def _dispatch_command(cmd: str, args: dict[str, Any]) -> dict[str, Any]: if not layer or not items: return {"ok": False, "detail": "layer and items required"} from services.ai_intel_store import inject_layer_data - result = inject_layer_data(layer, items) + result = inject_layer_data(layer, items, mode=args.get("mode", "append")) + if not result.get("ok"): + return result return {"ok": True, "data": result} if cmd == "create_layer": diff --git a/backend/tests/test_ai_intel_store_copy_on_write.py b/backend/tests/test_ai_intel_store_copy_on_write.py index 8c828e6..76ee168 100644 --- a/backend/tests/test_ai_intel_store_copy_on_write.py +++ b/backend/tests/test_ai_intel_store_copy_on_write.py @@ -1,4 +1,4 @@ -"""Regression coverage for copy-on-write OpenClaw layer injection.""" +"""Regression coverage for OpenClaw layer injection.""" from services import ai_intel_store from services.fetchers import _store @@ -13,47 +13,51 @@ def _publish_test_layer(monkeypatch, items): def test_append_does_not_mutate_previously_published_list(monkeypatch): before = _publish_test_layer(monkeypatch, [{"id": "existing"}]) - result = ai_intel_store.inject_layer_data( - "air_quality", - [{"id": "injected"}], - mode="append", + "air_quality", [{"id": "injected"}], mode="append" ) - after = _store.latest_data["air_quality"] - assert result == { - "ok": True, - "layer": "air_quality", - "injected": 1, - "mode": "append", - } + assert result["ok"] is True assert after is not before assert before == [{"id": "existing"}] assert [item["id"] for item in after] == ["existing", "injected"] - assert after[-1]["_injected"] is True - assert after[-1]["_source"] == "user:openclaw" def test_replace_does_not_mutate_previously_published_list(monkeypatch): before = _publish_test_layer( monkeypatch, - [ - {"id": "native"}, - {"id": "old-injected", "_injected": True}, - ], + [{"id": "native"}, {"id": "old-injected", "_injected": True}], ) - result = ai_intel_store.inject_layer_data( - "air_quality", - [{"id": "new-injected"}], - mode="replace", + "air_quality", [{"id": "new-injected"}], mode="replace" ) - after = _store.latest_data["air_quality"] assert result["ok"] is True assert after is not before assert before == [ - {"id": "native"}, - {"id": "old-injected", "_injected": True}, + {"id": "native"}, {"id": "old-injected", "_injected": True} ] assert [item["id"] for item in after] == ["native", "new-injected"] + + +def test_replace_rejects_invalid_items_without_clearing_existing_data(monkeypatch): + before = _publish_test_layer( + monkeypatch, + [{"id": "native"}, {"id": "old-injected", "_injected": True}], + ) + result = ai_intel_store.inject_layer_data( + "air_quality", ["not-a-mapping"], mode="replace" + ) + assert result == {"ok": False, "detail": "no valid items provided"} + assert _store.latest_data["air_quality"] is before + + +def test_unknown_mode_is_rejected_without_mutating_layer(monkeypatch): + before = _publish_test_layer(monkeypatch, [{"id": "existing"}]) + result = ai_intel_store.inject_layer_data( + "air_quality", [{"id": "injected"}], mode="overwrite" + ) + assert result == { + "ok": False, "detail": "mode must be 'append' or 'replace'" + } + assert _store.latest_data["air_quality"] is before diff --git a/backend/tests/test_injection_entrypoints.py b/backend/tests/test_injection_entrypoints.py new file mode 100644 index 0000000..0a0e71b --- /dev/null +++ b/backend/tests/test_injection_entrypoints.py @@ -0,0 +1,102 @@ +"""Regression coverage for public and OpenClaw telemetry injection paths.""" + +import pytest +from fastapi import HTTPException + +from routers import ai_intel +from services.fetchers import _store +from services.openclaw_channel import _dispatch_command + + +def _publish(monkeypatch, items): + published = list(items) + monkeypatch.setitem(_store.latest_data, "air_quality", published) + monkeypatch.setattr(_store, "bump_data_version", lambda: None) + return published + + +def test_openclaw_rejects_unknown_mode_without_mutation(monkeypatch): + before = _publish(monkeypatch, [{"id": "existing"}]) + result = _dispatch_command( + "inject_data", + {"layer": "air_quality", "items": [{"id": "new"}], "mode": "overwrite"}, + ) + assert result == {"ok": False, "detail": "mode must be 'append' or 'replace'"} + assert _store.latest_data["air_quality"] is before + + +def test_openclaw_all_invalid_replace_preserves_existing(monkeypatch): + before = _publish( + monkeypatch, + [{"id": "native"}, {"id": "old", "_injected": True}], + ) + result = _dispatch_command( + "inject_data", + {"layer": "air_quality", "items": ["invalid"], "mode": "replace"}, + ) + assert result == {"ok": False, "detail": "no valid items provided"} + assert _store.latest_data["air_quality"] is before + + +def test_openclaw_valid_replace_preserves_native_non_mapping_entries(monkeypatch): + _publish( + monkeypatch, + ["native-marker", {"id": "native"}, {"id": "old", "_injected": True}], + ) + result = _dispatch_command( + "inject_data", + {"layer": "air_quality", "items": [{"id": "new"}], "mode": "replace"}, + ) + assert result["ok"] is True + after = _store.latest_data["air_quality"] + assert after[0] == "native-marker" + assert after[1] == {"id": "native"} + assert after[2]["id"] == "new" + + +@pytest.mark.asyncio +async def test_rest_rejects_unknown_mode_without_mutation(monkeypatch): + before = _publish(monkeypatch, [{"id": "existing"}]) + body = ai_intel.InjectRequest( + layer="air_quality", items=[{"id": "new"}], mode="overwrite" + ) + with pytest.raises(HTTPException) as exc_info: + await ai_intel.inject_data(None, body) + assert exc_info.value.status_code == 400 + assert _store.latest_data["air_quality"] is before + + +@pytest.mark.asyncio +async def test_rest_empty_and_all_invalid_replace_preserve_existing(monkeypatch): + for invalid_items in ([], ["invalid"]): + before = _publish( + monkeypatch, + [{"id": "native"}, {"id": "old", "_injected": True}], + ) + body = ai_intel.InjectRequest( + layer="air_quality", items=invalid_items, mode="replace" + ) + with pytest.raises(HTTPException) as exc_info: + await ai_intel.inject_data(None, body) + assert exc_info.value.status_code == 400 + assert _store.latest_data["air_quality"] is before + + +@pytest.mark.asyncio +async def test_rest_valid_replace_uses_shared_helper(monkeypatch): + _publish( + monkeypatch, + ["native-marker", {"id": "native"}, {"id": "old", "_injected": True}], + ) + body = ai_intel.InjectRequest( + layer="air_quality", items=[{"id": "new"}], mode="replace" + ) + result = await ai_intel.inject_data(None, body) + assert result["ok"] is True + assert result["mode"] == "replace" + assert result["injected"] == 1 + assert result["total"] == 3 + after = _store.latest_data["air_quality"] + assert after[0] == "native-marker" + assert after[1] == {"id": "native"} + assert after[2]["id"] == "new"