Pin the internal auth port to the value nginx bound at startup

/auth grants anonymous admin to any request whose X-Server-Port matches networking.listen.internal, but it read that port off the live config while nginx binds its listeners once at container start and never reloads them, so any path that swaps the running config could move the trusted port without nginx moving with it. Saving networking.listen.internal equal to the external port applied immediately despite the restart-required warning, which handed unauthenticated admin to everything reaching the external port. Snapshot the port at app creation and compare against that instead, and reject a config whose two listeners share a port number, which nginx would refuse to start with anyway.

docs/docs/configuration/advanced/system.md

@@ -293,6 +293,10 @@ networking:
 
 This setting is for advanced users. For the majority of use cases it's recommended to change the `ports` section of your Docker compose file or use the Docker `run` `--publish` option instead, e.g. `-p 443:8971`. Changing Frigate's ports may break some integrations.
 
+The internal and external ports must be different port numbers, and Frigate will refuse to start otherwise. Requests arriving on the internal port are treated as authenticated admins, so pointing both at the same port would remove authentication from the external one.
+
+Nginx binds these ports when it starts, so port changes only take effect after Frigate restarts.
+
 :::
 
 ### Customizing the Nginx configuration

frigate/api/auth.py

@@ -31,7 +31,7 @@ from frigate.api.media_auth import (
     deny_response_for_media_uri,
     is_role_restricted,
 )
-from frigate.config import AuthConfig, NetworkingConfig, ProxyConfig
+from frigate.config import AuthConfig, ProxyConfig
 from frigate.const import CONFIG_DIR, JWT_SECRET_ENV_VAR, PASSWORD_HASH_ALGORITHM
 from frigate.models import User
 
@@ -620,18 +620,18 @@ def resolve_role(
 def auth(request: Request):
     auth_config: AuthConfig = request.app.frigate_config.auth
     proxy_config: ProxyConfig = request.app.frigate_config.proxy
-    networking_config: NetworkingConfig = request.app.frigate_config.networking
 
     success_response = Response("", status_code=202)
 
-    # handle case where internal port is a string with ip:port
-    internal_port = networking_config.listen.internal
-    if type(internal_port) is str:
-        internal_port = int(internal_port.split(":")[-1])
-
     # dont require auth if the request is on the internal port
-    # this header is set by Frigate's nginx proxy, so it cant be spoofed
-    if int(request.headers.get("x-server-port", default=0)) == internal_port:
+    # this header is set by Frigate's nginx proxy, so it cant be spoofed.
+    # the port is the boot-time snapshot rather than the live config value:
+    # nginx's listeners are fixed at container start, so an in-memory config
+    # change must never move the port that is trusted here
+    if (
+        int(request.headers.get("x-server-port", default=0))
+        == request.app.auth_internal_port
+    ):
         success_response.headers["remote-user"] = "anonymous"
         success_response.headers["remote-role"] = "admin"
         return success_response

frigate/api/fastapi_app.py

@@ -152,6 +152,8 @@ def create_fastapi_app(
     app.include_router(debug_replay.router)
     # App Properties
     app.frigate_config = frigate_config
+    # snapshot the port nginx bound at startup, the live config can be swapped
+    app.auth_internal_port = frigate_config.networking.listen.internal_port
     app.genai_manager = GenAIClientManager(frigate_config)
     app.embeddings = embeddings
     app.detected_frames_processor = detected_frames_processor

frigate/config/network.py

@@ -1,10 +1,18 @@
-from pydantic import Field
+from pydantic import Field, model_validator
 
 from .base import FrigateBaseModel
 
 __all__ = ["IPv6Config", "ListenConfig", "NetworkingConfig"]
 
 
+def parse_listen_port(value: int | str) -> int:
+    """Return the port number from a bare port or an "address:port" value."""
+    if isinstance(value, str):
+        return int(value.split(":")[-1])
+
+    return value
+
+
 class IPv6Config(FrigateBaseModel):
     enabled: bool = Field(
         default=False,
@@ -25,6 +33,21 @@ class ListenConfig(FrigateBaseModel):
         description="External listening port for Frigate (default 8971).",
     )
 
+    @property
+    def internal_port(self) -> int:
+        return parse_listen_port(self.internal)
+
+    @property
+    def external_port(self) -> int:
+        return parse_listen_port(self.external)
+
+    @model_validator(mode="after")
+    def validate_distinct_ports(self) -> "ListenConfig":
+        if self.internal_port == self.external_port:
+            raise ValueError("internal and external must listen on different ports")
+
+        return self
+
 
 class NetworkingConfig(FrigateBaseModel):
     ipv6: IPv6Config = Field(

frigate/test/http_api/test_http_auth_internal_port.py

@@ -0,0 +1,174 @@
+"""Tests that the internal port trusted by /auth cannot be moved at runtime."""
+
+import os
+import tempfile
+import unittest
+from unittest.mock import MagicMock, Mock, patch
+
+import ruamel.yaml
+from fastapi import Request
+
+from frigate.api.auth import get_allowed_cameras_for_filter, get_current_user
+from frigate.api.fastapi_app import create_fastapi_app
+from frigate.config import FrigateConfig
+from frigate.config.camera.updater import CameraConfigUpdatePublisher
+from frigate.const import JWT_SECRET_ENV_VAR
+from frigate.models import Event, Recordings, ReviewSegment
+from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
+
+
+@patch.dict(os.environ, {JWT_SECRET_ENV_VAR: "test-secret"})
+class TestAuthInternalPort(BaseTestHttp):
+    """/auth grants anonymous admin by port, so that port must stay put.
+
+    nginx binds its listeners once at container start and never reloads them,
+    but /api/config/set can swap the live config object mid-process. If /auth
+    read the port off the live config, saving networking.listen.internal would
+    hand unauthenticated admin to whoever can reach the external port.
+    """
+
+    def setUp(self):
+        super().setUp(models=[Event, Recordings, ReviewSegment])
+        self.minimal_config = {
+            "mqtt": {"host": "mqtt"},
+            "auth": {"enabled": True},
+            "networking": {"listen": {"internal": 5000, "external": 8971}},
+            "cameras": {
+                "front_door": {
+                    "ffmpeg": {
+                        "inputs": [
+                            {"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]}
+                        ]
+                    },
+                    "detect": {
+                        "height": 1080,
+                        "width": 1920,
+                        "fps": 5,
+                    },
+                }
+            },
+        }
+
+    def _create_app(self):
+        mock_publisher = Mock(spec=CameraConfigUpdatePublisher)
+        mock_publisher.publisher = MagicMock()
+
+        app = create_fastapi_app(
+            FrigateConfig(**self.minimal_config),
+            self.db,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            mock_publisher,
+            None,
+            enforce_default_admin=False,
+        )
+
+        async def mock_get_current_user(request: Request):
+            return {
+                "username": request.headers.get("remote-user"),
+                "role": request.headers.get("remote-role"),
+            }
+
+        async def mock_get_allowed_cameras_for_filter(request: Request):
+            return list(self.minimal_config.get("cameras", {}).keys())
+
+        app.dependency_overrides[get_current_user] = mock_get_current_user
+        app.dependency_overrides[get_allowed_cameras_for_filter] = (
+            mock_get_allowed_cameras_for_filter
+        )
+
+        return app
+
+    def _write_config_file(self):
+        """Write the minimal config to a temp YAML file and return the path."""
+        yaml = ruamel.yaml.YAML()
+        f = tempfile.NamedTemporaryFile(mode="w", suffix=".yml", delete=False)
+        yaml.dump(self.minimal_config, f)
+        f.close()
+        return f.name
+
+    def test_internal_port_is_anonymous_admin(self):
+        app = self._create_app()
+
+        with AuthTestClient(app) as client:
+            resp = client.get("/auth", headers={"x-server-port": "5000"})
+
+            self.assertEqual(resp.status_code, 202)
+            self.assertEqual(resp.headers["remote-user"], "anonymous")
+            self.assertEqual(resp.headers["remote-role"], "admin")
+
+    def test_external_port_requires_auth(self):
+        app = self._create_app()
+
+        with AuthTestClient(app) as client:
+            resp = client.get("/auth", headers={"x-server-port": "8971"})
+
+            self.assertEqual(resp.status_code, 401)
+
+    def test_swapped_config_does_not_move_the_trusted_port(self):
+        """The live config is not what /auth trusts.
+
+        Stands in for every path that can rebind app.frigate_config while the
+        process runs, whatever restart flag the caller claimed.
+        """
+        app = self._create_app()
+
+        swapped = FrigateConfig(
+            **{
+                **self.minimal_config,
+                "networking": {"listen": {"internal": 8971, "external": 5000}},
+            }
+        )
+        app.frigate_config = swapped
+
+        with AuthTestClient(app) as client:
+            resp = client.get("/auth", headers={"x-server-port": "8971"})
+            self.assertEqual(resp.status_code, 401)
+
+            # nginx is still listening where it was told to at boot
+            resp = client.get("/auth", headers={"x-server-port": "5000"})
+            self.assertEqual(resp.status_code, 202)
+            self.assertEqual(resp.headers["remote-role"], "admin")
+
+    @patch("frigate.api.app.find_config_file")
+    def test_config_set_rejects_internal_matching_external(self, mock_find_config):
+        """Saving the internal port onto the external one is refused outright."""
+        config_path = self._write_config_file()
+        mock_find_config.return_value = config_path
+
+        try:
+            app = self._create_app()
+
+            with AuthTestClient(app) as client:
+                resp = client.put(
+                    "/config/set",
+                    json={
+                        "config_data": {"networking": {"listen": {"internal": 8971}}},
+                        "update_topic": "config/networking",
+                        "requires_restart": 1,
+                    },
+                )
+
+                self.assertEqual(resp.status_code, 400)
+                self.assertFalse(resp.json()["success"])
+
+                # the rejected save must not have reached the live config
+                self.assertEqual(
+                    app.frigate_config.networking.listen.internal_port, 5000
+                )
+
+                resp = client.get("/auth", headers={"x-server-port": "8971"})
+                self.assertEqual(resp.status_code, 401)
+
+            with open(config_path) as f:
+                self.assertNotIn("8971", f.read().split("external")[0])
+        finally:
+            os.unlink(config_path)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

frigate/test/test_network_config.py

@@ -0,0 +1,41 @@
+"""Tests for networking config validation."""
+
+import unittest
+
+from pydantic import ValidationError
+
+from frigate.config.network import ListenConfig
+
+
+class TestListenConfig(unittest.TestCase):
+    def test_defaults_are_distinct(self):
+        listen = ListenConfig()
+
+        self.assertEqual(listen.internal_port, 5000)
+        self.assertEqual(listen.external_port, 8971)
+
+    def test_address_and_port_string_is_parsed(self):
+        listen = ListenConfig(internal="127.0.0.1:5000", external="0.0.0.0:8971")
+
+        self.assertEqual(listen.internal_port, 5000)
+        self.assertEqual(listen.external_port, 8971)
+
+    def test_identical_ports_rejected(self):
+        with self.assertRaises(ValidationError):
+            ListenConfig(internal=8971, external=8971)
+
+    def test_same_port_on_different_addresses_rejected(self):
+        # nginx would accept these as distinct listeners, but /auth decides on
+        # the port alone, so the external one would inherit anonymous admin
+        with self.assertRaises(ValidationError):
+            ListenConfig(internal="127.0.0.1:8971", external="0.0.0.0:8971")
+
+    def test_distinct_ports_accepted(self):
+        listen = ListenConfig(internal=5001, external="0.0.0.0:8971")
+
+        self.assertEqual(listen.internal_port, 5001)
+        self.assertEqual(listen.external_port, 8971)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)