scope classification attributes to allowed cameras

docs/static/frigate-api.yaml

@@ -1476,10 +1476,12 @@ paths:
         - Classification
       summary: Get custom classification attributes
       description: |-
-        **Access:** Admin role required.
+        **Access:** Any authenticated user.
 
         Returns custom classification attributes for a given object type.
             Only includes models with classification_type set to 'attribute'.
+            Callers without access to every camera only receive values that have been
+            recorded on the cameras they can access.
             By default returns a flat sorted list of all attribute labels.
             If group_by_model is true, returns attributes grouped by model name.
       operationId: get_custom_attributes_classification_attributes_get
@@ -1510,8 +1512,8 @@ paths:
               schema:
                 $ref: '#/components/schemas/HTTPValidationError'
       security:
-        - frigateAdminAuth: []
-      x-required-role: admin
+        - frigateUserAuth: []
+      x-required-role: any
   /classification/{name}/train:
     get:
       tags:

frigate/api/auth.py

@@ -86,6 +86,7 @@ def require_admin_by_default():
         "/categorized_object_names",
         "/plus/models",
         "/recognized_license_plates",
+        "/classification/attributes",
         "/timeline",
         "/timeline/hourly",
         "/recordings/storage",

frigate/api/classification.py

@@ -11,10 +11,14 @@ from typing import Any
 import cv2
 from fastapi import APIRouter, Depends, Request, UploadFile
 from fastapi.responses import JSONResponse
-from peewee import DoesNotExist
+from peewee import DoesNotExist, fn
 from playhouse.shortcuts import model_to_dict
 
-from frigate.api.auth import require_role
+from frigate.api.auth import (
+    allow_any_authenticated,
+    get_allowed_cameras_for_filter,
+    require_role,
+)
 from frigate.api.defs.request.classification_body import (
     AudioTranscriptionBody,
     DeleteFaceImagesBody,
@@ -739,18 +743,81 @@ def get_classification_dataset(name: str):
     )
 
 
+def get_observed_attributes(
+    model_attributes: dict[str, list[str]],
+    object_labels: set[str],
+    allowed_cameras: list[str],
+) -> dict[str, set[str]]:
+    """Get the attribute values recorded on the given cameras.
+
+    Args:
+        model_attributes: Labels each attribute model can emit, keyed by model name
+        object_labels: Object types those models run on
+        allowed_cameras: Cameras the caller has access to
+
+    Returns:
+        Values seen for each model, keyed by model name
+    """
+    if not model_attributes or not object_labels or not allowed_cameras:
+        return {}
+
+    model_names = list(model_attributes.keys())
+
+    query = (
+        Event.select(
+            *[
+                fn.json_extract(Event.data, f'$."{model_name}"')
+                for model_name in model_names
+            ]
+        )
+        .where(
+            (Event.camera << allowed_cameras) & (Event.label << sorted(object_labels))
+        )
+        .distinct()
+        .tuples()
+    )
+
+    targets = {
+        model_name: set(attributes)
+        for model_name, attributes in model_attributes.items()
+    }
+    observed: dict[str, set[str]] = {model_name: set() for model_name in model_names}
+
+    for row in query.iterator():
+        found = False
+
+        for model_name, value in zip(model_names, row):
+            if isinstance(value, str) and value not in observed[model_name]:
+                observed[model_name].add(value)
+                found = True
+
+        if found and all(
+            observed[model_name] >= targets[model_name] for model_name in model_names
+        ):
+            break
+
+    return observed
+
+
 @router.get(
     "/classification/attributes",
+    dependencies=[Depends(allow_any_authenticated())],
     summary="Get custom classification attributes",
     description="""Returns custom classification attributes for a given object type.
     Only includes models with classification_type set to 'attribute'.
+    Callers without access to every camera only receive values that have been
+    recorded on the cameras they can access.
     By default returns a flat sorted list of all attribute labels.
     If group_by_model is true, returns attributes grouped by model name.""",
 )
 def get_custom_attributes(
-    request: Request, object_type: str = None, group_by_model: bool = False
+    request: Request,
+    object_type: str = None,
+    group_by_model: bool = False,
+    allowed_cameras: list[str] = Depends(get_allowed_cameras_for_filter),
 ):
     models_with_attributes = {}
+    objects_by_model = {}
 
     for (
         model_key,
@@ -781,6 +848,32 @@ def get_custom_attributes(
         if attributes:
             model_name = model_config.name or model_key
             models_with_attributes[model_name] = sorted(attributes)
+            objects_by_model[model_name] = model_objects
+
+    # the dataset holds every label a model can emit, including ones never
+    # applied to an event, so callers without full camera access are limited to
+    # the values actually recorded on the cameras they can see
+    all_cameras = set(request.app.frigate_config.cameras.keys())
+
+    if models_with_attributes and not all_cameras.issubset(allowed_cameras):
+        observed = get_observed_attributes(
+            models_with_attributes,
+            set().union(*objects_by_model.values()),
+            allowed_cameras,
+        )
+        models_with_attributes = {
+            model_name: [
+                attribute
+                for attribute in attributes
+                if attribute in observed.get(model_name, set())
+            ]
+            for model_name, attributes in models_with_attributes.items()
+        }
+        models_with_attributes = {
+            model_name: attributes
+            for model_name, attributes in models_with_attributes.items()
+            if attributes
+        }
 
     if group_by_model:
         return JSONResponse(content=models_with_attributes)

frigate/test/http_api/test_http_classification_attributes.py

@@ -0,0 +1,189 @@
+"""Tests for GET /classification/attributes."""
+
+import os
+import shutil
+import unittest
+
+from frigate.api.auth import get_allowed_cameras_for_filter
+from frigate.const import CLIPS_DIR
+from frigate.models import Event, Recordings, ReviewSegment
+from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
+
+# "limited_user" only reaches front_door, so it never sees the values that were
+# recorded on back_door.
+_CONFIG = {
+    "mqtt": {"host": "mqtt"},
+    "auth": {"roles": {"limited_user": ["front_door"]}},
+    "classification": {
+        "custom": {
+            "delivery_service": {
+                "enabled": True,
+                "object_config": {
+                    "objects": ["car"],
+                    "classification_type": "attribute",
+                },
+            }
+        }
+    },
+    "cameras": {
+        "front_door": {
+            "ffmpeg": {
+                "inputs": [{"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]}]
+            },
+            "detect": {"height": 1080, "width": 1920, "fps": 5},
+        },
+        "back_door": {
+            "ffmpeg": {
+                "inputs": [{"path": "rtsp://10.0.0.2:554/video", "roles": ["detect"]}]
+            },
+            "detect": {"height": 1080, "width": 1920, "fps": 5},
+        },
+    },
+}
+
+
+class TestClassificationAttributesAccess(BaseTestHttp):
+    """The attribute list is read from the training dataset on disk, which holds
+    every label a model can emit regardless of which camera recorded it. Callers
+    without full camera access are cut back to the values on their own cameras,
+    so these tests pin that scoping.
+    """
+
+    def setUp(self):
+        super().setUp([Event, ReviewSegment, Recordings])
+        self.minimal_config = _CONFIG
+        self.app = super().create_app()
+        self.model_dir = os.path.join(CLIPS_DIR, "delivery_service")
+
+        for category in ("DHL", "Amazon", "Hermes", "none"):
+            os.makedirs(
+                os.path.join(self.model_dir, "dataset", category), exist_ok=True
+            )
+
+    def tearDown(self):
+        shutil.rmtree(self.model_dir, ignore_errors=True)
+        self.app.dependency_overrides.clear()
+        super().tearDown()
+
+    def _insert_event(self, event_id: str, camera: str, attribute: str | None):
+        data = {"type": "object", "score": 0.9}
+
+        if attribute is not None:
+            data["delivery_service"] = attribute
+
+        Event.insert(
+            id=event_id,
+            label="car",
+            camera=camera,
+            start_time=100,
+            end_time=200,
+            top_score=0.9,
+            score=0.9,
+            false_positive=False,
+            zones=[],
+            thumbnail="",
+            has_clip=True,
+            has_snapshot=True,
+            region=[],
+            box=[],
+            area=0,
+            retain_indefinitely=False,
+            ratio=1.0,
+            plus_id=None,
+            model_hash="",
+            detector_type="cpu",
+            model_type="ssd",
+            data=data,
+        ).execute()
+
+    def _get(self, role: str, **params):
+        # the base class resolves every camera by default, so drop the override
+        # to exercise the real role to allowed-cameras resolution
+        self.app.dependency_overrides.pop(get_allowed_cameras_for_filter, None)
+
+        with AuthTestClient(self.app) as client:
+            return client.get(
+                "/classification/attributes",
+                params=params,
+                headers={"remote-user": "test", "remote-role": role},
+            )
+
+    def _insert_split_events(self):
+        self._insert_event("front", "front_door", "DHL")
+        self._insert_event("back", "back_door", "Amazon")
+
+    def test_admin_gets_every_trained_label(self):
+        self._insert_split_events()
+        assert self._get("admin").json() == ["Amazon", "DHL", "Hermes"]
+
+    def test_viewer_gets_every_trained_label(self):
+        self._insert_split_events()
+        assert self._get("viewer").json() == ["Amazon", "DHL", "Hermes"]
+
+    def test_restricted_role_only_gets_its_own_cameras(self):
+        self._insert_split_events()
+        assert self._get("limited_user").json() == ["DHL"]
+
+    def test_restricted_role_grouped_by_model(self):
+        self._insert_split_events()
+        assert self._get("limited_user", group_by_model="true").json() == {
+            "delivery_service": ["DHL"]
+        }
+
+    def test_restricted_role_with_no_recorded_values(self):
+        self._insert_event("back", "back_door", "Amazon")
+        assert self._get("limited_user").json() == []
+        assert self._get("limited_user", group_by_model="true").json() == {}
+
+    def test_restricted_role_ignores_events_without_the_attribute(self):
+        self._insert_event("front", "front_door", None)
+        assert self._get("limited_user").json() == []
+
+    def test_restricted_role_with_a_dotted_model_name(self):
+        # model names are unrestricted config keys, and an unquoted "." in the
+        # json path would be read as a nested lookup and match nothing
+        self.app.frigate_config.classification.custom["delivery.service"] = (
+            self.app.frigate_config.classification.custom.pop("delivery_service")
+        )
+        self.app.frigate_config.classification.custom[
+            "delivery.service"
+        ].name = "delivery.service"
+        os.rename(self.model_dir, os.path.join(CLIPS_DIR, "delivery.service"))
+        self.model_dir = os.path.join(CLIPS_DIR, "delivery.service")
+
+        data = {"type": "object", "score": 0.9, "delivery.service": "DHL"}
+        Event.insert(
+            id="front",
+            label="car",
+            camera="front_door",
+            start_time=100,
+            end_time=200,
+            top_score=0.9,
+            score=0.9,
+            false_positive=False,
+            zones=[],
+            thumbnail="",
+            has_clip=True,
+            has_snapshot=True,
+            region=[],
+            box=[],
+            area=0,
+            retain_indefinitely=False,
+            ratio=1.0,
+            plus_id=None,
+            model_hash="",
+            detector_type="cpu",
+            model_type="ssd",
+            data=data,
+        ).execute()
+
+        assert self._get("limited_user").json() == ["DHL"]
+
+    def test_object_type_filters_out_unrelated_models(self):
+        self._insert_split_events()
+        assert self._get("limited_user", object_type="person").json() == []
+        assert self._get("limited_user", object_type="car").json() == ["DHL"]
+
+
+if __name__ == "__main__":
+    unittest.main()