check ownership before stopping a camera watch from chat

The stop_camera_watch chat tool cancelled the active VLM watch job without checking who started it, so any authenticated user could cancel another user's job by calling /chat/completion directly. The tool now uses the same admin-or-creator check as DELETE /vlm/monitor, and chat-started jobs record their creator so non-admins can still stop their own.

frigate/api/chat.py

@@ -45,6 +45,7 @@ from frigate.genai.prompts import (
 )
 from frigate.genai.utils import build_assistant_message_for_conversation
 from frigate.jobs.vlm_watch import (
+    VLMWatchJob,
     get_vlm_watch_job,
     start_vlm_watch_job,
     stop_vlm_watch_job,
@@ -765,7 +766,7 @@ async def _execute_tool_internal(
     elif tool_name == "start_camera_watch":
         return await _execute_start_camera_watch(request, arguments)
     elif tool_name == "stop_camera_watch":
-        return _execute_stop_camera_watch()
+        return _execute_stop_camera_watch(request)
     elif tool_name == "get_profile_status":
         return _execute_get_profile_status(request)
     elif tool_name == "get_recap":
@@ -819,6 +820,7 @@ async def _execute_start_camera_watch(
             dispatcher=request.app.dispatcher,
             labels=labels,
             zones=zones,
+            username=request.headers.get("remote-user", ""),
         )
     except RuntimeError as e:
         logger.exception("Failed to start VLM watch job: %s", e)
@@ -834,7 +836,18 @@ async def _execute_start_camera_watch(
     }
 
 
-def _execute_stop_camera_watch() -> dict[str, Any]:
+def _can_cancel_watch_job(request: Request, job: VLMWatchJob) -> bool:
+    """Admins can cancel any watch job; other users only jobs they started."""
+    role = request.headers.get("remote-role", "viewer")
+    username = request.headers.get("remote-user", "")
+    return role == "admin" or username == job.username
+
+
+def _execute_stop_camera_watch(request: Request) -> dict[str, Any]:
+    job = get_vlm_watch_job()
+    if job is not None and not _can_cancel_watch_job(request, job):
+        return {"success": False, "message": "Not authorized to cancel this watch job."}
+
     cancelled = stop_vlm_watch_job()
     if cancelled:
         return {"success": True, "message": "Watch job cancelled."}
@@ -1548,11 +1561,7 @@ async def cancel_vlm_monitor(request: Request) -> JSONResponse:
             status_code=404,
         )
 
-    role = request.headers.get("remote-role", "viewer")
-    username = request.headers.get("remote-user", "")
-
-    # Admin can cancel any job; other users can only cancel jobs they started.
-    if role != "admin" and username != job.username:
+    if not _can_cancel_watch_job(request, job):
         return JSONResponse(
             content={
                 "success": False,

frigate/test/test_chat_stop_camera_watch.py

@@ -0,0 +1,44 @@
+"""Tests for the stop_camera_watch chat tool."""
+
+import unittest
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from frigate.api.chat import _execute_stop_camera_watch
+from frigate.jobs.vlm_watch import VLMWatchJob
+
+
+def _request(user: str, role: str) -> SimpleNamespace:
+    return SimpleNamespace(headers={"remote-user": user, "remote-role": role})
+
+
+class TestStopCameraWatch(unittest.TestCase):
+    def setUp(self):
+        job = VLMWatchJob(camera="front", condition="person", username="alice")
+        patch("frigate.api.chat.get_vlm_watch_job", return_value=job).start()
+        self.stop = patch(
+            "frigate.api.chat.stop_vlm_watch_job", return_value=True
+        ).start()
+        self.addCleanup(patch.stopall)
+
+    def test_other_user_cannot_cancel(self):
+        result = _execute_stop_camera_watch(_request("bob", "viewer"))
+
+        self.assertFalse(result["success"])
+        self.stop.assert_not_called()
+
+    def test_creator_can_cancel(self):
+        result = _execute_stop_camera_watch(_request("alice", "viewer"))
+
+        self.assertTrue(result["success"])
+        self.stop.assert_called_once()
+
+    def test_admin_can_cancel(self):
+        result = _execute_stop_camera_watch(_request("bob", "admin"))
+
+        self.assertTrue(result["success"])
+        self.stop.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()