read the exec override from an import time snapshot

environment_vars is exported into os.environ, and is_go2rtc_arbitrary_exec_allowed read os.environ live, so the config file could enable exec sources. Snapshot the variable at import, which runs before any config is loaded.

frigate/test/http_api/test_http_camera_access.py

@@ -390,7 +390,7 @@ class TestGo2rtcStreamAccess(BaseTestHttp):
         intent and forward the request to go2rtc instead of short-circuiting with 400."""
         app = self._make_app(_MULTI_CAMERA_CONFIG)
         mock_response = type("R", (), {"ok": True, "status_code": 200, "text": "ok"})()
-        with patch.dict(os.environ, {"GO2RTC_ALLOW_ARBITRARY_EXEC": "true"}):
+        with patch("frigate.util.services._GO2RTC_ARBITRARY_EXEC_ENV", "true"):
             with patch(
                 "frigate.api.camera.requests.put", return_value=mock_response
             ) as mock_put:
@@ -403,6 +403,20 @@ class TestGo2rtcStreamAccess(BaseTestHttp):
                 forwarded_src = mock_put.call_args.kwargs["params"]["src"]
                 assert forwarded_src == "exec:/tmp/something"
 
+    def test_add_stream_ignores_override_written_after_import(self):
+        """The override is read once at import. A value written into os.environ
+        later, which is what the config's environment_vars block does, must not
+        unlock restricted sources."""
+        app = self._make_app(_MULTI_CAMERA_CONFIG)
+        with patch.dict(os.environ, {"GO2RTC_ALLOW_ARBITRARY_EXEC": "true"}):
+            with patch("frigate.api.camera.requests.put") as mock_put:
+                with AuthTestClient(app) as client:
+                    resp = client.put("/go2rtc/streams/legit?src=exec:/tmp/something")
+        # A live go2rtc would also answer 400, so assert on the forward.
+        mock_put.assert_not_called()
+        assert resp.status_code == 400
+        assert resp.json().get("success") is False
+
     def test_stream_alias_blocked_when_owning_camera_disallowed(self):
         """limited_user cannot access a stream alias that belongs to a camera they
         are not allowed to see."""

frigate/util/services.py

@@ -965,12 +965,17 @@ def get_hailo_temps() -> dict[str, float]:
     return temps
 
 
+# Snapshot: environment_vars lands in os.environ after import and must not
+# be able to enable this.
+_GO2RTC_ARBITRARY_EXEC_ENV = os.environ.get("GO2RTC_ALLOW_ARBITRARY_EXEC")
+
+
 def is_go2rtc_arbitrary_exec_allowed() -> bool:
     """Read the GO2RTC_ALLOW_ARBITRARY_EXEC override from env, docker
     secrets, or the Home Assistant add-on options file."""
     raw: str | None = None
-    if "GO2RTC_ALLOW_ARBITRARY_EXEC" in os.environ:
-        raw = os.environ.get("GO2RTC_ALLOW_ARBITRARY_EXEC")
+    if _GO2RTC_ARBITRARY_EXEC_ENV is not None:
+        raw = _GO2RTC_ARBITRARY_EXEC_ENV
     elif (
         os.path.isdir("/run/secrets")
         and os.access("/run/secrets", os.R_OK)