From 49c373272649aee030b4e3a1e27ef0a754128f6e Mon Sep 17 00:00:00 2001 From: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com> Date: Tue, 7 Apr 2026 08:16:02 -0500 Subject: [PATCH] Improve environment var handling (#22796) * refactor env var handling - use shared helper - use left-to-right parser * add tests * formatting --- .../rootfs/usr/local/go2rtc/create_config.py | 21 +-- frigate/api/camera.py | 4 +- frigate/config/env.py | 72 +++++++++- frigate/test/test_env.py | 132 ++++++++++++++++++ 4 files changed, 212 insertions(+), 17 deletions(-) diff --git a/docker/main/rootfs/usr/local/go2rtc/create_config.py b/docker/main/rootfs/usr/local/go2rtc/create_config.py index fb701a9b6..71897be0a 100644 --- a/docker/main/rootfs/usr/local/go2rtc/create_config.py +++ b/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( - **FRIGATE_ENV_VARS + go2rtc_config["rtsp"]["username"] = substitute_frigate_vars( + go2rtc_config["rtsp"]["username"] ) if go2rtc_config.get("rtsp", {}).get("password") is not None: - go2rtc_config["rtsp"]["password"] = go2rtc_config["rtsp"]["password"].format( - **FRIGATE_ENV_VARS + go2rtc_config["rtsp"]["password"] = substitute_frigate_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. " diff --git a/frigate/api/camera.py b/frigate/api/camera.py index 1881bde6d..7a3b19439 100644 --- a/frigate/api/camera.py +++ b/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 diff --git a/frigate/config/env.py b/frigate/config/env.py index db094a8af..209dda67b 100644 --- a/frigate/config/env.py +++ b/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)] diff --git a/frigate/test/test_env.py b/frigate/test/test_env.py index fe2ce8d3e..37b81a656 100644 --- a/frigate/test/test_env.py +++ b/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):