Merge remote-tracking branch 'origin/master' into dev

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01E4iDiYSpWJD13YQUdNKC5o

frigate/api/app.py

@@ -81,6 +81,7 @@ from frigate.util.object_names import get_categorized_object_names
 from frigate.util.schema import get_config_schema
 from frigate.util.services import (
     Go2rtcStreamIssue,
+    find_restricted_go2rtc_exec_changes,
     find_restricted_go2rtc_stream_changes,
     get_nvidia_driver_info,
     process_logs,
@@ -576,26 +577,26 @@ def config_raw():
         )
 
 
-def _go2rtc_streams(raw_config: str) -> Any:
-    """Load go2rtc.streams with the same loader Frigate and go2rtc startup use.
+def _go2rtc_config(raw_config: str) -> dict[str, Any]:
+    """Load the go2rtc section with the same loader Frigate and go2rtc startup use.
 
     Raises ruamel.yaml.YAMLError when the config cannot be parsed.
     """
     loaded = ruamel.yaml.YAML().load(raw_config)
     go2rtc = loaded.get("go2rtc") if isinstance(loaded, dict) else None
-    return go2rtc.get("streams") if isinstance(go2rtc, dict) else None
+    return go2rtc if isinstance(go2rtc, dict) else {}
 
 
-def _restricted_go2rtc_streams_response(
+def _restricted_go2rtc_response(
     old_raw_config: str, new_raw_config: str
 ) -> JSONResponse | None:
-    """Reject config changes that add restricted go2rtc stream sources.
+    """Reject config changes that add restricted go2rtc sources or options.
 
     The same sources are rejected by PUT /go2rtc/streams, and go2rtc would
     otherwise load them from the saved config on the next restart.
     """
     try:
-        new_streams = _go2rtc_streams(new_raw_config)
+        new_go2rtc = _go2rtc_config(new_raw_config)
     except ruamel.yaml.YAMLError:
         logger.warning("Rejected config change that could not be parsed")
         return JSONResponse(
@@ -604,16 +605,21 @@ def _restricted_go2rtc_streams_response(
         )
 
     try:
-        old_streams = _go2rtc_streams(old_raw_config)
+        old_go2rtc = _go2rtc_config(old_raw_config)
     except ruamel.yaml.YAMLError:
         # with no readable saved config, every source is treated as new
-        old_streams = None
+        old_go2rtc = {}
 
-    issues = find_restricted_go2rtc_stream_changes(old_streams, new_streams)
-    if not issues:
+    issues = find_restricted_go2rtc_stream_changes(
+        old_go2rtc.get("streams"), new_go2rtc.get("streams")
+    )
+    exec_keys = find_restricted_go2rtc_exec_changes(old_go2rtc, new_go2rtc)
+    if not issues and not exec_keys:
         return None
 
     messages = []
+    if exec_keys:
+        messages.append(f"Restricted go2rtc option: {', '.join(exec_keys)}")
     for issue, label in (
         (Go2rtcStreamIssue.restricted, "Restricted go2rtc stream source in"),
         (Go2rtcStreamIssue.unknown_variable, "Unknown variable in go2rtc stream"),
@@ -717,9 +723,7 @@ def config_save(save_option: str, body: Any = Body(media_type="text/plain")):
     except OSError:
         old_raw_config = ""
 
-    restricted_response = _restricted_go2rtc_streams_response(
-        old_raw_config, new_config
-    )
+    restricted_response = _restricted_go2rtc_response(old_raw_config, new_config)
     if restricted_response is not None:
         return restricted_response
 
@@ -1172,7 +1176,7 @@ def config_set(request: Request, body: AppConfigSetBody):
                         ),
                         status_code=400,
                     )
-                restricted_response = _restricted_go2rtc_streams_response(
+                restricted_response = _restricted_go2rtc_response(
                     old_raw_config, new_raw_config
                 )
                 if restricted_response is not None:

frigate/api/chat.py

@@ -56,6 +56,7 @@ from frigate.genai.utils import (
 )
 from frigate.jobs.export import ExportQueueFullError, start_export_job
 from frigate.jobs.vlm_watch import (
+    VLMWatchJob,
     get_vlm_watch_job,
     start_vlm_watch_job,
     stop_vlm_watch_job,
@@ -999,7 +1000,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":
@@ -1060,6 +1061,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)
@@ -1075,7 +1077,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."}
@@ -1908,11 +1921,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/config/camera/ffmpeg.py

@@ -2,7 +2,7 @@ from enum import Enum
 
 from pydantic import Field, field_validator
 
-from frigate.util.config import resolve_ffmpeg_path
+from frigate.util.config import is_valid_ffmpeg_path, resolve_ffmpeg_path
 from frigate.util.env import EnvString
 
 from ..base import FrigateBaseModel
@@ -101,6 +101,16 @@ class FfmpegConfig(FrigateBaseModel):
         description="Default GPU index used for hardware acceleration if available.",
     )
 
+    @field_validator("path")
+    @classmethod
+    def validate_path(cls, v: str) -> str:
+        if not is_valid_ffmpeg_path(v):
+            raise ValueError(
+                "FFmpeg path must be a version alias or an absolute path without "
+                "spaces, quotes or shell characters"
+            )
+        return v
+
     @property
     def ffmpeg_path(self) -> str:
         return resolve_ffmpeg_path(self.path, "ffmpeg")

frigate/test/http_api/test_http_config_go2rtc_streams.py

@@ -94,6 +94,23 @@ class TestConfigGo2rtcStreamRestrictions(BaseTestHttp):
                 headers={"Content-Type": "text/plain"},
             )
 
+    def _set_go2rtc(self, go2rtc: dict):
+        with AuthTestClient(self.app) as client:
+            return client.put(
+                "/config/set",
+                json={"config_data": {"go2rtc": go2rtc}, "requires_restart": 1},
+            )
+
+    def _save_go2rtc(self, go2rtc: dict):
+        config = copy.deepcopy(_CONFIG)
+        config["go2rtc"].update(go2rtc)
+        with AuthTestClient(self.app) as client:
+            return client.post(
+                "/config/save?save_option=saveonly",
+                content=_dump(config),
+                headers={"Content-Type": "text/plain"},
+            )
+
     # --- config/set ---
 
     def test_set_rejects_injected_ffmpeg_source(self):
@@ -224,3 +241,108 @@ class TestConfigGo2rtcStreamRestrictions(BaseTestHttp):
             resp = self._set_streams({"new_stream": [src]})
         assert resp.status_code == 200, resp.text
         assert src in self._read_config()
+
+    # --- go2rtc ffmpeg and ngrok options ---
+
+    def test_save_rejects_ffmpeg_bin_override(self):
+        before = self._read_config()
+        resp = self._save_go2rtc({"ffmpeg": {"bin": "/bin/sh", "global": "-c 'id' x"}})
+        assert resp.status_code == 400
+        assert resp.json()["message"] == (
+            "Restricted go2rtc option: ffmpeg.bin, ffmpeg.global"
+        )
+        assert self._read_config() == before
+
+    def test_set_rejects_ffmpeg_bin_override(self):
+        before = self._read_config()
+        resp = self._set_go2rtc({"ffmpeg": {"bin": "/bin/sh"}})
+        assert resp.status_code == 400
+        assert self._read_config() == before
+
+    def test_save_rejects_ffmpeg_template_override(self):
+        before = self._read_config()
+        resp = self._save_go2rtc({"ffmpeg": {"rtsp": "-i {input} -y /config/x"}})
+        assert resp.status_code == 400
+        assert "ffmpeg.rtsp" in resp.json()["message"]
+        assert self._read_config() == before
+
+    def test_set_rejects_custom_ffmpeg_template(self):
+        # a stream can reference any template key with #video=<name>
+        before = self._read_config()
+        resp = self._set_go2rtc({"ffmpeg": {"evil": "-y /config/x"}})
+        assert resp.status_code == 400
+        assert self._read_config() == before
+
+    def test_save_rejects_ngrok(self):
+        before = self._read_config()
+        resp = self._save_go2rtc({"ngrok": {"command": "/bin/sh -c id"}})
+        assert resp.status_code == 400
+        assert resp.json()["message"] == "Restricted go2rtc option: ngrok.command"
+        assert self._read_config() == before
+
+    def test_set_rejects_ngrok(self):
+        before = self._read_config()
+        resp = self._set_go2rtc({"ngrok": {"command": "/bin/sh -c id"}})
+        assert resp.status_code == 400
+        assert self._read_config() == before
+
+    def test_save_keeps_existing_ffmpeg_template_on_disk(self):
+        go2rtc = {"ffmpeg": {"global": "-vaapi_device /dev/dri/renderD128"}}
+        config = copy.deepcopy(_CONFIG)
+        config["go2rtc"].update(go2rtc)
+        with open(self.config_path, "w") as f:
+            f.write(_dump(config))
+
+        resp = self._save_go2rtc(go2rtc)
+        assert resp.status_code == 200, resp.text
+
+        resp = self._save_go2rtc({"ffmpeg": {"global": "-hide_banner"}})
+        assert resp.status_code == 400
+
+    def test_exec_options_allowed_when_override_set(self):
+        with patch("frigate.util.services._GO2RTC_ARBITRARY_EXEC_ENV", "true"):
+            resp = self._save_go2rtc(
+                {
+                    "ffmpeg": {"bin": "/usr/bin/ffmpeg", "rtsp": "-i {input}"},
+                    "ngrok": {"command": "ngrok tcp 8555"},
+                }
+            )
+        assert resp.status_code == 200, resp.text
+        assert "ngrok tcp 8555" in self._read_config()
+
+    def test_save_rejects_unsafe_ffmpeg_path(self):
+        before = self._read_config()
+        config = copy.deepcopy(_CONFIG)
+        config["ffmpeg"] = {"path": "/bin/sh -c id;#"}
+        with AuthTestClient(self.app) as client:
+            resp = client.post(
+                "/config/save?save_option=saveonly",
+                content=_dump(config),
+                headers={"Content-Type": "text/plain"},
+            )
+        assert resp.status_code == 400
+        assert self._read_config() == before
+
+    # --- go2rtc ${VAR} expansion ---
+
+    def test_set_rejects_var_expansion_source(self):
+        before = self._read_config()
+        resp = self._set_streams({"new_stream": "${NOPE:exec:touch /tmp/pwned}"})
+        assert resp.status_code == 400
+        assert "Restricted go2rtc stream source" in resp.json()["message"]
+        assert self._read_config() == before
+
+    def test_set_rejects_escaped_var_expansion_source(self):
+        before = self._read_config()
+        resp = self._set_streams({"new_stream": "${{NOPE:exec:id}}"})
+        assert resp.status_code == 400
+        assert "Restricted go2rtc stream source" in resp.json()["message"]
+        assert self._read_config() == before
+
+    def test_set_allows_var_expansion_source_when_override_set(self):
+        with patch(
+            "frigate.util.services.is_go2rtc_arbitrary_exec_allowed",
+            return_value=True,
+        ):
+            resp = self._set_streams({"new_stream": "${NOPE:exec:id}"})
+        assert resp.status_code == 200, resp.text

frigate/test/test_chat_stop_camera_watch.py

@@ -0,0 +1,63 @@
+"""Tests for camera watch ownership in the chat tools."""
+
+import asyncio
+import unittest
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from frigate.api.chat import _execute_start_camera_watch, _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()
+
+
+class TestStartCameraWatch(unittest.TestCase):
+    @patch("frigate.api.chat.require_camera_access", new_callable=AsyncMock)
+    @patch("frigate.api.chat.start_vlm_watch_job", return_value="job1")
+    def test_records_caller_as_owner(self, start, _access):
+        request = _request("alice", "viewer")
+        request.app = MagicMock()
+        request.app.frigate_config.cameras = {"front": None}
+        request.app.genai_manager.chat_client.supports_vision = True
+
+        asyncio.run(
+            _execute_start_camera_watch(
+                request, {"camera": "front", "condition": "person"}
+            )
+        )
+
+        self.assertEqual(start.call_args.kwargs["username"], "alice")
+
+
+if __name__ == "__main__":
+    unittest.main()

frigate/test/test_config.py

@@ -15,6 +15,7 @@ from frigate.const import MODEL_CACHE_DIR
 from frigate.detectors import DetectorTypeEnum
 from frigate.detectors.device import build_detector_config, runner_names
 from frigate.util.builtin import deep_merge
+from frigate.util.config import resolve_ffmpeg_path
 
 
 class TestConfig(unittest.TestCase):
@@ -2377,5 +2378,54 @@ class TestAttributeFilterDefaults(unittest.TestCase):
         self.assertEqual(face_filter.min_score, 0.3)
 
 
+class TestFfmpegPath(unittest.TestCase):
+    # ffmpeg commands are split on spaces, so this would run /bin/sh
+    UNSAFE_PATH = "/bin/sh -c id;#"
+
+    def setUp(self):
+        self.minimal = {
+            "mqtt": {"host": "mqtt"},
+            "cameras": {
+                "back": {
+                    "ffmpeg": {
+                        "inputs": [
+                            {"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]}
+                        ]
+                    },
+                    "detect": {"height": 1080, "width": 1920, "fps": 5},
+                }
+            },
+        }
+
+    def test_custom_path_allowed(self):
+        config = deep_merge(
+            {"ffmpeg": {"path": "/usr/lib/jellyfin-ffmpeg"}}, self.minimal
+        )
+        frigate_config = FrigateConfig(**config)
+        self.assertEqual(
+            frigate_config.cameras["back"].ffmpeg.ffmpeg_path,
+            "/usr/lib/jellyfin-ffmpeg/bin/ffmpeg",
+        )
+
+    def test_unsafe_global_path_rejected(self):
+        for path in (self.UNSAFE_PATH, "/opt/my ffmpeg", "/opt/'x'", "/opt/x\n"):
+            config = deep_merge({"ffmpeg": {"path": path}}, self.minimal)
+            with self.assertRaises(ValidationError, msg=path):
+                FrigateConfig(**config)
+
+    def test_unsafe_camera_path_rejected(self):
+        config = deep_merge(
+            {"cameras": {"back": {"ffmpeg": {"path": self.UNSAFE_PATH}}}},
+            self.minimal,
+        )
+        with self.assertRaises(ValidationError):
+            FrigateConfig(**config)
+
+    def test_resolve_falls_back_for_unsafe_path(self):
+        self.assertEqual(
+            resolve_ffmpeg_path(self.UNSAFE_PATH), resolve_ffmpeg_path("default")
+        )
+
+
 if __name__ == "__main__":
     unittest.main(verbosity=2)

frigate/test/test_go2rtc_exec_config.py

@@ -0,0 +1,88 @@
+"""Tests for the go2rtc ffmpeg and ngrok options applied at startup."""
+
+import unittest
+from unittest.mock import patch
+
+from frigate.util.services import (
+    apply_go2rtc_ffmpeg_config,
+    is_restricted_go2rtc_api_source,
+    is_restricted_go2rtc_source,
+)
+
+FFMPEG_PATH = "/usr/lib/ffmpeg/7.0/bin/ffmpeg"
+
+
+class TestApplyGo2rtcFfmpegConfig(unittest.TestCase):
+    def setUp(self):
+        # the override is snapshotted from the environment at import time
+        exec_patch = patch("frigate.util.services._GO2RTC_ARBITRARY_EXEC_ENV", None)
+        exec_patch.start()
+        self.addCleanup(exec_patch.stop)
+
+    def test_sets_bin_when_missing(self):
+        config = {}
+        assert apply_go2rtc_ffmpeg_config(config, FFMPEG_PATH) == []
+        assert config["ffmpeg"] == {"bin": FFMPEG_PATH}
+
+    def test_forces_bin_override(self):
+        config = {"ffmpeg": {"bin": "/bin/sh", "global": "-c 'id' x"}}
+        assert apply_go2rtc_ffmpeg_config(config, FFMPEG_PATH) == ["ffmpeg.bin"]
+        assert config["ffmpeg"]["bin"] == FFMPEG_PATH
+
+    def test_keeps_saved_templates(self):
+        config = {"ffmpeg": {"global": "-vaapi_device /dev/dri/renderD128"}}
+        assert apply_go2rtc_ffmpeg_config(config, FFMPEG_PATH) == []
+        assert config["ffmpeg"]["global"] == "-vaapi_device /dev/dri/renderD128"
+
+    def test_drops_ngrok(self):
+        config = {"ngrok": {"command": "/bin/sh -c id"}}
+        assert apply_go2rtc_ffmpeg_config(config, FFMPEG_PATH) == ["ngrok"]
+        assert "ngrok" not in config
+
+    def test_allows_bin_and_ngrok_when_override_set(self):
+        config = {
+            "ffmpeg": {"bin": "/usr/bin/ffmpeg"},
+            "ngrok": {"command": "ngrok tcp 8555"},
+        }
+        with patch("frigate.util.services._GO2RTC_ARBITRARY_EXEC_ENV", "true"):
+            assert apply_go2rtc_ffmpeg_config(config, FFMPEG_PATH) == []
+        assert config["ffmpeg"]["bin"] == "/usr/bin/ffmpeg"
+        assert config["ngrok"] == {"command": "ngrok tcp 8555"}
+
+
+class TestRestrictedGo2rtcVarExpansion(unittest.TestCase):
+    """go2rtc expands `${VAR:default}` over the raw config before parsing, and
+    the default can name an exec source, so a `${` source is restricted."""
+
+    def setUp(self):
+        exec_patch = patch(
+            "frigate.util.services.is_go2rtc_arbitrary_exec_allowed",
+            return_value=False,
+        )
+        self.mock_exec_allowed = exec_patch.start()
+        self.addCleanup(exec_patch.stop)
+
+    def test_var_expansion_source_is_restricted(self):
+        for source in (
+            "${NOPE:exec:touch /tmp/x}",
+            "rtsp://host/${CAM}",
+            "  ${NOPE:exec:id}",
+        ):
+            self.assertTrue(is_restricted_go2rtc_source(source), source)
+            self.assertTrue(is_restricted_go2rtc_api_source(source), source)
+
+    def test_plain_sources_are_not_restricted(self):
+        for source in (
+            "rtsp://user:pass@host/stream",
+            "ffmpeg:front#video=h264#height=720",
+        ):
+            self.assertFalse(is_restricted_go2rtc_source(source), source)
+            self.assertFalse(is_restricted_go2rtc_api_source(source), source)
+
+    def test_var_expansion_allowed_when_override_set(self):
+        self.mock_exec_allowed.return_value = True
+        self.assertFalse(is_restricted_go2rtc_source("${NOPE:exec:id}"))
+
+
+if __name__ == "__main__":
+    unittest.main()

frigate/util/config.py

@@ -87,18 +87,31 @@ def _warn_ignored_ffmpeg_path(path: str) -> None:
     )
 
 
+# ffmpeg commands are built as strings and split on whitespace, so a custom path
+# must stay a single plain token
+_CUSTOM_FFMPEG_PATH = re.compile(r"/[\w.+@/-]+")
+
+
+def is_valid_ffmpeg_path(path: str) -> bool:
+    """Check that a custom ffmpeg path has no whitespace, quotes or shell syntax."""
+    return not path.startswith("/") or _CUSTOM_FFMPEG_PATH.fullmatch(path) is not None
+
+
 def resolve_ffmpeg_path(path: str, binary: str = "ffmpeg") -> str:
     """Resolve an ffmpeg version alias or custom path to a binary path.
 
     A bare version alias that is no longer bundled (for example one that was
     dropped when the default version changed) falls back to the default
     bundled version so existing configs keep working across an upgrade or a
-    revert. Custom install paths (anything absolute) are used as-is, except
-    one in a runtime-user-writable tree while FRIGATE_ROOT_SERVICES makes
-    frigate root; see RUNTIME_USER_WRITABLE_DIRS.
+    revert. Custom install paths (anything absolute) are used as-is, unless
+    they fail `is_valid_ffmpeg_path` (which falls back to the default), or
+    they are in a runtime-user-writable tree while FRIGATE_ROOT_SERVICES
+    makes frigate root; see RUNTIME_USER_WRITABLE_DIRS.
     """
-    if path == "default" or (
-        not path.startswith("/") and path not in INCLUDED_FFMPEG_VERSIONS
+    if (
+        path == "default"
+        or not is_valid_ffmpeg_path(path)
+        or (not path.startswith("/") and path not in INCLUDED_FFMPEG_VERSIONS)
     ):
         version = DEFAULT_FFMPEG_VERSION
     elif path in INCLUDED_FFMPEG_VERSIONS:

frigate/util/services.py

@@ -1101,9 +1101,15 @@ def is_go2rtc_arbitrary_exec_allowed() -> bool:
 
 
 def is_restricted_go2rtc_source(stream_source: str) -> bool:
-    """Check if a stream source is a restricted type (echo, expr, or exec)
-    and the GO2RTC_ALLOW_ARBITRARY_EXEC override is not set."""
-    if not stream_source.strip().startswith(("echo:", "expr:", "exec:")):
+    """Check if a stream source is restricted and the
+    GO2RTC_ALLOW_ARBITRARY_EXEC override is not set.
+
+    Restricted sources are the echo, expr and exec schemes, and any source
+    using go2rtc's `${VAR:default}` expansion, which go2rtc runs over the raw
+    config before parsing and whose default can name an exec source.
+    """
+    source = stream_source.strip()
+    if not source.startswith(("echo:", "expr:", "exec:")) and "${" not in source:
         return False
     return not is_go2rtc_arbitrary_exec_allowed()
 
@@ -1221,6 +1227,74 @@ def find_restricted_go2rtc_stream_changes(
     return issues
 
 
+# go2rtc runs `ffmpeg.bin` and `ngrok.command`, and every other `ffmpeg` key is a
+# template of ffmpeg arguments that `#video=<name>` and friends can reference
+_GO2RTC_EXEC_SECTIONS = ("ffmpeg", "ngrok")
+
+
+def find_restricted_go2rtc_exec_changes(
+    old_go2rtc: dict[str, Any] | None, new_go2rtc: dict[str, Any] | None
+) -> list[str]:
+    """Return the go2rtc ffmpeg and ngrok keys added or changed through the API.
+
+    These are only accepted with arbitrary exec enabled. Keys already in the
+    saved config are left alone, the same as stream sources.
+    """
+    if is_go2rtc_arbitrary_exec_allowed():
+        return []
+
+    old_go2rtc = old_go2rtc if isinstance(old_go2rtc, dict) else {}
+    new_go2rtc = new_go2rtc if isinstance(new_go2rtc, dict) else {}
+    changed: list[str] = []
+
+    for section in _GO2RTC_EXEC_SECTIONS:
+        old = old_go2rtc.get(section)
+        new = new_go2rtc.get(section)
+
+        if not isinstance(new, dict):
+            if new is not None and new != old:
+                changed.append(section)
+            continue
+
+        old = old if isinstance(old, dict) else {}
+        changed.extend(
+            f"{section}.{key}" for key, value in new.items() if old.get(key) != value
+        )
+
+    return changed
+
+
+def apply_go2rtc_ffmpeg_config(
+    go2rtc_config: dict[str, Any], ffmpeg_path: str
+) -> list[str]:
+    """Set go2rtc's ffmpeg binary for startup.
+
+    Unless arbitrary exec is enabled, `ffmpeg.bin` is forced to Frigate's ffmpeg
+    and `ngrok` is removed. Templates in the saved config are kept.
+
+    Returns:
+        The keys that were overridden or removed
+    """
+    allowed = is_go2rtc_arbitrary_exec_allowed()
+    restricted: list[str] = []
+
+    ffmpeg = go2rtc_config.get("ffmpeg")
+    if not isinstance(ffmpeg, dict):
+        ffmpeg = go2rtc_config["ffmpeg"] = {}
+
+    if ffmpeg.get("bin") is None:
+        ffmpeg["bin"] = ffmpeg_path
+    elif ffmpeg["bin"] != ffmpeg_path and not allowed:
+        restricted.append("ffmpeg.bin")
+        ffmpeg["bin"] = ffmpeg_path
+
+    if "ngrok" in go2rtc_config and not allowed:
+        restricted.append("ngrok")
+        del go2rtc_config["ngrok"]
+
+    return restricted
+
+
 def ffprobe_stream(ffmpeg, path: str, detailed: bool = False) -> sp.CompletedProcess:
     """Run ffprobe on stream."""
     clean_path = escape_special_characters(path)