diff --git a/frigate/api/app.py b/frigate/api/app.py index d38876c34c..60905be632 100644 --- a/frigate/api/app.py +++ b/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( - existing_paths - ): + if is_masked_camera_path(path) and index < len(existing_paths): input_obj["path"] = existing_paths[index] diff --git a/frigate/const.py b/frigate/const.py index 6ef8919be2..3b118682a1 100644 --- a/frigate/const.py +++ b/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 diff --git a/frigate/test/http_api/test_http_config_set.py b/frigate/test/http_api/test_http_config_set.py index f31d382c51..6d9374e76d 100644 --- a/frigate/test/http_api/test_http_config_set.py +++ b/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() diff --git a/frigate/test/test_camera_pw.py b/frigate/test/test_camera_pw.py index af4cecbd22..0a058b9269 100644 --- a/frigate/test/test_camera_pw.py +++ b/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")) diff --git a/frigate/util/builtin.py b/frigate/util/builtin.py index 7b93ca8183..42ca6217a6 100644 --- a/frigate/util/builtin.py +++ b/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.""" - 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.""" + + # 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: diff --git a/web/src/utils/credentialMask.ts b/web/src/utils/credentialMask.ts index 34f0996a90..56eb390437 100644 --- a/web/src/utils/credentialMask.ts +++ b/web/src/utils/credentialMask.ts @@ -1,40 +1,28 @@ -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|[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); - -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(CREDENTIAL_QUERY_PATTERN, "$1*") + .replace(URL_USERINFO_PATTERN, "://*:*@"); }; + +export const hasCredentials = (value: string): boolean => + !!value && (isMaskedPath(value) || maskCredentials(value) !== value);