disable open_file_cache for review thumbnails

Review thumbnails are rewritten in place while a review is active, and nginx's open_file_cache reused the stale file size for up to 2 minutes. A larger rewrite was served truncated with a 200, and a smaller one aborted the response mid-transfer, which broke notification images. The review thumbnail location now skips the cache.

docker/main/rootfs/etc/s6-overlay/s6-rc.d/certsync/run

@@ -36,8 +36,8 @@ else
     lefile="/config/tls/fullchain.pem"
 fi
 
-tls_enabled=`python3 /usr/local/nginx/get_nginx_settings.py | jq -r .tls.enabled`
-listen_external_port=`python3 /usr/local/nginx/get_nginx_settings.py | jq -r .listen.external_port`
+tls_enabled=`python3 -I /usr/local/nginx/get_nginx_settings.py | jq -r .tls.enabled`
+listen_external_port=`python3 -I /usr/local/nginx/get_nginx_settings.py | jq -r .listen.external_port`
 
 while true
 do

docker/main/rootfs/etc/s6-overlay/s6-rc.d/frigate/run

@@ -11,30 +11,13 @@ if [[ "$(id -u)" -eq 0 ]]; then
     fi
 fi
 
-# /root survives s6-setuidgid and breaks cache writes after the drop; set
-# before opt_in_out so the opt-out marker lands where the service will look
-if [[ "$runs_as_root" -eq 0 ]]; then
-    export HOME=/config
-fi
-
-# detector runtimes installed at first start (pip install --user) live under
-# $HOME/.local; the dynamic loader only reads LD_LIBRARY_PATH at exec time
-export LD_LIBRARY_PATH="${LD_LIBRARY_PATH:+${LD_LIBRARY_PATH}:}${HOME}/.local/lib"
-export PATH="${PATH}:${HOME}/.local/bin"
-
-# opt out of openvino telemetry
-if [ -e /usr/local/bin/opt_in_out ]; then
-  /usr/local/bin/opt_in_out --opt_out > /dev/null 2>&1
-fi
-
-# Logs should be sent to stdout so that s6 can collect them
-
 # Tell S6-Overlay not to restart this service
 s6-svc -O .
 
+# Runs as root with root's HOME and LD_LIBRARY_PATH; -I skips user site-packages
 function set_libva_version() {
     local ffmpeg_path
-    ffmpeg_path=$(python3 /usr/local/ffmpeg/get_ffmpeg_path.py)
+    ffmpeg_path=$(python3 -I /usr/local/ffmpeg/get_ffmpeg_path.py)
     LIBAVFORMAT_VERSION_MAJOR=$("$ffmpeg_path" -version | grep -Po "libavformat\W+\K\d+")
     export LIBAVFORMAT_VERSION_MAJOR
 }
@@ -42,14 +25,32 @@ function set_libva_version() {
 echo "[INFO] Preparing Frigate..."
 set_libva_version
 
+# HOME=/root survives s6-setuidgid, so any run that is not root gets
+# HOME=/config explicitly. The runtime user owns /config, so these paths go
+# only to the service process, never to root. Detector runtimes installed at
+# first start (pip install --user) live under $HOME/.local.
+run_as=()
+runtime_home="${HOME}"
+if [[ "$runs_as_root" -eq 0 ]]; then
+    runtime_home=/config
+    if [[ "$(id -u)" -eq 0 ]]; then
+        run_as=(s6-setuidgid frigate)
+    fi
+fi
+run_as+=(env
+    "HOME=${runtime_home}"
+    "LD_LIBRARY_PATH=${LD_LIBRARY_PATH:+${LD_LIBRARY_PATH}:}${runtime_home}/.local/lib"
+    "PATH=${PATH}:${runtime_home}/.local/bin")
+
+# opt out of openvino telemetry; the marker lands under the service's HOME
+if [ -e /usr/local/bin/opt_in_out ]; then
+  "${run_as[@]}" /usr/local/bin/opt_in_out --opt_out > /dev/null 2>&1 || true
+fi
+
 echo "[INFO] Starting Frigate..."
 
 cd /opt/frigate || echo "[ERROR] Failed to change working directory to /opt/frigate"
 
 # Replace the bash process with the Frigate process, redirecting stderr to stdout
 exec 2>&1
-if [[ "$(id -u)" -ne 0 || "$runs_as_root" -eq 1 ]]; then
-    exec python3 -u -m frigate
-else
-    exec s6-setuidgid frigate python3 -u -m frigate
-fi
+exec "${run_as[@]}" python3 -u -m frigate

docker/main/rootfs/etc/s6-overlay/s6-rc.d/go2rtc/run

@@ -59,38 +59,33 @@ function get_ip_and_port_from_supervisor() {
 
 function set_libva_version() {
     local ffmpeg_path
-    ffmpeg_path=$(python3 /usr/local/ffmpeg/get_ffmpeg_path.py)
+    ffmpeg_path=$(python3 -I /usr/local/ffmpeg/get_ffmpeg_path.py)
     LIBAVFORMAT_VERSION_MAJOR=$("$ffmpeg_path" -version | grep -Po "libavformat\W+\K\d+")
     export LIBAVFORMAT_VERSION_MAJOR
 }
 
 set_libva_version
 
-if [[ -f "/dev/shm/go2rtc.yaml" ]]; then
-    echo "[INFO] Removing stale config from last run..."
-    rm /dev/shm/go2rtc.yaml
+# create_config.py creates the file exclusively and fails if anything
+# reappears at this path
+rm -f /dev/shm/go2rtc.yaml
+
+echo "[INFO] Preparing new go2rtc config..."
+
+if [[ -n "${SUPERVISOR_TOKEN:-}" ]]; then
+    # Running as a Home Assistant Add-on, infer the IP address and port
+    get_ip_and_port_from_supervisor
 fi
 
-if [[ ! -f "/dev/shm/go2rtc.yaml" ]]; then
-    echo "[INFO] Preparing new go2rtc config..."
-
-    if [[ -n "${SUPERVISOR_TOKEN:-}" ]]; then
-        # Running as a Home Assistant Add-on, infer the IP address and port
-        get_ip_and_port_from_supervisor
-    fi
-
-    python3 /usr/local/go2rtc/create_config.py
-else
-    echo "[WARNING] Unable to remove existing go2rtc config. Changes made to your frigate config file may not be recognized. Please remove the /dev/shm/go2rtc.yaml from your docker host manually."
-fi
+python3 -I /usr/local/go2rtc/create_config.py
 
 # HomeKit persistence. The helper is symlink-safe; hand off to go2rtc only when dropping.
 readonly homekit_config_path="/config/go2rtc_homekit.yml"
 if [[ "$(id -u)" -eq 0 && "$runs_as_root" -eq 0 ]]; then
-    python3 /usr/local/go2rtc/prepare_homekit.py "${homekit_config_path}" --chown
-    chown go2rtc:go2rtc /dev/shm/go2rtc.yaml 2>/dev/null || true
+    python3 -I /usr/local/go2rtc/prepare_homekit.py "${homekit_config_path}" --chown
+    chown -h go2rtc:go2rtc /dev/shm/go2rtc.yaml 2>/dev/null || true
 else
-    python3 /usr/local/go2rtc/prepare_homekit.py "${homekit_config_path}"
+    python3 -I /usr/local/go2rtc/prepare_homekit.py "${homekit_config_path}"
 fi
 
 readonly config_path="/config"

docker/main/rootfs/etc/s6-overlay/s6-rc.d/nginx/run

@@ -124,7 +124,7 @@ fi
 mkdir -p /etc/letsencrypt/www 2>/dev/null || true
 
 # nginx settings are read once; both templates consume them
-nginx_settings=$(python3 /usr/local/nginx/get_nginx_settings.py)
+nginx_settings=$(python3 -I /usr/local/nginx/get_nginx_settings.py)
 
 # build templates for optional FRIGATE_BASE_PATH environment variable
 echo "$nginx_settings" | \

docker/main/rootfs/usr/local/go2rtc/create_config.py

@@ -198,9 +198,13 @@ if config.get("birdseye", {}).get("restream", False):
     else:
         go2rtc_config["streams"] = {"birdseye": ffmpeg_cmd}
 
-# Write go2rtc_config to /dev/shm/go2rtc.yaml
-with open("/dev/shm/go2rtc.yaml", "w") as f:
+# The run script removes the old file first. Exclusive create never follows a
+# link the go2rtc user planted in sticky /dev/shm. The config contains camera
+# credentials, so it is not world-readable.
+fd = os.open(
+    "/dev/shm/go2rtc.yaml",
+    os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
+    0o640,
+)
+with os.fdopen(fd, "w") as f:
     yaml.dump(go2rtc_config, f)
-
-# config contains camera credentials; do not leave it world-readable
-os.chmod("/dev/shm/go2rtc.yaml", 0o640)

docker/main/rootfs/usr/local/nginx/conf/auth_location.conf

@@ -19,6 +19,7 @@ location /auth {
     # this cannot be spoofed
     proxy_set_header X-Go2rtc-Stream-Proxy $go2rtc_stream_proxy;
     proxy_set_header X-Server-Port $server_port;
+    proxy_set_header X-Forwarded-Proto $scheme;
     proxy_set_header Content-Length "";
     # Pass along auth related info
     proxy_set_header Authorization $http_authorization;

docker/main/rootfs/usr/local/nginx/conf/nginx.conf

@@ -176,6 +176,12 @@ http {
             add_header Cache-Control "public";
             autoindex on;
             root /media/frigate;
+
+            # review thumbnails are rewritten in place while a review is active,
+            # so a cached file size truncates the response
+            location /clips/review/ {
+                open_file_cache off;
+            }
         }
 
         location /cache/ {

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,29 @@ 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. Valid values are kept so a fallback never widens
+# the unauthenticated internal listener beyond the configured address.
+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
+
+if listen_config["internal_port"] == listen_config["external_port"]:
+    print(
+        "[ERROR] networking.listen.internal and external share a port",
+        file=sys.stderr,
+    )
+    sys.exit(1)
 
 base_path = os.environ.get("FRIGATE_BASE_PATH", "")
 

docs/docs/configuration/advanced/reference.md

@@ -117,7 +117,9 @@ auth:
   # Optional: Cookie to store the JWT token for native auth (default: shown below)
   cookie_name: frigate_token
   # Optional: Set secure flag on cookie. (default: shown below)
-  # NOTE: This should be set to True if you are using TLS
+  # NOTE: Frigate sets the flag automatically when serving TLS. Set this to True
+  # when a reverse proxy terminates TLS in front of Frigate. A proxy that connects
+  # to the TLS port must serve Frigate to browsers over HTTPS
   cookie_secure: False
   # Optional: Session length in seconds (default: shown below)
   session_length: 86400 # 24 hours

frigate/api/app.py

@@ -61,6 +61,7 @@ from frigate.util.builtin import (
     clean_camera_user_pass,
     deep_merge,
     flatten_config_data,
+    is_masked_camera_path,
     load_labels,
     process_config_query_string,
     split_config_key_path,
@@ -71,6 +72,7 @@ from frigate.util.config import (
     find_config_file,
     redact_credential,
 )
+from frigate.util.env import substitute_frigate_vars
 from frigate.util.live_streams import (
     generated_transcode_streams,
     sync_transcode_streams,
@@ -402,6 +404,8 @@ def config(request: Request):
 
         config["go2rtc"]["streams"][stream_name] = cleaned
 
+    _redact_go2rtc_credentials(config["go2rtc"])
+
     config["plus"] = {"enabled": request.app.frigate_config.plus_api.is_active()}
 
     for index, model in enumerate(config_obj.models):
@@ -428,6 +432,28 @@ def config(request: Request):
     return JSONResponse(content=config)
 
 
+def _redact_go2rtc_credentials(go2rtc: dict[str, Any]) -> None:
+    """Redact go2rtc secrets outside of stream URLs, in place."""
+    for section in ("rtsp", "api", "mqtt"):
+        if isinstance(go2rtc.get(section), dict):
+            redact_credential(go2rtc[section], "username")
+            redact_credential(go2rtc[section], "password")
+
+    if isinstance(go2rtc.get("ngrok"), dict):
+        redact_credential(go2rtc["ngrok"], "command")
+
+    if isinstance(go2rtc.get("homekit"), dict):
+        for accessory in go2rtc["homekit"].values():
+            if isinstance(accessory, dict):
+                for key in ("pin", "device_private", "pairings"):
+                    redact_credential(accessory, key)
+
+    # publish destinations carry stream keys in the URL path
+    if isinstance(go2rtc.get("publish"), dict):
+        for name in list(go2rtc["publish"]):
+            redact_credential(go2rtc["publish"], name)
+
+
 @router.get("/profiles", dependencies=[Depends(allow_any_authenticated())])
 def get_profiles(request: Request):
     """List all available profiles and the currently active profile."""
@@ -742,17 +768,24 @@ def config_save(save_option: str, body: Any = Body(media_type="text/plain")):
         )
 
 
-def _restore_masked_camera_paths(config_data: dict, config: FrigateConfig) -> None:
-    """Substitute incoming `*:*` masked credentials with the in-memory ones.
+def _restore_masked_camera_paths(config_data: dict, saved: dict) -> str | None:
+    """Substitute incoming masked camera paths with the saved ones.
 
     The /config response masks ffmpeg input credentials, so the settings UI
     sends the masked path back when sibling fields (e.g. hwaccel_args) are
-    edited.  Without this we'd write `rtsp://*:*@host` into YAML and lose
-    the real credentials.  Mutates `config_data` in place.
+    edited. Each masked path is matched to the saved input that masks to the
+    same value, with roles breaking a tie, so deleting or reordering inputs
+    keeps each path's own credentials. The saved path is restored as written,
+    keeping any {FRIGATE_*} placeholder. Mutates `config_data` in place and
+    returns an error message when a masked path has no single match.
     """
     cameras = config_data.get("cameras")
     if not isinstance(cameras, dict):
-        return
+        return None
+
+    saved_cameras = saved.get("cameras")
+    if not isinstance(saved_cameras, dict):
+        saved_cameras = {}
 
     for camera_name, camera_data in cameras.items():
         if not isinstance(camera_data, dict):
@@ -760,20 +793,122 @@ def _restore_masked_camera_paths(config_data: dict, config: FrigateConfig) -> No
         inputs = camera_data.get("ffmpeg", {}).get("inputs")
         if not isinstance(inputs, list):
             continue
-        existing = config.cameras.get(camera_name)
-        if existing is None:
-            continue
-        existing_paths = [inp.path for inp in existing.ffmpeg.inputs]
+
+        saved_camera = saved_cameras.get(camera_name)
+        saved_ffmpeg = (
+            saved_camera.get("ffmpeg") if isinstance(saved_camera, dict) else None
+        )
+        saved_inputs = [
+            inp
+            for inp in (
+                saved_ffmpeg.get("inputs") if isinstance(saved_ffmpeg, dict) else None
+            )
+            or []
+            if isinstance(inp, dict) and isinstance(inp.get("path"), str)
+        ]
+
         for index, input_obj in enumerate(inputs):
             if not isinstance(input_obj, dict):
                 continue
             path = input_obj.get("path")
-            if not isinstance(path, str):
+            if not isinstance(path, str) or not is_masked_camera_path(path):
                 continue
-            if ("://*:*@" in path or "user=*&password=*" in path) and index < len(
-                existing_paths
-            ):
-                input_obj["path"] = existing_paths[index]
+            candidates = [
+                inp
+                for inp in saved_inputs
+                if clean_camera_user_pass(_resolve_frigate_vars(inp["path"])) == path
+            ]
+            if len({inp["path"] for inp in candidates}) > 1:
+                candidates = [
+                    inp
+                    for inp in candidates
+                    if inp.get("roles") == input_obj.get("roles")
+                ]
+            real_paths = {inp["path"] for inp in candidates}
+            if len(real_paths) != 1:
+                return (
+                    f"Input {index} of camera {camera_name} has masked credentials "
+                    "that could not be matched; resend it with real credentials"
+                )
+            input_obj["path"] = real_paths.pop()
+
+    return None
+
+
+def _resolve_frigate_vars(value: str) -> str:
+    """Resolve {FRIGATE_*} placeholders the way /config shows them."""
+    try:
+        return substitute_frigate_vars(value)
+    except ValueError:
+        return value
+
+
+def _saved_config_from_memory(config: FrigateConfig) -> dict[str, Any]:
+    """Shape the in-memory camera inputs like the saved YAML."""
+    return {
+        "cameras": {
+            name: {
+                "ffmpeg": {
+                    "inputs": [
+                        {"path": inp.path, "roles": [role.value for role in inp.roles]}
+                        for inp in camera.ffmpeg.inputs
+                    ]
+                }
+            }
+            for name, camera in config.cameras.items()
+        },
+    }
+
+
+def _config_data_to_updates(
+    config_data: dict, saved: dict[str, Any]
+) -> tuple[dict[str, Any], str | None]:
+    """Flatten a /config/set payload after restoring masked credentials.
+
+    `saved` is the config the masked values were taken from. Returns the flat
+    updates, or an error message when a masked value can't be restored.
+    """
+    error = _restore_masked_camera_paths(config_data, saved)
+    if error:
+        return {}, error
+
+    updates = flatten_config_data(config_data)
+    # Convert None values to empty strings for deletion (e.g., when deleting masks)
+    updates = {k: ("" if v is None else v) for k, v in updates.items()}
+    # Drop sentinel-valued fields so untouched credential placeholders don't
+    # clobber the saved YAML value.
+    updates = {k: v for k, v in updates.items() if v != REDACTED_CREDENTIAL_SENTINEL}
+
+    error = _find_masked_update(updates)
+    if error:
+        return {}, error
+
+    return updates, None
+
+
+def _holds_masked_value(value: Any) -> bool:
+    if isinstance(value, str):
+        return REDACTED_CREDENTIAL_SENTINEL in value or is_masked_camera_path(value)
+    if isinstance(value, dict):
+        return any(_holds_masked_value(v) for v in value.values())
+    if isinstance(value, list):
+        return any(_holds_masked_value(v) for v in value)
+    return False
+
+
+def _find_masked_update(updates: dict[str, Any]) -> str | None:
+    """Return an error naming the first update that still holds a masked value.
+
+    Runs after the restores so a masked path or redaction sentinel that could
+    not be restored is never written over the real credential.
+    """
+    for key, value in updates.items():
+        if _holds_masked_value(value):
+            return (
+                f"{key} contains masked or saved credentials; "
+                "resend it with real values"
+            )
+    return None
 
 
 def _config_set_in_memory(request: Request, body: AppConfigSetBody) -> JSONResponse:
@@ -786,13 +921,13 @@ def _config_set_in_memory(request: Request, body: AppConfigSetBody) -> JSONRespo
     try:
         updates = {}
         if body.config_data:
-            _restore_masked_camera_paths(body.config_data, request.app.frigate_config)
-            updates = flatten_config_data(body.config_data)
-            updates = {k: ("" if v is None else v) for k, v in updates.items()}
-            # Drop any field whose value is still the redaction sentinel
-            updates = {
-                k: v for k, v in updates.items() if v != REDACTED_CREDENTIAL_SENTINEL
-            }
+            updates, error = _config_data_to_updates(
+                body.config_data, _saved_config_from_memory(request.app.frigate_config)
+            )
+            if error:
+                return JSONResponse(
+                    content={"success": False, "message": error}, status_code=400
+                )
 
         if not updates:
             return JSONResponse(
@@ -939,22 +1074,26 @@ def config_set(request: Request, body: AppConfigSetBody):
                 # Filter out empty keys but keep blank values for non-empty keys
                 query_string = {k: v for k, v in query_string.items() if k}
 
+                error = None
+
                 if query_string:
                     updates = process_config_query_string(query_string)
+                    error = _find_masked_update(updates)
                 elif body.config_data:
-                    _restore_masked_camera_paths(
-                        body.config_data, request.app.frigate_config
+                    try:
+                        saved_config = ruamel.yaml.YAML(typ="safe").load(old_raw_config)
+                    except ruamel.yaml.YAMLError:
+                        saved_config = None
+                    updates, error = _config_data_to_updates(
+                        body.config_data,
+                        saved_config if isinstance(saved_config, dict) else {},
+                    )
+
+                if error:
+                    return JSONResponse(
+                        content={"success": False, "message": error},
+                        status_code=400,
                     )
-                    updates = flatten_config_data(body.config_data)
-                    # Convert None values to empty strings for deletion (e.g., when deleting masks)
-                    updates = {k: ("" if v is None else v) for k, v in updates.items()}
-                    # Drop sentinel-valued fields so untouched credential
-                    # placeholders don't clobber the saved YAML value.
-                    updates = {
-                        k: v
-                        for k, v in updates.items()
-                        if v != REDACTED_CREDENTIAL_SENTINEL
-                    }
 
                 if not updates:
                     return JSONResponse(

frigate/api/auth.py

@@ -471,8 +471,37 @@ def create_encoded_jwt(user, role, expiration, secret):
     )
 
 
+def _user_session(user: User) -> tuple[str, int | None]:
+    changed_at = user.password_changed_at
+
+    # floored like iat, so a login in the same second as the change stays valid
+    return user.role, int(changed_at.timestamp()) if changed_at else None
+
+
+def load_user_sessions() -> dict[str, tuple[str, int | None]]:
+    """Map each username to its role and password change time for /auth."""
+    return {user.username: _user_session(user) for user in User.select()}
+
+
+def refresh_user_session(request: Request, username: str) -> None:
+    """Sync the /auth session map after a user is created, changed or deleted."""
+    user = User.get_or_none(User.username == username)
+    if user is None:
+        request.app.user_sessions.pop(username, None)
+    else:
+        request.app.user_sessions[username] = _user_session(user)
+
+
+def is_secure_request(request: Request) -> bool:
+    """Check the cookie setting and nginx's forwarded scheme for HTTPS."""
+    # nginx sets X-Forwarded-Proto to its own scheme, replacing any client value
+    return (
+        request.app.frigate_config.auth.cookie_secure
+        or request.headers.get("x-forwarded-proto") == "https"
+    )
+
+
 def set_jwt_cookie(response: Response, cookie_name, encoded_jwt, max_age, secure):
-    # TODO: ideally this would set secure as well, but that requires TLS
     # Starlette sets SameSite=Lax by default. The cookie is still sent to
     # same-site iframes (e.g. Home Assistant on the same host or domain), but
     # not to cross-site ones. CSRF is also mitigated by requiring a custom
@@ -750,7 +779,6 @@ def auth(request: Request):
     fail_response.headers["location"] = "/login"
 
     JWT_COOKIE_NAME = request.app.frigate_config.auth.cookie_name
-    JWT_COOKIE_SECURE = request.app.frigate_config.auth.cookie_secure
     JWT_REFRESH = request.app.frigate_config.auth.refresh_time
     JWT_SESSION_LENGTH = request.app.frigate_config.auth.session_length
 
@@ -776,20 +804,27 @@ def auth(request: Request):
         if "sub" not in token.claims:
             logger.debug("user not set in jwt token")
             return fail_response
-        if "role" not in token.claims:
-            logger.debug("role not set in jwt token")
-            return fail_response
         if "exp" not in token.claims:
             logger.debug("exp not set in jwt token")
             return fail_response
 
         user = token.claims.get("sub")
-        role = token.claims.get("role")
+        session = request.app.user_sessions.get(user)
+        if session is None:
+            logger.debug("jwt user %s no longer exists", user)
+            return fail_response
+
+        role, password_changed_at = session
+        if password_changed_at is not None and (
+            int(token.claims.get("iat", 0)) < password_changed_at
+        ):
+            logger.debug("jwt token issued before password change")
+            return fail_response
 
-        # the token keeps the role it was issued with, so a role removed from
-        # the config since then must send the user back through login
         if role not in auth_config.roles:
-            logger.debug("jwt role %s is not in the config", role)
+            logger.debug(
+                "jwt user %s has role %s, which is not in the config", user, role
+            )
             return fail_response
 
         current_time = int(time.time())
@@ -812,25 +847,6 @@ def auth(request: Request):
         # if the jwt cookie is expiring soon
         if jwt_source == "cookie" and expiration - JWT_REFRESH <= current_time:
             logger.debug("jwt token expiring soon, refreshing cookie")
-
-            # Check if password has been changed since token was issued
-            # If so, force re-login by rejecting the refresh
-            try:
-                user_obj = User.get_by_id(user)
-                if user_obj.password_changed_at is not None:
-                    token_iat = int(token.claims.get("iat", 0))
-                    password_changed_timestamp = int(
-                        user_obj.password_changed_at.timestamp()
-                    )
-                    if token_iat < password_changed_timestamp:
-                        logger.debug(
-                            "jwt token issued before password change, rejecting refresh"
-                        )
-                        return fail_response
-            except DoesNotExist:
-                logger.debug("user not found")
-                return fail_response
-
             new_expiration = current_time + JWT_SESSION_LENGTH
             new_encoded_jwt = create_encoded_jwt(
                 user, role, new_expiration, request.app.jwt_token
@@ -840,7 +856,7 @@ def auth(request: Request):
                 JWT_COOKIE_NAME,
                 new_encoded_jwt,
                 JWT_SESSION_LENGTH,
-                JWT_COOKIE_SECURE,
+                is_secure_request(request),
             )
 
         success_response.headers["remote-user"] = user
@@ -929,7 +945,6 @@ def login(request: Request, body: AppPostLoginBody):
         )
 
     JWT_COOKIE_NAME = request.app.frigate_config.auth.cookie_name
-    JWT_COOKIE_SECURE = request.app.frigate_config.auth.cookie_secure
     JWT_SESSION_LENGTH = request.app.frigate_config.auth.session_length
     user = body.user
     password = body.password
@@ -945,13 +960,16 @@ def login(request: Request, body: AppPostLoginBody):
 
     password_hash = db_user.password_hash
     if verify_password(password, password_hash):
-        role = getattr(db_user, "role", "viewer")
-        config_roles_set = set(request.app.frigate_config.auth.roles.keys())
-        if role not in config_roles_set:
+        role = db_user.role
+        if role not in request.app.frigate_config.auth.roles:
             logger.warning(
-                f"User {db_user.username} has an invalid role {role}, falling back to 'viewer'."
+                "Login refused for user '%s' with role '%s', which is not in the config",
+                user,
+                role,
             )
-            role = "viewer"
+            failed_logins.record(user, time.time(), known=True)
+            return JSONResponse(content={"message": "Login failed"}, status_code=401)
+
         expiration = int(time.time()) + JWT_SESSION_LENGTH
         encoded_jwt = create_encoded_jwt(user, role, expiration, request.app.jwt_token)
         response = Response("", 200)
@@ -960,7 +978,7 @@ def login(request: Request, body: AppPostLoginBody):
             JWT_COOKIE_NAME,
             encoded_jwt,
             JWT_SESSION_LENGTH,
-            JWT_COOKIE_SECURE,
+            is_secure_request(request),
         )
         # Clear admin_first_time_login flag after successful admin login so the
         # UI stops showing the first-time login documentation link.
@@ -1026,9 +1044,11 @@ def create_user(
             User.username: body.username,
             User.password_hash: password_hash,
             User.role: role,
+            User.password_changed_at: datetime.now(),
             User.notification_tokens: [],
         }
     ).execute()
+    refresh_user_session(request, body.username)
     request.app.config_publisher.publisher.publish("config/auth", None)
     return JSONResponse(content={"username": body.username})
 
@@ -1047,6 +1067,7 @@ def delete_user(request: Request, username: str):
         )
 
     User.delete_by_id(username)
+    refresh_user_session(request, username)
     request.app.config_publisher.publisher.publish("config/auth", None)
     return JSONResponse(content={"success": True})
 
@@ -1114,13 +1135,13 @@ async def update_password(
             User.password_changed_at: datetime.now(),
         }
     ).where(User.username == username).execute()
+    refresh_user_session(request, username)
 
     response = JSONResponse(content={"success": True})
 
     # If user changed their own password, issue a new JWT to keep them logged in
     if current_username == username:
         JWT_COOKIE_NAME = request.app.frigate_config.auth.cookie_name
-        JWT_COOKIE_SECURE = request.app.frigate_config.auth.cookie_secure
         JWT_SESSION_LENGTH = request.app.frigate_config.auth.session_length
 
         expiration = int(time.time()) + JWT_SESSION_LENGTH
@@ -1133,7 +1154,7 @@ async def update_password(
             JWT_COOKIE_NAME,
             encoded_jwt,
             JWT_SESSION_LENGTH,
-            JWT_COOKIE_SECURE,
+            is_secure_request(request),
         )
 
     return response
@@ -1173,6 +1194,7 @@ async def update_role(
         )
 
     User.set_by_id(username, {User.role: body.role})
+    refresh_user_session(request, username)
     request.app.config_publisher.publisher.publish("config/auth", None)
     return JSONResponse(content={"success": True})
 

frigate/api/fastapi_app.py

@@ -196,7 +196,9 @@ def create_fastapi_app(
             key_bytes = str(secret).encode("utf-8")
 
         app.jwt_token = OctKey.import_key(key_bytes)
+        app.user_sessions = auth.load_user_sessions()
     else:
         app.jwt_token = None
+        app.user_sessions = {}
 
     return app

frigate/api/motion_search.py

@@ -9,7 +9,10 @@ from pydantic import BaseModel, Field
 
 from frigate.api.auth import require_camera_access
 from frigate.api.defs.tags import Tags
+from frigate.config import FrigateConfig
 from frigate.jobs.motion_search import (
+    MotionSearchCapacityError,
+    MotionSearchUserBusyError,
     cancel_motion_search_job,
     get_motion_search_job,
     start_motion_search_job,
@@ -95,6 +98,20 @@ class MotionSearchStatusResponse(BaseModel):
     progress: float | None = None
 
 
+def _per_user_identity(request: Request, config: FrigateConfig) -> str | None:
+    """Return the requesting user, or None when callers share one identity.
+
+    Internal-port requests are all "anonymous", and with auth disabled and no
+    proxy user header every browser is "viewer".
+    """
+    username = request.headers.get("remote-user")
+    if not username or username == "anonymous":
+        return None
+    if config.auth.enabled or config.proxy.header_map.user:
+        return username
+    return None
+
+
 @router.post(
     "/{camera_name}/search/motion",
     response_model=MotionSearchStartResponse,
@@ -138,18 +155,35 @@ async def start_motion_search(
             status_code=400,
         )
 
-    # Start the job using the jobs module
-    job_id = start_motion_search_job(
-        config=config,
-        camera_name=camera_name,
-        start_time=body.start_time,
-        end_time=body.end_time,
-        polygon_points=body.polygon_points,
-        threshold=body.threshold,
-        min_area=body.min_area,
-        parallel=body.parallel,
-        max_results=body.max_results,
-    )
+    try:
+        job_id = start_motion_search_job(
+            config=config,
+            camera_name=camera_name,
+            start_time=body.start_time,
+            end_time=body.end_time,
+            polygon_points=body.polygon_points,
+            threshold=body.threshold,
+            min_area=body.min_area,
+            parallel=body.parallel,
+            max_results=body.max_results,
+            username=_per_user_identity(request, config),
+        )
+    except MotionSearchUserBusyError:
+        return JSONResponse(
+            content={
+                "success": False,
+                "message": "A motion search is already running for this user",
+            },
+            status_code=409,
+        )
+    except MotionSearchCapacityError:
+        return JSONResponse(
+            content={
+                "success": False,
+                "message": "Too many motion searches are running",
+            },
+            status_code=429,
+        )
 
     return JSONResponse(
         content={

frigate/api/notification.py

@@ -1,10 +1,8 @@
 """Notification apis."""
 
-import ipaddress
 import logging
 import os
 from typing import Any
-from urllib.parse import urlparse
 
 from cryptography.hazmat.primitives import serialization
 from fastapi import APIRouter, Depends, Request
@@ -13,6 +11,7 @@ from py_vapid import Vapid01, utils
 
 from frigate.api.auth import allow_any_authenticated
 from frigate.api.defs.tags import Tags
+from frigate.comms.webpush import validate_push_endpoint
 from frigate.const import CONFIG_DIR
 from frigate.models import User
 
@@ -20,76 +19,13 @@ logger = logging.getLogger(__name__)
 
 router = APIRouter(tags=[Tags.notifications])
 
-# Push endpoints are opaque URLs but stay well under this in practice
-MAX_ENDPOINT_LENGTH = 2048
-
-# Suffixes that only ever resolve on the local network
-INTERNAL_HOST_SUFFIXES = (".local", ".localdomain", ".internal", ".home.arpa")
-
-
-def _validate_push_endpoint(endpoint: Any) -> str | None:
-    """Return a reason the endpoint is unusable, or None when it is valid.
-
-    Subscriptions are issued by the browser vendor's push service, so a valid
-    endpoint is always a public https URL. Anything else is either a broken
-    registration or an attempt to aim the notification sender somewhere it
-    should not reach.
-    """
-    if not isinstance(endpoint, str) or not endpoint:
-        return "endpoint must be a url"
-
-    if len(endpoint) > MAX_ENDPOINT_LENGTH:
-        return "endpoint is too long"
-
-    try:
-        parsed = urlparse(endpoint)
-        port = parsed.port
-    except ValueError:
-        return "endpoint is not a valid url"
-
-    if parsed.scheme != "https":
-        return "endpoint must use https"
-
-    if parsed.username or parsed.password:
-        return "endpoint must not include credentials"
-
-    if port is not None and port != 443:
-        return "endpoint must use the default https port"
-
-    hostname = parsed.hostname
-
-    if not hostname:
-        return "endpoint must include a hostname"
-
-    try:
-        address = ipaddress.ip_address(hostname)
-    except ValueError:
-        address = None
-
-    if address is not None:
-        # A push service is never reachable at an address only this network can
-        # route, so anything non-global is a misconfiguration at best
-        if not address.is_global:
-            return "endpoint must not use a private address"
-    elif hostname == "localhost" or "." not in hostname:
-        return "endpoint must use a fully qualified hostname"
-    elif hostname.endswith(INTERNAL_HOST_SUFFIXES):
-        return "endpoint must not use an internal hostname"
-
-    # The subscription token lives in the path, and webpush.py assumes there is
-    # a separator after the host when it builds the VAPID audience
-    if len(parsed.path) <= 1:
-        return "endpoint must include a subscription path"
-
-    return None
-
 
 def _validate_subscription(sub: Any) -> str | None:
     """Return a reason the subscription is unusable, or None when it is valid."""
     if not isinstance(sub, dict):
         return "subscription must be an object"
 
-    reason = _validate_push_endpoint(sub.get("endpoint"))
+    reason = validate_push_endpoint(sub.get("endpoint"))
 
     if reason:
         return reason

frigate/app.py

@@ -622,6 +622,7 @@ class FrigateApp:
                     username="admin",
                     role="admin",
                     password_hash=password_hash,
+                    password_changed_at=datetime.datetime.now(),
                     notification_tokens=[],
                 ).execute()
 

frigate/comms/dispatcher.py

@@ -490,7 +490,12 @@ class Dispatcher:
     def publish(self, topic: str, payload: Any, retain: bool = False) -> None:
         """Handle publishing to communicators."""
         for comm in self.comms:
-            comm.publish(topic, payload, retain)
+            try:
+                comm.publish(topic, payload, retain)
+            except Exception:
+                logger.exception(
+                    "Error publishing %s to %s", topic, type(comm).__name__
+                )
 
     def publish_local(self, topic: str, payload: Any) -> None:
         """Publish to every communicator except MQTT.
@@ -501,7 +506,12 @@ class Dispatcher:
             if isinstance(comm, MqttClient):
                 continue
 
-            comm.publish(topic, payload, False)
+            try:
+                comm.publish(topic, payload, False)
+            except Exception:
+                logger.exception(
+                    "Error publishing %s to %s", topic, type(comm).__name__
+                )
 
     def _publish_notices(self) -> None:
         if self.notice_registry is None:

frigate/comms/inter_process.py

@@ -50,7 +50,13 @@ class InterProcessCommunicator(Communicator):
 
                     if isinstance(raw, list):
                         (topic, value) = raw
-                        response = self._dispatcher(topic, value)
+
+                        # REP owes a reply, so a handler error must not end the thread
+                        try:
+                            response = self._dispatcher(topic, value)
+                        except Exception:
+                            logger.exception("Error handling IPC topic %s", topic)
+                            response = None
                     else:
                         logging.warning(
                             f"Received unexpected data type in ZMQ recv_json: {type(raw)}"

frigate/comms/webpush.py

@@ -1,6 +1,7 @@
 """Handle sending notifications for Frigate via Firebase."""
 
 import datetime
+import ipaddress
 import json
 import logging
 import os
@@ -10,7 +11,9 @@ from collections.abc import Callable, Iterator
 from dataclasses import dataclass
 from multiprocessing.synchronize import Event as MpEvent
 from typing import Any
+from urllib.parse import urlparse
 
+import requests
 from py_vapid import Vapid01
 from pywebpush import WebPusher
 from titlecase import titlecase
@@ -41,14 +44,124 @@ class PushNotification:
     ttl: int = 0
 
 
+# Push endpoints are opaque URLs but stay well under this in practice
+MAX_ENDPOINT_LENGTH = 2048
+
+# Suffixes that only ever resolve on the local network
+INTERNAL_HOST_SUFFIXES = (".local", ".localdomain", ".internal", ".home.arpa")
+
+
+def validate_push_endpoint(endpoint: Any) -> str | None:
+    """Return a reason the endpoint is unusable, or None when it is valid.
+
+    Subscriptions are issued by the browser vendor's push service, so a valid
+    endpoint is always a public https URL. Anything else is either a broken
+    registration or an attempt to aim the notification sender somewhere it
+    should not reach.
+    """
+    if not isinstance(endpoint, str) or not endpoint:
+        return "endpoint must be a url"
+
+    if len(endpoint) > MAX_ENDPOINT_LENGTH:
+        return "endpoint is too long"
+
+    try:
+        parsed = urlparse(endpoint)
+        port = parsed.port
+    except ValueError:
+        return "endpoint is not a valid url"
+
+    if parsed.scheme != "https":
+        return "endpoint must use https"
+
+    if parsed.username or parsed.password:
+        return "endpoint must not include credentials"
+
+    if port is not None and port != 443:
+        return "endpoint must use the default https port"
+
+    hostname = parsed.hostname
+
+    if not hostname:
+        return "endpoint must include a hostname"
+
+    try:
+        address = ipaddress.ip_address(hostname)
+    except ValueError:
+        address = None
+
+    if address is not None:
+        # A push service is never reachable at an address only this network can
+        # route, so anything non-global is a misconfiguration at best
+        if not address.is_global:
+            return "endpoint must not use a private address"
+    elif hostname == "localhost" or "." not in hostname:
+        return "endpoint must use a fully qualified hostname"
+    elif "" in hostname.split("."):
+        return "endpoint hostname must not contain empty labels"
+    elif hostname.endswith(INTERNAL_HOST_SUFFIXES):
+        return "endpoint must not use an internal hostname"
+
+    # The subscription token lives in the path
+    if len(parsed.path) <= 1:
+        return "endpoint must include a subscription path"
+
+    return None
+
+
+class _NoRedirectSession(requests.Session):
+    """Session that never follows redirects.
+
+    A push endpoint could otherwise redirect the send to an internal URL.
+    """
+
+    def request(self, *args: Any, **kwargs: Any) -> requests.Response:
+        kwargs["allow_redirects"] = False
+        return super().request(*args, **kwargs)
+
+
+_push_session = _NoRedirectSession()
+
+
+def _vapid_audience(endpoint: str) -> str:
+    parsed = urlparse(endpoint)
+    return f"{parsed.scheme}://{parsed.netloc}"
+
+
 def _build_web_pushers(user: str, subs: list[dict[str, Any]]) -> list[WebPusher]:
     """Build pushers for a user's stored subscriptions, skipping unusable ones."""
     pushers: list[WebPusher] = []
 
     for sub in subs:
+        # Stored rows may predate registration validation, so only skip the
+        # endpoints that would break building the VAPID audience
+        endpoint = sub.get("endpoint") if isinstance(sub, dict) else None
+
+        if not isinstance(endpoint, str):
+            logger.warning(
+                "Skipping notification subscription for %s: no endpoint", user
+            )
+            continue
+
+        try:
+            hostname = urlparse(endpoint).hostname or ""
+        except ValueError:
+            logger.warning(
+                "Skipping notification subscription for %s: unparsable endpoint", user
+            )
+            continue
+
+        if "" in hostname.split("."):
+            logger.warning(
+                "Skipping notification subscription for %s: invalid endpoint host %r",
+                user,
+                hostname,
+            )
+            continue
+
         # WebPusher decodes the stored keys and raises on malformed ones
         try:
-            pushers.append(WebPusher(sub))
+            pushers.append(WebPusher(sub, requests_session=_push_session))
         except Exception:
             logger.warning("Skipping invalid notification subscription for %s", user)
 
@@ -131,7 +244,7 @@ class WebPushClient(Communicator):
             for pushers in self.web_pushers.values():
                 for push in pushers:
                     endpoint: str = push.subscription_info["endpoint"]
-                    endpoints.add(endpoint[0 : endpoint.index("/", 10)])
+                    endpoints.add(_vapid_audience(endpoint))
 
             # create new claim
             for endpoint in endpoints:
@@ -320,9 +433,8 @@ class WebPushClient(Communicator):
 
                 for pusher in self.web_pushers[notification.user]:
                     endpoint = pusher.subscription_info["endpoint"]
-                    headers = self.claim_headers[
-                        endpoint[: endpoint.index("/", 10)]
-                    ].copy()
+
+                    headers = self.claim_headers[_vapid_audience(endpoint)].copy()
                     headers["urgency"] = "high"
 
                     resp = pusher.send(

frigate/config/auth.py

@@ -27,7 +27,7 @@ class AuthConfig(FrigateBaseModel):
     cookie_secure: bool = Field(
         default=False,
         title="Secure cookie flag",
-        description="Set the secure flag on the auth cookie; should be true when using TLS.",
+        description="Always set the secure flag on the auth cookie. Frigate sets it automatically when serving TLS; enable this when a reverse proxy terminates TLS in front of Frigate. A proxy that connects to the TLS port must serve Frigate to browsers over HTTPS.",
     )
     session_length: int = Field(
         default=86400,

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,20 @@ class ListenConfig(FrigateBaseModel):
         description="External listening port for Frigate (default 8971).",
     )
 
+    # Runs before pydantic coerces a bool to an int port, which
+    # get_nginx_settings.py rejects when it reads the raw YAML
+    @field_validator("internal", "external", mode="before")
+    @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":

frigate/const.py

@@ -117,7 +117,19 @@ SUPPORTED_RK_SOCS = ["rk3562", "rk3566", "rk3568", "rk3576", "rk3588"]
 
 REGEX_CAMERA_NAME = r"^[a-zA-Z0-9_-]+$"
 REGEX_RTSP_CAMERA_USER_PASS = r":\/\/[a-zA-Z0-9_-]+:[\S]+@"
-REGEX_HTTP_CAMERA_USER_PASS = r"user=[a-zA-Z0-9_-]+&password=[\S]+"
+# A "user:" userinfo may hold "/", "#" or "@" in the password, so it runs to the
+# last "@" before the next URL. Anything else stops at the authority end. Group 1
+# is the userinfo and group 2 the rest of the URL.
+REGEX_URL_USERINFO = (
+    r"://((?:[^\s/?#:\[\]]*:(?:(?!://)\S)*|[^\s/?#]+))@((?:(?!://)\S)*)"
+)
+_CREDENTIAL_QUERY_KEYS = (
+    r"(?:user(?:name)?|usr|login|pass(?:word|wd)?|pwd"
+    r"|(?:[a-z_]*_)?token|(?:[a-z_]*_)?secret|client_private|api_?key)"
+)
+REGEX_CREDENTIAL_QUERY_KEY = rf"(?i)[?&]{_CREDENTIAL_QUERY_KEYS}="
+REGEX_CREDENTIAL_QUERY_PARAM = rf"(?i)([?&]{_CREDENTIAL_QUERY_KEYS}=)[^&\s]*"
+REGEX_MASKED_CREDENTIAL = rf"(?i)://\*:\*@|[?&]{_CREDENTIAL_QUERY_KEYS}=\*(?![^&\s])"
 REGEX_JSON = re.compile(r"^\s*\{")
 
 # Known Driver Names

frigate/jobs/motion_search.py

@@ -8,7 +8,7 @@ from collections.abc import Callable, Generator, Iterable
 from concurrent.futures import Future, ThreadPoolExecutor, as_completed
 from dataclasses import asdict, dataclass, field
 from datetime import datetime
-from typing import Any, cast
+from typing import Any
 
 import cv2
 import numpy as np
@@ -17,10 +17,6 @@ from frigate.comms.inter_process import InterProcessRequestor
 from frigate.config import FrigateConfig
 from frigate.const import STREAM_TYPE_MAIN, UPDATE_JOB_STATE
 from frigate.jobs.job import Job
-from frigate.jobs.manager import (
-    get_job_by_id,
-    set_current_job,
-)
 from frigate.jobs.motion_search_batch import (
     build_segment_time_map,
     coalesce_runs,
@@ -53,6 +49,12 @@ PROGRESS_BROADCAST_INTERVAL = 1.0
 # Output frame rate for the fixed-cadence fallback used on long-GOP cameras
 # (where keyframe sampling is too sparse). Keyframe cameras ignore this.
 FALLBACK_SAMPLE_FPS = 2.0
+# Each parallel job runs up to 4 decode workers, so this bounds total ffmpeg load.
+MAX_ACTIVE_JOBS = 3
+# Finished jobs stay pollable for this long before eviction (seconds).
+FINISHED_JOB_TTL = 600.0
+# A cancelled job stops holding a slot after this long, even if its thread is wedged.
+CANCEL_GRACE_SECONDS = 60.0
 
 
 @dataclass
@@ -98,6 +100,8 @@ class MotionSearchJob(Job):
     min_area: float = 5.0
     parallel: bool = False
     max_results: int = 25
+    username: str | None = None
+    cancelled_at: float | None = None
 
     # Track progress
     total_frames_processed: int = 0
@@ -112,6 +116,8 @@ class MotionSearchJob(Job):
     def to_dict(self) -> dict[str, Any]:
         """Convert to dictionary for WebSocket transmission."""
         d = asdict(self)
+        d.pop("username")
+        d.pop("cancelled_at")
         if self.metrics:
             d["metrics"] = self.metrics.to_dict()
         return d
@@ -579,7 +585,7 @@ class MotionSearchRunner(threading.Thread):
 
         # Decide keyframe vs fixed-cadence sampling once from the first run's GOP
         # (keyframe structure is a per-camera constant).
-        first_pts = probe_vod_keyframe_pts(ffprobe_path, first_url)
+        first_pts = probe_vod_keyframe_pts(ffprobe_path, first_url, self._should_stop)
         self.use_keyframe = keyframe_sampling_eligible(first_pts)
 
         logger.debug(
@@ -641,7 +647,9 @@ class MotionSearchRunner(threading.Thread):
         time_map = build_segment_time_map(run)
 
         if self.use_keyframe:
-            kf_pts = probe_vod_keyframe_pts(self.ffprobe_path, vod_url)
+            kf_pts = probe_vod_keyframe_pts(
+                self.ffprobe_path, vod_url, self._should_stop
+            )
             frames = list(
                 iter_vod_frames(
                     self.ffmpeg_path,
@@ -777,13 +785,15 @@ class MotionSearchRunner(threading.Thread):
                         run_results, frames = completed.pop(next_idx)
                         if self._merge_run(runs[next_idx], run_results, frames, state):
                             self.internal_stop_event.set()
-                            for pending in futures:
-                                pending.cancel()
                             break
                         next_idx += 1
 
                     if self.internal_stop_event.is_set():
                         break
+
+                # Queued runs would otherwise still execute when the pool exits.
+                for pending in futures:
+                    pending.cancel()
         else:
             for run in runs:
                 if self._should_stop():
@@ -833,11 +843,36 @@ class MotionSearchRunner(threading.Thread):
         return deduplicated
 
 
-# Module-level state for managing per-camera jobs
+class MotionSearchUserBusyError(Exception):
+    """The user already has an active motion search job."""
+
+
+class MotionSearchCapacityError(Exception):
+    """The global limit of active motion search jobs is reached."""
+
+
 _motion_search_jobs: dict[str, tuple[MotionSearchJob, threading.Event]] = {}
 _jobs_lock = threading.Lock()
 
 
+def _evict_finished_jobs(now: float) -> None:
+    """Drop jobs that finished more than FINISHED_JOB_TTL ago. Caller holds the lock."""
+    expired = [
+        job_id
+        for job_id, (job, _) in _motion_search_jobs.items()
+        if job.end_time is not None and now - job.end_time > FINISHED_JOB_TTL
+    ]
+    for job_id in expired:
+        del _motion_search_jobs[job_id]
+
+
+def _is_active(job: MotionSearchJob, now: float) -> bool:
+    """True while the job's thread may still be running and holding a slot."""
+    if job.end_time is not None:
+        return False
+    return job.cancelled_at is None or now - job.cancelled_at < CANCEL_GRACE_SECONDS
+
+
 def stop_all_motion_search_jobs() -> None:
     """Cancel all running motion search jobs for clean shutdown."""
     with _jobs_lock:
@@ -857,10 +892,13 @@ def start_motion_search_job(
     min_area: float = 5.0,
     parallel: bool = False,
     max_results: int = 25,
+    username: str | None = None,
 ) -> str:
     """Start a new motion search job.
 
-    Returns the job ID.
+    Returns the job ID. Raises MotionSearchUserBusyError when ``username`` already
+    has an active job and MotionSearchCapacityError when MAX_ACTIVE_JOBS are active.
+    Pass ``username=None`` when callers share an identity to skip the per-user check.
     """
     job = MotionSearchJob(
         camera=camera_name,
@@ -871,17 +909,38 @@ def start_motion_search_job(
         min_area=min_area,
         parallel=parallel,
         max_results=max_results,
+        username=username,
     )
 
     cancel_event = threading.Event()
 
     with _jobs_lock:
+        now = datetime.now().timestamp()
+        _evict_finished_jobs(now)
+
+        # A cancelled job keeps its thread until the runner sets end_time, so it
+        # still counts toward the global cap. Its user may start a new search.
+        active = [j for j, _ in _motion_search_jobs.values() if _is_active(j, now)]
+        if username is not None and any(
+            j.username == username and j.status != JobStatusTypesEnum.cancelled
+            for j in active
+        ):
+            raise MotionSearchUserBusyError()
+        if len(active) >= MAX_ACTIVE_JOBS:
+            raise MotionSearchCapacityError()
+
         _motion_search_jobs[job.id] = (job, cancel_event)
 
-    set_current_job(job)
-
-    runner = MotionSearchRunner(job, config, cancel_event)
-    runner.start()
+    runner: MotionSearchRunner | None = None
+    try:
+        runner = MotionSearchRunner(job, config, cancel_event)
+        runner.start()
+    except Exception:
+        if runner is not None:
+            runner.requestor.stop()
+        with _jobs_lock:
+            del _motion_search_jobs[job.id]
+        raise
 
     logger.debug(
         "Started motion search job %s for camera %s: "
@@ -904,10 +963,7 @@ def get_motion_search_job(job_id: str) -> MotionSearchJob | None:
     """Get a motion search job by ID."""
     with _jobs_lock:
         job_entry = _motion_search_jobs.get(job_id)
-        if job_entry:
-            return job_entry[0]
-    # Check completed jobs via manager
-    return cast(MotionSearchJob | None, get_job_by_id("motion_search", job_id))
+    return job_entry[0] if job_entry else None
 
 
 def cancel_motion_search_job(job_id: str) -> bool:
@@ -928,6 +984,7 @@ def cancel_motion_search_job(job_id: str) -> bool:
 
         cancel_event.set()
         job.status = JobStatusTypesEnum.cancelled
+        job.cancelled_at = datetime.now().timestamp()
         job_payload = job.to_dict()
         logger.info("Cancelled motion search job %s", job_id)
 

frigate/jobs/motion_search_decode.py

@@ -10,6 +10,7 @@ import json
 import logging
 import subprocess as sp
 import tempfile
+import time
 from collections.abc import Callable, Generator
 from typing import IO
 
@@ -264,11 +265,16 @@ def iter_vod_frames(
     )
 
 
-def probe_vod_keyframe_pts(ffprobe_path: str, vod_url: str) -> list[float]:
+KEYFRAME_PROBE_TIMEOUT = 120.0
+
+
+def probe_vod_keyframe_pts(
+    ffprobe_path: str, vod_url: str, should_stop: Callable[[], bool]
+) -> list[float]:
     """Return keyframe presentation timestamps (VOD stream time) in order.
 
     Reads packet flags via ffprobe over the VOD URL (no decode). Returns [] on
-    any failure so the caller can fall back.
+    any failure, timeout, or stop request so the caller can fall back.
     """
     cmd = [
         ffprobe_path,
@@ -286,16 +292,27 @@ def probe_vod_keyframe_pts(ffprobe_path: str, vod_url: str) -> list[float]:
         "json",
     ]
     try:
-        completed = sp.run(cmd, capture_output=True, text=True, timeout=120)
-    except (OSError, sp.SubprocessError):
+        proc = sp.Popen(cmd, stdout=sp.PIPE, stderr=sp.DEVNULL, text=True)
+    except OSError:
         logger.warning("ffprobe failed for VOD keyframe probe")
         return []
 
-    if completed.returncode != 0 or not completed.stdout:
+    deadline = time.monotonic() + KEYFRAME_PROBE_TIMEOUT
+    while True:
+        try:
+            stdout, _ = proc.communicate(timeout=0.5)
+            break
+        except sp.TimeoutExpired:
+            if should_stop() or time.monotonic() > deadline:
+                proc.kill()
+                proc.communicate()
+                return []
+
+    if proc.returncode != 0 or not stdout:
         return []
 
     try:
-        packets = json.loads(completed.stdout).get("packets", [])
+        packets = json.loads(stdout).get("packets", [])
     except json.JSONDecodeError:
         return []
 

frigate/test/http_api/test_http_auth_jwt_role.py

@@ -1,22 +1,29 @@
-"""Tests that /auth only accepts a JWT whose role is still in the config."""
+"""Tests that /auth takes the role and revocation state from the user, not the JWT."""
 
 import os
 import time
+from datetime import datetime
+from types import SimpleNamespace
 from unittest.mock import MagicMock, Mock, patch
 
-from frigate.api.auth import create_encoded_jwt
+from joserfc import jwt
+
+from frigate.api.auth import create_encoded_jwt, hash_password
 from frigate.api.fastapi_app import create_fastapi_app
+from frigate.app import FrigateApp
 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.models import Event, Recordings, ReviewSegment, User
 from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
 
+PASSWORD = "correct-horse-battery"
+
 
 @patch.dict(os.environ, {JWT_SECRET_ENV_VAR: "test-secret"})
 class TestAuthJwtRole(BaseTestHttp):
     def setUp(self):
-        super().setUp(models=[Event, Recordings, ReviewSegment])
+        super().setUp(models=[Event, Recordings, ReviewSegment, User])
         self.minimal_config = {
             "mqtt": {"host": "mqtt"},
             "auth": {"enabled": True, "roles": {"garage": ["front_door"]}},
@@ -36,6 +43,19 @@ class TestAuthJwtRole(BaseTestHttp):
                 }
             },
         }
+        User.delete().execute()
+        self._insert_user("admin", "admin")
+        self._insert_user("bob", "garage")
+
+    def _insert_user(self, username: str, role: str):
+        User.insert(
+            {
+                User.username: username,
+                User.role: role,
+                User.password_hash: hash_password(PASSWORD, iterations=1),
+                User.notification_tokens: [],
+            }
+        ).execute()
 
     def _create_app(self):
         mock_publisher = Mock(spec=CameraConfigUpdatePublisher)
@@ -55,27 +75,243 @@ class TestAuthJwtRole(BaseTestHttp):
             enforce_default_admin=False,
         )
 
-    def _auth(self, app, role: str):
-        token = create_encoded_jwt("bob", role, int(time.time()) + 3600, app.jwt_token)
-
-        with AuthTestClient(app) as client:
-            return client.get(
-                "/auth",
-                headers={
-                    "x-server-port": "8971",
-                    "authorization": f"Bearer {token}",
-                },
+    def _token(self, app, user: str, role: str, iat: int | None = None) -> str:
+        if iat is None:
+            return create_encoded_jwt(
+                user, role, int(time.time()) + 3600, app.jwt_token
             )
 
+        return jwt.encode(
+            {"alg": "HS256"},
+            {"sub": user, "role": role, "exp": int(time.time()) + 3600, "iat": iat},
+            app.jwt_token,
+        )
+
+    def _auth(self, client, token: str, source: str = "bearer"):
+        headers = {"x-server-port": "8971"}
+        if source == "bearer":
+            headers["authorization"] = f"Bearer {token}"
+        else:
+            client.cookies.set("frigate_token", token)
+
+        resp = client.get("/auth", headers=headers)
+        client.cookies.clear()
+        return resp
+
     def test_configured_role_is_accepted(self):
-        resp = self._auth(self._create_app(), "garage")
+        app = self._create_app()
+        with AuthTestClient(app) as client:
+            resp = self._auth(client, self._token(app, "bob", "garage"))
 
         self.assertEqual(resp.status_code, 202)
         self.assertEqual(resp.headers["remote-user"], "bob")
         self.assertEqual(resp.headers["remote-role"], "garage")
 
+    def test_role_claim_is_ignored(self):
+        app = self._create_app()
+        with AuthTestClient(app) as client:
+            resp = self._auth(client, self._token(app, "bob", "admin"))
+
+        self.assertEqual(resp.status_code, 202)
+        self.assertEqual(resp.headers["remote-role"], "garage")
+
     def test_role_removed_from_config_is_rejected(self):
-        resp = self._auth(self._create_app(), "removed_role")
+        User.update(role="removed_role").where(User.username == "bob").execute()
+        app = self._create_app()
+        with AuthTestClient(app) as client:
+            for source in ("bearer", "cookie"):
+                resp = self._auth(client, self._token(app, "bob", "garage"), source)
+                self.assertEqual(resp.status_code, 401)
+                self.assertNotIn("remote-role", resp.headers)
+
+    def test_login_refused_for_role_removed_from_config(self):
+        User.update(role="removed_role").where(User.username == "bob").execute()
+        with AuthTestClient(self._create_app()) as client:
+            resp = client.post("/login", json={"user": "bob", "password": PASSWORD})
 
         self.assertEqual(resp.status_code, 401)
-        self.assertNotIn("remote-role", resp.headers)
+        self.assertNotIn("set-cookie", resp.headers)
+
+    def test_demoted_user_gets_new_role_on_next_request(self):
+        app = self._create_app()
+        token = self._token(app, "bob", "admin")
+        User.update(role="admin").where(User.username == "bob").execute()
+        app.user_sessions["bob"] = ("admin", None)
+
+        with AuthTestClient(app) as client:
+            self.assertEqual(self._auth(client, token).headers["remote-role"], "admin")
+            resp = client.put("/users/bob/role", json={"role": "viewer"})
+            self.assertEqual(resp.status_code, 200)
+
+            for source in ("bearer", "cookie"):
+                resp = self._auth(client, token, source)
+                self.assertEqual(resp.status_code, 202)
+                self.assertEqual(resp.headers["remote-role"], "viewer")
+
+    def test_deleted_user_is_rejected(self):
+        app = self._create_app()
+        token = self._token(app, "bob", "garage")
+
+        with AuthTestClient(app) as client:
+            self.assertEqual(self._auth(client, token).status_code, 202)
+            self.assertEqual(client.delete("/users/bob").status_code, 200)
+
+            for source in ("bearer", "cookie"):
+                self.assertEqual(self._auth(client, token, source).status_code, 401)
+
+    def test_recreated_user_rejects_token_from_deleted_account(self):
+        app = self._create_app()
+        token = self._token(app, "bob", "garage", iat=int(time.time()) - 10)
+
+        with AuthTestClient(app) as client:
+            self.assertEqual(client.delete("/users/bob").status_code, 200)
+            resp = client.post(
+                "/users",
+                json={"username": "bob", "password": PASSWORD, "role": "viewer"},
+            )
+            self.assertEqual(resp.status_code, 200)
+
+            for source in ("bearer", "cookie"):
+                self.assertEqual(self._auth(client, token, source).status_code, 401)
+
+    def test_created_user_is_accepted(self):
+        app = self._create_app()
+        with AuthTestClient(app) as client:
+            resp = client.post(
+                "/users",
+                json={"username": "carol", "password": PASSWORD, "role": "viewer"},
+            )
+            self.assertEqual(resp.status_code, 200)
+            resp = self._auth(client, self._token(app, "carol", "viewer"))
+
+        self.assertEqual(resp.status_code, 202)
+        self.assertEqual(resp.headers["remote-role"], "viewer")
+
+    def test_password_change_revokes_older_tokens(self):
+        app = self._create_app()
+        old_token = self._token(app, "bob", "garage", iat=int(time.time()) - 10)
+
+        with AuthTestClient(app) as client:
+            self.assertEqual(self._auth(client, old_token).status_code, 202)
+            resp = client.put(
+                "/users/bob/password",
+                json={"password": "a-brand-new-password"},
+            )
+            self.assertEqual(resp.status_code, 200)
+
+            for source in ("bearer", "cookie"):
+                self.assertEqual(self._auth(client, old_token, source).status_code, 401)
+
+            changed_at = app.user_sessions["bob"][1]
+            resp = self._auth(client, self._token(app, "bob", "garage", changed_at))
+            self.assertEqual(resp.status_code, 202)
+
+    def test_login_in_the_same_second_as_password_change_is_valid(self):
+        app = self._create_app()
+        changed_at = datetime.now().replace(microsecond=500000)
+        new_token = self._token(app, "bob", "garage", iat=int(changed_at.timestamp()))
+
+        with (
+            patch("frigate.api.auth.datetime") as mock_datetime,
+            AuthTestClient(app) as client,
+        ):
+            mock_datetime.now.return_value = changed_at
+            resp = client.put(
+                "/users/bob/password",
+                json={"password": "a-brand-new-password"},
+            )
+            self.assertEqual(resp.status_code, 200)
+
+            for source in ("bearer", "cookie"):
+                self.assertEqual(self._auth(client, new_token, source).status_code, 202)
+
+    def test_own_password_change_cookie_stays_valid(self):
+        app = self._create_app()
+        with AuthTestClient(app) as client:
+            resp = client.put(
+                "/users/bob/password",
+                json={"password": "a-brand-new-password", "old_password": PASSWORD},
+                headers={"remote-user": "bob", "remote-role": "garage"},
+            )
+            self.assertEqual(resp.status_code, 200)
+            new_token = resp.cookies["frigate_token"]
+
+            resp = self._auth(client, new_token, "cookie")
+
+        self.assertEqual(resp.status_code, 202)
+        self.assertEqual(resp.headers["remote-role"], "garage")
+
+    def test_admin_password_reset_revokes_older_tokens(self):
+        config = FrigateConfig(
+            **{**self.minimal_config, "auth": {"reset_admin_password": True}}
+        )
+        FrigateApp.init_auth(SimpleNamespace(config=config))
+
+        app = self._create_app()
+        old_token = self._token(app, "admin", "admin", iat=int(time.time()) - 10)
+
+        with AuthTestClient(app) as client:
+            for source in ("bearer", "cookie"):
+                self.assertEqual(self._auth(client, old_token, source).status_code, 401)
+
+    def test_auth_does_not_query_users(self):
+        app = self._create_app()
+
+        # expires inside the refresh window so the cookie refresh runs too
+        token = jwt.encode(
+            {"alg": "HS256"},
+            {"sub": "bob", "role": "garage", "exp": int(time.time()) + 60, "iat": 0},
+            app.jwt_token,
+        )
+
+        with (
+            patch.object(User, "select", side_effect=AssertionError("db query")),
+            AuthTestClient(app) as client,
+        ):
+            client.cookies.set("frigate_token", token)
+            resp = client.get("/auth", headers={"x-server-port": "8971"})
+
+        self.assertEqual(resp.status_code, 202)
+        self.assertIn("frigate_token=", resp.headers["set-cookie"])
+
+    def _login_cookie(self, headers: dict | None = None) -> str:
+        with AuthTestClient(self._create_app()) as client:
+            resp = client.post(
+                "/login",
+                json={"user": "bob", "password": PASSWORD},
+                headers=headers or {},
+            )
+
+        self.assertEqual(resp.status_code, 200)
+        return resp.headers["set-cookie"].lower()
+
+    def test_login_cookie_is_secure_over_tls(self):
+        cookie = self._login_cookie({"x-forwarded-proto": "https"})
+        self.assertIn("; secure", cookie)
+
+    def test_login_cookie_is_not_secure_over_http(self):
+        cookie = self._login_cookie({"x-forwarded-proto": "http"})
+        self.assertNotIn("; secure", cookie)
+
+    def test_cookie_secure_config_forces_secure(self):
+        self.minimal_config["auth"]["cookie_secure"] = True
+        cookie = self._login_cookie()
+        self.assertIn("; secure", cookie)
+
+    def test_refreshed_cookie_is_secure_over_tls(self):
+        app = self._create_app()
+        token = jwt.encode(
+            {"alg": "HS256"},
+            {"sub": "bob", "exp": int(time.time()) + 60, "iat": int(time.time())},
+            app.jwt_token,
+        )
+
+        with AuthTestClient(app) as client:
+            client.cookies.set("frigate_token", token)
+            resp = client.get(
+                "/auth",
+                headers={"x-server-port": "8971", "x-forwarded-proto": "https"},
+            )
+
+        self.assertEqual(resp.status_code, 202)
+        self.assertIn("; secure", resp.headers["set-cookie"].lower())

frigate/test/http_api/test_http_config_set.py

@@ -14,6 +14,7 @@ from frigate.config.camera.updater import (
     CameraConfigUpdateTopic,
 )
 from frigate.config.holder import ConfigHolder
+from frigate.const import REDACTED_CREDENTIAL_SENTINEL
 from frigate.models import Event, Recordings, ReviewSegment
 from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
 
@@ -503,6 +504,358 @@ class TestConfigSetWildcardPropagation(BaseTestHttp):
         finally:
             os.unlink(config_path)
 
+    camera_path = "rtsp://10.0.0.1:554/video?password=s3cret&user=admin"
+
+    def _add_credentials(self):
+        """Give the config a credentialed camera path and go2rtc secrets."""
+        self.minimal_config["cameras"]["front_door"]["ffmpeg"]["inputs"][0]["path"] = (
+            self.camera_path
+        )
+        self.minimal_config["go2rtc"] = {
+            "rtsp": {"username": "rtspuser", "password": "rtsppass"},
+            "api": {"username": "apiuser", "password": "apipass"},
+            "mqtt": {"username": "mquser", "password": "mqpass"},
+            "ngrok": {"command": "ngrok tcp 8555 --authtoken ngroksecret"},
+            "homekit": {"front_door": {"pin": "19550224"}},
+            "publish": {"front_door": "rtmp://a.rtmp.youtube.com/live2/streamkey"},
+            "webrtc": {
+                "candidates": ["10.0.0.5:8555"],
+                "ice_servers": [
+                    {
+                        "urls": ["turn:turn.example.com:3478"],
+                        "username": "turnuser",
+                        "credential": "turnpass",
+                    }
+                ],
+            },
+        }
+
+    def _read_config_file(self, config_path):
+        with open(config_path) as f:
+            raw = f.read()
+        return raw, ruamel.yaml.YAML().load(raw)
+
+    @patch("frigate.api.app.find_config_file")
+    def test_masked_camera_path_round_trips(self, mock_find_config):
+        self._add_credentials()
+        config_path = self._write_config_file()
+        mock_find_config.return_value = config_path
+
+        try:
+            app, _ = self._create_app_with_publisher()
+
+            with AuthTestClient(app) as client:
+                camera = client.get("/config").json()["cameras"]["front_door"]
+                inputs = camera["ffmpeg"]["inputs"]
+                self.assertNotIn("s3cret", inputs[0]["path"])
+
+                resp = client.put(
+                    "/config/set",
+                    json={
+                        "config_data": {
+                            "cameras": {
+                                "front_door": {
+                                    "ffmpeg": {"inputs": inputs},
+                                    "detect": {"fps": 7},
+                                }
+                            }
+                        },
+                        "requires_restart": 0,
+                    },
+                )
+                self.assertEqual(resp.status_code, 200)
+
+            _, saved = self._read_config_file(config_path)
+            self.assertEqual(
+                saved["cameras"]["front_door"]["ffmpeg"]["inputs"][0]["path"],
+                self.camera_path,
+            )
+        finally:
+            os.unlink(config_path)
+
+    def _save(self, client, config_data):
+        return client.put(
+            "/config/set",
+            json={"config_data": config_data, "requires_restart": 0},
+        )
+
+    def _set_inputs(self, paths):
+        self.minimal_config["cameras"]["front_door"]["ffmpeg"]["inputs"] = [
+            {"path": path, "roles": [role]} for path, role in paths
+        ]
+
+    # config parsing rebuilds FRIGATE_ENV_VARS from its sources
+    @patch.dict("frigate.util.env._CONTAINER_ENV", {"FRIGATE_RTSP_PASSWORD": "s3cret"})
+    @patch.dict(
+        "frigate.util.env.FRIGATE_ENV_VARS", {"FRIGATE_RTSP_PASSWORD": "s3cret"}
+    )
+    @patch("frigate.api.app.find_config_file")
+    def test_masked_path_keeps_env_placeholder(self, mock_find_config):
+        path = "rtsp://admin:{FRIGATE_RTSP_PASSWORD}@10.0.0.1:554/live"
+        self._set_inputs([(path, "detect")])
+        config_path = self._write_config_file()
+        mock_find_config.return_value = config_path
+
+        try:
+            app, _ = self._create_app_with_publisher()
+
+            with AuthTestClient(app) as client:
+                camera = client.get("/config").json()["cameras"]["front_door"]
+                self.assertNotIn("s3cret", str(camera))
+
+                resp = self._save(
+                    client,
+                    {
+                        "cameras": {
+                            "front_door": {
+                                "ffmpeg": {"inputs": camera["ffmpeg"]["inputs"]},
+                                "detect": {"fps": 7},
+                            }
+                        }
+                    },
+                )
+                self.assertEqual(resp.status_code, 200)
+
+            raw, saved = self._read_config_file(config_path)
+            self.assertNotIn("s3cret", raw)
+            self.assertEqual(
+                saved["cameras"]["front_door"]["ffmpeg"]["inputs"][0]["path"], path
+            )
+        finally:
+            os.unlink(config_path)
+
+    @patch("frigate.api.app.find_config_file")
+    def test_over_masked_path_round_trips(self, mock_find_config):
+        path = "rtsp://10.0.0.1:554/live?x=a@b"
+        self._set_inputs([(path, "detect")])
+        config_path = self._write_config_file()
+        mock_find_config.return_value = config_path
+
+        try:
+            app, _ = self._create_app_with_publisher()
+
+            with AuthTestClient(app) as client:
+                inputs = client.get("/config").json()["cameras"]["front_door"][
+                    "ffmpeg"
+                ]["inputs"]
+                self.assertNotEqual(inputs[0]["path"], path)
+
+                resp = self._save(
+                    client,
+                    {"cameras": {"front_door": {"ffmpeg": {"inputs": inputs}}}},
+                )
+                self.assertEqual(resp.status_code, 200)
+
+            _, saved = self._read_config_file(config_path)
+            self.assertEqual(
+                saved["cameras"]["front_door"]["ffmpeg"]["inputs"][0]["path"], path
+            )
+        finally:
+            os.unlink(config_path)
+
+    @patch("frigate.api.app.find_config_file")
+    def test_unrestorable_masked_values_are_rejected(self, mock_find_config):
+        self._add_credentials()
+        config_path = self._write_config_file()
+        mock_find_config.return_value = config_path
+
+        try:
+            app, _ = self._create_app_with_publisher()
+
+            with AuthTestClient(app) as client:
+                for name, request in (
+                    (
+                        "index-keyed inputs",
+                        {
+                            "json": {
+                                "config_data": {
+                                    "cameras": {
+                                        "front_door": {
+                                            "ffmpeg": {
+                                                "inputs": {
+                                                    "0": {"path": "rtsp://*:*@h/x"}
+                                                }
+                                            }
+                                        }
+                                    }
+                                },
+                                "requires_restart": 0,
+                            }
+                        },
+                    ),
+                    (
+                        "go2rtc streams",
+                        {
+                            "json": {
+                                "config_data": {
+                                    "go2rtc": {"streams": {"cam": ["rtsp://*:*@h/x"]}}
+                                },
+                                "requires_restart": 0,
+                            }
+                        },
+                    ),
+                    (
+                        "query string",
+                        {
+                            "params": {
+                                "cameras.front_door.ffmpeg.inputs.0.path": (
+                                    "rtsp://*:*@h/x"
+                                ),
+                                "go2rtc.rtsp.password": REDACTED_CREDENTIAL_SENTINEL,
+                            },
+                            "json": {"requires_restart": 0},
+                        },
+                    ),
+                ):
+                    with self.subTest(name):
+                        resp = client.put("/config/set", **request)
+                        self.assertEqual(resp.status_code, 400)
+                        self.assertIn("credentials", resp.json()["message"])
+
+            raw, _ = self._read_config_file(config_path)
+            self.assertNotIn("*:*", raw)
+            self.assertNotIn(REDACTED_CREDENTIAL_SENTINEL, raw)
+        finally:
+            os.unlink(config_path)
+
+    @patch("frigate.api.app.find_config_file")
+    def test_go2rtc_credentials_redacted_and_round_trip(self, mock_find_config):
+        self._add_credentials()
+        config_path = self._write_config_file()
+        mock_find_config.return_value = config_path
+
+        try:
+            app, _ = self._create_app_with_publisher()
+
+            with AuthTestClient(app) as client:
+                response = client.get("/config").json()
+                body = str(response)
+                for secret in (
+                    "rtsppass",
+                    "apipass",
+                    "mqpass",
+                    "ngroksecret",
+                    "19550224",
+                    "streamkey",
+                ):
+                    self.assertNotIn(secret, body)
+
+                go2rtc = response["go2rtc"]
+                self.assertEqual(go2rtc["webrtc"]["candidates"], ["10.0.0.5:8555"])
+                # browsers need the TURN account to relay WebRTC
+                self.assertEqual(
+                    go2rtc["webrtc"]["ice_servers"][0]["credential"], "turnpass"
+                )
+
+                go2rtc.pop("ngrok")
+                go2rtc["webrtc"]["candidates"] = ["10.0.0.9:8555"]
+                self.assertEqual(
+                    self._save(client, {"go2rtc": go2rtc}).status_code, 200
+                )
+
+            raw, saved = self._read_config_file(config_path)
+            self.assertNotIn(REDACTED_CREDENTIAL_SENTINEL, raw)
+            self.assertEqual(saved["go2rtc"]["rtsp"]["password"], "rtsppass")
+            self.assertEqual(saved["go2rtc"]["mqtt"]["password"], "mqpass")
+            self.assertEqual(
+                saved["go2rtc"]["homekit"]["front_door"]["pin"], "19550224"
+            )
+            self.assertIn("streamkey", saved["go2rtc"]["publish"]["front_door"])
+            self.assertEqual(saved["go2rtc"]["webrtc"]["candidates"], ["10.0.0.9:8555"])
+            self.assertEqual(
+                saved["go2rtc"]["webrtc"]["ice_servers"][0]["credential"], "turnpass"
+            )
+        finally:
+            os.unlink(config_path)
+
+    @patch("frigate.api.app.find_config_file")
+    def test_masked_inputs_restore_by_value(self, mock_find_config):
+        main = "rtsp://u:MAIN@10.0.0.1:554/main"
+        sub = "rtsp://u:SUB@10.0.0.1:554/sub"
+
+        for name, edit, expected in (
+            ("delete first", lambda inputs: [inputs[1]], [sub]),
+            ("reorder", lambda inputs: [inputs[1], inputs[0]], [sub, main]),
+        ):
+            with self.subTest(name):
+                self._set_inputs([(main, "record"), (sub, "detect")])
+                config_path = self._write_config_file()
+                mock_find_config.return_value = config_path
+
+                try:
+                    app, _ = self._create_app_with_publisher()
+
+                    with AuthTestClient(app) as client:
+                        inputs = client.get("/config").json()["cameras"]["front_door"][
+                            "ffmpeg"
+                        ]["inputs"]
+                        resp = self._save(
+                            client,
+                            {
+                                "cameras": {
+                                    "front_door": {"ffmpeg": {"inputs": edit(inputs)}}
+                                }
+                            },
+                        )
+                        self.assertEqual(resp.status_code, 200)
+
+                    _, saved = self._read_config_file(config_path)
+                    saved_inputs = saved["cameras"]["front_door"]["ffmpeg"]["inputs"]
+                    self.assertEqual([i["path"] for i in saved_inputs], expected)
+                finally:
+                    os.unlink(config_path)
+
+    @patch("frigate.api.app.find_config_file")
+    def test_ambiguous_or_unmatched_masked_input(self, mock_find_config):
+        self._set_inputs(
+            [
+                ("rtsp://u:one@10.0.0.1:554/live", "record"),
+                ("rtsp://u:two@10.0.0.1:554/live", "detect"),
+            ]
+        )
+        config_path = self._write_config_file()
+        mock_find_config.return_value = config_path
+
+        try:
+            app, _ = self._create_app_with_publisher()
+
+            with AuthTestClient(app) as client:
+                inputs = client.get("/config").json()["cameras"]["front_door"][
+                    "ffmpeg"
+                ]["inputs"]
+
+                # both inputs mask to the same value and roles no longer tell
+                # them apart
+                remaining = dict(inputs[1], roles=["detect", "record"])
+                resp = self._save(
+                    client,
+                    {"cameras": {"front_door": {"ffmpeg": {"inputs": [remaining]}}}},
+                )
+                self.assertEqual(resp.status_code, 400)
+
+                inputs[0]["path"] = "rtsp://*:*@10.0.0.99:554/live"
+                resp = self._save(
+                    client,
+                    {"cameras": {"front_door": {"ffmpeg": {"inputs": inputs}}}},
+                )
+                self.assertEqual(resp.status_code, 400)
+
+                # roles break the tie when the second input stays as it was
+                resp = self._save(
+                    client,
+                    {"cameras": {"front_door": {"ffmpeg": {"inputs": [inputs[1]]}}}},
+                )
+                self.assertEqual(resp.status_code, 200)
+
+            raw, saved = self._read_config_file(config_path)
+            self.assertNotIn("*:*", raw)
+            self.assertEqual(
+                saved["cameras"]["front_door"]["ffmpeg"]["inputs"][0]["path"],
+                "rtsp://u:two@10.0.0.1:554/live",
+            )
+        finally:
+            os.unlink(config_path)
+
 
 if __name__ == "__main__":
     unittest.main()

frigate/test/http_api/test_http_go2rtc_proxy_auth.py

@@ -33,6 +33,7 @@ class TestGo2rtcProxyAuth(BaseTestHttp):
         self.minimal_config = _CONFIG
         self.app = super().create_app()
         self.app.jwt_token = OctKey.import_key(b"0" * 32)
+        self.app.user_sessions = {"admin": ("admin", None), "viewer": ("viewer", None)}
 
         get_patch = patch(
             "frigate.api.auth.requests.get",

frigate/test/http_api/test_http_motion_search.py

@@ -0,0 +1,121 @@
+"""Tests for motion search job limits."""
+
+from datetime import datetime
+from unittest.mock import patch
+
+from frigate.jobs import motion_search
+from frigate.models import Event, Recordings, ReviewSegment
+from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
+
+URL = "/front_door/search/motion"
+BODY = {
+    "start_time": 1000,
+    "end_time": 2000,
+    "polygon_points": [[0, 0], [1, 0], [1, 1]],
+}
+
+
+class TestHttpMotionSearch(BaseTestHttp):
+    def setUp(self):
+        super().setUp([Event, Recordings, ReviewSegment])
+        self.app = self.create_app()
+
+        motion_search._motion_search_jobs.clear()
+        self.addCleanup(motion_search._motion_search_jobs.clear)
+
+        runner_patch = patch("frigate.jobs.motion_search.MotionSearchRunner")
+        self.runner = runner_patch.start()
+        self.addCleanup(runner_patch.stop)
+
+    def start(self, client: AuthTestClient, user: str, body: dict = BODY):
+        return client.post(URL, json=body, headers={"remote-user": user})
+
+    def test_second_job_for_same_user_rejected(self):
+        with AuthTestClient(self.app) as client:
+            self.assertEqual(self.start(client, "alice").status_code, 200)
+            self.assertEqual(self.start(client, "alice").status_code, 409)
+            self.assertEqual(self.start(client, "bob").status_code, 200)
+
+    def test_user_can_start_again_after_job_finishes(self):
+        with AuthTestClient(self.app) as client:
+            job_id = self.start(client, "alice").json()["job_id"]
+            job = motion_search.get_motion_search_job(job_id)
+            job.end_time = datetime.now().timestamp()
+            self.assertEqual(self.start(client, "alice").status_code, 200)
+            self.assertIsNotNone(motion_search.get_motion_search_job(job_id))
+
+    def test_cancelled_job_frees_user_but_holds_global_slot_until_grace(self):
+        with AuthTestClient(self.app) as client:
+            job_id = self.start(client, "alice").json()["job_id"]
+            with patch("frigate.jobs.motion_search.InterProcessRequestor"):
+                self.assertTrue(motion_search.cancel_motion_search_job(job_id))
+            job = motion_search.get_motion_search_job(job_id)
+            self.assertIsNotNone(job.cancelled_at)
+            self.assertIsNone(job.end_time)
+
+            self.assertEqual(self.start(client, "alice").status_code, 200)
+            self.assertEqual(self.start(client, "bob").status_code, 200)
+            self.assertEqual(self.start(client, "carol").status_code, 429)
+
+            job.cancelled_at -= motion_search.CANCEL_GRACE_SECONDS + 1
+            self.assertEqual(self.start(client, "carol").status_code, 200)
+
+    def test_start_failure_releases_slot(self):
+        self.runner.return_value.start.side_effect = RuntimeError("boom")
+        with AuthTestClient(self.app) as client:
+            with self.assertRaises(RuntimeError):
+                self.start(client, "alice")
+            self.assertEqual(motion_search._motion_search_jobs, {})
+            self.runner.return_value.requestor.stop.assert_called_once()
+
+            self.runner.return_value.start.side_effect = None
+            self.assertEqual(self.start(client, "alice").status_code, 200)
+
+    def test_shared_identity_skips_per_user_limit(self):
+        with AuthTestClient(self.app) as client:
+            self.assertEqual(self.start(client, "anonymous").status_code, 200)
+            self.assertEqual(self.start(client, "anonymous").status_code, 200)
+
+        motion_search._motion_search_jobs.clear()
+        self.minimal_config["auth"] = {"enabled": False}
+        app = self.create_app()
+        with AuthTestClient(app) as client:
+            self.assertEqual(self.start(client, "viewer").status_code, 200)
+            self.assertEqual(self.start(client, "viewer").status_code, 200)
+
+    def test_proxy_user_header_keeps_per_user_limit_without_auth(self):
+        self.minimal_config["auth"] = {"enabled": False}
+        self.minimal_config["proxy"] = {"header_map": {"user": "x-forwarded-user"}}
+        app = self.create_app()
+        with AuthTestClient(app) as client:
+            self.assertEqual(self.start(client, "alice").status_code, 200)
+            self.assertEqual(self.start(client, "alice").status_code, 409)
+
+    def test_global_cap_enforced(self):
+        with AuthTestClient(self.app) as client:
+            for i in range(motion_search.MAX_ACTIVE_JOBS):
+                self.assertEqual(self.start(client, f"user{i}").status_code, 200)
+            self.assertEqual(self.start(client, "late").status_code, 429)
+
+    def test_finished_jobs_evicted_after_ttl(self):
+        with AuthTestClient(self.app) as client:
+            old_id = self.start(client, "alice").json()["job_id"]
+            recent_id = self.start(client, "bob").json()["job_id"]
+            running_id = self.start(client, "carol").json()["job_id"]
+
+        now = 10_000.0
+        motion_search.get_motion_search_job(old_id).end_time = (
+            now - motion_search.FINISHED_JOB_TTL - 1
+        )
+        motion_search.get_motion_search_job(recent_id).end_time = now - 1
+        motion_search._evict_finished_jobs(now)
+
+        self.assertIsNone(motion_search.get_motion_search_job(old_id))
+        self.assertIsNotNone(motion_search.get_motion_search_job(recent_id))
+        self.assertIsNotNone(motion_search.get_motion_search_job(running_id))
+
+    def test_username_and_cancel_time_not_broadcast(self):
+        job = motion_search.MotionSearchJob(username="alice", cancelled_at=1.0)
+        payload = job.to_dict()
+        self.assertNotIn("username", payload)
+        self.assertNotIn("cancelled_at", payload)

frigate/test/test_camera_pw.py

@@ -1,8 +1,13 @@
 """Test camera user and password cleanup."""
 
+import time
 import unittest
 
-from frigate.util.builtin import clean_camera_user_pass, escape_special_characters
+from frigate.util.builtin import (
+    clean_camera_user_pass,
+    escape_special_characters,
+    is_masked_camera_path,
+)
 
 
 class TestUserPassCleanup(unittest.TestCase):
@@ -45,3 +50,90 @@ class TestUserPassMasking(unittest.TestCase):
         escaped = clean_camera_user_pass(self.rtsp_log_message)
         print(f"The escaped is {escaped}")
         assert escaped == "Did you mean file:rtsp://*:*@192.168.1.3:554"
+
+    def test_masks_userinfo_and_credential_query_forms(self):
+        cases = {
+            "rtsp://first.last@mail.com:p%40w@10.0.0.2/live": "rtsp://*:*@10.0.0.2/live",
+            "rtsp://user+tag:pw@10.0.0.2/live": "rtsp://*:*@10.0.0.2/live",
+            "rtsp://:pw@10.0.0.2/live": "rtsp://*:*@10.0.0.2/live",
+            "tapo://cloudpassword@10.0.0.2": "tapo://*:*@10.0.0.2",
+            "rtsp://u:p@ss@10.0.0.2/live": "rtsp://*:*@10.0.0.2/live",
+            "rtsp://user:pa/ss@10.0.0.2/live": "rtsp://*:*@10.0.0.2/live",
+            "rtsp://u:p@ss#x@10.0.0.2/live": "rtsp://*:*@10.0.0.2/live",
+            "rtsp://admin:2024/pw@10.0.0.2/x": "rtsp://*:*@10.0.0.2/x",
+            "rtsp://admin:12#abc@10.0.0.2/x": "rtsp://*:*@10.0.0.2/x",
+            "rtsp://admin:7?x@10.0.0.2/x": "rtsp://*:*@10.0.0.2/x",
+            "rtsps://u:1/2@h/x": "rtsps://*:*@h/x",
+            "rtsp://admin:pre&token=suf@10.0.0.2/x": "rtsp://*:*@",
+            "rtsp://admin:a?pwd=b@10.0.0.2/x": "rtsp://*:*@",
+            "rtsp://10.0.0.2:554/live?password=pre@suf": "rtsp://*:*@",
+            "http://10.0.0.2:88/cgi?cmd=snap&usr=admin&pwd=P@ss": "http://*:*@",
+            "ffmpeg:http://h:80/flv?user=u&password=p@ss#video=copy": (
+                "ffmpeg:http://*:*@"
+            ),
+            "rtsp://u:p@10.0.0.2:554/live?token=t": "rtsp://*:*@10.0.0.2:554/live?token=*",
+            "http://h/x?api_key=a&apikey=b&access_token=c": (
+                "http://h/x?api_key=*&apikey=*&access_token=*"
+            ),
+            "http://h/cgi?user=admin&password=x": "http://h/cgi?user=*&password=*",
+            "http://h/cgi?password=x&user=admin&channel=1": (
+                "http://h/cgi?password=*&user=*&channel=1"
+            ),
+            "http://h/x?usr=a&pwd=b": "http://h/x?usr=*&pwd=*",
+            "http://h/x?username=a&password=b": "http://h/x?username=*&password=*",
+            "rtsp://host/live?user=admin&password=p@ss": (
+                "rtsp://host/live?user=*&password=*"
+            ),
+            "ring:?refresh_token=abc.def": "ring:?refresh_token=*",
+            "nest:?client_id=1&client_secret=s&refresh_token=t&project_id=p": (
+                "nest:?client_id=1&client_secret=*&refresh_token=*&project_id=p"
+            ),
+            "homekit://10.0.0.2:51827?client_id=c&client_private=k&device_id=d": (
+                "homekit://10.0.0.2:51827?client_id=c&client_private=*&device_id=d"
+            ),
+            "exec:ffmpeg -i rtsp://u:p@h/s -c copy -f rtsp {output}": (
+                "exec:ffmpeg -i rtsp://*:*@h/s -c copy -f rtsp {output}"
+            ),
+            "ffmpeg -i rtsp://u:p@h/1 -i http://h/snap?user=a&password=b -f null": (
+                "ffmpeg -i rtsp://*:*@h/1 -i http://h/snap?user=*&password=* -f null"
+            ),
+        }
+
+        for line, expected in cases.items():
+            with self.subTest(line=line):
+                masked = clean_camera_user_pass(line)
+                self.assertEqual(masked, expected)
+                self.assertTrue(is_masked_camera_path(masked))
+                self.assertEqual(clean_camera_user_pass(masked), masked)
+
+    def test_safe_strings_unchanged(self):
+        for line in (
+            "rtsp://10.0.0.2:554/live",
+            "ffmpeg:front_door#video=copy",
+            "rtsp://127.0.0.1:8554/front_door?video=copy&audio=aac",
+            "homekit://10.0.0.2:51827?device_id=d",
+            "http://10.0.0.1/snap.jpg?email=me@x.com",
+            "http://h/x?auth=a&author=b&authorization=c&user_agent=d&autoken=e",
+            "http://h/x?private=1&passthrough=1",
+            "Camera front_door is not responding",
+        ):
+            with self.subTest(line=line):
+                self.assertEqual(clean_camera_user_pass(line), line)
+                self.assertFalse(is_masked_camera_path(line))
+
+    def test_commas_outside_credentials_survive(self):
+        """ffprobe splits multiple paths on commas left after masking."""
+        for line in (
+            "rtsp://a/x,rtsp://u:p@b/y",
+            "rtsp://a:554/x,rtsp://u:p@b/y",
+        ):
+            with self.subTest(line=line):
+                self.assertIn(",", clean_camera_user_pass(line))
+        self.assertNotIn(",", clean_camera_user_pass("rtsp://u:p,q@b/y"))
+
+    def test_long_line_masks_in_linear_time(self):
+        """Log lines run through the mask, so a long one must stay fast."""
+        start = time.perf_counter()
+        clean_camera_user_pass("rtsp://u:p" * 20000)
+        clean_camera_user_pass("rtsp://" + "a." * 40000)
+        self.assertLess(time.perf_counter() - start, 1)

frigate/test/test_ipc_handler_errors.py

@@ -0,0 +1,76 @@
+"""Tests that IPC handler errors cannot stall inter-process communication."""
+
+import os
+import tempfile
+import unittest
+from unittest.mock import MagicMock, patch
+
+import zmq
+
+from frigate.comms.dispatcher import Dispatcher
+from frigate.comms.inter_process import InterProcessCommunicator
+
+
+class TestInterProcessHandlerErrors(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.address = f"ipc://{os.path.join(self.tmp.name, 'comms')}"
+
+        with patch("frigate.comms.inter_process.SOCKET_REP_REQ", self.address):
+            self.communicator = InterProcessCommunicator()
+
+        self.context = zmq.Context()
+        self.requestor = self.context.socket(zmq.REQ)
+        self.requestor.setsockopt(zmq.RCVTIMEO, 5000)
+        self.requestor.connect(self.address)
+
+    def tearDown(self):
+        self.requestor.close(linger=0)
+        self.context.destroy(linger=0)
+        self.communicator.stop()
+        self.tmp.cleanup()
+
+    def test_handler_exception_still_replies_and_keeps_reading(self):
+        def dispatcher(topic, value):
+            if topic == "boom":
+                raise ValueError("handler failed")
+            return value
+
+        self.communicator.subscribe(dispatcher)
+        self.communicator.start()
+
+        with self.assertLogs("frigate.comms.inter_process", level="ERROR"):
+            self.requestor.send_json(("boom", None))
+            self.assertEqual(self.requestor.recv_json(), [])
+
+        self.requestor.send_json(("echo", "ok"))
+        self.assertEqual(self.requestor.recv_json(), "ok")
+        self.assertTrue(self.communicator.reader_thread.is_alive())
+
+
+class TestDispatcherPublishErrors(unittest.TestCase):
+    def test_failing_communicator_does_not_block_the_others(self):
+        failing = MagicMock()
+        failing.publish.side_effect = ValueError("publish failed")
+        healthy = MagicMock()
+
+        with (
+            patch("frigate.comms.dispatcher.CameraActivityManager"),
+            patch("frigate.comms.dispatcher.AudioActivityManager"),
+        ):
+            dispatcher = Dispatcher(
+                MagicMock(), MagicMock(), MagicMock(), {}, [failing, healthy]
+            )
+
+        healthy.publish.reset_mock()
+
+        with self.assertLogs("frigate.comms.dispatcher", level="ERROR"):
+            dispatcher.publish("reviews", "{}")
+            dispatcher.publish_local("notices", "[]")
+
+        healthy.publish.assert_any_call("reviews", "{}", False)
+        healthy.publish.assert_any_call("notices", "[]", False)
+
+
+if __name__ == "__main__":
+    unittest.main()

frigate/test/test_motion_search_decode.py

@@ -1,5 +1,6 @@
 """Tests for the motion search hardware-accelerated decode helpers."""
 
+import subprocess as sp
 import unittest
 from types import SimpleNamespace
 from unittest import mock
@@ -140,6 +141,11 @@ class TestVodDecodeCommand(unittest.TestCase):
 
 
 class TestProbeVodKeyframePts(unittest.TestCase):
+    def _proc(self, stdout="", returncode=0):
+        proc = mock.Mock(returncode=returncode)
+        proc.communicate.return_value = (stdout, None)
+        return proc
+
     def test_parses_keyframe_packets(self):
         sample = (
             '{"packets":['
@@ -147,19 +153,37 @@ class TestProbeVodKeyframePts(unittest.TestCase):
             '{"pts_time":"1.000000","flags":"___"},'
             '{"pts_time":"2.000000","flags":"K__"}]}'
         )
-        completed = mock.Mock(stdout=sample, returncode=0)
         with mock.patch(
-            "frigate.jobs.motion_search_decode.sp.run", return_value=completed
+            "frigate.jobs.motion_search_decode.sp.Popen",
+            return_value=self._proc(sample),
         ):
-            pts = probe_vod_keyframe_pts("ffprobe", "http://x/index.m3u8")
+            pts = probe_vod_keyframe_pts(
+                "ffprobe", "http://x/index.m3u8", lambda: False
+            )
         self.assertEqual(pts, [0.0, 2.0])
 
     def test_returns_empty_on_failure(self):
         with mock.patch(
-            "frigate.jobs.motion_search_decode.sp.run",
+            "frigate.jobs.motion_search_decode.sp.Popen",
             side_effect=OSError("boom"),
         ):
-            self.assertEqual(probe_vod_keyframe_pts("ffprobe", "http://x"), [])
+            self.assertEqual(
+                probe_vod_keyframe_pts("ffprobe", "http://x", lambda: False), []
+            )
+
+    def test_stop_request_kills_probe(self):
+        proc = self._proc()
+        proc.communicate.side_effect = [
+            sp.TimeoutExpired("ffprobe", 0.5),
+            ("", None),
+        ]
+        with mock.patch(
+            "frigate.jobs.motion_search_decode.sp.Popen", return_value=proc
+        ):
+            self.assertEqual(
+                probe_vod_keyframe_pts("ffprobe", "http://x", lambda: True), []
+            )
+        proc.kill.assert_called_once()
 
 
 class TestProbeVideoDimensions(unittest.TestCase):

frigate/test/test_network_config.py

@@ -36,6 +36,73 @@ 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",
+            " 5000 ": 5000,
+            "5000\n": 5000,
+            " 127.0.0.1:5000 ": "127.0.0.1:5000",
+            "localhost:5000": "localhost:5000",
+            "frigate-host.lan:5000": "frigate-host.lan:5000",
+            "*:5000": "*: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;",
+            "::1:5000",
+            "[127.0.0.1]:5000",
+            "[fe80::1%eth0]:5000",
+            "local host:5000",
+            "localhost;:5000",
+            "local{host}:5000",
+            "host#:5000",
+            "'localhost':5000",
+            '"localhost":5000',
+            "-host:5000",
+            "host-:5000",
+            "host..lan:5000",
+            "**:5000",
+            "*.lan:5000",
+            "unix:/tmp/sock:5000",
+            "5000 5001",
+            ":5000",
+            "127.0.0.1:",
+            "127.0.0.1:+5000",
+            0,
+            65536,
+            -1,
+        ]
+
+        for value in cases:
+            with self.subTest(value=value):
+                with self.assertRaises(ValidationError):
+                    ListenConfig(internal=value)
+
+    def test_bool_listen_value_rejected(self):
+        # get_nginx_settings.py sees the raw bool and falls back to the default,
+        # so accepting it as port 1 would split nginx and the auth port check
+        for value in (True, False):
+            with self.subTest(value=value):
+                with self.assertRaises(ValidationError):
+                    ListenConfig(internal=value)
+
 
 if __name__ == "__main__":
     unittest.main(verbosity=2)

frigate/test/test_webpush_registration.py

@@ -3,8 +3,13 @@
 import unittest
 from base64 import urlsafe_b64encode
 
-from frigate.api.notification import _validate_push_endpoint, _validate_subscription
-from frigate.comms.webpush import _build_web_pushers
+from frigate.api.notification import _validate_subscription
+from frigate.comms.webpush import (
+    _build_web_pushers,
+    _NoRedirectSession,
+    _vapid_audience,
+    validate_push_endpoint,
+)
 
 VALID_ENDPOINTS = [
     "https://fcm.googleapis.com/fcm/send/dGhpcy1pcy1hLXRva2Vu",
@@ -26,11 +31,11 @@ class TestValidatePushEndpoint(unittest.TestCase):
     def test_accepts_real_push_service_endpoints(self):
         for endpoint in VALID_ENDPOINTS:
             with self.subTest(endpoint=endpoint):
-                self.assertIsNone(_validate_push_endpoint(endpoint))
+                self.assertIsNone(validate_push_endpoint(endpoint))
 
     def test_rejects_http(self):
         self.assertIsNotNone(
-            _validate_push_endpoint("http://fcm.googleapis.com/fcm/send/token")
+            validate_push_endpoint("http://fcm.googleapis.com/fcm/send/token")
         )
 
     def test_rejects_non_http_schemes(self):
@@ -40,7 +45,7 @@ class TestValidatePushEndpoint(unittest.TestCase):
             "//example.com/token",
         ):
             with self.subTest(endpoint=endpoint):
-                self.assertIsNotNone(_validate_push_endpoint(endpoint))
+                self.assertIsNotNone(validate_push_endpoint(endpoint))
 
     def test_rejects_localhost(self):
         for endpoint in (
@@ -50,7 +55,7 @@ class TestValidatePushEndpoint(unittest.TestCase):
             "https://[::1]/token",
         ):
             with self.subTest(endpoint=endpoint):
-                self.assertIsNotNone(_validate_push_endpoint(endpoint))
+                self.assertIsNotNone(validate_push_endpoint(endpoint))
 
     def test_rejects_private_addresses(self):
         for endpoint in (
@@ -61,7 +66,7 @@ class TestValidatePushEndpoint(unittest.TestCase):
             "https://0.0.0.0/token",
         ):
             with self.subTest(endpoint=endpoint):
-                self.assertIsNotNone(_validate_push_endpoint(endpoint))
+                self.assertIsNotNone(validate_push_endpoint(endpoint))
 
     def test_rejects_internal_hostnames(self):
         for endpoint in (
@@ -71,16 +76,16 @@ class TestValidatePushEndpoint(unittest.TestCase):
             "https://host.home.arpa/token",
         ):
             with self.subTest(endpoint=endpoint):
-                self.assertIsNotNone(_validate_push_endpoint(endpoint))
+                self.assertIsNotNone(validate_push_endpoint(endpoint))
 
     def test_rejects_non_default_port(self):
         self.assertIsNotNone(
-            _validate_push_endpoint("https://fcm.googleapis.com:8080/fcm/send/token")
+            validate_push_endpoint("https://fcm.googleapis.com:8080/fcm/send/token")
         )
 
     def test_rejects_embedded_credentials(self):
         self.assertIsNotNone(
-            _validate_push_endpoint(
+            validate_push_endpoint(
                 "https://user:pass@fcm.googleapis.com/fcm/send/token"
             )
         )
@@ -88,26 +93,26 @@ class TestValidatePushEndpoint(unittest.TestCase):
     def test_rejects_endpoint_without_path(self):
         for endpoint in ("https://fcm.googleapis.com", "https://fcm.googleapis.com/"):
             with self.subTest(endpoint=endpoint):
-                self.assertIsNotNone(_validate_push_endpoint(endpoint))
+                self.assertIsNotNone(validate_push_endpoint(endpoint))
 
-    def test_rejects_endpoint_that_breaks_audience_parsing(self):
-        # webpush.py locates the host by searching for a separator after index
-        # 10, which raises ValueError when the url has no path at all
-        endpoint = "https://fcm.googleapis.com"
-
-        with self.assertRaises(ValueError):
-            endpoint.index("/", 10)
-
-        self.assertIsNotNone(_validate_push_endpoint(endpoint))
+    def test_rejects_empty_hostname_labels(self):
+        for endpoint in (
+            "https://./x",
+            "https://fcm..googleapis.com/token",
+            "https://.fcm.googleapis.com/token",
+            "https://fcm.googleapis.com./token",
+        ):
+            with self.subTest(endpoint=endpoint):
+                self.assertIsNotNone(validate_push_endpoint(endpoint))
 
     def test_rejects_missing_or_non_string_endpoint(self):
         for endpoint in (None, "", 5, {"url": "https://example.com/token"}):
             with self.subTest(endpoint=endpoint):
-                self.assertIsNotNone(_validate_push_endpoint(endpoint))
+                self.assertIsNotNone(validate_push_endpoint(endpoint))
 
     def test_rejects_overlong_endpoint(self):
         self.assertIsNotNone(
-            _validate_push_endpoint(f"https://fcm.googleapis.com/{'a' * 4096}")
+            validate_push_endpoint(f"https://fcm.googleapis.com/{'a' * 4096}")
         )
 
 
@@ -148,10 +153,6 @@ class TestValidateSubscription(unittest.TestCase):
                 self.assertIsNotNone(_validate_subscription(sub))
 
 
-if __name__ == "__main__":
-    unittest.main()
-
-
 class TestBuildWebPushers(unittest.TestCase):
     def test_skips_subscriptions_with_malformed_keys(self):
         bad = _subscription(VALID_ENDPOINTS[0])
@@ -163,3 +164,52 @@ class TestBuildWebPushers(unittest.TestCase):
 
         self.assertEqual(len(pushers), 1)
         self.assertEqual(pushers[0].receiver_key, b"\x04" + bytes(64))
+
+    def _valid_keys_subscription(self, endpoint: str) -> dict:
+        sub = _subscription(endpoint)
+        sub["keys"]["p256dh"] = urlsafe_b64encode(b"\x04" + bytes(64)).decode()
+        return sub
+
+    def test_skips_stored_rows_that_break_the_audience(self):
+        good = self._valid_keys_subscription(VALID_ENDPOINTS[0])
+        rows = [
+            self._valid_keys_subscription("https://./secret-token"),
+            self._valid_keys_subscription("https://[::1/secret-token"),
+            {"keys": good["keys"]},
+            "not-a-dict",
+            good,
+        ]
+
+        with self.assertLogs("frigate.comms.webpush", level="WARNING") as logs:
+            pushers = _build_web_pushers("viewer", rows)
+
+        self.assertEqual(len(logs.output), 4)
+        self.assertIn("'.'", logs.output[0])
+        self.assertNotIn("secret-token", "".join(logs.output))
+        self.assertEqual(len(pushers), 1)
+        self.assertEqual(pushers[0].subscription_info["endpoint"], VALID_ENDPOINTS[0])
+        self.assertIsInstance(pushers[0].mod_or_session, _NoRedirectSession)
+
+    def test_keeps_stored_self_hosted_endpoints(self):
+        rows = [
+            self._valid_keys_subscription("https://ntfy.example.com:8443/up123"),
+            self._valid_keys_subscription("https://ntfy.example.com/"),
+        ]
+
+        pushers = _build_web_pushers("viewer", rows)
+
+        self.assertEqual(len(pushers), 2)
+
+
+class TestVapidAudience(unittest.TestCase):
+    def test_uses_scheme_and_host(self):
+        self.assertEqual(
+            _vapid_audience(VALID_ENDPOINTS[0]), "https://fcm.googleapis.com"
+        )
+        self.assertEqual(
+            _vapid_audience(VALID_ENDPOINTS[4]), "https://fcm.googleapis.com:443"
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

frigate/test/test_webpush_send_guard.py

@@ -0,0 +1,45 @@
+"""Tests that push sends never follow redirects."""
+
+import unittest
+
+import requests
+from requests.adapters import BaseAdapter
+
+from frigate.comms.webpush import _NoRedirectSession
+
+ENDPOINT = "https://push.example.com/token"
+
+
+class RedirectAdapter(BaseAdapter):
+    def __init__(self):
+        super().__init__()
+        self.calls = 0
+
+    def send(self, request, **kwargs):
+        self.calls += 1
+        response = requests.Response()
+        response.status_code = 307
+        response.headers["Location"] = "http://127.0.0.1:5000/api/restart"
+        response.url = request.url
+        response.request = request
+        return response
+
+    def close(self):
+        pass
+
+
+class TestNoRedirectSession(unittest.TestCase):
+    def test_does_not_follow_redirects(self):
+        adapter = RedirectAdapter()
+        session = _NoRedirectSession()
+        session.mount("https://", adapter)
+        session.mount("http://", adapter)
+
+        response = session.post(ENDPOINT, data=b"payload")
+
+        self.assertEqual(response.status_code, 307)
+        self.assertEqual(adapter.calls, 1)
+
+
+if __name__ == "__main__":
+    unittest.main()

frigate/util/builtin.py

@@ -21,8 +21,11 @@ import numpy as np
 from ruamel.yaml import YAML
 
 from frigate.const import (
-    REGEX_HTTP_CAMERA_USER_PASS,
+    REGEX_CREDENTIAL_QUERY_KEY,
+    REGEX_CREDENTIAL_QUERY_PARAM,
+    REGEX_MASKED_CREDENTIAL,
     REGEX_RTSP_CAMERA_USER_PASS,
+    REGEX_URL_USERINFO,
     STREAM_TYPE_MAIN,
     STREAM_TYPE_SUB,
 )
@@ -117,9 +120,24 @@ def deep_merge(dct1: dict, dct2: dict, override=False, merge_lists=False) -> dic
 
 
 def clean_camera_user_pass(line: str) -> str:
-    """Removes user and password from line."""
-    rtsp_cleaned = re.sub(REGEX_RTSP_CAMERA_USER_PASS, "://*:*@", line)
-    return re.sub(REGEX_HTTP_CAMERA_USER_PASS, "user=*&password=*", rtsp_cleaned)
+    """Mask URL userinfo and credential query parameters in line."""
+
+    line = re.sub(REGEX_URL_USERINFO, _mask_userinfo, line)
+    return re.sub(REGEX_CREDENTIAL_QUERY_PARAM, r"\1*", line)
+
+
+def _mask_userinfo(match: re.Match[str]) -> str:
+    # A credential key in the userinfo means a query password holds "@" or a
+    # password holds a query key. Neither split is safe, so mask the whole URL.
+    if re.search(REGEX_CREDENTIAL_QUERY_KEY, match.group(1)):
+        return "://*:*@"
+
+    return f"://*:*@{match.group(2)}"
+
+
+def is_masked_camera_path(path: str) -> bool:
+    """Return True if path holds credentials masked by clean_camera_user_pass."""
+    return re.search(REGEX_MASKED_CREDENTIAL, path) is not None
 
 
 def escape_special_characters(path: str) -> str:

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,49 @@ def redact_credential(obj: dict[str, Any], key: str) -> None:
         obj.pop(key, None)
 
 
+LISTEN_HOSTNAME = re.compile(
+    r"[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?"
+    r"(?:\.[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?)*"
+)
+
+
+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, "host:port", "*:port", or "[IPv6]:port", where host is an
+    IPv4 address or an RFC 1123 hostname. 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).strip().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
+
+    if host != "*" and not LISTEN_HOSTNAME.fullmatch(host):
+        raise ValueError(f"invalid listen address: {value!r}")
+
+    return f"{host}:{port}", port
+
+
 def find_config_file() -> str:
     """Return the path of the config file to use.
 

frigate/video/detect.py

@@ -338,7 +338,12 @@ def process_frames(
             continue
 
         if datetime.now().astimezone(UTC) > next_region_update:
-            region_grid = requestor.send_data(REQUEST_REGION_GRID, camera_config.name)
+            updated_grid = requestor.send_data(REQUEST_REGION_GRID, camera_config.name)
+
+            # A failed request replies empty, so keep the current grid
+            if updated_grid:
+                region_grid = updated_grid
+
             next_region_update = get_tomorrow_at_time(2)
 
         try:

web/e2e/specs/chat.spec.ts

@@ -218,6 +218,50 @@ test.describe("Chat — streaming @medium", () => {
       frigateApp.page.getByText(/searching for people/i),
     ).toBeVisible({ timeout: 5_000 });
   });
+
+  test("assistant markdown renders only same-origin images", async ({
+    frigateApp,
+  }) => {
+    const reply =
+      "![leaked alt](https://evil.example/x.png?d=secret) " +
+      "![local](/api/events/abc/thumbnail.webp) " +
+      "[offsite](https://evil.example/page)";
+    await installChatStreamOverride(frigateApp, [
+      { type: "content", delta: reply },
+      {
+        type: "messages",
+        messages: [
+          { role: "system", content: "sys" },
+          { role: "user", content: "show me" },
+          { role: "assistant", content: reply },
+        ],
+      },
+      { type: "done" },
+    ]);
+    const offsiteRequests: string[] = [];
+    frigateApp.page.on("request", (req) => {
+      if (req.url().includes("evil.example")) offsiteRequests.push(req.url());
+    });
+    await frigateApp.goto("/chat");
+    const input = frigateApp.page.getByPlaceholder(/ask/i);
+    await expect(input).toBeVisible({ timeout: 10_000 });
+    await input.fill("show me");
+    await input.press("Enter");
+
+    await expect(frigateApp.page.getByText("leaked alt")).toBeVisible({
+      timeout: 10_000,
+    });
+    await expect(
+      frigateApp.page.locator('img[src*="evil.example"]'),
+    ).toHaveCount(0);
+    await expect(
+      frigateApp.page.locator('img[src="/api/events/abc/thumbnail.webp"]'),
+    ).toHaveCount(1);
+    const link = frigateApp.page.getByRole("link", { name: "offsite" });
+    await expect(link).toHaveAttribute("target", "_blank");
+    await expect(link).toHaveAttribute("rel", "noopener noreferrer");
+    expect(offsiteRequests).toEqual([]);
+  });
 });
 
 test.describe("Chat — stop @medium", () => {

web/e2e/specs/settings/camera-ffmpeg-streams.spec.ts

@@ -32,7 +32,11 @@ type CameraInput = {
   input_args?: string;
 };
 
-async function installRoutes(page: Page, frontDoorInputs: CameraInput[]) {
+async function installRoutes(
+  page: Page,
+  frontDoorInputs: CameraInput[],
+  rawInputs: CameraInput[] = frontDoorInputs,
+) {
   const config = configFactory({
     go2rtc: { streams: GO2RTC_STREAMS },
     cameras: {
@@ -56,7 +60,7 @@ async function installRoutes(page: Page, frontDoorInputs: CameraInput[]) {
   await page.route("**/api/config/raw_paths", (route) =>
     route.fulfill({
       json: {
-        cameras: { front_door: { ffmpeg: { inputs: frontDoorInputs } } },
+        cameras: { front_door: { ffmpeg: { inputs: rawInputs } } },
         go2rtc: { streams: GO2RTC_STREAMS },
       },
     }),
@@ -200,4 +204,64 @@ test.describe("camera ffmpeg input source toggle @medium", () => {
       payload?.config_data?.cameras?.front_door?.ffmpeg?.inputs?.[0];
     expect(input?.input_args).not.toBe("preset-rtsp-restream");
   });
+
+  test("a masked path reveals its own saved path after an earlier input is deleted", async ({
+    frigateApp,
+  }) => {
+    const main = "rtsp://admin:mainpw@10.0.0.1:554/main";
+    const sub = "rtsp://admin:subpw@10.0.0.1:554/sub";
+    await installRoutes(
+      frigateApp.page,
+      [
+        { path: "rtsp://*:*@10.0.0.1:554/main", roles: ["record"] },
+        { path: "rtsp://*:*@10.0.0.1:554/sub", roles: ["detect"] },
+      ],
+      [
+        { path: main, roles: ["record"] },
+        { path: sub, roles: ["detect"] },
+      ],
+    );
+    await frigateApp.goto("/settings?page=cameraFfmpeg&camera=front_door");
+
+    const paths = frigateApp.page.getByRole("textbox", { name: "Input path" });
+    await expect(paths).toHaveCount(2);
+
+    await frigateApp.page
+      .getByRole("button", { name: "Delete" })
+      .first()
+      .click();
+    await expect(paths).toHaveCount(1);
+
+    await paths.focus();
+    await expect(paths).toHaveValue(sub);
+  });
+
+  test("inputs with the same masked path each reveal their own path by role", async ({
+    frigateApp,
+  }) => {
+    const record = "rtsp://admin:one@10.0.0.1:554/live";
+    const detect = "rtsp://viewer:two@10.0.0.1:554/live";
+    const masked = "rtsp://*:*@10.0.0.1:554/live";
+    await installRoutes(
+      frigateApp.page,
+      [
+        { path: masked, roles: ["record"] },
+        { path: masked, roles: ["detect"] },
+      ],
+      [
+        { path: record, roles: ["record"] },
+        { path: detect, roles: ["detect"] },
+      ],
+    );
+    await frigateApp.goto("/settings?page=cameraFfmpeg&camera=front_door");
+
+    const paths = frigateApp.page.getByRole("textbox", { name: "Input path" });
+    await expect(paths).toHaveCount(2);
+
+    await paths.nth(0).focus();
+    await expect(paths.nth(0)).toHaveValue(record);
+
+    await paths.nth(1).focus();
+    await expect(paths.nth(1)).toHaveValue(detect);
+  });
 });

web/public/locales/en/config/global.json

@@ -40,7 +40,7 @@
     },
     "cookie_secure": {
       "label": "Secure cookie flag",
-      "description": "Set the secure flag on the auth cookie; should be true when using TLS."
+      "description": "Always set the secure flag on the auth cookie. Frigate sets it automatically when serving TLS; enable this when a reverse proxy terminates TLS in front of Frigate. A proxy that connects to the TLS port must serve Frigate to browsers over HTTPS."
     },
     "session_length": {
       "label": "Session length",

web/public/locales/en/views/motionSearch.json

@@ -56,6 +56,8 @@
     "noROI": "Please draw a region of interest",
     "noTimeRange": "Please select a time range",
     "invalidTimeRange": "End time must be after start time",
+    "alreadyRunning": "You already have a motion search running. Wait for it to finish or cancel it.",
+    "tooManySearches": "Too many motion searches are running. Try again when one finishes.",
     "searchFailed": "Search failed: {{message}}",
     "polygonTooSmall": "Polygon must have at least 3 points",
     "unknown": "Unknown error"

web/src/components/chat/ChatMessage.tsx

@@ -18,6 +18,18 @@ import { cn } from "@/lib/utils";
 import { ChatAttachmentChip } from "@/components/chat/ChatAttachmentChip";
 import { parseAttachedEvent } from "@/utils/chatUtil";
 import type { ChatStats, ShowStatsMode } from "@/types/chat";
+import { baseUrl } from "@/api/baseUrl";
+
+// Assistant output can carry prompt-injected URLs, and the browser fetches
+// images without a click, so only same-origin images render.
+function isSameOrigin(url: string | undefined): boolean {
+  if (!url) return false;
+  try {
+    return new URL(url, baseUrl).origin === window.location.origin;
+  } catch {
+    return false;
+  }
+}
 
 type MessageBubbleProps = {
   role: "user" | "assistant";
@@ -223,6 +235,23 @@ export function MessageBubble({
                     {...props}
                   />
                 ),
+                img: ({ node: _n, src, alt, ...props }) =>
+                  typeof src === "string" && isSameOrigin(src) ? (
+                    <img src={src} alt={alt} {...props} />
+                  ) : (
+                    <>{alt}</>
+                  ),
+                a: ({ node: _n, href, ...props }) =>
+                  isSameOrigin(href) ? (
+                    <a href={href} {...props} />
+                  ) : (
+                    <a
+                      href={href}
+                      target="_blank"
+                      rel="noopener noreferrer"
+                      {...props}
+                    />
+                  ),
               }}
             >
               {content}

web/src/components/config-form/theme/widgets/CameraPathWidget.tsx

@@ -1,5 +1,7 @@
 import type { WidgetProps } from "@rjsf/utils";
 import useSWR from "swr";
+import get from "lodash/get";
+import isEqual from "lodash/isEqual";
 import { useMemo, useState, type FocusEvent } from "react";
 import { useTranslation } from "react-i18next";
 import { LuEye, LuEyeOff } from "react-icons/lu";
@@ -21,6 +23,7 @@ type RawPathsResponse = {
       ffmpeg?: {
         inputs?: Array<{
           path?: string;
+          roles?: string[];
         }>;
       };
     }
@@ -66,17 +69,35 @@ export function CameraPathWidget(props: WidgetProps) {
     shouldFetchRawPaths ? "config/raw_paths" : null,
   );
 
+  const rawValue = typeof value === "string" ? value : "";
+
+  const roles =
+    inputIndex === undefined
+      ? undefined
+      : get(formContext?.formData, ["inputs", inputIndex, "roles"]);
+
+  // Match by masked value with roles breaking a tie, as the backend restores
+  // on save, so deleting or reordering inputs never reveals another input's path
   const rawPath = useMemo(() => {
-    if (!cameraName || inputIndex === undefined) {
+    if (!cameraName || !isMaskedPath(rawValue)) {
       return undefined;
     }
 
-    const path =
-      rawPaths?.cameras?.[cameraName]?.ffmpeg?.inputs?.[inputIndex]?.path;
-    return typeof path === "string" ? path : undefined;
-  }, [cameraName, inputIndex, rawPaths]);
+    let matches = (
+      rawPaths?.cameras?.[cameraName]?.ffmpeg?.inputs ?? []
+    ).filter(
+      (input): input is { path: string; roles?: string[] } =>
+        typeof input.path === "string" &&
+        maskCredentials(input.path) === rawValue,
+    );
+    if (new Set(matches.map((input) => input.path)).size > 1) {
+      matches = matches.filter((input) => isEqual(input.roles, roles));
+    }
+
+    const paths = new Set(matches.map((input) => input.path));
+    return paths.size === 1 ? [...paths][0] : undefined;
+  }, [cameraName, rawPaths, rawValue, roles]);
 
-  const rawValue = typeof value === "string" ? value : "";
   const resolvedValue =
     isMaskedPath(rawValue) && rawPath ? rawPath : (rawValue ?? "");
   const canReveal =

web/src/utils/credentialMask.ts

@@ -1,40 +1,37 @@
-const MASKED_AUTH_PATTERN = /:\/\/\*:\*@/i;
-const MASKED_QUERY_PATTERN = /(?:[?&])user=\*&password=\*/i;
+// Mirrors clean_camera_user_pass and REGEX_MASKED_CREDENTIAL in the backend.
+const CREDENTIAL_QUERY_KEYS =
+  "(?:user(?:name)?|usr|login|pass(?:word|wd)?|pwd|(?:[a-z_]*_)?token|(?:[a-z_]*_)?secret|client_private|api_?key)";
+const URL_USERINFO_PATTERN =
+  /:\/\/((?:[^\s/?#:[\]]*:(?:(?!:\/\/)\S)*|[^\s/?#]+))@((?:(?!:\/\/)\S)*)/g;
+const CREDENTIAL_QUERY_KEY_PATTERN = new RegExp(
+  `[?&]${CREDENTIAL_QUERY_KEYS}=`,
+  "i",
+);
+const CREDENTIAL_QUERY_PATTERN = new RegExp(
+  `([?&]${CREDENTIAL_QUERY_KEYS}=)[^&\\s]*`,
+  "gi",
+);
+const MASKED_PATTERN = new RegExp(
+  `://\\*:\\*@|[?&]${CREDENTIAL_QUERY_KEYS}=\\*(?![^&\\s])`,
+  "i",
+);
 
 export const isMaskedPath = (value: string): boolean =>
-  MASKED_AUTH_PATTERN.test(value) || MASKED_QUERY_PATTERN.test(value);
-
-export const hasCredentials = (value: string): boolean => {
-  if (!value) {
-    return false;
-  }
-
-  if (isMaskedPath(value)) {
-    return true;
-  }
-
-  try {
-    const parsed = new URL(value);
-    if (parsed.username || parsed.password) {
-      return true;
-    }
-
-    return (
-      parsed.searchParams.has("user") && parsed.searchParams.has("password")
-    );
-  } catch {
-    return /:\/\/[^:@/\s]+:[^@/\s]+@/.test(value);
-  }
-};
+  MASKED_PATTERN.test(value);
 
 export const maskCredentials = (value: string): string => {
   if (!value) {
     return value;
   }
 
-  const maskedAuth = value.replace(/:\/\/[^:@/\s]+:[^@/\s]*@/g, "://*:*@");
-
-  return maskedAuth
-    .replace(/([?&]user=)[^&]*/gi, "$1*")
-    .replace(/([?&]password=)[^&]*/gi, "$1*");
+  return value
+    .replace(URL_USERINFO_PATTERN, (_match, userinfo: string, rest: string) =>
+      CREDENTIAL_QUERY_KEY_PATTERN.test(userinfo)
+        ? "://*:*@"
+        : `://*:*@${rest}`,
+    )
+    .replace(CREDENTIAL_QUERY_PATTERN, "$1*");
 };
+
+export const hasCredentials = (value: string): boolean =>
+  !!value && (isMaskedPath(value) || maskCredentials(value) !== value);

web/src/views/motion-search/MotionSearchView.tsx

@@ -892,6 +892,20 @@ export default function MotionSearchView({
         setIsSearching(false);
       }
     } catch (error) {
+      const status = axios.isAxiosError(error)
+        ? error.response?.status
+        : undefined;
+
+      if (status === 409 || status === 429) {
+        toast.error(
+          t(
+            status === 409 ? "errors.alreadyRunning" : "errors.tooManySearches",
+          ),
+        );
+        setIsSearching(false);
+        return;
+      }
+
       let errorMessage = t("errors.unknown");
 
       if (axios.isAxiosError<{ message?: string; detail?: string }>(error)) {