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

"""Creates a go2rtc config file."""

import json
import os
import sys
from typing import Any

from ruamel.yaml import YAML

sys.path.insert(0, "/opt/frigate")
from frigate.const import BIRDSEYE_PIPE
from frigate.ffmpeg_presets import parse_preset_hardware_acceleration_encode
from frigate.util.config import (
    find_config_file,
    is_valid_ffmpeg_path,
    resolve_ffmpeg_path,
)
from frigate.util.env import apply_config_env_vars, substitute_frigate_vars
from frigate.util.live_streams import raw_transcode_streams
from frigate.util.services import (
    apply_go2rtc_ffmpeg_config,
    is_go2rtc_arbitrary_exec_allowed,
    is_restricted_go2rtc_source,
)

sys.path.remove("/opt/frigate")

yaml = YAML()

config_file = find_config_file()

try:
    with open(config_file) as f:
        raw_config = f.read()

    if config_file.endswith((".yaml", ".yml")):
        config: dict[str, Any] = yaml.load(raw_config)
    elif config_file.endswith(".json"):
        config: dict[str, Any] = json.loads(raw_config)
except FileNotFoundError:
    config: dict[str, Any] = {}

# No validator runs here, so install environment_vars ourselves. FRIGATE_
# names only: anything else lands in os.environ, where the exec gate reads
# GO2RTC_ALLOW_ARBITRARY_EXEC.
config_env_vars = config.get("environment_vars")
apply_config_env_vars(
    {
        key: value
        for key, value in config_env_vars.items()
        if str(key).startswith("FRIGATE_")
    }
    if isinstance(config_env_vars, dict)
    else {}
)

go2rtc_config: dict[str, Any] = config.get("go2rtc", {})

# Need to enable CORS for go2rtc so the frigate integration / card work automatically
if go2rtc_config.get("api") is None:
    go2rtc_config["api"] = {"origin": "*"}
elif go2rtc_config["api"].get("origin") is None:
    go2rtc_config["api"]["origin"] = "*"

# Need to set default location for HA config
if go2rtc_config.get("hass") is None:
    go2rtc_config["hass"] = {"config": "/homeassistant"}

# we want to ensure that logs are easy to read
if go2rtc_config.get("log") is None:
    go2rtc_config["log"] = {"format": "text"}
elif go2rtc_config["log"].get("format") is None:
    go2rtc_config["log"]["format"] = "text"

# ensure there is a default webrtc config
if go2rtc_config.get("webrtc") is None:
    go2rtc_config["webrtc"] = {}

if go2rtc_config["webrtc"].get("candidates") is None:
    default_candidates = []
    # use internal candidate if it was discovered when running through the add-on
    internal_candidate = os.environ.get("FRIGATE_GO2RTC_WEBRTC_CANDIDATE_INTERNAL")
    if internal_candidate is not None:
        default_candidates.append(internal_candidate)
    # should set default stun server so webrtc can work
    default_candidates.append("stun:8555")

    go2rtc_config["webrtc"]["candidates"] = default_candidates

if go2rtc_config.get("rtsp", {}).get("username") is not None:
    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"] = substitute_frigate_vars(
        go2rtc_config["rtsp"]["password"]
    )

# ensure ffmpeg path is set correctly
path = config.get("ffmpeg", {}).get("path", "default")
if not is_valid_ffmpeg_path(path):
    print(
        "[ERROR] ffmpeg path contains spaces, quotes or shell characters, using the default ffmpeg instead."
    )
ffmpeg_path = resolve_ffmpeg_path(path, "ffmpeg", "go2rtc")

for key in apply_go2rtc_ffmpeg_config(go2rtc_config, ffmpeg_path):
    print(
        f"[ERROR] go2rtc option '{key}' was ignored because it is disabled by default for security. "
        f"Set GO2RTC_ALLOW_ARBITRARY_EXEC=true to enable arbitrary exec."
    )


for name in list(go2rtc_config.get("streams", {})):
    stream = go2rtc_config["streams"][name]

    if isinstance(stream, str):
        try:
            formatted_stream = substitute_frigate_vars(stream)
            if is_restricted_go2rtc_source(formatted_stream):
                print(
                    f"[ERROR] Stream '{name}' uses a restricted source (echo/expr/exec or ${{VAR}}) which is disabled by default for security. "
                    f"Set GO2RTC_ALLOW_ARBITRARY_EXEC=true to enable arbitrary exec sources."
                )
                del go2rtc_config["streams"][name]
                continue
            go2rtc_config["streams"][name] = formatted_stream
        except ValueError as e:
            print(
                "[ERROR] Invalid substitution found, see https://docs.frigate.video/configuration/restream#advanced-restream-configurations for more info."
            )
            sys.exit(e)

    elif isinstance(stream, list):
        filtered_streams = []
        for i, stream_item in enumerate(stream):
            try:
                formatted_stream = substitute_frigate_vars(stream_item)
                if is_restricted_go2rtc_source(formatted_stream):
                    print(
                        f"[ERROR] Stream '{name}' item {i + 1} uses a restricted source (echo/expr/exec or ${{VAR}}) which is disabled by default for security. "
                        f"Set GO2RTC_ALLOW_ARBITRARY_EXEC=true to enable arbitrary exec sources."
                    )
                    continue

                filtered_streams.append(formatted_stream)
            except ValueError as e:
                print(
                    "[ERROR] Invalid substitution found, see https://docs.frigate.video/configuration/restream#advanced-restream-configurations for more info."
                )
                sys.exit(e)

        if filtered_streams:
            go2rtc_config["streams"][name] = filtered_streams
        else:
            print(
                f"[ERROR] Stream '{name}' was removed because all sources were restricted (echo/expr/exec or ${{VAR}}). "
                f"Set GO2RTC_ALLOW_ARBITRARY_EXEC=true to enable arbitrary exec sources."
            )
            del go2rtc_config["streams"][name]

    elif isinstance(stream, dict):
        # The map form ({"url": ...}) lets go2rtc resolve the source
        # recursively, so it is effectively a dynamic way to generate the URL
        # for a stream. That can only be backed by an exec source, so it cannot
        # be allowed unless arbitrary exec is explicitly enabled. When it is
        # enabled, leave the map untouched for go2rtc to resolve.
        if not is_go2rtc_arbitrary_exec_allowed():
            print(
                f"[ERROR] Stream '{name}' uses a dynamic source format which is disabled by default for security. "
                f"Set GO2RTC_ALLOW_ARBITRARY_EXEC=true to enable arbitrary exec sources."
            )
            del go2rtc_config["streams"][name]
            continue

# add transcoded live streams; a user stream with the same name wins here and
# fails Frigate's config validation
transcoded_streams = raw_transcode_streams(config)

if transcoded_streams:
    if go2rtc_config.get("streams") is None:
        go2rtc_config["streams"] = {}

    for name, source in transcoded_streams.items():
        go2rtc_config["streams"].setdefault(name, source)

# add birdseye restream stream if enabled
if config.get("birdseye", {}).get("restream", False):
    birdseye: dict[str, Any] = config.get("birdseye")

    input = f"-f rawvideo -pix_fmt yuv420p -video_size {birdseye.get('width', 1280)}x{birdseye.get('height', 720)} -r 10 -i {BIRDSEYE_PIPE}"
    ffmpeg_cmd = f"exec:{parse_preset_hardware_acceleration_encode(ffmpeg_path, config.get('ffmpeg', {}).get('hwaccel_args', ''), input, '-rtsp_transport tcp -f rtsp {output}')}"

    if go2rtc_config.get("streams"):
        go2rtc_config["streams"]["birdseye"] = ffmpeg_cmd
    else:
        go2rtc_config["streams"] = {"birdseye": ffmpeg_cmd}

# The run script removes the old file first. Exclusive create never follows a
# link the go2rtc user planted in sticky /dev/shm. The config contains camera
# credentials, so it is not world-readable.
fd = os.open(
    "/dev/shm/go2rtc.yaml",
    os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
    0o640,
)
with os.fdopen(fd, "w") as f:
    yaml.dump(go2rtc_config, f)