Compare commits

...

9 Commits

Author SHA1 Message Date
Josh Hawkins
349de9189d
Merge b5a360be39 into e84a89ef3e 2026-06-16 06:11:41 +08:00
Josh Hawkins
e84a89ef3e
fix camera audio availability detection on mobile live grid (#23488)
Some checks are pending
CI / AMD64 Extra Build (push) Blocked by required conditions
CI / ARM Extra Build (push) Blocked by required conditions
CI / Synaptics Build (push) Blocked by required conditions
CI / AMD64 Build (push) Waiting to run
CI / ARM Build (push) Waiting to run
CI / Jetson Jetpack 6 (push) Waiting to run
CI / Assemble and push default build (push) Blocked by required conditions
2026-06-15 07:26:19 -06:00
Josh Hawkins
ba29e141da
Docs tweaks (#23487)
* docs tweaks

* tweak

* title tweak
2026-06-15 07:03:37 -06:00
Nicolas Mowen
32e433cafc
Allow GenAI providers to be initialized lazily (#23482)
Some checks are pending
CI / AMD64 Build (push) Waiting to run
CI / ARM Build (push) Waiting to run
CI / Jetson Jetpack 6 (push) Waiting to run
CI / AMD64 Extra Build (push) Blocked by required conditions
CI / ARM Extra Build (push) Blocked by required conditions
CI / Synaptics Build (push) Blocked by required conditions
CI / Assemble and push default build (push) Blocked by required conditions
* allow GenAI providers to be initialized even if they failed on previous attempts

* mypy
2026-06-14 11:40:33 -05:00
Josh Hawkins
bc816926a5
Replace export ffmpeg argument blocklist with a structural allowlist (#23478)
Some checks are pending
CI / Synaptics Build (push) Blocked by required conditions
CI / AMD64 Build (push) Waiting to run
CI / ARM Build (push) Waiting to run
CI / Jetson Jetpack 6 (push) Waiting to run
CI / AMD64 Extra Build (push) Blocked by required conditions
CI / ARM Extra Build (push) Blocked by required conditions
CI / Assemble and push default build (push) Blocked by required conditions
* use allowlist for custom export ffmpeg args

* reject brackets in export filtergraph validation instead of stripping link labels
2026-06-13 16:43:22 -06:00
Josh Hawkins
b79ad9871a
generate the API docs OpenAPI spec from the app with per-endpoint auth requirements (#23476) 2026-06-13 16:04:22 -06:00
Josh Hawkins
8be7a97fa6
guard norfair distance against non-finite and zero-area boxes that could crash autotracking cameras (#23475) 2026-06-13 16:23:30 -05:00
Josh Hawkins
b5a360be39 add test 2026-04-17 17:18:11 -05:00
Josh Hawkins
54a7c5015e fix birdseye layout calculation
replace the two pass layout with a single pass pixel space algorithm
2026-04-17 17:18:04 -05:00
22 changed files with 4018 additions and 1637 deletions

View File

@ -125,5 +125,7 @@ jobs:
run: devcontainer up --workspace-folder .
- name: Run mypy in devcontainer
run: devcontainer exec --workspace-folder . bash -lc "python3 -u -m mypy --config-file frigate/mypy.ini frigate"
- name: Check API spec is up to date
run: devcontainer exec --workspace-folder . bash -lc "python3 generate_api_auth_spec.py --check"
- name: Run unit tests in devcontainer
run: devcontainer exec --workspace-folder . bash -lc "python3 -u -m unittest"

View File

@ -235,6 +235,14 @@ ruff check frigate/
# Type check
python3 -u -m mypy --config-file frigate/mypy.ini frigate
# Regenerate the OpenAPI spec after adding, changing, or removing an API
# endpoint or its auth dependency — outputs docs/static/frigate-api.yaml,
# annotated with each endpoint's auth requirement (admin / any / camera /
# public). NEVER edit that file by hand. CI runs the --check variant and fails
# if it is out of date. (from repo root)
python3 generate_api_auth_spec.py
python3 generate_api_auth_spec.py --check
```
### Frontend (from web/ directory)
@ -316,6 +324,8 @@ async def get_events(request: Request, limit: int = 100):
# Implementation
```
After adding, changing, or removing an endpoint (or its auth dependency), regenerate the OpenAPI spec with `python3 generate_api_auth_spec.py` so `docs/static/frigate-api.yaml` stays in sync and the endpoint's auth requirement is documented. CI enforces this via the `--check` variant; never edit that file by hand.
### Configuration Access
```python

View File

@ -6,10 +6,16 @@ import NavPath from "@site/src/components/NavPath";
In addition to Frigate's Live camera dashboard, Birdseye allows a portable heads-up view of your cameras to see what is going on around your property / space without having to watch all cameras that may have nothing happening. Birdseye allows specific modes that intelligently show and disappear based on what you care about.
Birdseye can be viewed by adding the "Birdseye" camera to a Camera Group in the Web UI. Add a Camera Group by pressing the "+" icon on the Live page, and choose "Birdseye" as one of the cameras.
Birdseye can be viewed by adding the "Birdseye" camera to a Camera Group in the Web UI. Add a Camera Group by pressing the pencil icon in the sidebar on the Live page, and choose "Birdseye" as one of the cameras.
Birdseye can also be used in Home Assistant dashboards, cast to media devices, etc.
:::note
Each camera tile in Birdseye is composed from the frames of the stream assigned the `detect` role, so a camera's image quality in Birdseye matches its detect stream resolution rather than a higher-resolution recording stream. If a camera looks low quality in Birdseye, increasing the detect width and height (or assigning the `detect` role to a higher-resolution stream) is what affects it. See [setting up camera inputs](./cameras.md#setting-up-camera-inputs) for how roles are assigned.
:::
## Birdseye Behavior
### Birdseye Modes
@ -35,10 +41,10 @@ To include a camera in Birdseye view only for specific circumstances, or exclude
**Per-camera overrides:** Navigate to <NavPath path="Settings > Camera configuration > Birdseye" /> to override the mode or disable Birdseye for a specific camera.
| Field | Description |
|-------|-------------|
| **Enable Birdseye** | Whether this camera appears in Birdseye view |
| **Tracking mode** | When to show the camera: `continuous`, `motion`, or `objects` |
| Field | Description |
| ------------------- | ------------------------------------------------------------- |
| **Enable Birdseye** | Whether this camera appears in Birdseye view |
| **Tracking mode** | When to show the camera: `continuous`, `motion`, or `objects` |
</TabItem>
<TabItem value="yaml">
@ -72,8 +78,8 @@ By default birdseye shows all cameras that have had the configured activity in t
Navigate to <NavPath path="Settings > System > Birdseye" />.
| Field | Description |
|-------|-------------|
| Field | Description |
| ------------------------ | --------------------------------------------------------------------------- |
| **Inactivity threshold** | Seconds of inactivity before a camera is hidden from Birdseye (default: 30) |
</TabItem>
@ -100,9 +106,9 @@ The resolution and aspect ratio of birdseye can be configured. Resolution will i
Navigate to <NavPath path="Settings > System > Birdseye" />.
| Field | Description |
|-------|-------------|
| **Width** | Birdseye output width in pixels (default: 1280) |
| Field | Description |
| ---------- | ----------------------------------------------- |
| **Width** | Birdseye output width in pixels (default: 1280) |
| **Height** | Birdseye output height in pixels (default: 720) |
</TabItem>
@ -161,8 +167,8 @@ It is possible to limit the number of cameras shown on birdseye at one time. Whe
Navigate to <NavPath path="Settings > System > Birdseye" />.
| Field | Description |
|-------|-------------|
| Field | Description |
| ------------------------ | ----------------------------------------------------------------------------------- |
| **Layout > Max cameras** | Maximum number of cameras shown at once (e.g., `1` for only the most active camera) |
</TabItem>
@ -187,8 +193,8 @@ By default birdseye tries to fit 2 cameras in each row and then double in size u
Navigate to <NavPath path="Settings > System > Birdseye" />.
| Field | Description |
|-------|-------------|
| Field | Description |
| --------------------------- | -------------------------------------------------------- |
| **Layout > Scaling factor** | Camera scaling factor between 1.0 and 5.0 (default: 2.0) |
</TabItem>

View File

@ -9,11 +9,54 @@ import NavPath from "@site/src/components/NavPath";
Frigate can be configured through the **Settings UI** or by editing the YAML configuration file directly. The Settings UI is the recommended approach — it provides validation and a guided experience for all configuration options.
It is recommended to start with a minimal configuration and add to it as described in [the getting started guide](../guides/getting_started.md).
## Using the Settings UI
The Settings UI groups every configuration option into sections that are listed in the left-hand menu. Each section presents a guided form with validation, so you don't need to remember the structure of the YAML or look up option names by hand.
### Global vs. camera-level configuration
Settings are organized into two scopes:
- **Global configuration** — values under <NavPath path="Settings > Global configuration" /> apply to every camera by default. This is where you set the baseline behavior for object detection, recording, snapshots, motion, and so on.
- **Camera configuration** — values under <NavPath path="Settings > Camera configuration" /> apply to a single camera. Use the camera selector button at the top of these pages to choose which camera you are editing.
When a camera-level section is left untouched, the camera simply inherits the global values. Changing a value on a camera page **overrides** the global value for that camera only — the global setting and every other camera are unaffected. This mirrors how the YAML works, where a value set under `cameras.<name>` takes precedence over the same value set at the top level.
To undo an override and go back to inheriting from the parent scope, use the reset button at the bottom of the section:
- On a camera section, the button is labeled **Reset to Global** and restores the camera to the global value.
- On a global section, the button is labeled **Reset to Default** and restores Frigate's built-in default.
Resetting asks for confirmation and cannot be undone once applied.
### Saving changes and the Save All button
Edits are not applied until you save them. As soon as you change a value, the UI tracks it as a pending change:
- The edited section shows a **Modified** badge, and the changed fields are highlighted.
- A **You have unsaved changes** notice appears above the section's **Save** and **Undo** buttons. **Save** commits just that section; **Undo** discards its pending edits.
Because pending changes can span multiple sections — and multiple cameras — the header provides a **Save All** button that writes every pending change at once. Next to it, **Review pending changes** opens a summary that lists each pending edit with its scope (Global or a specific camera), the affected field, and the new value, so you can confirm exactly what will be written before committing. **Undo All** discards every pending change across all sections.
### Restart-required indicators
Most settings take effect immediately, but some require Frigate to restart before they apply. Fields that require a restart are marked with a small restart icon and a **Restart required** tooltip next to the field label.
When you save a change that touches one of these fields, Frigate confirms the save and reminds you that a restart is needed (for example, _"Settings saved successfully. Restart Frigate to apply your changes."_). The notification includes a one-click **Restart Frigate** action so you can apply the change right away, or you can continue editing and restart later.
### The colored dots in the camera configuration menu
When you are working under <NavPath path="Settings > Camera configuration" />, small colored dots can appear next to a section's name in the menu. They give you an at-a-glance summary of that section's state for the selected camera:
- **Blue dot** — this section **overrides the global configuration**. One or more values in the section have been set specifically for this camera and differ from the global defaults.
- **Profile-colored dot** — when you are viewing a [camera profile](./profiles.md), a dot in that profile's assigned color indicates the section is **overridden by that profile**. Each profile is given its own distinct color so you can tell at a glance which sections it changes.
- **Amber dot** — this section has **unsaved changes**. It appears alongside the **Modified** badge whenever you have pending edits in the section that haven't been saved yet.
Hover over any dot to see a tooltip describing what it means. Open a section to see exactly which fields are overridden — the section header indicates how many fields differ from the global (or base) configuration.
## Configuration File Location
For users who prefer to edit the YAML configuration file directly:
For users who prefer to edit the YAML configuration file directly, it is recommended to start with a minimal configuration and add to it as described in [the getting started guide](../guides/getting_started.md).
- **Home Assistant App:** `/addon_configs/<addon_directory>/config.yml` — see [directory list](#accessing-app-config-dir)
- **All other installations:** Map to `/config/config.yml` inside the container

View File

@ -371,7 +371,7 @@ When your browser runs into problems playing back your camera streams, it will l
- Verify your camera's H.264/AAC settings (see [Frigate's camera settings recommendations](#camera-settings-recommendations)).
- Check go2rtc configuration for transcoding (e.g., audio to AAC/OPUS).
- Test with a different stream via the UI dropdown (if `live -> streams` is configured).
- For WebRTC-specific issues, ensure port 8555 is forwarded and candidates are set (see (WebRTC Extra Configuration)(#webrtc-extra-configuration)).
- For WebRTC-specific issues, ensure port 8555 is forwarded and candidates are set (see [WebRTC Extra Configuration](#webrtc-extra-configuration)).
- If your cameras are streaming at a high resolution, your browser may be struggling to load all of the streams before the buffering timeout occurs. Frigate prioritizes showing a true live view as quickly as possible. If the fallback occurs often, change your live view settings to use a lower bandwidth substream.
3. **It doesn't seem like my cameras are streaming on the Live dashboard. Why?**

View File

@ -8,10 +8,13 @@ import ConfigTabs from "@site/src/components/ConfigTabs";
import TabItem from "@theme/TabItem";
import NavPath from "@site/src/components/NavPath";
# Supported Hardware
### Supported hardware
Object detection is what allows Frigate to identify _what_ is in your camera's view — people, cars, animals, and more — rather than just reacting to pixel changes. When Frigate's motion detection finds activity in a frame, that region is sent to an **object detector**, which returns the objects it recognizes along with their location and a confidence score. These detections are what drive tracked objects, alerts, detections, and notifications.
Object detection is computationally intensive, so Frigate is designed to run it on a dedicated AI accelerator or GPU rather than the CPU. A **detector** is the specific hardware-and-model backend Frigate uses to run inference. Choosing a detector that matches your hardware is one of the most important steps in getting good performance, and the right choice depends on what device Frigate is running on.
:::info
Frigate supports multiple different detectors that work on different types of hardware:
**Most Hardware**

View File

@ -7,13 +7,17 @@ import ConfigTabs from "@site/src/components/ConfigTabs";
import TabItem from "@theme/TabItem";
import NavPath from "@site/src/components/NavPath";
Frigate can save a snapshot image to `/media/frigate/clips` for each object that is detected named as `<camera>-<id>-clean.webp`. They are also accessible [via the api](../integrations/api/event-snapshot-events-event-id-snapshot-jpg-get.api.mdx)
A snapshot is a single still image that captures a tracked object at its best moment — the clearest frame Frigate saw while following that object across the scene. Unlike a [recording](./record.md), which is continuous video, a snapshot is one representative image saved per tracked object once tracking ends.
Snapshots are accessible in the UI in the Explore pane. This allows for quick submission to the Frigate+ service.
When snapshots are enabled, Frigate saves one image to `/media/frigate/clips` for each tracked object, named `<camera>-<id>-clean.webp`. A clean image is always stored without any annotations (no timestamp, bounding boxes, or cropping) so you have an unmodified copy of the original frame. Annotations like bounding boxes and timestamps are applied on demand when a snapshot is requested [via the HTTP API](../integrations/api/event-snapshot-events-event-id-snapshot-jpg-get.api.mdx) — see [Rendering](#rendering) below.
To only save snapshots for objects that enter a specific zone, [see the zone docs](./zones.md#restricting-snapshots-to-specific-zones)
A few things to keep in mind:
Snapshots sent via MQTT are configured separately under the camera MQTT settings, not here.
- Snapshots are saved per tracked object, so a camera with no detected objects produces no snapshots even if recording is enabled.
- Snapshots and recordings are configured and retained independently — enabling one does not enable the other.
- Snapshots are accessible in the UI in the Explore pane, which allows for quick submission to the Frigate+ service.
- To only save snapshots for objects that enter a specific zone, [see the zone docs](./zones.md#restricting-snapshots-to-specific-zones).
- Snapshots sent via MQTT are configured separately under the camera MQTT settings, not here.
## Enabling Snapshots

View File

@ -19,6 +19,12 @@ You can open History from several places:
Use the **Back** button to return where you came from, or the **Live** button to jump to the current camera's live view.
:::tip
If you see **"No recordings found for this time"**, the most common causes are: recording was not enabled for that camera at the time of the event; the retention window has since expired and those segments were removed; or storage ran low and Frigate deleted them early to free space. See [Recording](/configuration/record) to verify your retention settings.
:::
## Timeline, Events, and Detail
A toggle (a drawer on mobile) switches the side panel between three modes:

View File

@ -11,7 +11,7 @@ This page describes how to _use_ the Live view. For how to _configure_ live stre
## The dashboard at a glance
The default **All Cameras** dashboard shows every camera, with a filmstrip of recent **alerts** scrolling across the top. Clicking an alert opens it in [Review](/usage/review); each card also has a check button to mark it reviewed without leaving the dashboard.
The default **All Cameras** dashboard shows every camera, with a filmstrip of recent **alerts** scrolling across the top. Clicking an alert opens it in [Review](/usage/review); each card also has a check button to mark it reviewed without leaving the dashboard. Only **alerts** appear in the filmstrip — to suppress a label or zone from showing there, configure it as a detection instead (see [Alerts and Detections](/configuration/review#alerts-and-detections)).
By default Frigate uses **smart streaming**: a camera's image updates roughly once per minute while nothing is happening, and switches to a full live stream the moment activity is detected. This conserves bandwidth and resources. You can change this per camera or per group (see [Streaming settings](#streaming-settings-and-the-right-click-menu) below), and the behavior is explained in detail under [Live view technologies](/configuration/live#live-view-technologies).
@ -58,7 +58,7 @@ You can optionally overlay live streaming statistics (stream type, bandwidth, la
## Streaming settings and the right-click menu
Right-clicking (or long-pressing) a camera tile opens a context menu with quick controls: an **audio volume** control for streams that support audio, **Mute / Unmute all cameras**, **show or hide streaming statistics**, the **debug view**, **notification** options, and — for admins — turning the camera on or off.
Right-clicking (or long-pressing) a camera tile opens a context menu with quick controls: an **audio volume** control for streams that support audio, **Mute / Unmute all cameras**, **show or hide streaming statistics**, the **debug view**, **notification** options, and — for admins — turning the camera on or off. If the audio control doesn't appear, see [Audio Support](/configuration/live#audio-support) — audio requires go2rtc configured with a compatible codec.
A **Low-bandwidth mode** notice may also appear in the context menu with a **Reset** option appears when Frigate has fallen back to the lower-quality jsmpeg stream — see the [Live view FAQ](/configuration/live#live-view-faq) for why this happens.

View File

@ -11,7 +11,7 @@ This page describes how to _use_ the Review view. For how alerts and detections
:::info
Review items are only created for a camera when **recording is enabled** for that camera. See [Recording](/configuration/record).
Review items are only created for a camera when **object tracking and recording are enabled** for that camera. See [Recording](/configuration/record).
:::
@ -39,7 +39,7 @@ Review items are shown as a grid of thumbnail cards next to a vertical activity
- The object chip on each card is **gray** when the item is unreviewed and turns **green** once it has been reviewed.
- The **Mark these items as reviewed** button marks everything currently shown as reviewed at once.
Reviewed state is tracked per user, so marking an item reviewed does not hide it for other users.
Reviewed state is tracked per user, so marking an item reviewed does not hide it for other users. Marking an item reviewed does not delete anything — the footage and the review item itself remain until they expire via retention.
## Selecting and acting on multiple items

File diff suppressed because it is too large Load Diff

View File

@ -5,6 +5,7 @@ import json
import logging
import os
import re
import time
from typing import Any, AsyncGenerator, Callable, Optional
import numpy as np
@ -50,6 +51,10 @@ def register_genai_provider(key: GenAIProviderEnum) -> Callable:
class GenAIClient:
"""Generative AI client for Frigate."""
# Minimum seconds between re-initialization attempts when the provider was
# offline at startup
REINIT_INTERVAL = 60.0
def __init__(
self,
genai_config: GenAIConfig,
@ -60,6 +65,34 @@ class GenAIClient:
self.timeout = timeout
self.validate_model = validate_model
self.provider = self._init_provider()
self._last_init_attempt = time.monotonic()
def ensure_provider(self) -> bool:
"""Ensure a provider is available, retrying initialization if needed.
Providers can fail to initialize at startup when their backing service
isn't online yet (common when both are started together). This retries
``_init_provider`` lazily throttled to ``REINIT_INTERVAL`` so the
client recovers on its own once the service is reachable, without a
config reload.
Returns True if a provider is available.
"""
if self.provider is not None:
return True
now = time.monotonic()
if now - self._last_init_attempt < self.REINIT_INTERVAL:
return False
self._last_init_attempt = now
self.provider = self._init_provider()
if self.provider is not None:
logger.info(
"GenAI provider %s is now available",
self.genai_config.provider,
)
return self.provider is not None
def generate_review_description(
self,

View File

@ -62,7 +62,9 @@ class GenAIClientManager:
def _get_client(self, name: str) -> "Optional[GenAIClient]":
"""Return the client for *name*, creating it on first access."""
if name in self._clients:
return self._clients[name]
client = self._clients[name]
client.ensure_provider()
return client
from frigate.genai import PROVIDERS
@ -78,7 +80,7 @@ class GenAIClientManager:
return None
try:
client: "GenAIClient" = provider_cls(genai_cfg)
client = provider_cls(genai_cfg)
except Exception as e:
logger.exception(
"Failed to create GenAI client for provider %s: %s",

View File

@ -595,112 +595,92 @@ class BirdsEyeFrameManager:
) -> Optional[list[list[Any]]]:
"""Calculate the optimal layout for 2+ cameras."""
def map_layout(
camera_layout: list[list[Any]], row_height: int
) -> tuple[int, int, Optional[list[list[Any]]]]:
"""Map the calculated layout."""
candidate_layout = []
starting_x = 0
x = 0
max_width = 0
y = 0
def find_available_x(
current_x: int,
width: int,
reserved_ranges: list[tuple[int, int]],
max_width: int,
) -> Optional[int]:
"""Find the first horizontal slot that does not collide with reservations."""
x = current_x
for row in camera_layout:
final_row = []
max_width = max(max_width, x)
x = starting_x
for cameras in row:
camera_dims = self.cameras[cameras[0]]["dimensions"].copy()
camera_aspect = cameras[1]
for reserved_start, reserved_end in sorted(reserved_ranges):
if x >= reserved_end:
continue
if camera_dims[1] > camera_dims[0]:
scaled_height = int(row_height * 2)
scaled_width = int(scaled_height * camera_aspect)
starting_x = scaled_width
else:
scaled_height = row_height
scaled_width = int(scaled_height * camera_aspect)
if x + width <= reserved_start:
return x
# layout is too large
if (
x + scaled_width > self.canvas.width
or y + scaled_height > self.canvas.height
):
return x + scaled_width, y + scaled_height, None
x = max(x, reserved_end)
final_row.append((cameras[0], (x, y, scaled_width, scaled_height)))
x += scaled_width
if x + width <= max_width:
return x
y += row_height
candidate_layout.append(final_row)
if max_width == 0:
max_width = x
return max_width, y, candidate_layout
canvas_aspect_x, canvas_aspect_y = self.canvas.get_aspect(coefficient)
camera_layout: list[list[Any]] = []
camera_layout.append([])
starting_x = 0
x = starting_x
y = 0
y_i = 0
max_y = 0
for camera in cameras_to_add:
camera_dims = self.cameras[camera]["dimensions"].copy()
camera_aspect_x, camera_aspect_y = self.canvas.get_camera_aspect(
camera, camera_dims[0], camera_dims[1]
)
if camera_dims[1] > camera_dims[0]:
portrait = True
else:
portrait = False
if (x + camera_aspect_x) <= canvas_aspect_x:
# insert if camera can fit on current row
camera_layout[y_i].append(
(
camera,
camera_aspect_x / camera_aspect_y,
)
)
if portrait:
starting_x = camera_aspect_x
else:
max_y = max(
max_y,
camera_aspect_y,
)
x += camera_aspect_x
else:
# move on to the next row and insert
y += max_y
y_i += 1
camera_layout.append([])
x = starting_x
if x + camera_aspect_x > canvas_aspect_x:
return None
camera_layout[y_i].append(
(
camera,
camera_aspect_x / camera_aspect_y,
)
)
x += camera_aspect_x
if y + max_y > canvas_aspect_y:
return None
row_height = int(self.canvas.height / coefficient)
total_width, total_height, standard_candidate_layout = map_layout(
camera_layout, row_height
)
def map_layout(row_height: int) -> tuple[int, int, Optional[list[list[Any]]]]:
"""Lay out cameras row by row while reserving portrait spans for the next row."""
candidate_layout: list[list[Any]] = []
reserved_ranges: dict[int, list[tuple[int, int]]] = {}
current_row: list[Any] = []
row_index = 0
row_y = 0
row_x = 0
max_width = 0
max_height = 0
for camera in cameras_to_add:
camera_dims = self.cameras[camera]["dimensions"].copy()
camera_aspect_x, camera_aspect_y = self.canvas.get_camera_aspect(
camera, camera_dims[0], camera_dims[1]
)
portrait = camera_dims[1] > camera_dims[0]
scaled_height = row_height * 2 if portrait else row_height
scaled_width = int(scaled_height * (camera_aspect_x / camera_aspect_y))
while True:
x = find_available_x(
row_x,
scaled_width,
reserved_ranges.get(row_index, []),
self.canvas.width,
)
if x is not None and row_y + scaled_height <= self.canvas.height:
current_row.append(
(camera, (x, row_y, scaled_width, scaled_height))
)
row_x = x + scaled_width
max_width = max(max_width, row_x)
max_height = max(max_height, row_y + scaled_height)
if portrait:
reserved_ranges.setdefault(row_index + 1, []).append(
(x, row_x)
)
break
if current_row:
candidate_layout.append(current_row)
current_row = []
row_index += 1
row_y = row_index * row_height
row_x = 0
if row_y + scaled_height > self.canvas.height:
overflow_width = max(max_width, scaled_width)
overflow_height = row_y + scaled_height
return overflow_width, overflow_height, None
if current_row:
candidate_layout.append(current_row)
return max_width, max_height, candidate_layout
row_height = max(1, int(self.canvas.height / coefficient))
total_width, total_height, standard_candidate_layout = map_layout(row_height)
if not standard_candidate_layout:
# if standard layout didn't work
@ -709,9 +689,9 @@ class BirdsEyeFrameManager:
total_width / self.canvas.width,
total_height / self.canvas.height,
)
row_height = int(row_height / scale_down_percent)
row_height = max(1, int(row_height / scale_down_percent))
total_width, total_height, standard_candidate_layout = map_layout(
camera_layout, row_height
row_height
)
if not standard_candidate_layout:
@ -725,8 +705,8 @@ class BirdsEyeFrameManager:
1 / (total_width / self.canvas.width),
1 / (total_height / self.canvas.height),
)
row_height = int(row_height * scale_up_percent)
_, _, scaled_layout = map_layout(camera_layout, row_height)
row_height = max(1, int(row_height * scale_up_percent))
_, _, scaled_layout = map_layout(row_height)
if scaled_layout:
return scaled_layout

View File

@ -48,6 +48,22 @@ def ptz_moving_at_frame_time(frame_time, ptz_start_time, ptz_stop_time):
)
def transform_is_finite(coord_transformations) -> bool:
"""Return True if a norfair coordinate transform contains only finite values.
A near-singular homography (common when the motion estimator can't find
enough stable features during zoom on a low-texture scene) can produce
inf/nan matrix entries. norfair accumulates the homography across frames, so
a single bad transform poisons every subsequent one and propagates nan into
the tracker's distance function, crashing the camera process.
"""
for attr in ("homography_matrix", "inverse_homography_matrix", "movement_vector"):
value = getattr(coord_transformations, attr, None)
if value is not None and not np.all(np.isfinite(value)):
return False
return True
class PtzMotionEstimator:
def __init__(self, config: CameraConfig, ptz_metrics: PTZMetrics) -> None:
self.frame_manager = SharedMemoryFrameManager()
@ -135,6 +151,19 @@ class PtzMotionEstimator:
)
self.coord_transformations = None
# A degenerate homography can yield non-finite transform values that
# norfair would accumulate and feed to the tracker as nan estimates.
# Drop the bad transform and request a reset so the estimator rebuilds
# a fresh reference frame instead of poisoning every following frame.
if self.coord_transformations is not None and not transform_is_finite(
self.coord_transformations
):
logger.warning(
f"Autotracker: motion estimator produced a non-finite transform for {camera} at frame time {frame_time}, resetting"
)
self.coord_transformations = None
self.ptz_metrics.reset.set()
try:
logger.debug(
f"{camera}: Motion estimator transformation: {self.coord_transformations.rel_to_abs([[0, 0]])}"

View File

@ -42,33 +42,118 @@ TIMELAPSE_DATA_INPUT_ARGS = "-an -skip_frame nokey"
# Captures the floating-point factor so we can scale expected duration.
SETPTS_FACTOR_RE = re.compile(r"setpts=([0-9]*\.?[0-9]+)\*PTS")
# ffmpeg flags that can read from or write to arbitrary files
BLOCKED_FFMPEG_ARGS = frozenset(
# Allowlisted flags that take no value.
_VALUELESS_FLAGS = frozenset({"-an", "-sn", "-dn"})
# Allowlisted filter flags. Their value is validated as a filtergraph and may
# only reference filters in _SAFE_FILTERS.
_FILTER_FLAGS = frozenset({"-vf", "-af", "-filter"})
# Allowlisted flags that take exactly one value (encoder / muxer-safe options).
_VALUE_FLAGS = frozenset(
{
"-i",
"-filter_script",
"-filter_complex",
"-lavfi",
"-vf",
"-af",
"-filter",
"-vstats_file",
"-passlogfile",
"-sdp_file",
"-dump_attachment",
"-attach",
"-c",
"-codec",
"-b",
"-crf",
"-qp",
"-q",
"-qscale",
"-preset",
"-tune",
"-profile",
"-level",
"-pix_fmt",
"-r",
"-g",
"-keyint_min",
"-sc_threshold",
"-bf",
"-refs",
"-qmin",
"-qmax",
"-maxrate",
"-minrate",
"-bufsize",
"-movflags",
"-threads",
"-aspect",
"-fps_mode",
"-vsync",
"-skip_frame",
}
)
_ALLOWED_FLAGS = _VALUELESS_FLAGS | _FILTER_FLAGS | _VALUE_FLAGS
# Filters that cannot read files, load plugins, or open network sources.
_SAFE_FILTERS = frozenset(
{
"setpts",
"fps",
"scale",
"format",
"transpose",
"hflip",
"vflip",
"crop",
"pad",
"setsar",
"setdar",
}
)
# Conservative shape for a non-filter flag value. Excludes "/" (paths /
# filtergraph division), whitespace, brackets, and a leading "-" so a value
# can never be a path or swallow a following flag. ":" is permitted for values
# like "16:9".
_SAFE_VALUE_RE = re.compile(r"^[A-Za-z0-9_.:+][A-Za-z0-9_.:+-]*$")
# Substrings inside a filtergraph that indicate a file-reading filter option.
# "movie=" also matches "amovie=" as a substring.
_BLOCKED_FILTER_VALUE_MARKERS = ("movie=", "textfile=", "filename=", "fontfile=")
def _base_flag(token: str) -> str:
"""Return a flag's base name, lowercased and without its stream specifier.
e.g. "-c:v" -> "-c", "-filter:a:0" -> "-filter".
"""
return token.lower().split(":", 1)[0]
def _validate_filtergraph(value: str) -> tuple[bool, str]:
"""Validate a filtergraph value, allowing only filters in _SAFE_FILTERS."""
# None of the safe filters need any of these
if any(token in value for token in ("://", "..", "[", "]")):
return False, "Invalid filter graph in custom ffmpeg arguments"
lowered = value.lower()
if any(marker in lowered for marker in _BLOCKED_FILTER_VALUE_MARKERS):
return False, "File-reading filters are not allowed in custom ffmpeg arguments"
# Filters are separated by "," within a chain and ";" between chains. Safe
# filters never use unescaped "," or ";" in their arguments, so splitting on
# them to recover filter names cannot hide a disallowed filter.
for spec in re.split(r"[;,]", value):
spec = spec.strip()
if not spec:
continue
name = spec.split("=", 1)[0].strip().lower()
if name not in _SAFE_FILTERS:
return False, f"Filter not allowed in custom ffmpeg arguments: {name}"
return True, ""
def validate_ffmpeg_args(args: str) -> tuple[bool, str]:
"""Validate that user-provided ffmpeg args don't allow input/output injection.
"""Validate user-provided custom export ffmpeg args with an allowlist.
Blocks:
- The -i flag and other flags that read/write arbitrary files
- Filter flags (can read files via movie=/amovie= source filters)
- Absolute/relative file paths (potential extra outputs)
- URLs and ffmpeg protocol references (data exfiltration)
Every token must be an allowlisted flag or the value of one; filter values
may only reference safe filters; and no token may become a bare input or
output URL. This structurally prevents arbitrary file read/write, network
exfiltration/SSRF, and resource-exhaustion via the export endpoint.
Admin users skip this validation entirely since they are trusted.
"""
@ -76,26 +161,36 @@ def validate_ffmpeg_args(args: str) -> tuple[bool, str]:
return True, ""
tokens = args.split()
for token in tokens:
# Block flags that could inject inputs or write to arbitrary files
if token.lower() in BLOCKED_FFMPEG_ARGS:
i = 0
while i < len(tokens):
token = tokens[i]
# A bare (non-flag) token here would be parsed by ffmpeg as an input or
# output URL. Only the server sets inputs/outputs, never the user.
if not token.startswith("-"):
return False, f"Unexpected argument in custom ffmpeg arguments: {token}"
base = _base_flag(token)
if base not in _ALLOWED_FLAGS:
return False, f"Forbidden ffmpeg argument: {token}"
# Block tokens that look like file paths (potential output injection)
if (
token.startswith("/")
or token.startswith("./")
or token.startswith("../")
or token.startswith("~")
):
return False, "File paths are not allowed in custom ffmpeg arguments"
if base in _VALUELESS_FLAGS:
i += 1
continue
# Block URLs and ffmpeg protocol references (e.g. http://, tcp://, pipe:, file:)
if "://" in token or token.startswith("pipe:") or token.startswith("file:"):
return (
False,
"Protocol references are not allowed in custom ffmpeg arguments",
)
# Remaining flags consume exactly one value.
if i + 1 >= len(tokens):
return False, f"Missing value for ffmpeg argument: {token}"
value = tokens[i + 1]
if base in _FILTER_FLAGS:
valid, message = _validate_filtergraph(value)
if not valid:
return False, message
elif not _SAFE_VALUE_RE.match(value):
return False, f"Invalid value for {token}: {value}"
i += 2
return True, ""

View File

@ -1,11 +1,64 @@
"""Test camera user and password cleanup."""
"""Tests for Birdseye canvas sizing and layout behavior."""
import unittest
from multiprocessing import Event
from frigate.output.birdseye import get_canvas_shape
from frigate.config import FrigateConfig
from frigate.output.birdseye import BirdsEyeFrameManager, get_canvas_shape
class TestBirdseye(unittest.TestCase):
def _build_manager(
self, camera_dimensions: dict[str, tuple[int, int]]
) -> BirdsEyeFrameManager:
config = {
"mqtt": {"host": "mqtt"},
"birdseye": {"width": 1280, "height": 720},
"cameras": {},
}
for order, (camera, dimensions) in enumerate(
camera_dimensions.items(), start=1
):
config["cameras"][camera] = {
"ffmpeg": {
"inputs": [
{
"path": f"rtsp://10.0.0.1:554/{camera}",
"roles": ["detect"],
}
]
},
"detect": {
"width": dimensions[0],
"height": dimensions[1],
"fps": 5,
},
"birdseye": {"order": order},
}
return BirdsEyeFrameManager(FrigateConfig(**config), Event())
def _assert_no_overlaps(
self, layout: list[list[tuple[str, tuple[int, int, int, int]]]]
):
rectangles = [position for row in layout for _, position in row]
for index, rect in enumerate(rectangles):
x1, y1, width1, height1 = rect
for other in rectangles[index + 1 :]:
x2, y2, width2, height2 = other
overlap = (
x1 < x2 + width2
and x2 < x1 + width1
and y1 < y2 + height2
and y2 < y1 + height1
)
self.assertFalse(
overlap,
msg=f"Overlapping rectangles found: {rect} and {other}",
)
def test_16x9(self):
"""Test 16x9 aspect ratio works as expected for birdseye."""
width = 1280
@ -45,3 +98,104 @@ class TestBirdseye(unittest.TestCase):
canvas_width, canvas_height = get_canvas_shape(width, height)
assert canvas_width == width # width will be the same
assert canvas_height != height
def test_portrait_camera_does_not_overlap_next_row(self):
"""Portrait cameras should reserve their real horizontal position on the next row."""
manager = self._build_manager(
{
"cam_a": (1280, 720),
"cam_p": (360, 640),
"cam_b": (1280, 720),
"cam_c": (640, 480),
}
)
layout = manager.calculate_layout(["cam_a", "cam_p", "cam_b", "cam_c"], 3)
self.assertIsNotNone(layout)
assert layout is not None
self._assert_no_overlaps(layout)
cam_c = [
position for row in layout for camera, position in row if camera == "cam_c"
][0]
self.assertEqual(cam_c[0], 0)
def test_portrait_reservation_only_applies_to_next_row(self):
"""Portrait reservations should not push later rows after the span ends."""
manager = self._build_manager(
{
"cam_a": (1280, 720),
"cam_p": (360, 640),
"cam_b": (1280, 720),
"cam_c": (1280, 720),
"cam_d": (1280, 720),
"cam_e": (1280, 720),
}
)
layout = manager.calculate_layout(
["cam_a", "cam_p", "cam_b", "cam_c", "cam_d", "cam_e"],
3,
)
self.assertIsNotNone(layout)
assert layout is not None
self._assert_no_overlaps(layout)
cam_e = [
position for row in layout for camera, position in row if camera == "cam_e"
][0]
self.assertEqual(cam_e[0], 0)
def test_multiple_portraits_reserve_distinct_ranges(self):
"""Multiple portrait cameras in one row should reserve separate spans below them."""
manager = self._build_manager(
{
"cam_a": (640, 480),
"cam_p1": (360, 640),
"cam_p2": (360, 640),
"cam_b": (640, 480),
"cam_c": (1280, 720),
"cam_d": (640, 480),
}
)
layout = manager.calculate_layout(
["cam_a", "cam_p1", "cam_p2", "cam_b", "cam_c", "cam_d"],
4,
)
self.assertIsNotNone(layout)
assert layout is not None
self._assert_no_overlaps(layout)
def test_two_landscapes_then_portrait_then_two_landscapes(self):
"""A portrait after two landscapes should reserve only its own tail span."""
manager = self._build_manager(
{
"cam_a": (1280, 720),
"cam_b": (1280, 720),
"cam_p": (360, 640),
"cam_c": (1280, 720),
"cam_d": (1280, 720),
}
)
layout = manager.calculate_layout(
["cam_a", "cam_b", "cam_p", "cam_c", "cam_d"],
3,
)
self.assertIsNotNone(layout)
assert layout is not None
self._assert_no_overlaps(layout)
cam_c = [
position for row in layout for camera, position in row if camera == "cam_c"
][0]
cam_d = [
position for row in layout for camera, position in row if camera == "cam_d"
][0]
self.assertEqual(cam_c[0], 0)
self.assertEqual(cam_d[0], cam_c[0] + cam_c[2])

132
frigate/test/test_export.py Normal file
View File

@ -0,0 +1,132 @@
import unittest
from frigate.record.export import validate_ffmpeg_args
class TestValidateFfmpegArgs(unittest.TestCase):
"""Tests for the non-admin custom export ffmpeg arg validator.
The validator uses a structural allowlist: every token must be an
allowlisted flag or the value of one, filter values are restricted to a
safe set of filters, and no token may become a bare input/output URL.
"""
def assertRejected(self, args: str) -> None:
valid, message = validate_ffmpeg_args(args)
self.assertFalse(valid, f"expected {args!r} to be rejected")
self.assertNotEqual(message, "")
def assertAllowed(self, args: str) -> None:
valid, message = validate_ffmpeg_args(args)
self.assertTrue(valid, f"expected {args!r} to be allowed, got: {message}")
self.assertEqual(message, "")
# --- legitimate use cases must keep working ---------------------------
def test_timelapse_setpts_allowed(self):
# The whole reason -vf cannot simply be blocked: timelapse exports.
self.assertAllowed("-vf setpts=PTS/60 -r 25")
self.assertAllowed("-vf setpts=0.04*PTS -r 30") # server default
self.assertAllowed("-filter:v setpts=PTS/60 -r 25")
def test_default_input_args_allowed(self):
self.assertAllowed("")
self.assertAllowed("-an -skip_frame nokey")
def test_encoding_args_allowed(self):
self.assertAllowed("-c:v libx264 -crf 23 -preset fast")
self.assertAllowed("-c:v copy -c:a copy")
self.assertAllowed("-c:v libx264 -b:v 2M -maxrate 2M -bufsize 4M")
self.assertAllowed("-movflags +faststart")
self.assertAllowed("-pix_fmt yuv420p -r 30 -g 30")
def test_safe_filters_allowed(self):
self.assertAllowed("-vf scale=640:480")
self.assertAllowed("-vf scale=640:480,setpts=0.5*PTS")
self.assertAllowed("-vf format=yuv420p")
self.assertAllowed("-vf transpose=1")
self.assertAllowed("-vf hflip")
self.assertAllowed("-vf fps=15")
self.assertAllowed("-vf setsar=1 -an")
self.assertAllowed("-vf setdar=16/9")
# --- the reported advisory and file-read class ------------------------
def test_reported_advisory_rejected(self):
self.assertRejected(
"-filter:v drawtext=textfile=/etc/passwd:fontcolor=white:fontsize=20"
)
def test_file_reading_filters_rejected(self):
self.assertRejected("-vf movie=/etc/passwd")
self.assertRejected("-vf drawtext=textfile=/etc/passwd")
self.assertRejected("-vf subtitles=/etc/passwd")
# marker embedded as an option of an otherwise-allowed filter name
self.assertRejected("-vf scale=movie=/etc/passwd")
def test_filtergraph_brackets_rejected(self):
# link labels aren't needed for safe filters; rejecting "[" / "]" keeps
# filtergraph validation linear (no ReDoS on attacker input)
self.assertRejected("-vf [in]scale=640:480[out]")
self.assertRejected("-vf " + "[" * 5000)
def test_preset_file_read_rejected(self):
# cwd-anchored traversal slipped past the old startswith() path check
self.assertRejected("-fpre frigate/../../../etc/passwd")
self.assertRejected("-fpre evil.preset")
self.assertRejected("-vpre x")
self.assertRejected("-apre x")
self.assertRejected("-pre x")
def test_slash_option_file_read_rejected(self):
# ffmpeg "-/option file" reads the option value from a file
self.assertRejected("-/filter:v graph.txt")
self.assertRejected("-/filter_complex graph.txt")
# --- network / SSRF class ---------------------------------------------
def test_schemeless_protocol_rejected(self):
self.assertRejected("-f mpegts tcp:10.0.0.5:4444")
self.assertRejected("tcp:10.0.0.5:4444")
self.assertRejected("udp:10.0.0.5:4444")
self.assertRejected("-progress http:attacker.example.com:80/p")
# --- file-write class --------------------------------------------------
def test_tee_write_rejected(self):
self.assertRejected("-c:v libx264 -map 0 -f tee [f=mpegts]/tmp/owned.ts")
self.assertRejected("-f tee [f=mpegts]/etc/frigate/x.ts")
self.assertRejected("tee:/tmp/x")
def test_bare_output_token_rejected(self):
self.assertRejected("evil.mp4")
self.assertRejected("-c copy evil.mp4")
self.assertRejected("x/../escaped.mkv")
def test_file_producing_muxers_rejected(self):
self.assertRejected("-f hls -hls_segment_filename pwn%03d.ts out.m3u8")
self.assertRejected("-f md5 victim.txt")
self.assertRejected("-f segment seg%03d.ts")
def test_write_flags_rejected(self):
self.assertRejected("-progress evil.log")
self.assertRejected("-stats_enc_pre evil.csv")
self.assertRejected("-report")
# --- resource exhaustion / misc ---------------------------------------
def test_dos_input_flags_rejected(self):
self.assertRejected("-stream_loop -1")
self.assertRejected("-readrate 0.001")
def test_disallowed_flags_rejected(self):
self.assertRejected("-map 0")
self.assertRejected("-i /etc/passwd")
self.assertRejected("-attach evil.bin")
self.assertRejected("-dump_attachment evil.bin")
self.assertRejected("/etc/passwd")
self.assertRejected("-metadata comment=x")
if __name__ == "__main__":
unittest.main()

View File

@ -0,0 +1,91 @@
import math
import unittest
import numpy as np
from norfair.camera_motion import (
HomographyTransformation,
TranslationTransformation,
)
from frigate.ptz.autotrack import transform_is_finite
from frigate.track.norfair_tracker import distance
class TestNorfairDistance(unittest.TestCase):
"""Regression tests for the tracker distance guard.
norfair raises a hard ValueError on any nan distance, which kills the camera
process. During autotracking, an ill-conditioned homography can hand the
tracker a non-finite or degenerate estimate box, so distance() must never
return nan for any input.
"""
def setUp(self) -> None:
# boxes are [[x1, y1], [x2, y2]]
self.detection = np.array([[805.0, 402.0], [864.0, 521.0]])
self.estimate = np.array([[800.0, 400.0], [860.0, 520.0]])
def test_finite_boxes_give_finite_distance(self) -> None:
d = distance(self.detection, self.estimate)
self.assertTrue(math.isfinite(d))
def test_inf_estimate_corner_does_not_return_nan(self) -> None:
estimate = np.array([[np.inf, 400.0], [860.0, 520.0]])
d = distance(self.detection, estimate)
self.assertFalse(math.isnan(d))
self.assertEqual(d, float("inf"))
def test_nan_estimate_corner_does_not_return_nan(self) -> None:
# the actual autotracking crash: a positive-only guard would miss this
# because nan <= 0 is False
estimate = np.array([[np.nan, 400.0], [860.0, 520.0]])
d = distance(self.detection, estimate)
self.assertFalse(math.isnan(d))
self.assertEqual(d, float("inf"))
def test_zero_area_estimate_does_not_return_nan(self) -> None:
estimate = np.array([[900.0, 500.0], [900.0, 500.0]])
d = distance(self.detection, estimate)
self.assertFalse(math.isnan(d))
self.assertEqual(d, float("inf"))
def test_zero_area_detection_does_not_return_nan(self) -> None:
detection = np.array([[805.0, 402.0], [805.0, 521.0]])
d = distance(detection, self.estimate)
self.assertFalse(math.isnan(d))
self.assertEqual(d, float("inf"))
def test_inverted_estimate_corners_do_not_return_nan(self) -> None:
# Kalman estimates can occasionally cross corners (x2 < x1)
estimate = np.array([[860.0, 520.0], [800.0, 400.0]])
d = distance(self.detection, estimate)
self.assertFalse(math.isnan(d))
self.assertEqual(d, float("inf"))
class TestTransformIsFinite(unittest.TestCase):
def test_finite_homography_is_finite(self) -> None:
matrix = np.array([[1.0, 0.0, 5.0], [0.0, 1.0, 3.0], [0.0, 0.0, 1.0]])
self.assertTrue(transform_is_finite(HomographyTransformation(matrix)))
def test_finite_translation_is_finite(self) -> None:
self.assertTrue(
transform_is_finite(TranslationTransformation(np.array([12.0, -4.0])))
)
def test_non_finite_homography_is_not_finite(self) -> None:
transform = HomographyTransformation(np.eye(3))
# simulate accumulation overflowing to a non-finite matrix
transform.homography_matrix = np.array(
[[1.0, 0.0, np.inf], [0.0, 1.0, 0.0], [0.0, 0.0, 1.0]]
)
self.assertFalse(transform_is_finite(transform))
def test_nan_translation_is_not_finite(self) -> None:
self.assertFalse(
transform_is_finite(TranslationTransformation(np.array([np.nan, 0.0])))
)
if __name__ == "__main__":
unittest.main()

View File

@ -45,6 +45,17 @@ def distance(detection: np.ndarray, estimate: np.ndarray) -> float:
estimate_dim = np.diff(estimate, axis=0).flatten()
detection_dim = np.diff(detection, axis=0).flatten()
# Guard against degenerate or non-finite boxes
if (
not np.all(np.isfinite(estimate_dim))
or not np.all(np.isfinite(detection_dim))
or estimate_dim[0] <= 0
or estimate_dim[1] <= 0
or detection_dim[0] <= 0
or detection_dim[1] <= 0
):
return float("inf")
# get bottom center positions
detection_position = np.array(
[np.average(detection[:, 0]), np.max(detection[:, 1])]

606
generate_api_auth_spec.py Normal file
View File

@ -0,0 +1,606 @@
"""Generate the OpenAPI spec from the app, annotated with auth requirements.
This generator builds the FastAPI application, exports its OpenAPI document via
``app.openapi()``, and enriches every operation with authentication metadata:
* a ``components.securitySchemes`` block,
* a per-operation ``security`` requirement (so the docs render a lock badge),
* an ``x-required-role`` extension for machine readers, and
* a short bold ``Access:`` note prepended to each operation description.
The committed docs/static/frigate-api.yaml is the output of this script. It is
generated rather than hand-maintained so it stays complete and current; the docs
build (docusaurus-plugin-openapi-docs) consumes it as-is.
The access level for an endpoint is determined by BOTH its route-level
dependency (``require_role``/``allow_any_authenticated``/``allow_public``/
``require_camera_access``) AND the global "secure by default" admin dependency,
which is bypassed only for the paths listed in ``require_admin_by_default``.
Those exempt lists are read directly from the function's closure so this script
stays in lockstep with ``frigate/api/auth.py`` instead of duplicating them.
Many handlers enforce per-camera access by calling ``require_camera_access``
inside the handler body rather than as a route dependency, which dependency
introspection cannot see. We recover those from the handler's bytecode (see
``_handler_enforces_camera``) and promote an otherwise "any authenticated"
operation to camera-scoped.
Usage (from the repository root):
python3 generate_api_auth_spec.py # write the spec
python3 generate_api_auth_spec.py --check # CI guard: fail if stale
The process exits non-zero if the generated document fails structural
validation, or (in --check mode) if the committed spec is out of date.
"""
import argparse
import difflib
import inspect
import io
import logging
import sys
from pathlib import Path
from fastapi import FastAPI
from fastapi.routing import APIRoute
from ruamel.yaml import YAML
from ruamel.yaml.scalarstring import LiteralScalarString
from frigate.api import app as main_app
from frigate.api import (
auth,
camera,
chat,
classification,
debug_replay,
event,
export,
media,
motion_search,
notification,
preview,
record,
review,
)
from frigate.api.auth import require_admin_by_default
logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("generate_api_auth_spec")
REPO_ROOT = Path(__file__).resolve().parent
OUTPUT_SPEC = REPO_ROOT / "docs" / "static" / "frigate-api.yaml"
HTTP_METHODS = {"get", "post", "put", "delete", "patch"}
# Banner written at the top of the generated spec.
HEADER = (
"# Generated by generate_api_auth_spec.py — do not edit by hand.\n"
"# Regenerate with: python3 generate_api_auth_spec.py\n"
"# The empty info.title is intentional: a docusaurus-openapi-docs convention\n"
"# that suppresses the generated API introduction page.\n"
)
# Post-processing applied on top of the raw app.openapi() export. These live
# only in the published spec, not in the app, so they are reproduced here.
SPEC_TITLE = ""
SPEC_SERVERS = [
{"url": "https://demo.frigate.video/api"},
{"url": "http://localhost:5001/api"},
]
# Access levels, ordered from least to most privileged. The string values are
# also what we emit as ``x-required-role``.
PUBLIC = "public"
AUTHENTICATED = "any"
CAMERA = "camera"
ADMIN = "admin"
ADMIN_SCHEME = "frigateAdminAuth"
USER_SCHEME = "frigateUserAuth"
SECURITY_SCHEMES = {
ADMIN_SCHEME: {
"type": "apiKey",
"in": "cookie",
"name": "frigate_token",
"description": (
"Authenticated session whose resolved role is 'admin'. The session "
"is established via the JWT cookie issued by POST /login, or via "
"proxy auth headers (remote-user / remote-role) when Frigate runs "
"behind an authenticating reverse proxy."
),
},
USER_SCHEME: {
"type": "apiKey",
"in": "cookie",
"name": "frigate_token",
"description": (
"Any authenticated session (role 'viewer' or higher), established "
"via the JWT cookie issued by POST /login, or via proxy auth "
"headers when Frigate runs behind an authenticating reverse proxy."
),
},
}
# How each access level maps to a rendered note.
ACCESS_NOTES = {
PUBLIC: "**Access:** Public — no authentication required.",
AUTHENTICATED: "**Access:** Any authenticated user.",
CAMERA: "**Access:** Authenticated user with access to the referenced camera.",
ADMIN: "**Access:** Admin role required.",
}
def build_app() -> FastAPI:
"""Build a bare app with every router mounted.
This mirrors the router set wired up in frigate.api.fastapi_app. It omits
the global admin dependency and all runtime state; the OpenAPI route table
and the per-route dependencies are all we need to export and classify.
"""
app = FastAPI()
routers = [
auth.router,
camera.router,
chat.router,
classification.router,
review.router,
main_app.router,
preview.router,
notification.router,
export.router,
event.router,
media.router,
motion_search.router,
record.router,
debug_replay.router,
]
for router in routers:
app.include_router(router)
return app
def read_exempt_rules() -> tuple[set[str], tuple[str, ...]]:
"""Read the admin-exemption lists straight from the auth dependency closure.
Reading them here (rather than copying) keeps this generator in sync with
frigate/api/auth.py automatically.
"""
closure = inspect.getclosurevars(require_admin_by_default()).nonlocals
exempt_paths = set(closure["EXEMPT_PATHS"])
exempt_prefixes = tuple(closure["EXEMPT_PREFIXES"])
return exempt_paths, exempt_prefixes
def _first_segment(path: str) -> str:
return path.split("/", 2)[1] if path.startswith("/") and len(path) > 1 else ""
def _route_markers(route: APIRoute) -> tuple[set[str], list[str] | None]:
"""Return the set of recognized auth markers on a route's dependencies."""
markers: set[str] = set()
admin_roles: list[str] | None = None
for dep in route.dependant.dependencies:
call = dep.call
qualname = getattr(call, "__qualname__", "") or ""
name = getattr(call, "__name__", "") or ""
if "role_checker" in qualname:
markers.add(ADMIN)
try:
roles = inspect.getclosurevars(call).nonlocals.get("required_roles")
if roles:
admin_roles = list(roles)
except (TypeError, ValueError):
pass
elif name in ("require_camera_access", "require_go2rtc_stream_access"):
markers.add(CAMERA)
elif "auth_checker" in qualname:
markers.add(AUTHENTICATED)
elif "public_checker" in qualname:
markers.add(PUBLIC)
return markers, admin_roles
def _handler_enforces_camera(route: APIRoute) -> bool:
"""True if the route handler calls require_camera_access in its body.
Such calls are invisible to dependency introspection. We detect them from
the handler's compiled bytecode: a global name referenced anywhere in the
function appears in ``__code__.co_names``. This catches direct calls (all of
them, currently); a call hidden behind a helper function would be missed.
"""
code = getattr(route.endpoint, "__code__", None)
return bool(code and "require_camera_access" in code.co_names)
def classify_route(
route: APIRoute,
exempt_paths: set[str],
exempt_prefixes: tuple[str, ...],
) -> tuple[str, list[str] | None, str | None]:
"""Resolve the effective access level for a route.
Returns (access_level, roles, flag). ``flag`` is a human-readable note when
the result needed inference or revealed a possible inconsistency.
"""
level, roles, flag = _classify_base(route, exempt_paths, exempt_prefixes)
# In-body require_camera_access enforcement is invisible to dependency
# introspection. When the effective access would otherwise be "any
# authenticated", the handler's per-camera check is the real constraint, so
# promote it to camera-scoped. Admin/public are left alone: for admin the
# role is the binding requirement and the camera check is only defensive.
if level == AUTHENTICATED and _handler_enforces_camera(route):
return CAMERA, None, None
return level, roles, flag
def _classify_base(
route: APIRoute,
exempt_paths: set[str],
exempt_prefixes: tuple[str, ...],
) -> tuple[str, list[str] | None, str | None]:
"""Resolve the access level from route-level dependencies and exempt rules."""
markers, admin_roles = _route_markers(route)
path = route.path
is_camera_path = _first_segment(path) == "{camera_name}"
exempt = path in exempt_paths or path.startswith(exempt_prefixes) or is_camera_path
# Explicit route-level markers win, in order of specificity.
if ADMIN in markers:
return ADMIN, admin_roles or ["admin"], None
if CAMERA in markers:
return CAMERA, None, None
if AUTHENTICATED in markers:
if exempt:
return AUTHENTICATED, None, None
# The route opts in to any-authenticated, but the global admin check is
# not bypassed for this path, so admin is what actually gets enforced.
return (
ADMIN,
["admin"],
(
"route declares allow_any_authenticated but path is not exempt from "
"the global admin check; admin is effectively enforced"
),
)
if PUBLIC in markers:
if exempt:
return PUBLIC, None, None
return (
ADMIN,
["admin"],
(
"route declares allow_public but path is not exempt from the global "
"admin check; admin is effectively enforced"
),
)
# No explicit auth marker: governed purely by the global default.
if not exempt:
return ADMIN, ["admin"], None
# Exempt with no route dependency: the global admin check is bypassed and
# there is no route-level gate, so authorization (if any) happens inside the
# handler. Infer from the path shape and flag for confirmation.
if is_camera_path:
return (
CAMERA,
None,
(
"no route-level dependency; camera-scoped path, authorization "
"assumed to be enforced in the handler"
),
)
return (
AUTHENTICATED,
None,
(
"path is exempt from the global admin check but has no route-level "
"dependency; confirm authorization is enforced in the handler"
),
)
def build_access_map(
app: FastAPI,
exempt_paths: set[str],
exempt_prefixes: tuple[str, ...],
) -> dict[tuple[str, str], dict]:
"""Map (path, lowercase method) -> classification details."""
access_map: dict[tuple[str, str], dict] = {}
for route in app.routes:
if not isinstance(route, APIRoute):
continue
level, roles, flag = classify_route(route, exempt_paths, exempt_prefixes)
for method in route.methods:
if method in ("HEAD", "OPTIONS"):
continue
access_map[(route.path, method.lower())] = {
"level": level,
"roles": roles,
"flag": flag,
"path": route.path,
"method": method,
}
return access_map
def security_for(level: str) -> list:
"""Build the OpenAPI ``security`` value for an access level."""
if level == PUBLIC:
return []
if level == ADMIN:
return [{ADMIN_SCHEME: []}]
# AUTHENTICATED and CAMERA both require any authenticated session; the
# camera-specific scoping is conveyed in the note and x-required-role.
return [{USER_SCHEME: []}]
def required_role_value(level: str, roles: list[str] | None):
if level == ADMIN and roles and roles != ["admin"]:
return roles
return level
def annotate_description(operation: dict, note: str) -> None:
existing = operation.get("description")
if not existing:
operation["description"] = note
return
operation["description"] = LiteralScalarString(
f"{note}\n\n{str(existing).rstrip()}"
)
def base_document(raw: dict) -> dict:
"""Apply the docs pipeline post-processing with a stable top-level order."""
info = dict(raw.get("info", {}))
info["title"] = SPEC_TITLE
return {
"openapi": raw["openapi"],
"info": info,
"servers": [dict(server) for server in SPEC_SERVERS],
"paths": raw["paths"],
"components": raw.get("components", {}),
}
def enrich(spec: dict, access_map: dict) -> tuple[dict, list, list]:
"""Add security schemes and per-operation auth metadata in place."""
components = spec.setdefault("components", {})
components["securitySchemes"] = dict(SECURITY_SCHEMES)
counts: dict[str, int] = {}
flagged: list[dict] = []
unmatched: list[tuple[str, str]] = []
for path, path_item in spec["paths"].items():
for method, operation in path_item.items():
if method.lower() not in HTTP_METHODS:
continue
details = access_map.get((path, method.lower()))
if details is None:
unmatched.append((method.upper(), path))
continue
level = details["level"]
counts[level] = counts.get(level, 0) + 1
operation["security"] = security_for(level)
operation["x-required-role"] = required_role_value(level, details["roles"])
annotate_description(operation, ACCESS_NOTES[level])
if details["flag"]:
flagged.append(details)
return counts, flagged, unmatched
# Numeric defaults at or above this magnitude are treated as live Unix
# timestamps baked into the schema at import time (e.g. the /{camera_name}
# /recordings after/before params default to datetime.now()). They make the
# export non-deterministic and document a meaningless frozen epoch, so they are
# stripped. The proper fix is to default those route params to None and resolve
# "now" inside the handler.
VOLATILE_DEFAULT_THRESHOLD = 1_000_000_000
def strip_volatile_defaults(node, trail: str = "") -> list[tuple[str, float]]:
"""Remove epoch-like numeric ``default`` values so the export is stable.
Returns the (location, value) pairs that were removed, for reporting.
"""
removed: list[tuple[str, float]] = []
if isinstance(node, dict):
default = node.get("default")
if (
isinstance(default, (int, float))
and not isinstance(default, bool)
and default >= VOLATILE_DEFAULT_THRESHOLD
):
removed.append((trail, default))
del node["default"]
for key, value in node.items():
removed.extend(strip_volatile_defaults(value, f"{trail}/{key}"))
elif isinstance(node, list):
for index, value in enumerate(node):
removed.extend(strip_volatile_defaults(value, f"{trail}[{index}]"))
return removed
def to_block_scalars(node):
"""Recursively render multi-line strings as literal block scalars.
Produces readable, deterministic YAML (``|-`` blocks) instead of long
double-quoted lines with escaped newlines.
"""
if isinstance(node, dict):
return {key: to_block_scalars(value) for key, value in node.items()}
if isinstance(node, list):
return [to_block_scalars(value) for value in node]
if isinstance(node, str) and "\n" in node:
return LiteralScalarString(node)
return node
def _iter_refs(node):
if isinstance(node, dict):
for key, value in node.items():
if key == "$ref" and isinstance(value, str):
yield value
else:
yield from _iter_refs(value)
elif isinstance(node, list):
for value in node:
yield from _iter_refs(value)
def validate(spec: dict) -> list[str]:
"""Structural sanity checks on the generated document."""
problems: list[str] = []
schemas = set(spec.get("components", {}).get("schemas", {}))
defined_schemes = set(spec.get("components", {}).get("securitySchemes", {}))
for ref in _iter_refs(spec):
if ref.startswith("#/components/schemas/"):
name = ref.rsplit("/", 1)[-1]
if name not in schemas:
problems.append(f"dangling $ref: {ref}")
for path, path_item in spec.get("paths", {}).items():
for method, operation in path_item.items():
if method.lower() not in HTTP_METHODS or not isinstance(operation, dict):
continue
location = f"{method.upper()} {path}"
if "x-required-role" not in operation:
problems.append(f"missing x-required-role: {location}")
if "security" not in operation:
problems.append(f"missing security: {location}")
continue
for requirement in operation["security"]:
for scheme in requirement:
if scheme not in defined_schemes:
problems.append(
f"undefined security scheme {scheme}: {location}"
)
return sorted(set(problems))
def render(spec: dict) -> str:
"""Serialize the spec to the canonical YAML string (with the header)."""
yaml = YAML()
yaml.width = 80
yaml.indent(mapping=2, sequence=4, offset=2)
stream = io.StringIO()
yaml.dump(spec, stream)
return HEADER + stream.getvalue()
def build_spec() -> tuple[dict, dict, list, list, list]:
app = build_app()
exempt_paths, exempt_prefixes = read_exempt_rules()
access_map = build_access_map(app, exempt_paths, exempt_prefixes)
spec = base_document(app.openapi())
normalized = strip_volatile_defaults(spec)
counts, flagged, unmatched = enrich(spec, access_map)
spec = to_block_scalars(spec)
return spec, counts, flagged, unmatched, normalized
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Generate the annotated OpenAPI spec.")
parser.add_argument(
"--check",
action="store_true",
help="verify the committed spec is up to date without writing; "
"exit non-zero if it would change",
)
args = parser.parse_args(argv)
spec, counts, flagged, unmatched, normalized = build_spec()
problems = validate(spec)
rendered = render(spec)
if args.check:
return _check(rendered, problems)
if problems:
logger.error("Refusing to write — generated spec failed validation:")
for problem in problems:
logger.error(" %s", problem)
return 1
OUTPUT_SPEC.write_text(rendered)
_report(counts, flagged, unmatched, normalized)
logger.info("\nWrote %s", OUTPUT_SPEC.relative_to(REPO_ROOT))
return 0
def _check(rendered: str, problems: list[str]) -> int:
name = OUTPUT_SPEC.relative_to(REPO_ROOT)
if problems:
logger.error("Generated spec failed validation:")
for problem in problems:
logger.error(" %s", problem)
return 1
current = OUTPUT_SPEC.read_text() if OUTPUT_SPEC.exists() else ""
if current == rendered:
logger.info("%s is up to date", name)
return 0
logger.error(
"%s is out of date. Regenerate with: python3 %s",
name,
Path(__file__).name,
)
diff = difflib.unified_diff(
current.splitlines(),
rendered.splitlines(),
fromfile=f"{name} (committed)",
tofile=f"{name} (generated)",
lineterm="",
n=2,
)
for shown, line in enumerate(diff):
if shown >= 60:
logger.error(" ... (diff truncated)")
break
logger.error(" %s", line)
return 1
def _report(counts, flagged, unmatched, normalized) -> None:
logger.info("Access levels applied:")
for level in (PUBLIC, AUTHENTICATED, CAMERA, ADMIN):
logger.info(" %-14s %d", level, counts.get(level, 0))
logger.info(" %-14s %d", "total", sum(counts.values()))
if normalized:
logger.info("\nStripped volatile timestamp defaults (%d):", len(normalized))
for location, value in normalized:
logger.info(" %s = %s", location.lstrip("/"), value)
if flagged:
logger.info("\nFlagged for manual confirmation (%d):", len(flagged))
for item in flagged:
logger.info(" %-6s %s", item["method"], item["path"])
logger.info(" -> %s (%s)", item["level"], item["flag"])
if unmatched:
logger.info(
"\nOperations with no classification (%d) [unexpected]:", len(unmatched)
)
for method, path in unmatched:
logger.info(" %-6s %s", method, path)
if __name__ == "__main__":
sys.exit(main())

View File

@ -592,6 +592,7 @@ export default function LiveDashboardView({
resetPreferredLiveMode(camera.name)
}
config={config}
streamMetadata={streamMetadata}
>
<LivePlayer
cameraRef={cameraRef}