Container security hardening (phase 3, breaking) (#24081)

* Run the frigate service as the frigate user

* Run go2rtc as its own restricted user

* Run nginx as the frigate user with writable state in /tmp/nginx

* Disable bandwidth stats gracefully when not running as root

* Hand TensorRT model cache ownership to the runtime user

* Document non-root operation and per-hardware device access

* Create /media/frigate after the ownership sweep

* Assert non-root services, JWT migration, and escape hatch in CI

* only write the sweep sentinel when a media volume is mounted

* tolerate homekit config chown failures in the go2rtc run script

* chown the s6 log pipe so non-root nginx can reopen /dev/stdout

* set HOME to /config for non-root services

* run smoke nginx -t and the write probe as the runtime user

* re-own the nginx shm cache on service restart

* discard stdout for the unprivileged smoke nginx -t

* unwrap hard-wrapped prose in the installation docs

* report progress during the ownership sweep

* document EXTRA_GROUPS as the only device access path for dropped services

* expand the non-root device access docs with diagnosis steps and udev rules

* document network storage ownership and the remaining detector hardware

* skip lost+found during the ownership sweep

* hand /tmp/cache to the runtime user before services start

* make bundled models readable by the runtime user

* reload nginx by signaling the master instead of parsing its config as root

* harden root writes into unprivileged-owned paths

Restrict the sweep sentinel to a mount at or below /media/frigate so a
parent /media mount cannot bless a later-shadowed volume. Rebuild
/tmp/nginx root-owned each start so root's cp and tempio writes cannot
follow a symlink an unprivileged nginx planted in the previous run.

* collapse the duplicated sentinel comment

* add a service-runs-as-root helper for granular root services

* validate FRIGATE_ROOT_SERVICES and fail fast on unknown names

* let services listed in FRIGATE_ROOT_SERVICES skip the privilege drop

* record the root-services mode in the sentinel and sweep small trees each boot

* cache the runtime ids in the ownership helper

* chown recordings, previews, and exports to the runtime user at create

* chown the database files after init

* recommend FRIGATE_ROOT_SERVICES in the bandwidth stats warning

* assert granular root services in CI

* document FRIGATE_ROOT_SERVICES

* own every directory level created for a recording segment

* clear the cached runtime ids when ownership tests finish

* skip missing media paths in the per-boot ownership sweep

* clarify granular root services docs

* clean up

* install acl for device access grants

* grant runtime users access to mapped device nodes at boot

* assert device access grants in CI

* document automatic device access grants

* stop telling users device access needs host side setup

* clarify the non-root docs

* link the migration script to the repo

* group the manual device setup under one section

* harden against symlink attacks

/config is owned by the unprivileged runtime user after the ownership sweep, so root operations on files there could be redirected by a planted symlink.

- go2rtc HomeKit setup: replace the root yq/jq normalization and chown with an O_NOFOLLOW helper (prepare_homekit.py), so a symlink at go2rtc_homekit.yml can't redirect a root write or chown onto another file
- go2rtc binary override: ignore /config/go2rtc whenever the service runs as root, so a planted binary can't exec as root under FRIGATE_ROOT_SERVICES
- sweep sentinel: read and write it through safe-sentinel, which trusts only a root-owned regular file and never follows a symlink, so it can't be forged to skip the migration or symlinked to clobber a root file
- ownership sweep: chown with -execdir so a parent directory swapped for a symlink mid-walk can't redirect the chown out of the volume
- validate inputs: restrict DEVICE_ACL_PATHS to /dev, require nonzero numeric EXTRA_GROUPS, and reject PUID/PGID that collide with the go2rtc ids
- docs: correct the TLS key ownership note to match what actually happens

* tweak docs

* stop the ownership sweep chasing entries other mechanisms own

* keep custom binaries out of root services only under granular root
This commit is contained in:
Josh Hawkins
2026-09-12 07:30:04 -06:00
committed by Nicolas Mowen
parent 1693415375
commit 3be59c9c18
38 changed files with 1427 additions and 83 deletions
+11
View File
@@ -231,6 +231,17 @@ class FrigateApp:
migrate_db.close()
# a root frigate service creates these as root; wal and shm recreated
# later in the run are realigned by the per-boot /config sweep
for db_file in (
self.config.database.path,
f"{self.config.database.path}-wal",
f"{self.config.database.path}-shm",
self.config.database.path.replace("frigate.db", "backup.db"),
):
if os.path.exists(db_file):
chown_to_runtime(db_file)
def init_go2rtc(self) -> None:
for proc in psutil.process_iter(["pid", "name"]):
if proc.info["name"] == "go2rtc":
+2
View File
@@ -23,6 +23,7 @@ from frigate.ffmpeg_presets import (
)
from frigate.models import Previews
from frigate.util.image import copy_yuv_to_position, get_blank_yuv_frame, get_yuv_crop
from frigate.util.ownership import chown_to_runtime
logger = logging.getLogger(__name__)
@@ -185,6 +186,7 @@ class FFMpegConverter(threading.Thread):
if p.returncode == 0:
logger.debug("successfully saved preview")
chown_to_runtime(self.path)
self.requestor.send_data(
INSERT_PREVIEW,
{
+4
View File
@@ -35,6 +35,7 @@ from frigate.ffmpeg_presets import (
)
from frigate.models import Export, Previews, Recordings, ReviewSegment
from frigate.util.ffmpeg import run_ffmpeg_with_progress
from frigate.util.ownership import chown_to_runtime
from frigate.util.time import is_current_hour
logger = logging.getLogger(__name__)
@@ -941,6 +942,8 @@ class RecordingExporter(threading.Thread):
else:
video_path = f"{EXPORT_DIR}/{self.camera}_{filename_start_datetime}-{filename_end_datetime}_{cleaned_export_id}.mp4"
thumb_path = self.save_thumbnail(self.export_id)
if thumb_path:
chown_to_runtime(thumb_path)
export_values = {
Export.id: self.export_id,
@@ -1015,6 +1018,7 @@ class RecordingExporter(threading.Thread):
Path(thumb_path).unlink(missing_ok=True)
return
else:
chown_to_runtime(video_path)
self._emit_progress("finalizing", 100.0)
Export.update({Export.in_progress: False}).where(
Export.id == self.export_id
+8
View File
@@ -43,6 +43,7 @@ from frigate.const import (
from frigate.models import Recordings, ReviewSegment
from frigate.review.types import SeverityEnum
from frigate.util.media import get_keyframe_offsets
from frigate.util.ownership import chown_to_runtime
from frigate.util.services import get_video_properties
logger = logging.getLogger(__name__)
@@ -928,6 +929,11 @@ class RecordingMaintainer(threading.Thread):
)
os.makedirs(directory, exist_ok=True)
# own every level makedirs creates so the host user can prune recordings
level = directory
while level != RECORD_DIR:
chown_to_runtime(level)
level = os.path.dirname(level)
# file will be in utc due to path_time being in utc
file_name = f"{path_time.strftime('%M.%S.mp4')}"
@@ -966,6 +972,8 @@ class RecordingMaintainer(threading.Thread):
f"Copied {file_path} in {datetime.datetime.now().timestamp() - start_frame} seconds."
)
chown_to_runtime(file_path)
try:
# get the segment size of the cache file
# file without faststart is same size
+26
View File
@@ -0,0 +1,26 @@
"""Tests for bandwidth stats privilege handling."""
import unittest
from unittest.mock import MagicMock, patch
from frigate.util import services
class TestBandwidthStatsPrivileges(unittest.TestCase):
def setUp(self):
services._bandwidth_warning_logged = False
@patch("frigate.util.services.sp.run")
@patch("frigate.util.services.os.geteuid", return_value=1000)
def test_returns_empty_and_warns_once_without_root(self, _, sp_run):
config = MagicMock()
with self.assertLogs("frigate.util.services", level="WARNING") as logs:
assert services.get_bandwidth_stats(config) == {}
assert services.get_bandwidth_stats(config) == {}
sp_run.assert_not_called()
warnings = [m for m in logs.output if "require root" in m]
assert len(warnings) == 1
if __name__ == "__main__":
unittest.main()
+154
View File
@@ -0,0 +1,154 @@
"""Tests for custom ffmpeg path resolution and the root-mode guard."""
import unittest
from unittest.mock import patch
from frigate.const import DEFAULT_FFMPEG_VERSION
from frigate.util.config import (
_warn_ignored_ffmpeg_path,
frigate_service_is_granular_root,
resolve_ffmpeg_path,
)
BUNDLED = f"/usr/lib/ffmpeg/{DEFAULT_FFMPEG_VERSION}/bin/ffmpeg"
CUSTOM = "/config/custom-ffmpeg"
class TestConfigFfmpegRootGuard(unittest.TestCase):
"""A user-writable ffmpeg must not run as root under FRIGATE_ROOT_SERVICES."""
def setUp(self) -> None:
# the warning is memoized so it fires once per path, not per camera
_warn_ignored_ffmpeg_path.cache_clear()
def _resolve(self, path: str, *, euid: int, env: dict, binary: str = "ffmpeg"):
with (
patch("os.geteuid", return_value=euid),
patch.dict("os.environ", env, clear=True),
):
return resolve_ffmpeg_path(path, binary)
def test_custom_path_used_when_service_is_unprivileged(self) -> None:
self.assertEqual(
self._resolve(CUSTOM, euid=1000, env={}), f"{CUSTOM}/bin/ffmpeg"
)
def test_escape_hatch_keeps_working_exactly_as_before(self) -> None:
# FRIGATE_RUN_AS_ROOT never sweeps /config and leaves no unprivileged
# service, so a custom build there is as safe as it was pre-drop
self.assertEqual(
self._resolve(CUSTOM, euid=0, env={"FRIGATE_RUN_AS_ROOT": "true"}),
f"{CUSTOM}/bin/ffmpeg",
)
def test_custom_path_ignored_when_frigate_is_a_root_service(self) -> None:
self.assertEqual(
self._resolve(CUSTOM, euid=0, env={"FRIGATE_ROOT_SERVICES": "frigate"}),
BUNDLED,
)
def test_ffprobe_is_guarded_too(self) -> None:
self.assertEqual(
self._resolve(
CUSTOM,
euid=0,
env={"FRIGATE_ROOT_SERVICES": "frigate"},
binary="ffprobe",
),
f"/usr/lib/ffmpeg/{DEFAULT_FFMPEG_VERSION}/bin/ffprobe",
)
def test_another_root_service_does_not_trigger_the_guard(self) -> None:
# go2rtc running as root says nothing about who spawns ffmpeg
self.assertEqual(
self._resolve(CUSTOM, euid=0, env={"FRIGATE_ROOT_SERVICES": "go2rtc"}),
f"{CUSTOM}/bin/ffmpeg",
)
def test_root_services_has_no_effect_under_docker_user(self) -> None:
# docker's own user: means the service never had root to keep, which is
# also the case for get_ffmpeg_path.py in a --user container
self.assertEqual(
self._resolve(CUSTOM, euid=1000, env={"FRIGATE_ROOT_SERVICES": "frigate"}),
f"{CUSTOM}/bin/ffmpeg",
)
def test_path_outside_config_is_left_alone(self) -> None:
# only /config is runtime-user-owned; a root-owned tree stays usable
self.assertEqual(
self._resolve(
"/opt/custom-ffmpeg", euid=0, env={"FRIGATE_ROOT_SERVICES": "frigate"}
),
"/opt/custom-ffmpeg/bin/ffmpeg",
)
def test_traversal_out_of_config_does_not_evade_the_guard(self) -> None:
self.assertEqual(
self._resolve(
"/config/../config/custom-ffmpeg",
euid=0,
env={"FRIGATE_ROOT_SERVICES": "frigate"},
),
BUNDLED,
)
def test_media_tree_is_guarded_too(self) -> None:
# config.yml is uid-1000-writable, so ffmpeg.path can be pointed at any
# writable tree; /config alone would be an evasion, not a guard
self.assertEqual(
self._resolve(
"/media/frigate/evil", euid=0, env={"FRIGATE_ROOT_SERVICES": "frigate"}
),
BUNDLED,
)
def test_config_dir_itself_is_guarded(self) -> None:
self.assertEqual(
self._resolve("/config", euid=0, env={"FRIGATE_ROOT_SERVICES": "frigate"}),
BUNDLED,
)
def test_default_alias_is_unaffected(self) -> None:
self.assertEqual(
self._resolve("default", euid=0, env={"FRIGATE_ROOT_SERVICES": "frigate"}),
BUNDLED,
)
class TestFrigateServiceIsGranularRoot(unittest.TestCase):
"""Root via FRIGATE_ROOT_SERVICES only, never via the escape hatch."""
def _check(self, *, euid: int, env: dict) -> bool:
with (
patch("os.geteuid", return_value=euid),
patch.dict("os.environ", env, clear=True),
):
return frigate_service_is_granular_root()
def test_false_without_any_root_signal(self) -> None:
self.assertFalse(self._check(euid=0, env={}))
def test_escape_hatch_is_not_granular_root(self) -> None:
# the escape hatch restores old behavior wholesale, sweep included
self.assertFalse(self._check(euid=0, env={"FRIGATE_RUN_AS_ROOT": "true"}))
self.assertFalse(
self._check(
euid=0,
env={"FRIGATE_RUN_AS_ROOT": "true", "FRIGATE_ROOT_SERVICES": "frigate"},
)
)
def test_membership_ignores_whitespace_and_other_entries(self) -> None:
self.assertTrue(
self._check(euid=0, env={"FRIGATE_ROOT_SERVICES": "go2rtc, frigate"})
)
self.assertFalse(
self._check(euid=0, env={"FRIGATE_ROOT_SERVICES": "go2rtc,nginx"})
)
def test_substring_of_a_service_name_does_not_match(self) -> None:
self.assertFalse(self._check(euid=0, env={"FRIGATE_ROOT_SERVICES": "frigatee"}))
if __name__ == "__main__":
unittest.main()
+18
View File
@@ -14,6 +14,11 @@ class FakePwEntry:
# The devcontainer image exports FRIGATE_RUN_AS_ROOT, so any test that has to
# reach past the escape-hatch check pins the variable instead of inheriting it.
class TestGetRuntimeIds(unittest.TestCase):
def setUp(self) -> None:
ownership.get_runtime_ids.cache_clear()
# a value cached under this test's patches must not leak into later modules
self.addCleanup(ownership.get_runtime_ids.cache_clear)
@patch("frigate.util.ownership.os.geteuid", return_value=1000)
def test_returns_none_when_not_root(self, _):
assert ownership.get_runtime_ids() is None
@@ -35,8 +40,21 @@ class TestGetRuntimeIds(unittest.TestCase):
def test_returns_frigate_ids_as_root(self, *_):
assert ownership.get_runtime_ids() == (1500, 1500)
@patch.dict("os.environ", {"FRIGATE_RUN_AS_ROOT": "false"})
@patch("frigate.util.ownership.pwd.getpwnam", return_value=FakePwEntry())
@patch("frigate.util.ownership.os.geteuid", return_value=0)
def test_caches_lookup(self, _geteuid, getpwnam):
assert ownership.get_runtime_ids() == (1500, 1500)
assert ownership.get_runtime_ids() == (1500, 1500)
getpwnam.assert_called_once()
class TestChownToRuntime(unittest.TestCase):
def setUp(self) -> None:
ownership.get_runtime_ids.cache_clear()
# a value cached under this test's patches must not leak into later modules
self.addCleanup(ownership.get_runtime_ids.cache_clear)
@patch("frigate.util.ownership.os.chown")
@patch("frigate.util.ownership.get_runtime_ids", return_value=None)
def test_noop_when_no_runtime_ids(self, _, chown):
@@ -442,6 +442,71 @@ class TestSegmentPathTime(unittest.IsolatedAsyncioTestCase):
self.assertTrue(paths[1].endswith("30.23.mp4"), paths[1])
class TestMoveSegmentOwnership(unittest.IsolatedAsyncioTestCase):
"""Every directory level makedirs creates must be handed to the runtime user."""
def _build_maintainer(self) -> RecordingMaintainer:
camera_config = MagicMock()
camera_config.record.enabled = True
camera_config.record.continuous.days = 1
camera_config.record.motion.days = 0
config = MagicMock()
config.cameras = {"test_cam": camera_config}
maintainer = RecordingMaintainer.__new__(RecordingMaintainer)
maintainer.config = config
maintainer.end_time_cache = {}
maintainer.object_recordings_info = defaultdict(list)
maintainer.audio_recordings_info = defaultdict(list)
maintainer.recordings_publisher = MagicMock()
maintainer.last_segment_end = {("test_cam", "main"): 0.0}
return maintainer
async def test_move_segment_chowns_all_created_levels(self):
maintainer = self._build_maintainer()
maintainer.config.ffmpeg.ffmpeg_path = "ffmpeg"
start_time = datetime.datetime(2026, 6, 10, 14, 30, 22, tzinfo=datetime.UTC)
proc = MagicMock()
proc.returncode = 0
proc.wait = AsyncMock(return_value=0)
chown = MagicMock()
with tempfile.TemporaryDirectory() as tmpdir:
record_dir = os.path.join(tmpdir, "recordings")
cache_path = os.path.join(tmpdir, "test_cam@20260610143022+0000.mp4")
with open(cache_path, "wb") as f:
f.write(b"\x00" * 16)
with (
patch("frigate.record.maintainer.RECORD_DIR", record_dir),
patch(
"frigate.record.maintainer.asyncio.create_subprocess_exec",
AsyncMock(return_value=proc),
),
patch("frigate.record.maintainer.chown_to_runtime", chown),
):
result = await maintainer.move_segment(
"test_cam",
"main",
start_time,
start_time + datetime.timedelta(seconds=10),
10.0,
cache_path,
SegmentInfo(0, 0, 0, 0),
)
self.assertIsNotNone(result)
camera_dir = os.path.join(record_dir, "2026-06-10", "14", "test_cam")
hour_dir = os.path.dirname(camera_dir)
date_dir = os.path.dirname(hour_dir)
file_path = os.path.join(camera_dir, "30.22.mp4")
chowned = [call.args[0] for call in chown.call_args_list]
self.assertEqual(chowned, [camera_dir, hour_dir, date_dir, file_path])
class TestSegmentStartChaining(unittest.IsolatedAsyncioTestCase):
"""Contiguous segments must chain start times across filename truncation.
+52 -2
View File
@@ -4,11 +4,14 @@ import asyncio
import logging
import os
import shutil
from functools import cache
from typing import Any
from ruamel.yaml import YAML
from frigate.const import (
BASE_DIR,
CACHE_DIR,
CONFIG_DIR,
DEFAULT_FFMPEG_VERSION,
EXPORT_DIR,
@@ -43,13 +46,56 @@ DROPPED_DETECTOR_OPTIONS = {
}
# Trees the unprivileged runtime user can write. A root frigate service must
# not execute a binary from any of them; a compromised uid-1000 process could
# plant one and be root after the next restart.
RUNTIME_USER_WRITABLE_DIRS = (CONFIG_DIR, BASE_DIR, CACHE_DIR, "/dev/shm", "/tmp")
def frigate_service_is_granular_root() -> bool:
"""Report whether FRIGATE_ROOT_SERVICES runs frigate as root.
The escape hatch is excluded: it never sweeps /config and leaves no
unprivileged service running, so custom binaries stay as safe as they
were before the privilege drop.
"""
if os.geteuid() != 0:
return False
if os.environ.get("FRIGATE_RUN_AS_ROOT", "false") == "true":
return False
entries = os.environ.get("FRIGATE_ROOT_SERVICES", "").split(",")
return any("".join(entry.split()) == "frigate" for entry in entries)
def _is_runtime_user_writable(path: str) -> bool:
"""Report whether a path resolves inside a runtime-user-writable tree."""
resolved = os.path.realpath(path)
return any(
resolved == root or resolved.startswith(f"{root}{os.sep}")
for root in RUNTIME_USER_WRITABLE_DIRS
)
@cache
def _warn_ignored_ffmpeg_path(path: str) -> None:
"""Warn once per path; resolution runs per camera and per binary."""
logger.warning(
"Ignoring ffmpeg.path %s because FRIGATE_ROOT_SERVICES runs frigate as root and that location is writable by the unprivileged user; using the bundled build",
path,
)
def resolve_ffmpeg_path(path: str, binary: str = "ffmpeg") -> str:
"""Resolve an ffmpeg version alias or custom path to a binary path.
A bare version alias that is no longer bundled (for example one that was
dropped when the default version changed) falls back to the default
bundled version so existing configs keep working across an upgrade or a
revert. Custom install paths (anything absolute) are used as-is.
revert. Custom install paths (anything absolute) are used as-is, except
one in a runtime-user-writable tree while FRIGATE_ROOT_SERVICES makes
frigate root; see RUNTIME_USER_WRITABLE_DIRS.
"""
if path == "default" or (
not path.startswith("/") and path not in INCLUDED_FFMPEG_VERSIONS
@@ -58,7 +104,11 @@ def resolve_ffmpeg_path(path: str, binary: str = "ffmpeg") -> str:
elif path in INCLUDED_FFMPEG_VERSIONS:
version = path
else:
return f"{path}/bin/{binary}"
if not (frigate_service_is_granular_root() and _is_runtime_user_writable(path)):
return f"{path}/bin/{binary}"
_warn_ignored_ffmpeg_path(path)
version = DEFAULT_FFMPEG_VERSION
return f"/usr/lib/ffmpeg/{version}/bin/{binary}"
+4
View File
@@ -1,5 +1,6 @@
"""Helpers for aligning created files with the non-root runtime user."""
import functools
import logging
import os
import pwd
@@ -9,12 +10,15 @@ logger = logging.getLogger(__name__)
RUNTIME_USER = "frigate"
@functools.lru_cache(maxsize=1)
def get_runtime_ids() -> tuple[int, int] | None:
"""Return (uid, gid) that services run as, or None when chown is not applicable.
None when: not root (docker --user, so the host already mapped us),
FRIGATE_RUN_AS_ROOT=true (escape hatch must not mutate ownership),
or outside the Frigate container image (no frigate user).
The result is cached for the process lifetime because the runtime user
cannot change after boot.
"""
if os.geteuid() != 0:
return None
+15
View File
@@ -187,8 +187,23 @@ def get_physical_interfaces(interfaces) -> list:
return physical_interfaces
_bandwidth_warning_logged = False
def get_bandwidth_stats(config) -> dict[str, dict]:
"""Get bandwidth usages for each ffmpeg process id"""
global _bandwidth_warning_logged
if os.geteuid() != 0:
if not _bandwidth_warning_logged:
logger.warning(
"Network bandwidth stats require root (nethogs needs CAP_NET_ADMIN/CAP_NET_RAW) "
"and are disabled; set FRIGATE_ROOT_SERVICES=frigate (or FRIGATE_RUN_AS_ROOT=true) "
"or disable telemetry.stats.network_bandwidth to silence this warning"
)
_bandwidth_warning_logged = True
return {}
usages = {}
top_command = ["nethogs", "-t", "-v0", "-c5", "-d1"] + get_physical_interfaces(
config.telemetry.network_interfaces