Merge remote-tracking branch 'origin/master' into dev

frigate/test/http_api/test_http_camera_access.py

@@ -1,6 +1,7 @@
 from unittest.mock import patch
 
 from fastapi import HTTPException, Request
+from fastapi.testclient import TestClient
 
 from frigate.api.auth import (
     get_allowed_cameras_for_filter,
@@ -9,6 +10,33 @@ from frigate.api.auth import (
 from frigate.models import Event, Recordings, ReviewSegment
 from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
 
+# Minimal multi-camera config used by go2rtc stream access tests.
+# front_door has a stream alias "front_door_main"; back_door uses its own name.
+# The "limited_user" role is restricted to front_door only.
+_MULTI_CAMERA_CONFIG = {
+    "mqtt": {"host": "mqtt"},
+    "auth": {
+        "roles": {
+            "limited_user": ["front_door"],
+        }
+    },
+    "cameras": {
+        "front_door": {
+            "ffmpeg": {
+                "inputs": [{"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]}]
+            },
+            "detect": {"height": 1080, "width": 1920, "fps": 5},
+            "live": {"streams": {"default": "front_door_main"}},
+        },
+        "back_door": {
+            "ffmpeg": {
+                "inputs": [{"path": "rtsp://10.0.0.2:554/video", "roles": ["detect"]}]
+            },
+            "detect": {"height": 1080, "width": 1920, "fps": 5},
+        },
+    },
+}
+
 
 class TestCameraAccessEventReview(BaseTestHttp):
     def setUp(self):
@@ -190,3 +218,179 @@ class TestCameraAccessEventReview(BaseTestHttp):
             resp = client.get("/events/summary")
             summary_list = resp.json()
             assert len(summary_list) == 2
+
+
+class TestGo2rtcStreamAccess(BaseTestHttp):
+    """Tests for require_go2rtc_stream_access — the auth dependency on
+    GET /go2rtc/streams/{stream_name}.
+
+    go2rtc is not running in unit tests, so an authorized request returns
+    500 (the proxy call fails), while an unauthorized request returns 401/403
+    before the proxy is ever reached.
+    """
+
+    def _make_app(self, config_override: dict | None = None):
+        """Build a test app, optionally replacing self.minimal_config."""
+        if config_override is not None:
+            self.minimal_config = config_override
+        app = super().create_app()
+
+        # Allow tests to control the current user via request headers.
+        async def mock_get_current_user(request: Request):
+            username = request.headers.get("remote-user")
+            role = request.headers.get("remote-role")
+            if not username or not role:
+                from fastapi.responses import JSONResponse
+
+                return JSONResponse(
+                    content={"message": "No authorization headers."},
+                    status_code=401,
+                )
+            return {"username": username, "role": role}
+
+        app.dependency_overrides[get_current_user] = mock_get_current_user
+        return app
+
+    def setUp(self):
+        super().setUp([Event, ReviewSegment, Recordings])
+
+    def tearDown(self):
+        super().tearDown()
+
+    # ------------------------------------------------------------------
+    # Helpers
+    # ------------------------------------------------------------------
+
+    def _get_stream(
+        self, app, stream_name: str, role: str = "admin", user: str = "test"
+    ):
+        """Issue GET /go2rtc/streams/{stream_name} with the given role."""
+        with AuthTestClient(app) as client:
+            return client.get(
+                f"/go2rtc/streams/{stream_name}",
+                headers={"remote-user": user, "remote-role": role},
+            )
+
+    # ------------------------------------------------------------------
+    # Tests
+    # ------------------------------------------------------------------
+
+    def test_admin_can_access_any_stream(self):
+        """Admin role bypasses camera restrictions."""
+        app = self._make_app(_MULTI_CAMERA_CONFIG)
+        # front_door stream — go2rtc is not running so expect 500, not 401/403
+        resp = self._get_stream(app, "front_door", role="admin")
+        assert resp.status_code not in (401, 403), (
+            f"Admin should not be blocked; got {resp.status_code}"
+        )
+
+        # back_door stream
+        resp = self._get_stream(app, "back_door", role="admin")
+        assert resp.status_code not in (401, 403)
+
+    def test_missing_auth_headers_returns_401(self):
+        """Requests without auth headers must be rejected with 401."""
+        app = self._make_app(_MULTI_CAMERA_CONFIG)
+        # Use plain TestClient (not AuthTestClient) so no headers are injected.
+        with TestClient(app, raise_server_exceptions=False) as client:
+            resp = client.get("/go2rtc/streams/front_door")
+        assert resp.status_code == 401, f"Expected 401, got {resp.status_code}"
+
+    def test_unconfigured_role_can_access_any_stream(self):
+        """When no camera restrictions are configured for a role the user
+        should have access to all streams (no roles_dict entry ⇒ no restriction)."""
+        no_roles_config = {
+            "mqtt": {"host": "mqtt"},
+            "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},
+                },
+            },
+        }
+        app = self._make_app(no_roles_config)
+
+        # "myuser" role is not listed in roles_dict — should be allowed everywhere
+        for stream in ("front_door", "back_door"):
+            resp = self._get_stream(app, stream, role="myuser")
+            assert resp.status_code not in (401, 403), (
+                f"Unconfigured role should not be blocked on '{stream}'; "
+                f"got {resp.status_code}"
+            )
+
+    def test_restricted_role_can_access_allowed_camera(self):
+        """limited_user role (restricted to front_door) can access front_door stream."""
+        app = self._make_app(_MULTI_CAMERA_CONFIG)
+        resp = self._get_stream(app, "front_door", role="limited_user")
+        assert resp.status_code not in (401, 403), (
+            f"limited_user should be allowed on front_door; got {resp.status_code}"
+        )
+
+    def test_restricted_role_blocked_from_disallowed_camera(self):
+        """limited_user role (restricted to front_door) cannot access back_door stream."""
+        app = self._make_app(_MULTI_CAMERA_CONFIG)
+        resp = self._get_stream(app, "back_door", role="limited_user")
+        assert resp.status_code == 403, (
+            f"limited_user should be denied on back_door; got {resp.status_code}"
+        )
+
+    def test_stream_alias_allowed_for_owning_camera(self):
+        """Stream alias 'front_door_main' is owned by front_door; limited_user (who
+        is allowed front_door) should be permitted."""
+        app = self._make_app(_MULTI_CAMERA_CONFIG)
+        # front_door_main is the alias defined in live.streams for front_door
+        resp = self._get_stream(app, "front_door_main", role="limited_user")
+        assert resp.status_code not in (401, 403), (
+            f"limited_user should be allowed on alias front_door_main; "
+            f"got {resp.status_code}"
+        )
+
+    def test_stream_alias_blocked_when_owning_camera_disallowed(self):
+        """limited_user cannot access a stream alias that belongs to a camera they
+        are not allowed to see."""
+        # Give back_door a stream alias and restrict limited_user to front_door only
+        config = {
+            "mqtt": {"host": "mqtt"},
+            "auth": {
+                "roles": {
+                    "limited_user": ["front_door"],
+                }
+            },
+            "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},
+                    "live": {"streams": {"default": "back_door_main"}},
+                },
+            },
+        }
+        app = self._make_app(config)
+        resp = self._get_stream(app, "back_door_main", role="limited_user")
+        assert resp.status_code == 403, (
+            f"limited_user should be denied on alias back_door_main; "
+            f"got {resp.status_code}"
+        )

frigate/test/test_env.py

@@ -0,0 +1,105 @@
+"""Tests for environment variable handling."""
+
+import os
+import unittest
+
+from frigate.config.env import (
+    FRIGATE_ENV_VARS,
+    validate_env_string,
+    validate_env_vars,
+)
+
+
+class TestEnvString(unittest.TestCase):
+    def setUp(self):
+        self._original_env_vars = dict(FRIGATE_ENV_VARS)
+
+    def tearDown(self):
+        FRIGATE_ENV_VARS.clear()
+        FRIGATE_ENV_VARS.update(self._original_env_vars)
+
+    def test_substitution(self):
+        """EnvString substitutes FRIGATE_ env vars."""
+        FRIGATE_ENV_VARS["FRIGATE_TEST_HOST"] = "192.168.1.100"
+        result = validate_env_string("{FRIGATE_TEST_HOST}")
+        self.assertEqual(result, "192.168.1.100")
+
+    def test_substitution_in_url(self):
+        """EnvString substitutes vars embedded in a URL."""
+        FRIGATE_ENV_VARS["FRIGATE_CAM_USER"] = "admin"
+        FRIGATE_ENV_VARS["FRIGATE_CAM_PASS"] = "secret"
+        result = validate_env_string(
+            "rtsp://{FRIGATE_CAM_USER}:{FRIGATE_CAM_PASS}@10.0.0.1/stream"
+        )
+        self.assertEqual(result, "rtsp://admin:secret@10.0.0.1/stream")
+
+    def test_no_placeholder(self):
+        """Plain strings pass through unchanged."""
+        result = validate_env_string("192.168.1.1")
+        self.assertEqual(result, "192.168.1.1")
+
+    def test_unknown_var_raises(self):
+        """Referencing an unknown var raises KeyError."""
+        with self.assertRaises(KeyError):
+            validate_env_string("{FRIGATE_NONEXISTENT_VAR}")
+
+
+class TestEnvVars(unittest.TestCase):
+    def setUp(self):
+        self._original_env_vars = dict(FRIGATE_ENV_VARS)
+        self._original_environ = os.environ.copy()
+
+    def tearDown(self):
+        FRIGATE_ENV_VARS.clear()
+        FRIGATE_ENV_VARS.update(self._original_env_vars)
+        # Clean up any env vars we set
+        for key in list(os.environ.keys()):
+            if key not in self._original_environ:
+                del os.environ[key]
+
+    def _make_context(self, install: bool):
+        """Create a mock ValidationInfo with the given install flag."""
+
+        class MockContext:
+            def __init__(self, ctx):
+                self.context = ctx
+
+        mock = MockContext({"install": install})
+        return mock
+
+    def test_install_sets_os_environ(self):
+        """validate_env_vars with install=True sets os.environ."""
+        ctx = self._make_context(install=True)
+        validate_env_vars({"MY_CUSTOM_VAR": "value123"}, ctx)
+        self.assertEqual(os.environ.get("MY_CUSTOM_VAR"), "value123")
+
+    def test_install_updates_frigate_env_vars(self):
+        """validate_env_vars with install=True updates FRIGATE_ENV_VARS for FRIGATE_ keys."""
+        ctx = self._make_context(install=True)
+        validate_env_vars({"FRIGATE_MQTT_PASS": "secret"}, ctx)
+        self.assertEqual(FRIGATE_ENV_VARS["FRIGATE_MQTT_PASS"], "secret")
+
+    def test_install_skips_non_frigate_in_env_vars_dict(self):
+        """Non-FRIGATE_ keys are set in os.environ but not in FRIGATE_ENV_VARS."""
+        ctx = self._make_context(install=True)
+        validate_env_vars({"OTHER_VAR": "value"}, ctx)
+        self.assertEqual(os.environ.get("OTHER_VAR"), "value")
+        self.assertNotIn("OTHER_VAR", FRIGATE_ENV_VARS)
+
+    def test_no_install_does_not_set(self):
+        """validate_env_vars without install=True does not modify state."""
+        ctx = self._make_context(install=False)
+        validate_env_vars({"FRIGATE_SKIP": "nope"}, ctx)
+        self.assertNotIn("FRIGATE_SKIP", FRIGATE_ENV_VARS)
+        self.assertNotIn("FRIGATE_SKIP", os.environ)
+
+    def test_env_vars_available_for_env_string(self):
+        """Vars set via validate_env_vars are usable in validate_env_string."""
+        ctx = self._make_context(install=True)
+        validate_env_vars({"FRIGATE_BROKER": "mqtt.local"}, ctx)
+        result = validate_env_string("{FRIGATE_BROKER}")
+        self.assertEqual(result, "mqtt.local")
+
+
+if __name__ == "__main__":
+    unittest.main()