public_git/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-09-30 03:01:21 +02:00
Code Issues Packages Projects Releases Wiki Activity

Security hardening (maziggy/bambuddy-security #9)

Browse Source
This commit is contained in:
maziggy
2026-08-15 14:24:04 +02:00
parent 5bee8df5cd 16c8c6f2ea
commit b1610f70ae
2 changed files with 448 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/services/external_camera.py
+158 -51
View File
@@ -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/../../<anything that exists>, 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 = [
backend/tests/unit/test_external_camera_ssrf.py
+290
View File
@@ -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")