Fix go2rtc nested key dict

docker/main/rootfs/usr/local/go2rtc/create_config.py

@@ -17,7 +17,10 @@ from frigate.const import (
 )
 from frigate.ffmpeg_presets import parse_preset_hardware_acceleration_encode
 from frigate.util.config import find_config_file
-from frigate.util.services import is_restricted_go2rtc_source
+from frigate.util.services import (
+    is_go2rtc_arbitrary_exec_allowed,
+    is_restricted_go2rtc_source,
+)
 
 sys.path.remove("/opt/frigate")
 
@@ -159,6 +162,20 @@ for name in list(go2rtc_config.get("streams", {})):
             )
             del go2rtc_config["streams"][name]
 
+    elif isinstance(stream, dict):
+        # The map form ({"url": ...}) lets go2rtc resolve the source
+        # recursively, so it is effectively a dynamic way to generate the URL
+        # for a stream. That can only be backed by an exec source, so it cannot
+        # be allowed unless arbitrary exec is explicitly enabled. When it is
+        # enabled, leave the map untouched for go2rtc to resolve.
+        if not is_go2rtc_arbitrary_exec_allowed():
+            print(
+                f"[ERROR] Stream '{name}' uses a dynamic source format which is disabled by default for security. "
+                f"Set GO2RTC_ALLOW_ARBITRARY_EXEC=true to enable arbitrary exec sources."
+            )
+            del go2rtc_config["streams"][name]
+            continue
+
 # add birdseye restream stream if enabled
 if config.get("birdseye", {}).get("restream", False):
     birdseye: dict[str, Any] = config.get("birdseye")

frigate/util/services.py

@@ -556,7 +556,7 @@ def get_jetson_stats() -> Optional[dict[int, dict]]:
     return results
 
 
-def _go2rtc_arbitrary_exec_allowed() -> bool:
+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: Optional[str] = None
@@ -588,7 +588,7 @@ def is_restricted_go2rtc_source(stream_source: str) -> bool:
     and the GO2RTC_ALLOW_ARBITRARY_EXEC override is not set."""
     if not stream_source.strip().startswith(("echo:", "expr:", "exec:")):
         return False
-    return not _go2rtc_arbitrary_exec_allowed()
+    return not is_go2rtc_arbitrary_exec_allowed()
 
 
 def ffprobe_stream(ffmpeg, path: str, detailed: bool = False) -> sp.CompletedProcess: