Improve environment var handling (#22796)

* refactor env var handling

- use shared helper
- use left-to-right parser

* add tests

* formatting

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

@@ -9,6 +9,7 @@ from typing import Any
 from ruamel.yaml import YAML
 
 sys.path.insert(0, "/opt/frigate")
+from frigate.config.env import substitute_frigate_vars
 from frigate.const import (
     BIRDSEYE_PIPE,
     DEFAULT_FFMPEG_VERSION,
@@ -47,14 +48,6 @@ ALLOW_ARBITRARY_EXEC = allow_arbitrary_exec is not None and str(
     allow_arbitrary_exec
 ).lower() in ("true", "1", "yes")
 
-FRIGATE_ENV_VARS = {k: v for k, v in os.environ.items() if k.startswith("FRIGATE_")}
-# read docker secret files as env vars too
-if os.path.isdir("/run/secrets"):
-    for secret_file in os.listdir("/run/secrets"):
-        if secret_file.startswith("FRIGATE_"):
-            FRIGATE_ENV_VARS[secret_file] = (
-                Path(os.path.join("/run/secrets", secret_file)).read_text().strip()
-            )
 
 config_file = find_config_file()
 
@@ -103,13 +96,13 @@ if go2rtc_config["webrtc"].get("candidates") is None:
     go2rtc_config["webrtc"]["candidates"] = default_candidates
 
 if go2rtc_config.get("rtsp", {}).get("username") is not None:
-    go2rtc_config["rtsp"]["username"] = go2rtc_config["rtsp"]["username"].format(
+    go2rtc_config["rtsp"]["username"] = substitute_frigate_vars(
-        **FRIGATE_ENV_VARS
+        go2rtc_config["rtsp"]["username"]
     )
 
 if go2rtc_config.get("rtsp", {}).get("password") is not None:
-    go2rtc_config["rtsp"]["password"] = go2rtc_config["rtsp"]["password"].format(
+    go2rtc_config["rtsp"]["password"] = substitute_frigate_vars(
-        **FRIGATE_ENV_VARS
+        go2rtc_config["rtsp"]["password"]
     )
 
 # ensure ffmpeg path is set correctly
@@ -145,7 +138,7 @@ for name in list(go2rtc_config.get("streams", {})):
 
     if isinstance(stream, str):
         try:
-            formatted_stream = stream.format(**FRIGATE_ENV_VARS)
+            formatted_stream = substitute_frigate_vars(stream)
             if not ALLOW_ARBITRARY_EXEC and is_restricted_source(formatted_stream):
                 print(
                     f"[ERROR] Stream '{name}' uses a restricted source (echo/expr/exec) which is disabled by default for security. "
@@ -164,7 +157,7 @@ for name in list(go2rtc_config.get("streams", {})):
         filtered_streams = []
         for i, stream_item in enumerate(stream):
             try:
-                formatted_stream = stream_item.format(**FRIGATE_ENV_VARS)
+                formatted_stream = substitute_frigate_vars(stream_item)
                 if not ALLOW_ARBITRARY_EXEC and is_restricted_source(formatted_stream):
                     print(
                         f"[ERROR] Stream '{name}' item {i + 1} uses a restricted source (echo/expr/exec) which is disabled by default for security. "

frigate/api/camera.py

@@ -30,7 +30,7 @@ from frigate.config.camera.updater import (
     CameraConfigUpdateEnum,
     CameraConfigUpdateTopic,
 )
-from frigate.config.env import FRIGATE_ENV_VARS
+from frigate.config.env import substitute_frigate_vars
 from frigate.util.builtin import clean_camera_user_pass
 from frigate.util.camera_cleanup import cleanup_camera_db, cleanup_camera_files
 from frigate.util.config import find_config_file
@@ -126,7 +126,7 @@ def go2rtc_add_stream(request: Request, stream_name: str, src: str = ""):
         params = {"name": stream_name}
         if src:
             try:
-                params["src"] = src.format(**FRIGATE_ENV_VARS)
+                params["src"] = substitute_frigate_vars(src)
             except KeyError:
                 params["src"] = src
 

frigate/config/env.py

@@ -1,4 +1,5 @@
 import os
+import re
 from pathlib import Path
 from typing import Annotated
 
@@ -15,8 +16,77 @@ if os.path.isdir(secrets_dir) and os.access(secrets_dir, os.R_OK):
             )
 
 
+# Matches a FRIGATE_* identifier following an opening brace.
+_FRIGATE_IDENT_RE = re.compile(r"FRIGATE_[A-Za-z0-9_]+")
+
+
+def substitute_frigate_vars(value: str) -> str:
+    """Substitute `{FRIGATE_*}` placeholders in *value*.
+
+    Reproduces the subset of `str.format()` brace semantics that Frigate's
+    config has historically supported, while leaving unrelated brace content
+    (e.g. ffmpeg `%{localtime\\:...}` expressions) untouched:
+
+    * `{{` and `}}` collapse to literal `{` / `}` (the documented escape).
+    * `{FRIGATE_NAME}` is replaced from `FRIGATE_ENV_VARS`; an unknown name
+      raises `KeyError` to preserve the existing "Invalid substitution"
+      error path.
+    * A `{` that begins `{FRIGATE_` but is not a well-formed
+      `{FRIGATE_NAME}` placeholder raises `ValueError` (malformed
+      placeholder). Callers that catch `KeyError` to allow unknown-var
+      passthrough will still surface malformed syntax as an error.
+    * Any other `{` or `}` is treated as a literal and passed through.
+    """
+    out: list[str] = []
+    i = 0
+    n = len(value)
+    while i < n:
+        ch = value[i]
+        if ch == "{":
+            # Escaped literal `{{`.
+            if i + 1 < n and value[i + 1] == "{":
+                out.append("{")
+                i += 2
+                continue
+            # Possible `{FRIGATE_*}` placeholder.
+            if value.startswith("{FRIGATE_", i):
+                ident_match = _FRIGATE_IDENT_RE.match(value, i + 1)
+                if (
+                    ident_match is not None
+                    and ident_match.end() < n
+                    and value[ident_match.end()] == "}"
+                ):
+                    key = ident_match.group(0)
+                    if key not in FRIGATE_ENV_VARS:
+                        raise KeyError(key)
+                    out.append(FRIGATE_ENV_VARS[key])
+                    i = ident_match.end() + 1
+                    continue
+                # Looks like a FRIGATE placeholder but is malformed
+                # (no closing brace, illegal char, format spec, etc.).
+                raise ValueError(
+                    f"Malformed FRIGATE_ placeholder near {value[i : i + 32]!r}"
+                )
+            # Plain `{` — pass through (e.g. `%{localtime\:...}`).
+            out.append("{")
+            i += 1
+            continue
+        if ch == "}":
+            # Escaped literal `}}`.
+            if i + 1 < n and value[i + 1] == "}":
+                out.append("}")
+                i += 2
+                continue
+            out.append("}")
+            i += 1
+            continue
+        out.append(ch)
+        i += 1
+    return "".join(out)
+
+
 def validate_env_string(v: str) -> str:
-    return v.format(**FRIGATE_ENV_VARS)
+    return substitute_frigate_vars(v)
 
 
 EnvString = Annotated[str, AfterValidator(validate_env_string)]

frigate/test/test_env.py

@@ -2,6 +2,7 @@
 
 import os
 import unittest
+from unittest.mock import MagicMock, patch
 
 from frigate.config.env import (
     FRIGATE_ENV_VARS,
@@ -10,6 +11,71 @@ from frigate.config.env import (
 )
 
 
+class TestGo2RtcAddStreamSubstitution(unittest.TestCase):
+    """Covers the API path: PUT /go2rtc/streams/{stream_name}.
+
+    The route shells out to go2rtc via `requests.put`; we mock the HTTP call
+    and assert that the substituted `src` parameter handles the same mixed
+    {FRIGATE_*} + literal-brace strings as the config-loading path.
+    """
+
+    def setUp(self):
+        self._original_env_vars = dict(FRIGATE_ENV_VARS)
+
+    def tearDown(self):
+        FRIGATE_ENV_VARS.clear()
+        FRIGATE_ENV_VARS.update(self._original_env_vars)
+
+    def _call_route(self, src: str) -> str:
+        """Invoke go2rtc_add_stream and return the substituted src param."""
+        from frigate.api import camera as camera_api
+
+        captured = {}
+
+        def fake_put(url, params=None, timeout=None):
+            captured["params"] = params
+            resp = MagicMock()
+            resp.ok = True
+            resp.text = ""
+            resp.status_code = 200
+            return resp
+
+        with patch.object(camera_api.requests, "put", side_effect=fake_put):
+            camera_api.go2rtc_add_stream(
+                request=MagicMock(), stream_name="cam1", src=src
+            )
+        return captured["params"]["src"]
+
+    def test_mixed_localtime_and_frigate_var(self):
+        """%{localtime\\:...} alongside {FRIGATE_USER} substitutes only the var."""
+        FRIGATE_ENV_VARS["FRIGATE_USER"] = "admin"
+        src = (
+            "ffmpeg:rtsp://host/s#raw=-vf "
+            "drawtext=text=%{localtime\\:%Y-%m-%d}:user={FRIGATE_USER}"
+        )
+        self.assertEqual(
+            self._call_route(src),
+            "ffmpeg:rtsp://host/s#raw=-vf "
+            "drawtext=text=%{localtime\\:%Y-%m-%d}:user=admin",
+        )
+
+    def test_unknown_var_falls_back_to_raw_src(self):
+        """Existing route behavior: unknown {FRIGATE_*} keeps raw src."""
+        src = "rtsp://host/{FRIGATE_NONEXISTENT}/stream"
+        self.assertEqual(self._call_route(src), src)
+
+    def test_malformed_placeholder_rejected_via_api(self):
+        """Malformed FRIGATE placeholders raise (not silently passed through).
+
+        Regression: previously camera.py caught any KeyError and fell back
+        to the raw src, so `{FRIGATE_FOO:>5}` was silently accepted via the
+        API while config loading rejected it. The helper now raises
+        ValueError for malformed syntax to keep the two paths consistent.
+        """
+        with self.assertRaises(ValueError):
+            self._call_route("rtsp://host/{FRIGATE_FOO:>5}/stream")
+
+
 class TestEnvString(unittest.TestCase):
     def setUp(self):
         self._original_env_vars = dict(FRIGATE_ENV_VARS)
@@ -43,6 +109,72 @@ class TestEnvString(unittest.TestCase):
         with self.assertRaises(KeyError):
             validate_env_string("{FRIGATE_NONEXISTENT_VAR}")
 
+    def test_non_frigate_braces_passthrough(self):
+        """Braces that are not {FRIGATE_*} placeholders pass through untouched.
+
+        Regression test for ffmpeg drawtext expressions like
+        "%{localtime\\:%Y-%m-%d}" being mangled by str.format().
+        """
+        expr = (
+            "ffmpeg:rtsp://127.0.0.1/src#raw=-vf "
+            "drawtext=text=%{localtime\\:%Y-%m-%d_%H\\:%M\\:%S}"
+            ":x=5:fontcolor=white"
+        )
+        self.assertEqual(validate_env_string(expr), expr)
+
+    def test_double_brace_escape_preserved(self):
+        """`{{output}}` collapses to `{output}` (documented go2rtc escape)."""
+        result = validate_env_string(
+            "exec:ffmpeg -i /media/file.mp4 -f rtsp {{output}}"
+        )
+        self.assertEqual(result, "exec:ffmpeg -i /media/file.mp4 -f rtsp {output}")
+
+    def test_double_brace_around_frigate_var(self):
+        """`{{FRIGATE_FOO}}` stays literal — escape takes precedence."""
+        FRIGATE_ENV_VARS["FRIGATE_FOO"] = "bar"
+        self.assertEqual(validate_env_string("{{FRIGATE_FOO}}"), "{FRIGATE_FOO}")
+
+    def test_mixed_frigate_var_and_braces(self):
+        """A FRIGATE_ var alongside literal single braces substitutes only the var."""
+        FRIGATE_ENV_VARS["FRIGATE_USER"] = "admin"
+        result = validate_env_string(
+            "drawtext=text=%{localtime}:user={FRIGATE_USER}:x=5"
+        )
+        self.assertEqual(result, "drawtext=text=%{localtime}:user=admin:x=5")
+
+    def test_triple_braces_around_frigate_var(self):
+        """`{{{FRIGATE_FOO}}}` collapses like str.format(): `{bar}`."""
+        FRIGATE_ENV_VARS["FRIGATE_FOO"] = "bar"
+        self.assertEqual(validate_env_string("{{{FRIGATE_FOO}}}"), "{bar}")
+
+    def test_trailing_double_brace_after_var(self):
+        """`{FRIGATE_FOO}}}` collapses like str.format(): `bar}`."""
+        FRIGATE_ENV_VARS["FRIGATE_FOO"] = "bar"
+        self.assertEqual(validate_env_string("{FRIGATE_FOO}}}"), "bar}")
+
+    def test_leading_double_brace_then_var(self):
+        """`{{{FRIGATE_FOO}` collapses like str.format(): `{bar`."""
+        FRIGATE_ENV_VARS["FRIGATE_FOO"] = "bar"
+        self.assertEqual(validate_env_string("{{{FRIGATE_FOO}"), "{bar")
+
+    def test_malformed_unterminated_placeholder_raises(self):
+        """`{FRIGATE_FOO` (no closing brace) raises like str.format() did."""
+        FRIGATE_ENV_VARS["FRIGATE_FOO"] = "bar"
+        with self.assertRaises(ValueError):
+            validate_env_string("prefix-{FRIGATE_FOO")
+
+    def test_malformed_format_spec_raises(self):
+        """`{FRIGATE_FOO:>5}` (format spec) raises like str.format() did."""
+        FRIGATE_ENV_VARS["FRIGATE_FOO"] = "bar"
+        with self.assertRaises(ValueError):
+            validate_env_string("{FRIGATE_FOO:>5}")
+
+    def test_malformed_conversion_raises(self):
+        """`{FRIGATE_FOO!r}` (conversion) raises like str.format() did."""
+        FRIGATE_ENV_VARS["FRIGATE_FOO"] = "bar"
+        with self.assertRaises(ValueError):
+            validate_env_string("{FRIGATE_FOO!r}")
+
 
 class TestEnvVars(unittest.TestCase):
     def setUp(self):