diff --git a/backend/app/services/external_camera.py b/backend/app/services/external_camera.py index 9728abf59..cf710951f 100644 --- a/backend/app/services/external_camera.py +++ b/backend/app/services/external_camera.py @@ -9,9 +9,11 @@ to ensure they are well-formed before use. import asyncio import functools +import ipaddress import logging import re import shutil +import socket from collections.abc import AsyncGenerator, Callable from pathlib import Path from urllib.parse import urlparse @@ -22,13 +24,77 @@ from backend.app.core.logging_filters import redact_url_credentials logger = logging.getLogger(__name__) +# Protocols ffmpeg may use for an RTSP input. RTSP negotiates its media +# transport at runtime, so the transports have to be here alongside rtsp itself; +# tls and crypto cover encrypted variants. Everything ffmpeg would otherwise +# accept behind an -i — file, http, tcp to anywhere, concat — is left out, so a +# stream that references something outside itself cannot pull it in. +_RTSP_PROTOCOL_WHITELIST = "rtsp,rtp,udp,tcp,tls,crypto" + + +def _blocked_host_reason(hostname: str) -> str | None: + """Describe why *hostname* is a destination we refuse to fetch, or None to allow it. + + Camera URLs are user-supplied and reach the network — over aiohttp for the + HTTP types, and as an ``ffmpeg -i`` argument for RTSP — so this is where the + SSRF boundary sits. LAN addresses are deliberately allowed: cameras live on + the same network as Bambuddy, and blocking RFC-1918 would remove the feature + rather than protect it. What is left to refuse is the host talking to + itself, the unspecified address, link-local (which is where the cloud + metadata endpoint lives), and the metadata hostnames. + + IP literals are classified with ``ipaddress`` rather than compared against a + list of spellings, because 127.0.0.1, 127.0.0.2, 2130706433, 0177.0.0.1, + 127.1 and ::ffff:127.0.0.1 all arrive at loopback and a list of strings only + ever catches whichever one someone thought to write down. ``inet_aton`` + comes first because it accepts the legacy octal, decimal and short forms + that ``ip_address`` rejects — the C resolvers behind aiohttp and ffmpeg + accept them, so refusing to understand them here would only mean not seeing + where the request is actually going. + """ + host = hostname.lower() + + ip: ipaddress.IPv4Address | ipaddress.IPv6Address | None = None + try: + ip = ipaddress.ip_address(socket.inet_aton(host)) + except OSError: + try: + ip = ipaddress.ip_address(host) + except ValueError: + ip = None + + if ip is None: + # A name, not an address. It is not resolved here on purpose: aiohttp + # and ffmpeg each resolve independently afterwards, so a check here + # decides nothing about where they end up (DNS rebinding), while a + # lookup on every capture would break LAN cameras behind slow or + # intermittent local DNS. + if host == "localhost" or host.endswith(".localhost"): + return "localhost" + if host in ("metadata.google.internal", "metadata.google"): + return "a cloud metadata service" + return None + + # ::ffff:127.0.0.1 is loopback wearing an IPv6 spelling. + mapped = getattr(ip, "ipv4_mapped", None) + if mapped is not None: + ip = mapped + + if ip.is_loopback: + return "loopback" + if ip.is_unspecified: + return "the unspecified address" + if ip.is_link_local: + return "a link-local address (the cloud metadata range)" + return None + def _sanitize_camera_url(url: str, allowed_schemes: tuple[str, ...] = ("http", "https", "rtsp")) -> str | None: """Validate and sanitize camera URL, returning a safe reconstructed URL. - This validates that the URL is well-formed, uses an allowed scheme, - does not target cloud metadata services, and returns a reconstructed - URL from validated components. + This validates that the URL is well-formed, uses an allowed scheme, does not + target the host itself or a cloud metadata service, and returns a URL + reconstructed from the validated components. Note: This intentionally allows user-provided URLs as that is the purpose of external camera configuration. Local network IPs are @@ -51,37 +117,35 @@ def _sanitize_camera_url(url: str, allowed_schemes: tuple[str, ...] = ("http", " if scheme not in allowed_schemes: return None - # Block cloud metadata service endpoints (SSRF mitigation) - # These are dangerous destinations that should never be accessed hostname = parsed.hostname or "" - hostname_lower = hostname.lower() - blocked_hosts = ( - "169.254.169.254", # AWS/GCP/Azure metadata - "metadata.google.internal", # GCP metadata - "metadata.google", - "localhost", # Block localhost to prevent internal service access - "127.0.0.1", - "::1", - "0.0.0.0", # nosec B104 - ) - if hostname_lower in blocked_hosts: - logger.warning("Blocked camera URL targeting restricted host: %s", hostname) + if not hostname: return None - - # Block link-local addresses (169.254.x.x) - if hostname.startswith("169.254."): - logger.warning("Blocked camera URL targeting link-local address: %s", hostname) + blocked = _blocked_host_reason(hostname) + if blocked: + logger.warning("Blocked camera URL targeting %s: %s", blocked, hostname) return None # Reconstruct URL from validated components to break taint chain # This creates a new string from validated parts + # + # The credentials are carried across verbatim from netloc rather than + # via parsed.username/.password, which urlparse has already percent- + # decoded: re-emitting those would corrupt any password containing an + # @ or a :. They have to survive at all because most RTSP cameras — and + # a fair number of MJPEG ones — carry their login in the URL, and + # dropping it turns every one of them into an authentication failure. + netloc = parsed.netloc + userinfo = f"{netloc.rsplit('@', 1)[0]}@" if "@" in netloc else "" + # parsed.hostname has already stripped the brackets off an IPv6 literal; + # without them back the result is not a URL any client can parse. + host_str = f"[{hostname}]" if ":" in hostname else hostname port_str = f":{parsed.port}" if parsed.port else "" path = parsed.path or "" query = f"?{parsed.query}" if parsed.query else "" fragment = f"#{parsed.fragment}" if parsed.fragment else "" # Build sanitized URL from validated components - sanitized = f"{scheme}://{hostname}{port_str}{path}{query}{fragment}" + sanitized = f"{scheme}://{userinfo}{host_str}{port_str}{path}{query}{fragment}" return sanitized except ValueError: return None @@ -380,18 +444,18 @@ async def _capture_frame_uncoalesced( return None -async def _capture_usb_frame(device: str, timeout: int) -> bytes | None: - """Capture frame from USB camera using ffmpeg.""" - ffmpeg = get_ffmpeg_path() - if not ffmpeg: - logger.error("ffmpeg not found - required for USB camera capture") - return None +def _safe_usb_device_path(device: str) -> str | None: + """Rebuild a /dev/videoN path from a validated device number, or None. - # Validate device path - must be /dev/videoN format where N is 0-99 - # This prevents path traversal by using a strict allowlist approach - import re as regex_module + Validate device path - must be /dev/videoN format where N is 0-99. This + prevents path traversal by using a strict allowlist approach: the returned + path is built from an integer, which cannot carry a traversal, rather than + from any part of the caller's string. - device_match = regex_module.match(r"^/dev/video(\d{1,2})$", device) + Returns None if the device does not exist, so a caller cannot hand ffmpeg a + path to something that is not a device node. + """ + device_match = re.match(r"^/dev/video(\d{1,2})$", device) if not device_match: logger.error("Invalid USB device path format: %s", device) return None @@ -399,9 +463,6 @@ async def _capture_usb_frame(device: str, timeout: int) -> bytes | None: # Convert to integer to break taint chain - integers cannot contain path traversal # lgtm[py/path-injection] - device_num is validated integer 0-99 device_num = int(device_match.group(1)) # Safe: regex guarantees 1-2 digits - if device_num > 99: - logger.error("USB device number out of range: %s", device_num) - return None # Construct safe path from validated integer (completely untainted) safe_device_path = Path(f"/dev/video{device_num}") # lgtm[py/path-injection] @@ -410,8 +471,22 @@ async def _capture_usb_frame(device: str, timeout: int) -> bytes | None: logger.error("USB device does not exist: %s", safe_device_path) return None + return str(safe_device_path) # lgtm[py/path-injection] + + +async def _capture_usb_frame(device: str, timeout: int) -> bytes | None: + """Capture frame from USB camera using ffmpeg.""" + ffmpeg = get_ffmpeg_path() + if not ffmpeg: + logger.error("ffmpeg not found - required for USB camera capture") + return None + + safe_device = _safe_usb_device_path(device) + if not safe_device: + return None + # Use the safe path for ffmpeg - this is a hardcoded /dev/videoN path - device = str(safe_device_path) # lgtm[py/path-injection] + device = safe_device # lgtm[py/path-injection] # Use ffmpeg to grab a single frame from USB camera cmd = [ @@ -542,22 +617,34 @@ async def _capture_rtsp_frame(url: str, timeout: int) -> bytes | None: """Capture frame from RTSP using ffmpeg. For rtsps:// URLs, a local TLS proxy is used to avoid GnuTLS issues. + + Note: this function intentionally connects to user-configured URLs, the same + as the MJPEG and snapshot paths. The URL is sanitized and dangerous + destinations are blocked before it reaches ffmpeg. """ ffmpeg = get_ffmpeg_path() if not ffmpeg: logger.error("ffmpeg not found - required for RTSP capture") return None + # ffmpeg's -i accepts every protocol it was built with, so an unchecked URL + # here is a request to any host and scheme the caller names, not merely to a + # camera. Restricting the scheme to RTSP is what keeps this a camera fetch. + safe_url = _sanitize_camera_url(url, ("rtsp", "rtsps")) + if not safe_url: + logger.error("Invalid RTSP URL: %s...", redact_url_credentials(url)[:50]) + return None + # If rtsps://, use TLS proxy proxy_server = None - effective_url = url - if url.lower().startswith("rtsps://"): + effective_url = safe_url + if safe_url.lower().startswith("rtsps://"): try: from urllib.parse import urlparse from backend.app.services.camera import create_tls_proxy - parsed = urlparse(url) + parsed = urlparse(safe_url) target_port = parsed.port or 322 proxy_port, proxy_server = await create_tls_proxy(parsed.hostname, target_port) userinfo = "" @@ -566,17 +653,24 @@ async def _capture_rtsp_frame(url: str, timeout: int) -> bytes | None: if parsed.password: userinfo += f":{parsed.password}" userinfo += "@" + # Points at loopback deliberately, and is built after the check + # above rather than re-checked: the destination that mattered was + # the one the caller named, and it has already been vetted. effective_url = f"rtsp://{userinfo}127.0.0.1:{proxy_port}{parsed.path}" if parsed.query: effective_url += f"?{parsed.query}" except Exception as e: logger.warning("Failed to create TLS proxy for RTSP capture, falling back: %s", e) - effective_url = url + effective_url = safe_url cmd = [ ffmpeg, "-rtsp_transport", "tcp", + # Belt and braces on the scheme check above: a demuxer that follows a + # reference out of the stream cannot leave these protocols either. + "-protocol_whitelist", + _RTSP_PROTOCOL_WHITELIST, "-i", effective_url, "-frames:v", @@ -956,6 +1050,11 @@ async def _stream_rtsp( For rtsps:// URLs, a local TLS proxy (Python OpenSSL) is used instead of relying on ffmpeg's GnuTLS backend, which has compatibility issues with some printer firmwares. + + Note: this function intentionally connects to user-configured URLs. The URL + is sanitized and dangerous destinations are blocked before it reaches + ffmpeg — see ``_capture_rtsp_frame``, which guards the one-shot path the + same way. """ ffmpeg = get_ffmpeg_path() if not ffmpeg: @@ -964,16 +1063,21 @@ async def _stream_rtsp( from backend.app.services.camera import rtsp_socket_timeout_flag + safe_url = _sanitize_camera_url(url, ("rtsp", "rtsps")) + if not safe_url: + logger.error("Invalid RTSP stream URL: %s...", redact_url_credentials(url)[:50]) + return + # If the URL uses rtsps://, set up a TLS proxy so ffmpeg uses plain rtsp:// proxy_server = None - effective_url = url - if url.lower().startswith("rtsps://"): + effective_url = safe_url + if safe_url.lower().startswith("rtsps://"): try: from urllib.parse import urlparse from backend.app.services.camera import create_tls_proxy - parsed = urlparse(url) + parsed = urlparse(safe_url) target_port = parsed.port or 322 proxy_port, proxy_server = await create_tls_proxy(parsed.hostname, target_port) # Rewrite URL: rtsps://user:pass@host:port/path → rtsp://user:pass@127.0.0.1:proxy/path @@ -983,12 +1087,14 @@ async def _stream_rtsp( if parsed.password: userinfo += f":{parsed.password}" userinfo += "@" + # Loopback by design, and built after the check above rather than + # re-checked — see the same rewrite in _capture_rtsp_frame. effective_url = f"rtsp://{userinfo}127.0.0.1:{proxy_port}{parsed.path}" if parsed.query: effective_url += f"?{parsed.query}" except Exception as e: logger.warning("Failed to create TLS proxy for RTSP, falling back to direct: %s", e) - effective_url = url + effective_url = safe_url cmd = [ ffmpeg, @@ -996,6 +1102,8 @@ async def _stream_rtsp( "tcp", "-rtsp_flags", "prefer_tcp", + "-protocol_whitelist", + _RTSP_PROTOCOL_WHITELIST, # Socket I/O timeout name varies by ffmpeg version (#1504); see # `rtsp_socket_timeout_flag()` in services.camera. f"-{rtsp_socket_timeout_flag()}", @@ -1109,14 +1217,13 @@ async def _stream_usb( logger.error("ffmpeg not found - required for USB camera streaming") return - # Validate device path - if not device.startswith("/dev/video"): - logger.error("Invalid USB device path: %s", device) - return - - if not Path(device).exists(): - logger.error("USB device does not exist: %s", device) + # Same validation as the one-shot path: a prefix check accepted + # /dev/video/../../, which -f v4l2 would then refuse + # rather than the check refusing it. + safe_device = _safe_usb_device_path(device) + if not safe_device: return + device = safe_device # ffmpeg command to stream from USB camera (v4l2) cmd = [ diff --git a/backend/tests/unit/test_external_camera_ssrf.py b/backend/tests/unit/test_external_camera_ssrf.py new file mode 100644 index 000000000..92093b7b3 --- /dev/null +++ b/backend/tests/unit/test_external_camera_ssrf.py @@ -0,0 +1,290 @@ +"""The RTSP camera paths must not become a request generator for arbitrary hosts. + +`_sanitize_camera_url` is the SSRF boundary for user-configured camera URLs. It +was applied to the MJPEG and snapshot paths but not to the two RTSP ones, which +handed the URL to `ffmpeg -i` unchecked — and ffmpeg's `-i` speaks http, tcp, +file and everything else it was built with, so `camera_type=rtsp` was a way to +name any destination and any protocol. + +Wiring the guard in is only half of it. The guard rebuilt URLs from +`parsed.hostname`, which drops credentials and unbrackets IPv6 literals, and it +recognised loopback by comparing against four spellings of it. So these tests +pin three things at once: the RTSP paths refuse what they should, the guard +recognises a destination however it is written, and a real camera — which +usually means an authenticated one — still works. +""" + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from backend.app.services.external_camera import ( + _blocked_host_reason, + _capture_rtsp_frame, + _safe_usb_device_path, + _sanitize_camera_url, + _stream_rtsp, +) + +RTSP_SCHEMES = ("rtsp", "rtsps") +HTTP_SCHEMES = ("http", "https") + + +class TestTheHostsWeRefuse: + """Loopback, the unspecified address and link-local, however they are spelled.""" + + @pytest.mark.parametrize( + "host", + [ + "127.0.0.1", + "127.0.0.2", # the whole 127/8 range, not just .1 + "127.1", # short form + "2130706433", # decimal + "0177.0.0.1", # octal + "0x7f.0.0.1", # hex + "[::1]", + "[::ffff:127.0.0.1]", # loopback wearing an IPv6 spelling + "localhost", + "sub.localhost", + ], + ) + def test_loopback_is_refused(self, host): + assert _sanitize_camera_url(f"rtsp://{host}:554/live", RTSP_SCHEMES) is None + + @pytest.mark.parametrize("host", ["0.0.0.0", "[::]"]) # nosec B104 + def test_the_unspecified_address_is_refused(self, host): + assert _sanitize_camera_url(f"rtsp://{host}:554/live", RTSP_SCHEMES) is None + + @pytest.mark.parametrize( + "host", + [ + "169.254.169.254", # AWS/GCP/Azure metadata + "169.254.1.1", # the rest of the range, not just the metadata IP + "[fe80::1]", + "metadata.google.internal", + "metadata.google", + ], + ) + def test_link_local_and_metadata_are_refused(self, host): + assert _sanitize_camera_url(f"rtsp://{host}/live", RTSP_SCHEMES) is None + + def test_the_reason_is_reported_for_logging(self): + assert _blocked_host_reason("2130706433") == "loopback" + assert _blocked_host_reason("169.254.169.254") is not None + assert _blocked_host_reason("192.168.1.50") is None + + +class TestTheCamerasWeAllow: + """LAN is allowed on purpose — that is where cameras are.""" + + @pytest.mark.parametrize( + "url", + [ + "rtsp://192.168.1.50:554/live", + "rtsp://10.0.0.5/stream1", + "rtsp://172.16.4.9:8554/cam", + "rtsp://[fd00::1]:554/live", # unique-local IPv6 + "rtsp://cam.lan/live", + "rtsps://camera.example.com:322/stream", + ], + ) + def test_a_camera_url_survives(self, url): + assert _sanitize_camera_url(url, RTSP_SCHEMES) is not None + + def test_a_hostname_is_not_resolved(self): + """A name that would resolve to loopback still passes. + + Not an oversight: aiohttp and ffmpeg resolve independently afterwards, + so a lookup here decides nothing (DNS rebinding) while costing a DNS + round trip on every capture. Pinned so the omission stays deliberate. + """ + assert _sanitize_camera_url("rtsp://localtest.me/live", RTSP_SCHEMES) is not None + + +class TestWhatTheGuardMustNotDestroy: + """Most RTSP cameras carry their login in the URL. Stripping it would turn + every one of them into an authentication failure — a worse outage than the + hole being closed.""" + + def test_credentials_survive(self): + url = "rtsp://admin:hunter2@192.168.1.50:554/live" + assert _sanitize_camera_url(url, RTSP_SCHEMES) == url + + def test_percent_encoded_credentials_survive_byte_for_byte(self): + """urlparse's .username/.password are already decoded, so rebuilding + from them would corrupt any password containing an @ or a :.""" + url = "rtsp://ad%40min:p%3Ass%40word@192.168.1.50:554/live" + assert _sanitize_camera_url(url, RTSP_SCHEMES) == url + + def test_an_ipv6_literal_keeps_its_brackets(self): + """Without them the result is not a URL any client can parse.""" + assert _sanitize_camera_url("rtsp://[fd00::1]:554/live", RTSP_SCHEMES) == "rtsp://[fd00::1]:554/live" + + def test_http_cameras_keep_their_basic_auth_too(self): + url = "http://admin:hunter2@192.168.1.50/stream.mjpg" + assert _sanitize_camera_url(url, HTTP_SCHEMES) == url + + def test_port_query_and_fragment_survive(self): + url = "rtsp://192.168.1.50:8554/live?channel=2&subtype=1#frag" + assert _sanitize_camera_url(url, RTSP_SCHEMES) == url + + +class TestSchemeAllowlist: + """What keeps an ffmpeg input a camera fetch rather than a fetch.""" + + @pytest.mark.parametrize( + "url", + [ + "http://192.168.1.50:8080/internal", + "https://192.168.1.50/internal", + "tcp://192.168.1.50:22", + "file:///etc/passwd", + "concat:/etc/passwd", + "udp://192.168.1.50:1234", + "ftp://192.168.1.50/x", + ], + ) + def test_only_rtsp_reaches_the_rtsp_paths(self, url): + assert _sanitize_camera_url(url, RTSP_SCHEMES) is None + + def test_rtsp_does_not_reach_the_http_paths(self): + assert _sanitize_camera_url("rtsp://192.168.1.50/live", HTTP_SCHEMES) is None + + @pytest.mark.parametrize("url", ["", "not a url", "rtsp://", "://192.168.1.50/x"]) + def test_malformed_input_is_refused(self, url): + assert _sanitize_camera_url(url, RTSP_SCHEMES) is None + + +def _fake_ffmpeg(): + return patch("backend.app.services.external_camera.get_ffmpeg_path", return_value="/usr/bin/ffmpeg") + + +def _spawn_spy(returncode: int | None = 0, stdout: bytes = b"\xff\xd8" + b"\x00" * 200): + """Stand in for the ffmpeg subprocess, recording the argv it was handed. + + The streaming path reads until EOF, so stdout.read returns b"" and the + generator finishes immediately — these tests are about whether ffmpeg was + launched and with what, not about frame extraction. + """ + process = MagicMock() + process.returncode = returncode + process.communicate = AsyncMock(return_value=(stdout, b"")) + process.stdout.read = AsyncMock(return_value=b"") + process.stderr.read = AsyncMock(return_value=b"") + process.wait = AsyncMock(return_value=returncode) + process.kill = MagicMock() + process.terminate = MagicMock() + return patch( + "backend.app.services.external_camera.asyncio.create_subprocess_exec", + new=AsyncMock(return_value=process), + ) + + +class TestRtspCaptureRefusesUnsafeUrls: + """`_capture_rtsp_frame` — the one-shot path behind the test-connection + endpoint, which takes url and camera_type straight off the query string.""" + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "url", + [ + "http://127.0.0.1:8080/internal-service", # the reported PoC + "http://192.168.1.100:8080/any-image.jpg", + "file:///etc/passwd", + "rtsp://127.0.0.1:554/live", + "rtsp://2130706433:554/live", + "rtsp://169.254.169.254/live", + ], + ) + async def test_no_process_is_spawned(self, url): + with _fake_ffmpeg(), _spawn_spy() as spawn: + assert await _capture_rtsp_frame(url, timeout=5) is None + spawn.assert_not_awaited() + + @pytest.mark.asyncio + async def test_a_real_camera_still_captures(self): + with _fake_ffmpeg(), _spawn_spy() as spawn: + frame = await _capture_rtsp_frame("rtsp://admin:hunter2@192.168.1.50:554/live", timeout=5) + + assert frame is not None + cmd = spawn.await_args.args + assert "rtsp://admin:hunter2@192.168.1.50:554/live" in cmd, ( + "the camera's credentials must reach ffmpeg or every authenticated camera breaks" + ) + + @pytest.mark.asyncio + async def test_ffmpeg_is_confined_to_rtsp_protocols(self): + """Belt and braces behind the scheme check: a stream that references + something outside itself must not be able to pull it in.""" + with _fake_ffmpeg(), _spawn_spy() as spawn: + await _capture_rtsp_frame("rtsp://192.168.1.50:554/live", timeout=5) + + cmd = spawn.await_args.args + whitelist = cmd[cmd.index("-protocol_whitelist") + 1].split(",") + assert "rtsp" in whitelist + assert "file" not in whitelist + assert "http" not in whitelist + + +class TestRtspStreamRefusesUnsafeUrls: + """`_stream_rtsp` — the live-view path, and the one the report missed.""" + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "url", + [ + "http://127.0.0.1:8080/internal-service", + "rtsp://127.0.0.1:554/live", + "rtsp://[::ffff:127.0.0.1]:554/live", + "file:///etc/passwd", + ], + ) + async def test_no_process_is_spawned(self, url): + with _fake_ffmpeg(), _spawn_spy() as spawn: + frames = [frame async for frame in _stream_rtsp(url, fps=5)] + + assert frames == [] + spawn.assert_not_awaited() + + @pytest.mark.asyncio + async def test_a_real_camera_still_reaches_ffmpeg(self): + with _fake_ffmpeg(), _spawn_spy(returncode=None) as spawn: + [frame async for frame in _stream_rtsp("rtsp://admin:hunter2@192.168.1.50:554/live", fps=5)] + + spawn.assert_awaited_once() + cmd = spawn.await_args.args + assert "rtsp://admin:hunter2@192.168.1.50:554/live" in cmd + assert "-protocol_whitelist" in cmd + + +class TestUsbDevicePaths: + """The USB paths take a device path from the same request field, and the + streaming one used to check only that it started with /dev/video.""" + + @pytest.mark.parametrize( + "device", + [ + "/dev/video/../../etc/passwd", + "/dev/videos/../../etc/shadow", + "/dev/video0; rm -rf /", + "/etc/passwd", + "/dev/video100", # three digits is not a device number + "", + ], + ) + def test_a_path_that_is_not_a_device_node_is_refused(self, device): + assert _safe_usb_device_path(device) is None + + def test_a_missing_device_is_refused(self): + """Existence is part of the check — ffmpeg must never be pointed at a + path just because it is shaped like one.""" + with patch("backend.app.services.external_camera.Path") as path_cls: + path_cls.return_value.exists.return_value = False + assert _safe_usb_device_path("/dev/video0") is None + + def test_the_path_is_rebuilt_from_the_device_number(self): + with patch("backend.app.services.external_camera.Path") as path_cls: + path_cls.return_value.exists.return_value = True + path_cls.return_value.__str__.return_value = "/dev/video7" + assert _safe_usb_device_path("/dev/video7") == "/dev/video7" + path_cls.assert_called_once_with("/dev/video7")