mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-09-28 19:06:52 +03:00
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.
This commit is contained in:
Josh Hawkins
@@ -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):
|
||||
|
||||
@@ -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)
|
||||
|
||||
+192
-18
@@ -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
|
||||
|
||||
|
||||
+311
-2
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user