From f650ccb3b5efc1908fc9be1761a82cbc07387ed6 Mon Sep 17 00:00:00 2001 From: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com> Date: Sun, 18 Oct 2026 12:15:30 -0500 Subject: [PATCH] validate networking.listen values before rendering them into nginx config ListenConfig and get_nginx_settings.py only checked that the text after the last colon parsed as an int, and tempio rendered the raw string into the listen directive, so a value like "5000; location /x/ { alias /; } #:5000" injected nginx directives. Both now go through parse_listen_address, which accepts only a port, IPv4:port, or [IPv6]:port with a 1 to 65535 port and rebuilds the value from the parsed parts. get_nginx_settings.py falls back to the default port on an invalid value so nginx still serves the safe-mode UI. --- .../usr/local/nginx/get_nginx_settings.py | 27 ++++++------ frigate/config/network.py | 21 +++++---- frigate/test/test_network_config.py | 44 +++++++++++++++++++ frigate/util/config.py | 35 +++++++++++++++ 4 files changed, 103 insertions(+), 24 deletions(-) diff --git a/docker/main/rootfs/usr/local/nginx/get_nginx_settings.py b/docker/main/rootfs/usr/local/nginx/get_nginx_settings.py index 79cda36860..44d8a01782 100644 --- a/docker/main/rootfs/usr/local/nginx/get_nginx_settings.py +++ b/docker/main/rootfs/usr/local/nginx/get_nginx_settings.py @@ -8,7 +8,7 @@ from typing import Any from ruamel.yaml import YAML sys.path.insert(0, "/opt/frigate") -from frigate.util.config import find_config_file +from frigate.util.config import find_config_file, parse_listen_address sys.path.remove("/opt/frigate") @@ -35,20 +35,21 @@ ipv6_config: dict[str, Any] = networking_config.get("ipv6", {}) ipv6_config.setdefault("enabled", False) listen_config: dict[str, Any] = networking_config.get("listen", {}) -listen_config.setdefault("internal", 5000) -listen_config.setdefault("external", 8971) -# handle case where internal port is a string with ip:port -internal_port = listen_config["internal"] -if type(internal_port) is str: - internal_port = int(internal_port.split(":")[-1]) -listen_config["internal_port"] = internal_port +# Frigate's own validation runs elsewhere, and a bad value here must not stop +# nginx from serving the safe-mode UI, which uses the default ports +for key, default in (("internal", 5000), ("external", 8971)): + try: + value, port = parse_listen_address(listen_config.get(key, default)) + except ValueError: + print( + f"[WARN] Invalid networking.listen.{key}, using {default}", + file=sys.stderr, + ) + value, port = default, default -# handle case where external port is a string with ip:port -external_port = listen_config["external"] -if type(external_port) is str: - external_port = int(external_port.split(":")[-1]) -listen_config["external_port"] = external_port + listen_config[key] = value + listen_config[f"{key}_port"] = port base_path = os.environ.get("FRIGATE_BASE_PATH", "") diff --git a/frigate/config/network.py b/frigate/config/network.py index 1838b410bd..49f7c1984b 100644 --- a/frigate/config/network.py +++ b/frigate/config/network.py @@ -1,18 +1,12 @@ -from pydantic import Field, model_validator +from pydantic import Field, field_validator, model_validator + +from frigate.util.config import parse_listen_address 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, @@ -33,13 +27,18 @@ class ListenConfig(FrigateBaseModel): description="External listening port for Frigate (default 8971).", ) + @field_validator("internal", "external") + @classmethod + def validate_listen(cls, value: int | str) -> int | str: + return parse_listen_address(value)[0] + @property def internal_port(self) -> int: - return parse_listen_port(self.internal) + return parse_listen_address(self.internal)[1] @property def external_port(self) -> int: - return parse_listen_port(self.external) + return parse_listen_address(self.external)[1] @model_validator(mode="after") def validate_distinct_ports(self) -> "ListenConfig": diff --git a/frigate/test/test_network_config.py b/frigate/test/test_network_config.py index 7c78bf7bc4..f3e3943d2f 100644 --- a/frigate/test/test_network_config.py +++ b/frigate/test/test_network_config.py @@ -36,6 +36,50 @@ class TestListenConfig(unittest.TestCase): self.assertEqual(listen.internal_port, 5001) self.assertEqual(listen.external_port, 8971) + def test_valid_listen_values_are_rebuilt(self): + cases = { + 5000: 5000, + "5000": 5000, + "127.0.0.1:5000": "127.0.0.1:5000", + "0.0.0.0:5000": "0.0.0.0:5000", + "[::1]:5000": "[::1]:5000", + "[::]:5000": "[::]:5000", + "[2001:DB8:0:0::1]:5000": "[2001:db8::1]:5000", + } + + for value, expected in cases.items(): + with self.subTest(value=value): + listen = ListenConfig(internal=value) + + self.assertEqual(listen.internal, expected) + self.assertEqual(listen.internal_port, 5000) + + def test_invalid_listen_values_rejected(self): + cases = [ + "5000; location /x/ { alias /; autoindex on; } #:5000", + "127.0.0.1:5000; include /etc/passwd", + "5000 ssl", + " 5000", + "5000;", + "5000\n", + "localhost:5000", + "::1:5000", + "[127.0.0.1]:5000", + "[fe80::1%eth0]:5000", + ":5000", + "127.0.0.1:", + "127.0.0.1:+5000", + "127.0.0.01:5000", + 0, + 65536, + -1, + ] + + for value in cases: + with self.subTest(value=value): + with self.assertRaises(ValidationError): + ListenConfig(internal=value) + if __name__ == "__main__": unittest.main(verbosity=2) diff --git a/frigate/util/config.py b/frigate/util/config.py index 2cfe8612d2..04da7f72ee 100644 --- a/frigate/util/config.py +++ b/frigate/util/config.py @@ -1,8 +1,10 @@ """configuration utils.""" import asyncio +import ipaddress import logging import os +import re import shutil from functools import cache from typing import Any @@ -126,6 +128,39 @@ def redact_credential(obj: dict[str, Any], key: str) -> None: obj.pop(key, None) +def parse_listen_address(value: int | str) -> tuple[int | str, int]: + """Validate a networking.listen value and rebuild it from its parsed parts. + + Accepts a port, "IPv4:port", or "[IPv6]:port". nginx receives the rebuilt + value verbatim, so anything else raises ValueError. + + Returns: + The normalized value (an int for a bare port) and the port + """ + if isinstance(value, bool) or not isinstance(value, int | str): + raise ValueError(f"invalid listen value: {value!r}") + + host, sep, port_text = str(value).rpartition(":") + + if not re.fullmatch(r"[0-9]{1,5}", port_text) or not (1 <= int(port_text) <= 65535): + raise ValueError(f"invalid listen port: {value!r}") + + port = int(port_text) + + if not sep: + return port, port + + if host.startswith("[") and host.endswith("]"): + address = ipaddress.IPv6Address(host[1:-1]) + + if address.scope_id: + raise ValueError(f"listen address must not have a scope ID: {value!r}") + + return f"[{address}]:{port}", port + + return f"{ipaddress.IPv4Address(host)}:{port}", port + + def find_config_file() -> str: """Return the path of the config file to use.