mask every credential form in camera and go2rtc urls

The credential mask only matched a userinfo with a [A-Za-z0-9_-] username or the exact user=..&password=.. query order, so dotted or email usernames, password-only userinfo (tapo://pw@host), empty usernames, reversed or renamed query credentials, and token query strings (ring refresh_token, nest client_secret, homekit client_private) reached every authenticated user through /config, /go2rtc/streams, /stats and /metrics. clean_camera_user_pass now masks any userinfo and any credential-named query parameter, and the config save restore and the settings UI detect masked paths with the same patterns so a save never writes the mask back over the real credential.

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,
@@ -770,9 +771,7 @@ def _restore_masked_camera_paths(config_data: dict, config: FrigateConfig) -> No
             path = input_obj.get("path")
             if not isinstance(path, str):
                 continue
-            if ("://*:*@" in path or "user=*&password=*" in path) and index < len(
+            if is_masked_camera_path(path) and index < len(existing_paths):
-                existing_paths
-            ):
                 input_obj["path"] = existing_paths[index]
 
 

frigate/const.py

@@ -117,7 +117,13 @@ 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]+"
+REGEX_URL_USERINFO = r"://(?:[^\s/:]*:\S*|[^\s/]+)@"
+_CREDENTIAL_QUERY_KEYS = (
+    r"(?:user(?:name)?|usr|login|pass(?:word|wd)?|pwd"
+    r"|[a-z_]*token|[a-z_]*secret|[a-z_]*private|api_?key|auth)"
+)
+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/test/http_api/test_http_config_set.py

@@ -503,6 +503,71 @@ 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"},
+            "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)
+
 
 if __name__ == "__main__":
     unittest.main()

frigate/test/test_camera_pw.py

@@ -2,7 +2,11 @@
 
 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 +49,58 @@ 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",
+            "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",
+            "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."""
+        self.assertIn(",", clean_camera_user_pass("rtsp://a/x,rtsp://u:p@b/y"))
+        self.assertNotIn(",", clean_camera_user_pass("rtsp://u:p,q@b/y"))

frigate/util/builtin.py

@@ -21,8 +21,10 @@ import numpy as np
 from ruamel.yaml import YAML
 
 from frigate.const import (
-    REGEX_HTTP_CAMERA_USER_PASS,
+    REGEX_CREDENTIAL_QUERY_PARAM,
+    REGEX_MASKED_CREDENTIAL,
     REGEX_RTSP_CAMERA_USER_PASS,
+    REGEX_URL_USERINFO,
     STREAM_TYPE_MAIN,
     STREAM_TYPE_SUB,
 )
@@ -117,9 +119,16 @@ 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."""
+    """Mask URL userinfo and credential query parameters in line."""
-    rtsp_cleaned = re.sub(REGEX_RTSP_CAMERA_USER_PASS, "://*:*@", line)
+
-    return re.sub(REGEX_HTTP_CAMERA_USER_PASS, "user=*&password=*", rtsp_cleaned)
+    # Query values go first so an "@" inside one can't extend the userinfo match.
+    line = re.sub(REGEX_CREDENTIAL_QUERY_PARAM, r"\1*", line)
+    return re.sub(REGEX_URL_USERINFO, "://*:*@", line)
+
+
+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:

web/src/utils/credentialMask.ts

@@ -1,40 +1,28 @@
-const MASKED_AUTH_PATTERN = /:\/\/\*:\*@/i;
+// Mirrors clean_camera_user_pass and REGEX_MASKED_CREDENTIAL in the backend.
-const MASKED_QUERY_PATTERN = /(?:[?&])user=\*&password=\*/i;
+const CREDENTIAL_QUERY_KEYS =
+  "(?:user(?:name)?|usr|login|pass(?:word|wd)?|pwd|[a-z_]*token|[a-z_]*secret|[a-z_]*private|api_?key|auth)";
+const URL_USERINFO_PATTERN = /:\/\/(?:[^\s/:]*:\S*|[^\s/]+)@/g;
+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);
+  MASKED_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);
-  }
-};
 
 export const maskCredentials = (value: string): string => {
   if (!value) {
     return value;
   }
 
-  const maskedAuth = value.replace(/:\/\/[^:@/\s]+:[^@/\s]*@/g, "://*:*@");
+  return value
-
+    .replace(CREDENTIAL_QUERY_PATTERN, "$1*")
-  return maskedAuth
+    .replace(URL_USERINFO_PATTERN, "://*:*@");
-    .replace(/([?&]user=)[^&]*/gi, "$1*")
-    .replace(/([?&]password=)[^&]*/gi, "$1*");
 };
+
+export const hasCredentials = (value: string): boolean =>
+  !!value && (isMaskedPath(value) || maskCredentials(value) !== value);