Add OpenWebUI stability and privacy guards

dev/test_openwebui_filters.py

@@ -0,0 +1,171 @@
+#!/usr/bin/env python3
+"""Offline tests for the versioned OpenWebUI filters.
+
+The production container provides pydantic.  A tiny local stand-in keeps these
+logic tests dependency-free and prevents test setup from reaching the network.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import sys
+import tempfile
+import types
+import unittest
+from pathlib import Path
+
+
+class _BaseModel:
+    def __init__(self, **values):
+        annotations = {}
+        for base in reversed(type(self).__mro__):
+            annotations.update(getattr(base, "__annotations__", {}))
+        for name in annotations:
+            setattr(self, name, values.get(name, getattr(type(self), name, None)))
+
+
+fake_pydantic = types.ModuleType("pydantic")
+fake_pydantic.BaseModel = _BaseModel
+fake_pydantic.Field = lambda default=None, **kwargs: default
+sys.modules.setdefault("pydantic", fake_pydantic)
+
+FILTER_DIR = Path(__file__).parents[1] / "platform" / "openwebui" / "filters"
+
+
+def _load(name: str):
+    spec = importlib.util.spec_from_file_location(name, FILTER_DIR / f"{name}.py")
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+class StabilityGuardTests(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.module = _load("stability_guard")
+        self.guard = self.module.Filter()
+
+    async def test_large_tool_output_is_bounded(self):
+        body = {
+            "model": "qwen-fast",
+            "messages": [
+                {"role": "user", "content": "Prüfe das Log."},
+                {"role": "tool", "tool_call_id": "x", "content": "A" * 50000},
+            ],
+        }
+        result = await self.guard.inlet(body)
+        content = result["messages"][1]["content"]
+        self.assertLessEqual(len(content), self.guard.valves.max_single_tool_chars)
+        self.assertIn("Werkzeugausgabe gekürzt", content)
+
+    async def test_duplicate_calls_disable_tools(self):
+        call = {
+            "id": "call",
+            "type": "function",
+            "function": {"name": "search", "arguments": '{"q":"same"}'},
+        }
+        body = {
+            "model": "qwen-fast",
+            "tools": [{"type": "function", "function": {"name": "search"}}],
+            "tool_ids": ["server:mcp:web"],
+            "messages": [
+                {"role": "user", "content": "Suche genau einmal."},
+                {"role": "assistant", "tool_calls": [call]},
+                {"role": "tool", "tool_call_id": "1", "content": "nichts"},
+                {"role": "assistant", "tool_calls": [call]},
+                {"role": "tool", "tool_call_id": "2", "content": "nichts"},
+                {"role": "assistant", "tool_calls": [call]},
+            ],
+        }
+        result = await self.guard.inlet(body)
+        self.assertEqual(result["tools"], [])
+        self.assertEqual(result["tool_ids"], [])
+        self.assertIn("Weitere Werkzeugaufrufe", result["messages"][0]["content"])
+
+    async def test_old_context_is_compacted_before_current_turn(self):
+        self.guard.valves.default_context_tokens = 10000
+        self.guard.valves.hard_context_ratio = 0.8
+        self.guard.valves.reserved_output_tokens = 1000
+        body = {
+            "model": "unknown",
+            "messages": [
+                {"role": "system", "content": "Sicher arbeiten."},
+                {"role": "user", "content": "alt " * 15000},
+                {"role": "assistant", "content": "altantwort " * 8000},
+                {"role": "tool", "tool_call_id": "old", "content": "log " * 20000},
+                {"role": "user", "content": "Aktuelle wichtige Frage"},
+            ],
+        }
+        result = await self.guard.inlet(body)
+        self.assertEqual(result["messages"][-1]["content"], "Aktuelle wichtige Frage")
+        self.assertLess(len(result["messages"][1]["content"]), 3000)
+
+
+class MetricsTests(unittest.IsolatedAsyncioTestCase):
+    async def test_metrics_file_contains_no_chat_content_or_ids(self):
+        module = _load("local_performance_metrics")
+        metrics = module.Filter()
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "metrics.jsonl"
+            metrics.valves.metrics_path = str(path)
+            metadata = {"message_id": "secret-message-id", "chat_id": "secret-chat-id"}
+            await metrics.inlet(
+                {
+                    "model": "qwen-fast",
+                    "messages": [{"role": "user", "content": "private prompt"}],
+                    "tools": [{"name": "tool"}],
+                },
+                __metadata__=metadata,
+            )
+            await metrics.outlet(
+                {
+                    "model": "qwen-fast",
+                    "messages": [
+                        {
+                            "role": "assistant",
+                            "content": "private answer",
+                            "usage": {
+                                "prompt_tokens": 10,
+                                "completion_tokens": 4,
+                                "total_tokens": 14,
+                            },
+                        }
+                    ],
+                },
+                __metadata__=metadata,
+            )
+            raw = path.read_text()
+            record = json.loads(raw)
+            self.assertEqual(record["prompt_tokens"], 10)
+            self.assertNotIn("private", raw)
+            self.assertNotIn("secret", raw)
+
+
+class SecretRedactionTests(unittest.IsolatedAsyncioTestCase):
+    async def test_only_tool_and_assistant_content_is_redacted(self):
+        module = _load("secret_redaction")
+        guard = module.Filter()
+        token = "eyJ" + "A" * 24 + "." + "B" * 24 + "." + "C" * 16
+        body = {
+            "messages": [
+                {"role": "user", "content": f"Absichtlich lokal nutzen: {token}"},
+                {"role": "tool", "content": f'{{"api_key":"1234567890abcdef"}} {token}'},
+            ]
+        }
+        result = await guard.inlet(body)
+        self.assertIn(token, result["messages"][0]["content"])
+        self.assertNotIn(token, result["messages"][1]["content"])
+        self.assertNotIn("1234567890abcdef", result["messages"][1]["content"])
+
+        outlet = {
+            "messages": [
+                {"role": "assistant", "content": "Bearer " + "abcdefghijklmnopqrstuvwxyz"}
+            ]
+        }
+        result = await guard.outlet(outlet)
+        self.assertNotIn("abcdefghijklmnopqrstuvwxyz", result["messages"][0]["content"])
+
+
+if __name__ == "__main__":
+    unittest.main()