fix(presets): scope expand-prompt model resolution to owner (#3477)

* fix(presets): scope expand-prompt model resolution to owner

/api/presets/expand resolved its model endpoint with no owner, so in a
multi-user setup it could match another user's endpoint and use its URL
and decrypted api_key. Pass effective_user(request) to _resolve_model so
resolution is owner-scoped. Adds a regression test.

* fix(presets): scope teacher and audit model resolution to owner

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Alex Little <alexwilliamlittle@gmail.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: Kenny Van de Maele <kenny@kvandemaele.be>

tests/test_preset_expand_owner_scope.py

@@ -0,0 +1,86 @@
+"""Route-level owner-scope test for POST /api/presets/expand.
+
+`expand_character_prompt` resolves a model endpoint to run its LLM call. It must
+scope that lookup to the calling user, otherwise it can resolve another owner's
+ModelEndpoint (and its decrypted api_key) in a multi-user deployment. See #2283.
+"""
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+from routes.preset_routes import setup_preset_routes
+
+
+class _FakeRequest:
+    """Minimal stand-in: an async ``json()`` body plus a ``state`` namespace."""
+
+    def __init__(self, body, **state):
+        self._body = body
+        self.state = SimpleNamespace(**state)
+
+    async def json(self):
+        return self._body
+
+
+def _expand_endpoint():
+    router = setup_preset_routes(MagicMock())
+    for route in router.routes:
+        if getattr(route, "path", "") == "/api/presets/expand" and "POST" in getattr(route, "methods", set()):
+            return route.endpoint
+    raise AssertionError("POST /api/presets/expand route not registered")
+
+
+def _patch_model_pipeline(monkeypatch):
+    """Capture the owner passed to _resolve_model and stub the LLM call."""
+    seen = {}
+
+    def fake_resolve_model(spec, owner=None):
+        seen["spec"] = spec
+        seen["owner"] = owner
+        return ("http://endpoint.local/v1", "test-model", {})
+
+    async def fake_llm_call_async(url, model, messages, **kwargs):
+        return "  expanded prompt  "
+
+    monkeypatch.setattr("src.ai_interaction._resolve_model", fake_resolve_model)
+    monkeypatch.setattr("src.llm_core.llm_call_async", fake_llm_call_async)
+    return seen
+
+
+def test_expand_scopes_model_resolution_to_cookie_user(monkeypatch):
+    seen = _patch_model_pipeline(monkeypatch)
+    endpoint = _expand_endpoint()
+
+    req = _FakeRequest({"name": "Pirate", "prompt": "talks like a pirate", "model": "test-model"},
+                       current_user="alice")
+    result = asyncio.run(endpoint(req))
+
+    assert seen["owner"] == "alice"
+    assert seen["spec"] == "test-model"
+    assert result == {"success": True, "prompt": "expanded prompt"}
+
+
+def test_expand_attributes_bearer_token_to_its_owner(monkeypatch):
+    # effective_user (not get_current_user) resolves a bearer ody_ caller to the
+    # token's real owner instead of the sandbox "api" pseudo-user.
+    seen = _patch_model_pipeline(monkeypatch)
+    endpoint = _expand_endpoint()
+
+    req = _FakeRequest({"name": "Pirate", "model": ""},
+                       current_user="api", api_token=True, api_token_owner="bob")
+    asyncio.run(endpoint(req))
+
+    assert seen["owner"] == "bob"
+
+
+def test_expand_short_circuits_without_input(monkeypatch):
+    seen = _patch_model_pipeline(monkeypatch)
+    endpoint = _expand_endpoint()
+
+    req = _FakeRequest({}, current_user="alice")
+    result = asyncio.run(endpoint(req))
+
+    # Nothing to expand: no model resolution attempted.
+    assert result["success"] is False
+    assert "owner" not in seen

tests/test_teacher_audit_owner_scope.py

@@ -0,0 +1,64 @@
+"""Owner-scope tests for the remaining _resolve_model call sites.
+
+Both the teacher-escalation path and the skill-audit teacher resolution map a
+model spec to an endpoint (and its decrypted api_key). Like /presets/expand,
+that lookup must be scoped to the calling user, otherwise it can resolve another
+owner's ModelEndpoint in a multi-user deployment. See #2283.
+"""
+
+import asyncio
+
+import src.teacher_escalation as teacher_escalation
+import routes.skills_routes as skills_routes
+
+
+def test_call_teacher_scopes_model_resolution_to_owner(monkeypatch):
+    seen = {}
+
+    def fake_resolve_model(spec, owner=None):
+        seen["spec"] = spec
+        seen["owner"] = owner
+        return ("http://endpoint.local/v1", "teacher-model", {})
+
+    async def fake_llm_call_async(url, model, messages, **kwargs):
+        return "teacher reply"
+
+    monkeypatch.setattr("src.ai_interaction._resolve_model", fake_resolve_model)
+    monkeypatch.setattr("src.ai_interaction._TEACHER_SYSTEM_PROMPT", "sys", raising=False)
+    monkeypatch.setattr("src.llm_core.llm_call_async", fake_llm_call_async)
+
+    result = asyncio.run(
+        teacher_escalation._call_teacher("teacher-model", "prompt", owner="alice")
+    )
+
+    assert result == "teacher reply"
+    assert seen["owner"] == "alice"
+    assert seen["spec"] == "teacher-model"
+
+
+def test_audit_teacher_resolution_scoped_to_owner(monkeypatch):
+    seen = {}
+
+    def fake_resolve_endpoint(role, owner=None):
+        return ("http://worker.local/v1", "worker-model", {})
+
+    def fake_get_setting(key, default=None):
+        return {"teacher_enabled": True, "teacher_model": "teacher-model"}.get(key, default)
+
+    def fake_resolve_model(spec, owner=None):
+        seen["spec"] = spec
+        seen["owner"] = owner
+        return ("http://endpoint.local/v1", "teacher-model", {})
+
+    monkeypatch.setattr("src.endpoint_resolver.resolve_endpoint", fake_resolve_endpoint)
+    monkeypatch.setattr("src.settings.get_setting", fake_get_setting)
+    monkeypatch.setattr("src.ai_interaction._resolve_model", fake_resolve_model)
+    # list_model_ids is best-effort; force it to no-op so the worker model passes through.
+    monkeypatch.setattr("src.llm_core.list_model_ids", lambda url, headers=None: [])
+
+    url, model, headers, teacher = skills_routes._resolve_audit_models(owner="alice")
+
+    assert (url, model) == ("http://worker.local/v1", "worker-model")
+    assert teacher == ("http://endpoint.local/v1", "teacher-model", {})
+    assert seen["owner"] == "alice"
+    assert seen["spec"] == "teacher-model"