mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-11-15 01:53:02 +03:00
Add secrets.yaml and unify variable substitution sources (#24044)
* 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
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user