Parse config stream changes with Frigate's YAML loader

The go2rtc stream check on config/save and config/set parsed the config
with the safe loader and treated a parse error as an empty config, while
Frigate and the go2rtc startup script use the round-trip loader. Input
that only the round-trip loader accepts, such as a tagged root map, hid
new stream sources from the check.

The check now uses the same round-trip loader and rejects the request
when the new config cannot be parsed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

frigate/api/app.py

@@ -523,16 +523,13 @@ def config_raw():
         )
 
 
-def _load_yaml_dict(raw_config: str) -> dict[str, Any]:
-    try:
-        loaded = ruamel.yaml.YAML(typ="safe", pure=True).load(raw_config)
-    except ruamel.yaml.YAMLError:
-        return {}
-    return loaded if isinstance(loaded, dict) else {}
-
-
 def _go2rtc_streams(raw_config: str) -> Any:
-    go2rtc = _load_yaml_dict(raw_config).get("go2rtc")
+    """Load go2rtc.streams 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
 
 
@@ -544,9 +541,22 @@ def _restricted_go2rtc_streams_response(
     The same sources are rejected by PUT /go2rtc/streams, and go2rtc would
     otherwise load them from the saved config on the next restart.
     """
-    restricted = find_restricted_go2rtc_stream_changes(
+    try:
-        _go2rtc_streams(old_raw_config), _go2rtc_streams(new_raw_config)
+        new_streams = _go2rtc_streams(new_raw_config)
-    )
+    except ruamel.yaml.YAMLError:
+        logger.warning("Rejected config change that could not be parsed")
+        return JSONResponse(
+            content={"success": False, "message": "Unable to parse config"},
+            status_code=400,
+        )
+
+    try:
+        old_streams = _go2rtc_streams(old_raw_config)
+    except ruamel.yaml.YAMLError:
+        # with no readable saved config, every source is treated as new
+        old_streams = None
+
+    restricted = find_restricted_go2rtc_stream_changes(old_streams, new_streams)
     if not restricted:
         return None
 

frigate/test/http_api/test_http_config_go2rtc_streams.py

@@ -71,13 +71,13 @@ class TestConfigGo2rtcStreamRestrictions(BaseTestHttp):
                 },
             )
 
-    def _save_streams(self, streams: dict):
+    def _save_streams(self, streams: dict, prefix: str = ""):
         config = copy.deepcopy(_CONFIG)
         config["go2rtc"]["streams"].update(streams)
         with AuthTestClient(self.app) as client:
             return client.post(
                 "/config/save?save_option=saveonly",
-                content=_dump(config),
+                content=prefix + _dump(config),
                 headers={"Content-Type": "text/plain"},
             )
 
@@ -153,3 +153,17 @@ class TestConfigGo2rtcStreamRestrictions(BaseTestHttp):
         config = self._read_config()
         assert _EXISTING_SRC in config
         assert "rtsp://10.0.0.2:554/video" in config
+
+    def test_save_tagged_config_with_standard_source_allowed(self):
+        # a tag on the root map still passes validation and startup parsing
+        resp = self._save_streams(
+            {"new_stream": "rtsp://10.0.0.2:554/video"}, prefix="!config\n"
+        )
+        assert resp.status_code == 200, resp.text
+
+    def test_save_tagged_config_with_injected_source_rejected(self):
+        before = self._read_config()
+        resp = self._save_streams({"new_stream": [_INJECTED_SRC]}, prefix="!config\n")
+        assert resp.status_code == 400
+        assert "Restricted go2rtc stream source" in resp.json()["message"]
+        assert self._read_config() == before