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-05 15:21:42 +02:00
parent 684a328d6f
commit 16c8c6f2ea
3 changed files with 449 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+1 -1
View File
@@ -116,7 +116,7 @@ All notable changes to Bambuddy will be documented in this file.
### Security
- **Patched two build-time frontend dependencies flagged by `npm audit` (GHSA-r28c-9q8g-f849, GHSA-mh99-v99m-4gvg, GHSA-rgw5-rvv9-x895)** — `postcss` 8.5.15 → 8.5.23 fixes a path traversal in its source-map auto-loader (`sourceMappingURL`) that could disclose arbitrary `.map` files, and `brace-expansion` (pulled in transitively by `eslint` via `minimatch`) is bumped through the existing `overrides` block (`^5.0.7` → `^5.0.9`) for a denial-of-service via unbounded expansion. The first `brace-expansion` advisory was answered in 5.0.8 by capping the length of the combined result, but that cap covered only the accumulator the results are merged into and not the two intermediate arrays that feed it — so a small brace pattern could still exhaust the heap, fatally and beyond the reach of a `try`/`catch`, or stall the event loop for minutes. 5.0.9 bounds both arrays as they are built. Both packages are build/lint-time tooling only — neither is part of the shipped app, so no running Bambuddy install was exposed. `postcss` moved within its existing range; `brace-expansion` needed the pin because `npm audit fix` can't lift `eslint` to the patched transitive on its own.
- **Pinned `react-router` to its most-patched 7.x (7.18.1) and documented the one remaining, unreachable advisory (GHSA-qwww-vcr4-c8h2)** — Staying current on the 7.x line matters: 7.18.1 clears 14 advisories that older 7.x releases carry, several reachable in a browser SPA (open-redirect XSS in `<Link>`/`useNavigate`, route-matching DoS). The single advisory that still flags 7.18.1 — a CSRF bypass — applies only to React Router's **RSC mode**, which requires the server runtime (`@react-router/server`, not installed); Bambuddy is a Vite SPA using `BrowserRouter`, so the vulnerable path is unreachable. There is no non-major fix (the patch landed only in the 8.3.0 major, and `react-router-dom` has no 8.x — adopting it would mean migrating every import to `react-router` plus a React peer bump), so `react-router`/`react-router-dom` are pinned to 7.18.1 and the finding is carried as a documented, fail-closed exception in the CI audit gate: a *different* react-router advisory still fails CI, and the exemption is dropped automatically the moment a non-major fix ships. `npm audit fix --force` is deliberately avoided — its suggested "fix" is a downgrade to 7.11.0, which reintroduces those 14 advisories.
- **Security hardening (security-issue #9)**
## [1.2.5.2] - 2026-08-02
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")