restrict go2rtc ${VAR} expansion in stream sources (#24734)

go2rtc expands ${NAME:default} over the raw config text before parsing it, and falls back to the inline default when NAME is unset, so a stream source like ${NOPE:exec:<cmd>} passed the exec gate and go2rtc then ran the command. On master the API accepted the plain form, and startup accepted the ${{...}} form because .format collapsed it to ${...} after the check would have caught it. Any stream source containing ${ is now restricted unless GO2RTC_ALLOW_ARBITRARY_EXEC is set, the same as exec sources.

frigate/test/http_api/test_http_config_go2rtc_streams.py

@@ -313,3 +313,27 @@ class TestConfigGo2rtcStreamRestrictions(BaseTestHttp):
             )
         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_go2rtc_exec_config.py

@@ -4,7 +4,11 @@ import os
 import unittest
 from unittest.mock import patch
 
-from frigate.util.services import apply_go2rtc_ffmpeg_config
+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"
 
@@ -47,5 +51,39 @@ class TestApplyGo2rtcFfmpegConfig(unittest.TestCase):
         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()