administrator/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-09-28 04:48:58 +03:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Add secrets.yaml and unify variable substitution sources (#24044)

Browse Source 
* add secrets.yaml and merge substitution sources by precedence

FRIGATE_ENV_VARS was built once at import from container env and /run/secrets, and the environment_vars validator overwrote it unconditionally, so the block beat the deployment and nothing could be re-read. Sources are now separate dicts merged lowest to highest (environment_vars, secrets.yaml, container env, credentials directory), re-read at the top of every parse, and a collision warns once naming the winner. An undefined {FRIGATE_*} raises a ValueError subclass so pydantic reports the field instead of a KeyError traceback.

* use the shared substitution namespace in go2rtc config

The generator rebuilt the namespace itself from os.environ and a hardcoded /run/secrets, so it never saw environment_vars or CREDENTIALS_DIRECTORY, and str.format made any stray brace fatal. It now installs the FRIGATE_ names from environment_vars and substitutes streams the same way every other field does.

* 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.

* docs

* clarify docs
This commit is contained in:
Josh Hawkins
2026-09-12 07:30:04 -06:00
committed by Nicolas Mowen
parent c9c6a923ef
commit 853840dfd4
11 changed files with 604 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frigate/api/camera.py
+2 -2
View File
@@ -33,7 +33,7 @@ from frigate.config.camera.updater import (
CameraConfigUpdateEnum,
CameraConfigUpdateTopic,
)
from frigate.config.env import substitute_frigate_vars
from frigate.config.env import UnknownVariableError, substitute_frigate_vars
from frigate.models import User
from frigate.util.builtin import clean_camera_user_pass, get_record_segment_time
from frigate.util.camera_cleanup import cleanup_camera_db, cleanup_camera_files
@@ -166,7 +166,7 @@ def go2rtc_add_stream(request: Request, stream_name: str, src: str = ""):
if src:
try:
resolved_src = substitute_frigate_vars(src)
except KeyError:
except UnknownVariableError:
resolved_src = src
if is_restricted_go2rtc_source(resolved_src):
frigate/config/config.py
+4 -1
View File
@@ -63,7 +63,7 @@ from .classification import (
SemanticSearchModelEnum,
)
from .database import DatabaseConfig
from .env import EnvVars
from .env import EnvVars, reload_sources
from .logger import LoggerConfig
from .mqtt import MqttConfig
from .network import NetworkingConfig
@@ -1307,6 +1307,9 @@ class FrigateConfig(FrigateBaseModel):
@classmethod
def parse(cls, config, *, is_json=None, safe_load=False, **context):
# Pick up secrets.yaml edits without a restart.
reload_sources()
# If config is a file, read its contents.
if hasattr(config, "read"):
fname = getattr(config, "name", None)
frigate/config/env.py
+192 -18
View File
@@ -1,20 +1,193 @@
"""Environment variable and secrets handling for the Frigate config."""
import logging
import os
import re
from collections.abc import Mapping
from pathlib import Path
from typing import Annotated
from typing import Annotated, Any
from pydantic import AfterValidator, ValidationInfo
from ruamel.yaml import YAML, YAMLError
FRIGATE_ENV_VARS = {k: v for k, v in os.environ.items() if k.startswith("FRIGATE_")}
secrets_dir = os.environ.get("CREDENTIALS_DIRECTORY", "/run/secrets")
# read secret files as env vars too
if os.path.isdir(secrets_dir) and os.access(secrets_dir, os.R_OK):
for secret_file in os.listdir(secrets_dir):
if secret_file.startswith("FRIGATE_"):
FRIGATE_ENV_VARS[secret_file] = (
Path(os.path.join(secrets_dir, secret_file)).read_text().strip()
from frigate.const import CONFIG_DIR
logger = logging.getLogger(__name__)
class UnknownVariableError(ValueError):
"""Undefined {FRIGATE_*} placeholder. ValueError so pydantic names the field."""
# Substitution sources, lowest precedence first.
_CONFIG_ENV_VARS: dict[str, str] = {}
_SECRETS_FILE: dict[str, str] = {}
# Snapshot: apply_config_env_vars() writes os.environ after import.
_CONTAINER_ENV: dict[str, str] = {
k: v for k, v in os.environ.items() if k.startswith("FRIGATE_")
}
_CREDENTIALS_DIR: dict[str, str] = {}
_SOURCES: tuple[tuple[str, dict[str, str]], ...] = (
("environment_vars config block", _CONFIG_ENV_VARS),
("secrets.yaml", _SECRETS_FILE),
("container environment", _CONTAINER_ENV),
("credentials directory", _CREDENTIALS_DIR),
)
FRIGATE_ENV_VARS: dict[str, str] = {}
_WARNED_COLLISIONS: set[str] = set()
def _rebuild(warn: bool = True) -> None:
"""Merge the sources into FRIGATE_ENV_VARS.
warn=False is for the import-time call, before logging is configured.
"""
merged: dict[str, str] = {}
origin: dict[str, str] = {}
duplicated: set[str] = set()
for label, source in _SOURCES:
for key, value in source.items():
if key in merged and merged[key] != value:
duplicated.add(key)
merged[key] = value
origin[key] = label
if warn:
for key in sorted(duplicated - _WARNED_COLLISIONS):
_WARNED_COLLISIONS.add(key)
logger.warning(
"%s is defined in more than one place, using the value from %s",
key,
origin[key],
)
# In place: tests hold a reference to this dict.
FRIGATE_ENV_VARS.clear()
FRIGATE_ENV_VARS.update(merged)
def _load_credentials_dir() -> dict[str, str]:
"""Read FRIGATE_* files from the Docker or systemd credentials directory."""
directory = os.environ.get("CREDENTIALS_DIRECTORY", "/run/secrets")
values: dict[str, str] = {}
if not (os.path.isdir(directory) and os.access(directory, os.R_OK)):
return values
for name in os.listdir(directory):
if not name.startswith("FRIGATE_"):
continue
try:
values[name] = Path(os.path.join(directory, name)).read_text().strip()
except (OSError, UnicodeDecodeError):
logger.warning("Unable to read %s in %s, skipping", name, directory)
return values
def _secrets_file_path() -> str | None:
"""Locate secrets.yaml next to the config file."""
config_file = os.environ.get("CONFIG_FILE")
config_dir = os.path.dirname(config_file) if config_file else CONFIG_DIR
for name in ("secrets.yaml", "secrets.yml"):
path = os.path.join(config_dir, name)
if os.path.isfile(path):
return path
return None
def _load_secrets_file() -> dict[str, str]:
"""Read the flat FRIGATE_* map from secrets.yaml, if it exists."""
path = _secrets_file_path()
if path is None:
return {}
try:
with open(path) as f:
raw: Any = YAML(typ="safe").load(f)
except OSError as err:
raise ValueError(f"Unable to read {path}: {err.strerror}") from err
except YAMLError as err:
# The parser message can quote values, so only name a position.
mark = getattr(err, "problem_mark", None)
where = f" near line {mark.line + 1}" if mark is not None else ""
raise ValueError(f"{path} is not valid YAML{where}") from err
if raw is None:
return {}
if not isinstance(raw, dict):
raise ValueError(f"{path} must be a flat map of names to values")
values: dict[str, str] = {}
for key, value in raw.items():
name = str(key)
if isinstance(value, (dict, list)):
raise ValueError(f"{path} value for {name} must be a single value")
if not name.startswith("FRIGATE_"):
logger.warning(
"Ignoring %s in %s, names must start with FRIGATE_", name, path
)
continue
values[name] = "" if value is None else str(value)
return values
def reload_sources(warn: bool = True) -> None:
"""Re-read the file backed sources and rebuild the namespace."""
_CREDENTIALS_DIR.clear()
_CREDENTIALS_DIR.update(_load_credentials_dir())
try:
secrets = _load_secrets_file()
except ValueError as err:
# Keep the last good values; this runs at import and on every parse.
logger.error("Ignoring secrets file, %s", err)
else:
_SECRETS_FILE.clear()
_SECRETS_FILE.update(secrets)
_rebuild(warn)
def apply_config_env_vars(values: Mapping[str, object]) -> None:
"""Install the environment_vars block as the lowest priority source.
Unprefixed keys only set os.environ.
"""
for key, value in values.items():
resolved = str(value)
if key.startswith("FRIGATE_"):
_CONFIG_ENV_VARS[key] = resolved
else:
os.environ[key] = resolved
_rebuild()
# Export the winning value; auth reads FRIGATE_JWT_SECRET from os.environ.
for key in values:
if key.startswith("FRIGATE_"):
os.environ[key] = FRIGATE_ENV_VARS[key]
reload_sources(warn=False)
# Matches a FRIGATE_* identifier following an opening brace.
_FRIGATE_IDENT_RE = re.compile(r"FRIGATE_[A-Za-z0-9_]+")
@@ -29,12 +202,13 @@ def substitute_frigate_vars(value: str) -> str:
* `{{` 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.
raises `UnknownVariableError` 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.
placeholder). Callers that catch `UnknownVariableError` 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] = []
@@ -58,7 +232,10 @@ def substitute_frigate_vars(value: str) -> str:
):
key = ident_match.group(0)
if key not in FRIGATE_ENV_VARS:
raise KeyError(key)
raise UnknownVariableError(
f"{key} is not defined in the environment, "
"secrets.yaml, or the environment_vars config"
)
out.append(FRIGATE_ENV_VARS[key])
i = ident_match.end() + 1
continue
@@ -94,10 +271,7 @@ EnvString = Annotated[str, AfterValidator(validate_env_string)]
def validate_env_vars(v: dict[str, str], info: ValidationInfo) -> dict[str, str]:
if isinstance(info.context, dict) and info.context.get("install", False):
for k, val in v.items():
os.environ[k] = val
if k.startswith("FRIGATE_"):
FRIGATE_ENV_VARS[k] = val
apply_config_env_vars(v)
return v
frigate/test/http_api/test_http_camera_access.py
+15 -1
View File
@@ -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/test/test_env.py
+311 -2
View File
@@ -1,9 +1,13 @@
"""Tests for environment variable handling."""
import os
import tempfile
import unittest
from unittest.mock import MagicMock, patch
from pydantic import ValidationError
from frigate.config import FrigateConfig, env
from frigate.config.env import (
FRIGATE_ENV_VARS,
validate_env_string,
@@ -105,9 +109,10 @@ class TestEnvString(unittest.TestCase):
self.assertEqual(result, "192.168.1.1")
def test_unknown_var_raises(self):
"""Referencing an unknown var raises KeyError."""
with self.assertRaises(KeyError):
"""Referencing an unknown var raises UnknownVariableError."""
with self.assertRaises(env.UnknownVariableError) as ctx:
validate_env_string("{FRIGATE_NONEXISTENT_VAR}")
self.assertIn("FRIGATE_NONEXISTENT_VAR", str(ctx.exception))
def test_non_frigate_braces_passthrough(self):
"""Braces that are not {FRIGATE_*} placeholders pass through untouched.
@@ -176,6 +181,18 @@ class TestEnvString(unittest.TestCase):
validate_env_string("{FRIGATE_FOO!r}")
class TestUnknownVariableSurfacing(unittest.TestCase):
"""An undefined variable must reach the user as a config validation error."""
def test_unknown_var_is_a_validation_error(self):
"""Pydantic reports the field path instead of raising KeyError."""
with self.assertRaises(ValidationError) as ctx:
FrigateConfig.parse_object(
{"mqtt": {"host": "{FRIGATE_NOT_SET_ANYWHERE}"}, "cameras": {}}
)
self.assertIn("FRIGATE_NOT_SET_ANYWHERE", str(ctx.exception))
class TestEnvVars(unittest.TestCase):
def setUp(self):
self._original_env_vars = dict(FRIGATE_ENV_VARS)
@@ -184,6 +201,7 @@ class TestEnvVars(unittest.TestCase):
def tearDown(self):
FRIGATE_ENV_VARS.clear()
FRIGATE_ENV_VARS.update(self._original_env_vars)
env._CONFIG_ENV_VARS.clear()
# Clean up any env vars we set
for key in list(os.environ.keys()):
if key not in self._original_environ:
@@ -233,5 +251,296 @@ class TestEnvVars(unittest.TestCase):
self.assertEqual(result, "mqtt.local")
class TestVariableSources(unittest.TestCase):
"""Precedence between the sources that feed FRIGATE_ENV_VARS."""
def setUp(self):
self._original_env_vars = dict(env.FRIGATE_ENV_VARS)
self._original_os_environ = dict(os.environ)
def tearDown(self):
env.FRIGATE_ENV_VARS.clear()
env.FRIGATE_ENV_VARS.update(self._original_env_vars)
env._CONFIG_ENV_VARS.clear()
env._WARNED_COLLISIONS.clear()
os.environ.clear()
os.environ.update(self._original_os_environ)
def test_container_env_beats_config_env_vars(self):
"""A container env var wins over the same key in environment_vars."""
with patch.dict(env._CONTAINER_ENV, {"FRIGATE_MQTT_HOST": "from_env"}):
env.apply_config_env_vars({"FRIGATE_MQTT_HOST": "from_config"})
self.assertEqual(env.FRIGATE_ENV_VARS["FRIGATE_MQTT_HOST"], "from_env")
def test_credentials_dir_beats_container_env(self):
"""A credentials directory file wins over a container env var."""
with (
patch.dict(env._CONTAINER_ENV, {"FRIGATE_MQTT_HOST": "from_env"}),
patch.dict(env._CREDENTIALS_DIR, {"FRIGATE_MQTT_HOST": "from_creds"}),
):
env._rebuild()
self.assertEqual(env.FRIGATE_ENV_VARS["FRIGATE_MQTT_HOST"], "from_creds")
def test_config_env_vars_used_when_no_other_source(self):
"""environment_vars still resolves when nothing else defines the key."""
env.apply_config_env_vars({"FRIGATE_CAM_PASS": "hunter2"})
self.assertEqual(env.FRIGATE_ENV_VARS["FRIGATE_CAM_PASS"], "hunter2")
def test_config_env_vars_do_not_become_container_env(self):
"""environment_vars writes os.environ but must not gain env precedence."""
env.apply_config_env_vars({"FRIGATE_CAM_PASS": "from_config"})
self.assertEqual(os.environ["FRIGATE_CAM_PASS"], "from_config")
self.assertNotIn("FRIGATE_CAM_PASS", env._CONTAINER_ENV)
def test_non_frigate_config_env_vars_only_set_os_environ(self):
"""Unprefixed environment_vars keys reach os.environ but not substitution."""
env.apply_config_env_vars({"LIBVA_DRIVER_NAME": "i965"})
self.assertEqual(os.environ["LIBVA_DRIVER_NAME"], "i965")
self.assertNotIn("LIBVA_DRIVER_NAME", env.FRIGATE_ENV_VARS)
def test_collision_warns_once_naming_the_winner(self):
"""A key from two sources logs one warning naming the source that won."""
with patch.dict(env._CONTAINER_ENV, {"FRIGATE_MQTT_HOST": "from_env"}):
with self.assertLogs("frigate.config.env", level="WARNING") as logs:
env.apply_config_env_vars({"FRIGATE_MQTT_HOST": "from_config"})
self.assertEqual(len(logs.output), 1)
self.assertIn("FRIGATE_MQTT_HOST", logs.output[0])
self.assertIn("using the value from container environment", logs.output[0])
self.assertNotIn("environment_vars", logs.output[0])
with self.assertNoLogs("frigate.config.env", level="WARNING"):
env._rebuild()
class TestSecretsFile(unittest.TestCase):
"""Reading <config dir>/secrets.yaml."""
def setUp(self):
self._original_env_vars = dict(env.FRIGATE_ENV_VARS)
self._original_os_environ = dict(os.environ)
self._dir = tempfile.TemporaryDirectory()
self.addCleanup(self._dir.cleanup)
os.environ["CONFIG_FILE"] = os.path.join(self._dir.name, "config.yml")
def tearDown(self):
env.FRIGATE_ENV_VARS.clear()
env.FRIGATE_ENV_VARS.update(self._original_env_vars)
env._SECRETS_FILE.clear()
env._CONFIG_ENV_VARS.clear()
env._WARNED_COLLISIONS.clear()
os.environ.clear()
os.environ.update(self._original_os_environ)
def _write(self, contents: str, name: str = "secrets.yaml") -> None:
with open(os.path.join(self._dir.name, name), "w") as f:
f.write(contents)
def test_missing_file_is_not_an_error(self):
"""No secrets.yaml means no values and no exception."""
self.assertEqual(env._load_secrets_file(), {})
def test_flat_map_is_read(self):
"""A flat FRIGATE_* map loads."""
self._write("FRIGATE_CAM_USER: viewer\nFRIGATE_CAM_PASS: 'p@ss w0rd'\n")
self.assertEqual(
env._load_secrets_file(),
{"FRIGATE_CAM_USER": "viewer", "FRIGATE_CAM_PASS": "p@ss w0rd"},
)
def test_yml_extension_is_read(self):
"""secrets.yml works the same as secrets.yaml."""
self._write("FRIGATE_CAM_USER: viewer\n", name="secrets.yml")
self.assertEqual(env._load_secrets_file(), {"FRIGATE_CAM_USER": "viewer"})
def test_numeric_value_is_coerced_to_string(self):
"""Unquoted numbers become strings so they can be substituted."""
self._write("FRIGATE_MQTT_PORT: 1883\n")
self.assertEqual(env._load_secrets_file(), {"FRIGATE_MQTT_PORT": "1883"})
def test_unprefixed_key_is_ignored_with_a_warning(self):
"""Names must start with FRIGATE_, matching the credentials directory."""
self._write("cam_pass: hunter2\nFRIGATE_CAM_PASS: hunter2\n")
with self.assertLogs("frigate.config.env", level="WARNING") as logs:
values = env._load_secrets_file()
self.assertEqual(values, {"FRIGATE_CAM_PASS": "hunter2"})
self.assertIn("cam_pass", logs.output[0])
def test_non_mapping_document_raises(self):
"""A list or scalar document is a config error."""
self._write("- FRIGATE_CAM_PASS\n")
with self.assertRaises(ValueError):
env._load_secrets_file()
def test_nested_value_raises_naming_the_key(self):
"""Nesting is not supported and the error names the key."""
self._write("FRIGATE_CAMS:\n alley: hunter2\n")
with self.assertRaises(ValueError) as ctx:
env._load_secrets_file()
self.assertIn("FRIGATE_CAMS", str(ctx.exception))
def test_secrets_file_beats_config_env_vars(self):
"""secrets.yaml outranks the environment_vars block."""
self._write("FRIGATE_CAM_PASS: from_secrets\n")
env.apply_config_env_vars({"FRIGATE_CAM_PASS": "from_config"})
env._SECRETS_FILE.update(env._load_secrets_file())
env._rebuild()
self.assertEqual(env.FRIGATE_ENV_VARS["FRIGATE_CAM_PASS"], "from_secrets")
def test_container_env_beats_secrets_file(self):
"""The container environment outranks secrets.yaml."""
self._write("FRIGATE_CAM_PASS: from_secrets\n")
env._SECRETS_FILE.update(env._load_secrets_file())
with patch.dict(env._CONTAINER_ENV, {"FRIGATE_CAM_PASS": "from_env"}):
env._rebuild()
self.assertEqual(env.FRIGATE_ENV_VARS["FRIGATE_CAM_PASS"], "from_env")
class TestSecretsReload(unittest.TestCase):
"""secrets.yaml is re-read when a config is parsed."""
def setUp(self):
self._original_env_vars = dict(env.FRIGATE_ENV_VARS)
self._original_os_environ = dict(os.environ)
self._dir = tempfile.TemporaryDirectory()
self.addCleanup(self._dir.cleanup)
os.environ["CONFIG_FILE"] = os.path.join(self._dir.name, "config.yml")
def tearDown(self):
env.FRIGATE_ENV_VARS.clear()
env.FRIGATE_ENV_VARS.update(self._original_env_vars)
env._SECRETS_FILE.clear()
env._CONFIG_ENV_VARS.clear()
env._WARNED_COLLISIONS.clear()
os.environ.clear()
os.environ.update(self._original_os_environ)
env.reload_sources()
def test_new_secret_resolves_without_restart(self):
"""A key written after import is picked up by the next parse."""
with open(os.path.join(self._dir.name, "secrets.yaml"), "w") as f:
f.write("FRIGATE_MQTT_HOST: mqtt.internal\n")
config = FrigateConfig.parse_yaml(
'mqtt:\n host: "{FRIGATE_MQTT_HOST}"\ncameras: {}\n'
)
self.assertEqual(config.mqtt.host, "mqtt.internal")
def test_config_env_vars_survive_the_reload(self):
"""environment_vars is only installed when install=True, so it has to
outlive the reload that a later non-install parse triggers. This is
the /config/save path: a config that starts fine must still validate.
"""
env.apply_config_env_vars({"FRIGATE_MQTT_HOST": "from_config"})
config = FrigateConfig.parse_yaml(
'mqtt:\n host: "{FRIGATE_MQTT_HOST}"\ncameras: {}\n'
)
self.assertEqual(config.mqtt.host, "from_config")
self.assertEqual(env._CONFIG_ENV_VARS["FRIGATE_MQTT_HOST"], "from_config")
class TestSourceRobustness(unittest.TestCase):
"""Reload behavior, bad input, and the os.environ export."""
def setUp(self):
self._original_env_vars = dict(env.FRIGATE_ENV_VARS)
self._original_os_environ = dict(os.environ)
self._dir = tempfile.TemporaryDirectory()
self._creds = tempfile.TemporaryDirectory()
self.addCleanup(self._dir.cleanup)
self.addCleanup(self._creds.cleanup)
os.environ["CONFIG_FILE"] = os.path.join(self._dir.name, "config.yml")
os.environ["CREDENTIALS_DIRECTORY"] = self._creds.name
def tearDown(self):
env.FRIGATE_ENV_VARS.clear()
env.FRIGATE_ENV_VARS.update(self._original_env_vars)
env._SECRETS_FILE.clear()
env._CONFIG_ENV_VARS.clear()
env._CREDENTIALS_DIR.clear()
env._WARNED_COLLISIONS.clear()
os.environ.clear()
os.environ.update(self._original_os_environ)
env.reload_sources()
def _write_secrets(self, contents: str) -> None:
with open(os.path.join(self._dir.name, "secrets.yaml"), "w") as f:
f.write(contents)
def test_os_environ_gets_the_winning_value(self):
"""FRIGATE_JWT_SECRET and friends are read straight from os.environ."""
self._write_secrets("FRIGATE_JWT_SECRET: from_secrets\n")
env.reload_sources()
env.apply_config_env_vars({"FRIGATE_JWT_SECRET": "from_config"})
self.assertEqual(os.environ["FRIGATE_JWT_SECRET"], "from_secrets")
self.assertEqual(
os.environ["FRIGATE_JWT_SECRET"],
env.FRIGATE_ENV_VARS["FRIGATE_JWT_SECRET"],
)
def test_rebuild_without_warn_stays_quiet_then_warns_later(self):
"""The import-time rebuild must not consume the one-shot warning."""
self._write_secrets("FRIGATE_DUPE: from_secrets\n")
env.reload_sources()
env._CONFIG_ENV_VARS["FRIGATE_DUPE"] = "from_config"
with self.assertNoLogs("frigate.config.env", level="WARNING"):
env._rebuild(warn=False)
with self.assertLogs("frigate.config.env", level="WARNING") as logs:
env._rebuild()
self.assertIn("FRIGATE_DUPE", logs.output[0])
def test_malformed_secrets_file_keeps_last_good_values(self):
"""A typo must not raise, since this runs at import and on every parse."""
self._write_secrets("FRIGATE_CAM_PASS: hunter2\n")
env.reload_sources()
self._write_secrets("FRIGATE_CAMS:\n alley: hunter2\n")
with self.assertLogs("frigate.config.env", level="ERROR") as logs:
env.reload_sources()
self.assertIn("FRIGATE_CAMS", logs.output[0])
self.assertEqual(env.FRIGATE_ENV_VARS["FRIGATE_CAM_PASS"], "hunter2")
def test_duplicate_key_error_does_not_log_the_values(self):
"""ruamel's duplicate key message quotes both values; the log must not."""
self._write_secrets("FRIGATE_CAM_PASS: hunter2\nFRIGATE_CAM_PASS: hunter3\n")
with self.assertLogs("frigate.config.env", level="ERROR") as logs:
env.reload_sources()
self.assertNotIn("hunter2", logs.output[0])
self.assertNotIn("hunter3", logs.output[0])
self.assertIn("secrets.yaml", logs.output[0])
def test_deleted_secret_stops_resolving(self):
"""Removing a name takes effect on the next parse, not on restart."""
self._write_secrets("FRIGATE_GONE: hunter2\n")
env.reload_sources()
self.assertEqual(env.FRIGATE_ENV_VARS["FRIGATE_GONE"], "hunter2")
os.remove(os.path.join(self._dir.name, "secrets.yaml"))
env.reload_sources()
self.assertNotIn("FRIGATE_GONE", env.FRIGATE_ENV_VARS)
def test_reload_rereads_the_credentials_directory(self):
"""The credentials directory is refreshed, not just secrets.yaml."""
with open(os.path.join(self._creds.name, "FRIGATE_CRED"), "w") as f:
f.write("from_creds\n")
env.reload_sources()
self.assertEqual(env.FRIGATE_ENV_VARS["FRIGATE_CRED"], "from_creds")
def test_unreadable_credentials_entry_is_skipped(self):
"""A subdirectory must not take down validation on every parse."""
os.mkdir(os.path.join(self._creds.name, "FRIGATE_NOT_A_FILE"))
with open(os.path.join(self._creds.name, "FRIGATE_CRED"), "w") as f:
f.write("from_creds\n")
with self.assertLogs("frigate.config.env", level="WARNING") as logs:
env.reload_sources()
self.assertIn("FRIGATE_NOT_A_FILE", logs.output[0])
self.assertEqual(env.FRIGATE_ENV_VARS["FRIGATE_CRED"], "from_creds")
if __name__ == "__main__":
unittest.main()
frigate/util/services.py
+7 -2
View File
@@ -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)