diff --git a/backend/app/api/routes/camera.py b/backend/app/api/routes/camera.py index eef4e7c9c..c876c3724 100644 --- a/backend/app/api/routes/camera.py +++ b/backend/app/api/routes/camera.py @@ -22,12 +22,12 @@ from backend.app.core.auth import ( create_camera_stream_token, ) from backend.app.core.database import get_db -from backend.app.core.logging_filters import redact_url_credentials from backend.app.core.permissions import Permission from backend.app.models.printer import Printer from backend.app.models.user import User from backend.app.services.camera import ( capture_camera_frame, + close_tls_proxy, create_tls_proxy, generate_chamber_image_stream, get_camera_port, @@ -45,6 +45,7 @@ from backend.app.services.camera_fanout import ( shutdown_broadcaster, ) from backend.app.services.camera_profiles import get_camera_profile +from backend.app.utils.ffmpeg_output import summarize_ffmpeg_stderr logger = logging.getLogger(__name__) router = APIRouter(prefix="/printers", tags=["camera"]) @@ -407,37 +408,11 @@ async def _terminate_ffmpeg(process: asyncio.subprocess.Process, stream_id: str _spawned_ffmpeg_pids.pop(process.pid, None) -def _summarize_ffmpeg_stderr(text: str | None) -> str: - """Strip ffmpeg's boilerplate banner and keep only actionable lines. - - ffmpeg prints ~20 lines of version/build/configuration/lib headers before - any actual error message. Logging the full banner on every retry floods - the log (hundreds of lines per failed stream). This filter drops the - banner and caps output at the last 10 meaningful lines. - - Credentials are masked here rather than at each ``logger`` call because - this is the one funnel every stderr log in this module passes through. - ffmpeg echoes the RTSP input URL back in its ``Input #0`` line, which - carries the printer access code. - """ - if not text: - return "" - text = redact_url_credentials(text) or "" - banner_prefixes = ( - "ffmpeg version ", - " built with ", - " configuration:", - " libavutil ", - " libavcodec ", - " libavformat ", - " libavdevice ", - " libavfilter ", - " libswscale ", - " libswresample ", - " libpostproc ", - ) - meaningful = [ln for ln in text.splitlines() if ln.strip() and not ln.startswith(banner_prefixes)] - return "\n".join(meaningful[-10:]) +# The banner-stripping summariser moved to backend.app.utils.ffmpeg_output so +# the seven other places that log ffmpeg stderr could stop truncating it from +# the front (#2968). Imported under the private name this module has always +# used: _FfmpegStderrTail and the tests both reach for it by that name. +_summarize_ffmpeg_stderr = summarize_ffmpeg_stderr class _FfmpegStderrTail: @@ -851,8 +826,7 @@ async def generate_rtsp_mjpeg_stream( await stderr_tail.aclose() # Shut down the TLS proxy - proxy_server.close() - await proxy_server.wait_closed() + await close_tls_proxy(proxy_server) @router.post("/camera/stream-token") diff --git a/backend/app/services/archive.py b/backend/app/services/archive.py index e6a61c6c1..48c49d540 100644 --- a/backend/app/services/archive.py +++ b/backend/app/services/archive.py @@ -19,8 +19,9 @@ from backend.app.models.archive import PrintArchive from backend.app.models.filament import Filament from backend.app.models.printer import Printer from backend.app.utils.archive_paths import archive_dir as resolve_archive_dir +from backend.app.utils.ffmpeg_output import NO_FFMPEG_OUTPUT, summarize_ffmpeg_stderr from backend.app.utils.filename import clean_display_name -from backend.app.utils.safe_path import PathTraversalError, safe_join_under +from backend.app.utils.safe_path import PathTraversalError, assert_under, safe_join_under logger = logging.getLogger(__name__) @@ -1676,52 +1677,140 @@ class ArchiveService: # the first soft-delete pass so there is nothing left on disk. return True - dir_to_delete = self._resolve_archive_dir_for_delete(archive) + dirs_to_delete = self._resolve_archive_dirs_for_delete(archive) + recorded_paths = (archive.timelapse_path, archive.thumbnail_path) await _null_print_log_thumbnail_paths(self.db, archive_id) await _delete_related_queue_items(self.db, archive_id) archive.deleted_at = datetime.now(timezone.utc) await self.db.commit() - if dir_to_delete: - shutil.rmtree(dir_to_delete, ignore_errors=True) + for directory in dirs_to_delete: + shutil.rmtree(directory, ignore_errors=True) + self._purge_id_named_dir(archive_id, recorded_paths) return True - def _resolve_archive_dir_for_delete(self, archive: PrintArchive) -> Path | None: - """Return the on-disk directory that backs *archive*, after the same - two safety checks ``delete_archive`` enforces. + def _resolve_archive_dirs_for_delete(self, archive: PrintArchive) -> list[Path]: + """Directories belonging to *archive* alone, safe to remove whole. - Extracted so soft-delete and hard-delete share the path-resolution - rules. Returns ``None`` when nothing should be removed from disk - (no file_path, path outside archive_dir, or path not deep enough). + Shared by soft-delete and hard-delete so the two cannot drift apart + again — the previous helper said it was extracted for that reason, and + ``delete_archive`` was still doing its own copy of the same rules. + + An archive with a 3MF owns the directory its ``file_path`` sits in, + ``//_/``. Any archive may also + own ``archive/no_source//``, where a source 3MF uploaded onto a + no-3MF archive lands (#1531); that one was never removed, so deleting + such an archive freed the row and left the upload behind. + + Two directories are deliberately absent. ``/photos`` is the + legacy location *every* no-3MF archive wrote into at once, so removing + it on one delete would take the others' photos with it. And + ``/`` — the directory :func:`resolve_archive_dir` gives + an archive with no ``file_path`` — is handled by + :meth:`_purge_id_named_dir` instead, for the reason given there. """ - if not archive.file_path or not archive.file_path.strip(): - logger.error( - f"SECURITY: Refusing to delete files for archive {archive.id} - " - f"file_path is empty or invalid: '{archive.file_path}'" - ) - return None + candidates: list[Path] = [] + # Only when there is a path to derive it from. Without one, + # ``resolve_archive_dir`` returns the id-named directory, which must not + # be removed wholesale -- see _purge_id_named_dir. + if archive.file_path and archive.file_path.strip(): + candidates.append(resolve_archive_dir(archive)) + candidates.append(settings.archive_dir / "no_source" / str(archive.id)) - file_path = settings.base_dir / archive.file_path - if not file_path.exists(): - return None + resolved: list[Path] = [] + for candidate in candidates: + if candidate in resolved or not candidate.is_dir(): + continue + try: + relative_path = candidate.resolve().relative_to(settings.archive_dir.resolve()) + except ValueError: + # A genuine guard trip, unlike the empty ``file_path`` this used + # to shout about: the row points somewhere outside the archive + # tree, which only a corrupted import or hand-edited SQL can do. + logger.error( + f"SECURITY: Refusing to delete archive {archive.id} - " + f"path {candidate} is outside archive directory {settings.archive_dir}" + ) + continue + # Two deep, not one. An archive directory has been + # ``//_/`` since the first + # commit, so nothing legitimate sits one level down -- but the + # per-printer folder does, and it holds every print that printer + # ever made. Under the old ``< 1`` a row whose file_path had lost a + # path component took the whole folder with it. + if len(relative_path.parts) < 2: + logger.error( + f"SECURITY: Refusing to delete archive {archive.id} - " + f"path {candidate} is not deep enough inside archive directory" + ) + continue + resolved.append(candidate) + return resolved - archive_dir = file_path.parent + def _purge_id_named_dir(self, archive_id: int, recorded_paths: tuple[str | None, ...]) -> None: + """Remove one archive's own files from ``/``, carefully. + + Takes the recorded paths rather than the row because ``delete_archive`` + removes the row before it touches the disk, deliberately: a failed + commit must leave the files alone. Reading ``archive.timelapse_path`` + off a deleted instance afterwards would raise or silently refresh. + + That directory is where an archive with no 3MF keeps its timelapse and + its finish photos (:func:`resolve_archive_dir`). It is emphatically NOT + an ``rmtree`` target, because it shares a namespace with the per-printer + folders: a normal archive lives at + ``//_/``, so ``archive/1`` is + printer 1's folder *and* the directory the helper hands archive id 1. + Archive ids and printer ids are both small integers from unrelated + sequences, so on any install the first few archives collide with the + printers. Removing the directory would take every print that printer + ever made — measured on a scratch tree before this guard existed. + + So nothing is removed that has not been identified as this archive's. + ``photos`` is a fixed subdirectory name and an archive directory is + always ``_``, so the two cannot be confused; the video + is removed by the path the row itself records. The directory then goes + only if that left it empty, which a printer folder holding prints never + will. Anything unrecognised keeps it alive and is leaked rather than + guessed at — the safe direction for a recursive delete. + """ + directory = settings.archive_dir / str(archive_id) + if not directory.is_dir(): + return try: - relative_path = archive_dir.resolve().relative_to(settings.archive_dir.resolve()) + relative_path = directory.resolve().relative_to(settings.archive_dir.resolve()) except ValueError: - logger.error( - f"SECURITY: Refusing to delete archive {archive.id} - " - f"path {archive_dir} is outside archive directory {settings.archive_dir}" - ) - return None - if len(relative_path.parts) < 1: - logger.error( - f"SECURITY: Refusing to delete archive {archive.id} - " - f"path {archive_dir} is not deep enough inside archive directory" - ) - return None - return archive_dir + return + if len(relative_path.parts) != 1: + return + + shutil.rmtree(directory / "photos", ignore_errors=True) # SEC-PATH-OK: constant subdirectory + for recorded in recorded_paths: + if not recorded: + continue + try: + # Two checks, not one. safe_join_under rejects the absolute and + # ``..`` shapes and proves the result is inside the data + # directory; assert_under then narrows it to *this* archive's + # own directory, because a row whose timelapse_path names + # another archive's file must not take it with this delete. + # The column is written by Bambuddy from a filename the printer + # supplied over FTP, so it is not a trusted constant. + candidate = safe_join_under(settings.base_dir, recorded, http=False) + assert_under(directory, candidate, http=False) + except PathTraversalError: + continue + if candidate.is_file(): + candidate.unlink(missing_ok=True) + + try: + directory.rmdir() + except OSError: + # Not empty (a printer folder, or a file this archive did not + # record) or already gone. Both are fine: the point of rmdir over + # rmtree is that it cannot take anything with it. + pass async def delete_archive(self, archive_id: int) -> bool: """Delete an archive and its files.""" @@ -1729,46 +1818,12 @@ class ArchiveService: if not archive: return False - # Resolve the directory to delete BEFORE committing the DB change - dir_to_delete: Path | None = None - - if archive.file_path and archive.file_path.strip(): - file_path = settings.base_dir / archive.file_path - if file_path.exists(): - archive_dir = file_path.parent - - # Safety check 1: archive_dir must be inside archive_dir - try: - archive_dir.resolve().relative_to(settings.archive_dir.resolve()) - except ValueError: - logger.error( - f"SECURITY: Refusing to delete archive {archive_id} - " - f"path {archive_dir} is outside archive directory {settings.archive_dir}" - ) - await self.db.delete(archive) - await self.db.commit() - return True - - # Safety check 2: archive_dir must be at least 1 level deep inside archive_dir - try: - relative_path = archive_dir.resolve().relative_to(settings.archive_dir.resolve()) - if len(relative_path.parts) < 1: - logger.error( - f"SECURITY: Refusing to delete archive {archive_id} - " - f"path {archive_dir} is not deep enough inside archive directory" - ) - await self.db.delete(archive) - await self.db.commit() - return True - except ValueError: - pass # Already handled above - - dir_to_delete = archive_dir - else: - logger.error( - f"SECURITY: Refusing to delete files for archive {archive_id} - " - f"file_path is empty or invalid: '{archive.file_path}'" - ) + # Resolved BEFORE committing the DB change, since the row is what says + # where the files are. Shared with soft-delete rather than repeated + # here: this was a second copy of the same checks and it had already + # diverged from the one it was extracted from. + dirs_to_delete = self._resolve_archive_dirs_for_delete(archive) + recorded_paths = (archive.timelapse_path, archive.thumbnail_path) # NULL stale thumbnail_path on linked PrintLogEntries before the FK # SET-NULL cascade fires. The on-disk file is about to be removed by @@ -1783,8 +1838,9 @@ class ArchiveService: await self.db.commit() # Only delete files AFTER the DB commit succeeds to avoid orphaned records - if dir_to_delete: - shutil.rmtree(dir_to_delete, ignore_errors=True) + for directory in dirs_to_delete: + shutil.rmtree(directory, ignore_errors=True) + self._purge_id_named_dir(archive_id, recorded_paths) return True @@ -1922,7 +1978,7 @@ async def _convert_timelapse_to_mp4(archive_id: int, source_path: Path) -> None: logger.warning( "Timelapse conversion failed for archive %s: %s", archive_id, - stderr.decode()[-500:], + summarize_ffmpeg_stderr(stderr) or NO_FFMPEG_OUTPUT, ) if mp4_path.exists(): mp4_path.unlink() diff --git a/backend/app/services/camera.py b/backend/app/services/camera.py index ba4788d6e..ce7b0a970 100644 --- a/backend/app/services/camera.py +++ b/backend/app/services/camera.py @@ -17,7 +17,7 @@ import uuid from datetime import datetime from pathlib import Path -from backend.app.core.logging_filters import redact_url_credentials +from backend.app.utils.ffmpeg_output import NO_FFMPEG_OUTPUT, summarize_ffmpeg_stderr logger = logging.getLogger(__name__) @@ -231,7 +231,8 @@ async def create_tls_proxy(target_host: str, target_port: int) -> tuple[int, "as rewrites ``127.0.0.1:`` → ``:`` in client→server data so the printer recognises the stream path. - Returns ``(local_port, server)``. Caller must close the server when done. + Returns ``(local_port, server)``. Caller must close it with + :func:`close_tls_proxy` when done. """ ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) ssl_ctx.check_hostname = False @@ -240,7 +241,20 @@ async def create_tls_proxy(target_host: str, target_port: int) -> tuple[int, "as # Filled in after the server socket is created (handler only runs after). _local_port: list[int] = [0] + # Strong references to the in-flight connection handlers (#2968). + # ``asyncio.start_server`` wraps the callback in a task and keeps only a + # weak reference to it, so a handler still awaiting its two forwarders can + # be garbage-collected out from under itself — which is asyncio's + # "Task was destroyed but it is pending!", logged at ERROR with a traceback + # pointing here and no indication that it is a teardown race rather than a + # camera fault. Holding the set also gives close_tls_proxy something to + # cancel, so shutdown stops depending on ffmpeg having dropped its end. + handlers: set[asyncio.Task] = set() + async def _handle(client_reader: asyncio.StreamReader, client_writer: asyncio.StreamWriter): + current = asyncio.current_task() + if current is not None: + handlers.add(current) tls_writer = None try: tls_reader, tls_writer = await asyncio.wait_for( @@ -305,7 +319,19 @@ async def create_tls_proxy(target_host: str, target_port: int) -> tuple[int, "as ) except (ConnectionError, OSError, TimeoutError) as e: logger.debug("TLS proxy connection to %s:%s failed: %s", target_host, target_port, e) + except asyncio.CancelledError: + # close_tls_proxy cancelling us at shutdown, which is the only thing + # that cancels this task. Swallowing a cancellation is normally + # wrong because it hides the request from whoever made it; here we + # *are* whoever made it, the cleanup it exists to trigger is in the + # finally below, and nothing awaits this task's result. Asyncio's + # own done-callback for a connection handler treats a cancelled task + # differently from a completed one, so ending in the ordinary way + # keeps the teardown on one path across Python versions. + pass finally: + if current is not None: + handlers.discard(current) for w in (client_writer, tls_writer): if w and not w.is_closing(): try: @@ -315,10 +341,36 @@ async def create_tls_proxy(target_host: str, target_port: int) -> tuple[int, "as server = await asyncio.start_server(_handle, "127.0.0.1", 0) _local_port[0] = server.sockets[0].getsockname()[1] + server._bambuddy_proxy_handlers = handlers # type: ignore[attr-defined] logger.debug("TLS proxy for %s:%s listening on 127.0.0.1:%s", target_host, target_port, _local_port[0]) return _local_port[0], server +async def close_tls_proxy(server: "asyncio.Server") -> None: + """Shut a :func:`create_tls_proxy` server down without leaving tasks behind. + + ``server.close()`` stops the listener but leaves established connections + running, and ``wait_closed()`` is only as deterministic as the peer: it + waits for the handlers, and a handler waits for ffmpeg to drop its end of + the socket. By the time this is called ffmpeg has already been reaped, so + the connection is dead weight — cancelling it is both correct and the only + way to guarantee no handler outlives the server that owns it. + + ``Server.close_clients()`` would do this natively, but it landed in Python + 3.13 and Bambuddy supports 3.10, so the handler set is tracked by hand. + + Safe to call on a plain ``asyncio.Server`` from anywhere else: without the + attribute it degrades to the close/wait it replaces. + """ + handlers: set[asyncio.Task] = getattr(server, "_bambuddy_proxy_handlers", set()) + server.close() + for task in list(handlers): + task.cancel() + if handlers: + await asyncio.gather(*list(handlers), return_exceptions=True) + await server.wait_closed() + + def is_chamber_image_model(model: str | None) -> bool: """Check if printer uses chamber image protocol instead of RTSP. @@ -692,8 +744,7 @@ async def _capture_camera_frame_bytes_uncoalesced( ffmpeg = get_ffmpeg_path() if not ffmpeg: - proxy_server.close() - await proxy_server.wait_closed() + await close_tls_proxy(proxy_server) logger.error("ffmpeg not found for camera frame capture") return None @@ -740,9 +791,11 @@ async def _capture_camera_frame_bytes_uncoalesced( logger.info("Successfully captured camera frame bytes: %s bytes", len(stdout)) return stdout else: - # ffmpeg echoes the RTSP input URL, which carries the access code. - stderr_text = redact_url_credentials(stderr.decode()) if stderr else "Unknown error" - logger.error("ffmpeg frame bytes capture failed (code %s): %s", process.returncode, stderr_text[:200]) + # The summariser drops ffmpeg's banner and masks the access code + # the RTSP input URL carries; without it this line was 200 + # characters of build configuration (#2968). + stderr_text = summarize_ffmpeg_stderr(stderr) or NO_FFMPEG_OUTPUT + logger.error("ffmpeg frame bytes capture failed (code %s): %s", process.returncode, stderr_text) return None except FileNotFoundError: @@ -754,8 +807,7 @@ async def _capture_camera_frame_bytes_uncoalesced( finally: if process is not None: _active_capture_pids.discard(process.pid) - proxy_server.close() - await proxy_server.wait_closed() + await close_tls_proxy(proxy_server) async def extract_video_last_frame(video_path: Path, output_path: Path) -> bool: @@ -815,7 +867,7 @@ async def extract_video_last_frame(video_path: Path, output_path: Path) -> bool: logger.warning( "ffmpeg failed extracting last frame from %s: %s", video_path, - stderr.decode(errors="replace")[:500], + summarize_ffmpeg_stderr(stderr) or NO_FFMPEG_OUTPUT, ) return False if not output_path.exists() or output_path.stat().st_size == 0: diff --git a/backend/app/services/external_camera.py b/backend/app/services/external_camera.py index cf710951f..1fd561731 100644 --- a/backend/app/services/external_camera.py +++ b/backend/app/services/external_camera.py @@ -21,6 +21,7 @@ from urllib.parse import urlparse import aiohttp from backend.app.core.logging_filters import redact_url_credentials +from backend.app.utils.ffmpeg_output import NO_FFMPEG_OUTPUT, summarize_ffmpeg_stderr logger = logging.getLogger(__name__) @@ -517,7 +518,7 @@ async def _capture_usb_frame(device: str, timeout: int) -> bytes | None: stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout) if process.returncode != 0: - logger.error("ffmpeg USB capture failed: %s", stderr.decode()[:200]) + logger.error("ffmpeg USB capture failed: %s", summarize_ffmpeg_stderr(stderr) or NO_FFMPEG_OUTPUT) return None if not stdout or len(stdout) < 100: @@ -701,8 +702,8 @@ async def _capture_rtsp_frame(url: str, timeout: int) -> bytes | None: ) if process.returncode != 0: - # ffmpeg echoes the RTSP input URL, which carries the camera password. - logger.error("ffmpeg RTSP capture failed: %s", redact_url_credentials(stderr.decode())[:200]) + # The summariser masks the camera password the input URL carries. + logger.error("ffmpeg RTSP capture failed: %s", summarize_ffmpeg_stderr(stderr) or NO_FFMPEG_OUTPUT) return None if not stdout or len(stdout) < 100: @@ -1149,8 +1150,10 @@ async def _stream_rtsp( await asyncio.sleep(0.1) if process.returncode is not None: stderr = await process.stderr.read() - # ffmpeg echoes the RTSP input URL, which carries the camera password. - logger.error("ffmpeg RTSP stream failed immediately: %s", redact_url_credentials(stderr.decode())[:300]) + # The summariser masks the camera password the input URL carries. + logger.error( + "ffmpeg RTSP stream failed immediately: %s", summarize_ffmpeg_stderr(stderr) or NO_FFMPEG_OUTPUT + ) return buffer = b"" @@ -1262,7 +1265,9 @@ async def _stream_usb( await asyncio.sleep(0.5) if process.returncode is not None: stderr = await process.stderr.read() - logger.error("ffmpeg USB stream failed immediately: %s", stderr.decode()[:300]) + logger.error( + "ffmpeg USB stream failed immediately: %s", summarize_ffmpeg_stderr(stderr) or NO_FFMPEG_OUTPUT + ) return buffer = b"" diff --git a/backend/app/services/layer_timelapse.py b/backend/app/services/layer_timelapse.py index 72e0f2c2d..57f973db2 100644 --- a/backend/app/services/layer_timelapse.py +++ b/backend/app/services/layer_timelapse.py @@ -14,6 +14,7 @@ from pathlib import Path from backend.app.core.config import settings from backend.app.services.camera import apply_camera_rotation from backend.app.services.external_camera import capture_frame +from backend.app.utils.ffmpeg_output import NO_FFMPEG_OUTPUT, summarize_ffmpeg_stderr logger = logging.getLogger(__name__) @@ -189,7 +190,7 @@ class TimelapseSession: stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=300) if process.returncode != 0: - logger.error("ffmpeg timelapse stitch failed: %s", stderr.decode()[:500]) + logger.error("ffmpeg timelapse stitch failed: %s", summarize_ffmpeg_stderr(stderr) or NO_FFMPEG_OUTPUT) return False logger.info("Created timelapse video: %s (%s frames)", output_path, self.frame_count) diff --git a/backend/app/services/timelapse_processor.py b/backend/app/services/timelapse_processor.py index 5fae64ab2..ff1769112 100644 --- a/backend/app/services/timelapse_processor.py +++ b/backend/app/services/timelapse_processor.py @@ -7,6 +7,7 @@ import tempfile from pathlib import Path from backend.app.services.camera import get_ffmpeg_path +from backend.app.utils.ffmpeg_output import NO_FFMPEG_OUTPUT, summarize_ffmpeg_stderr logger = logging.getLogger(__name__) @@ -43,8 +44,12 @@ class TimelapseProcessor: stdout, stderr = await process.communicate() if process.returncode != 0: - logger.error("ffprobe failed: %s", stderr.decode()) - raise RuntimeError(f"ffprobe failed: {stderr.decode()}") + # Summarised once and used for both: the raise carried a second, + # bare ``stderr.decode()`` that could itself raise UnicodeDecodeError + # on the bytes ffprobe copies out of a broken file (#2968). + detail = summarize_ffmpeg_stderr(stderr) or NO_FFMPEG_OUTPUT + logger.error("ffprobe failed: %s", detail) + raise RuntimeError(f"ffprobe failed: {detail}") data = json.loads(stdout.decode()) video_stream = next( @@ -230,7 +235,7 @@ class TimelapseProcessor: _, stderr = await process.communicate() if process.returncode != 0: - logger.error("FFmpeg processing failed: %s", stderr.decode()) + logger.error("FFmpeg processing failed: %s", summarize_ffmpeg_stderr(stderr) or NO_FFMPEG_OUTPUT) return False return output_path.exists() diff --git a/backend/app/utils/ffmpeg_output.py b/backend/app/utils/ffmpeg_output.py new file mode 100644 index 000000000..47dffab3e --- /dev/null +++ b/backend/app/utils/ffmpeg_output.py @@ -0,0 +1,102 @@ +"""Turning an ffmpeg subprocess's stderr into a log line worth reading (#2968). + +ffmpeg opens every run with ~20 lines of version, build and library banner and +prints its diagnosis *last*. Truncating that from the front -- ``stderr[:200]``, +which is what most call sites did -- keeps the banner and throws the diagnosis +away. A reporter's H2D produced twelve of these, and every one of them read + + ffmpeg frame bytes capture failed (code 183): ffmpeg version 7.1.4-0+deb13u1 + Copyright (c) 2000-2026 the FFmpeg developers built with gcc 14 (Debian + 14.2.0-19) configuration: --prefix=/usr --extra-version=0+deb13u1 --toolch + +-- 200 characters that are identical on every install and say nothing about why +the capture failed. The exit code was the only usable byte in the whole line. + +The banner-stripping summariser this module holds was written for #925 and +lived as a private helper in ``api/routes/camera.py``, where the streaming +endpoint used it. Ten other places log ffmpeg or ffprobe stderr -- snapshot +capture, last-frame extraction, the layer-timelapse stitch, the archive's MP4 +conversion, external USB and RTSP capture and streaming, and timelapse +post-processing. Seven of them truncated from the front, two logged the whole +banner, and one already kept the tail. They all come here now, so they cannot +drift again. + +Redaction is part of the summary rather than each caller's job. ffmpeg echoes +its input URL back in the ``Input #0`` line, so a camera password or a printer +access code reaches stderr on any failure; seven of those ten logged it +unmasked. A helper that redacts is one that cannot be called wrong. + +Kept as a leaf module -- stdlib plus :mod:`core.logging_filters`, which is +itself stdlib-only -- so the services and the route can all reach it without +pulling a startup graph behind them. +""" + +from __future__ import annotations + +from backend.app.core.logging_filters import redact_url_credentials + +# What ffmpeg prints before it has anything to say. Every line of the banner is +# either the version line or an indented continuation, and a real diagnostic is +# never indented this way, so the match is on the exact prefixes rather than on +# indentation alone -- `` Duration: ...`` and `` Stream #0:0 ...`` are +# indented too and are worth keeping. +_BANNER_PREFIXES = ( + "ffmpeg version ", + "ffprobe version ", + " built with ", + " configuration:", + " libavutil ", + " libavcodec ", + " libavformat ", + " libavdevice ", + " libavfilter ", + " libswscale ", + " libswresample ", + " libpostproc ", +) + +# How much of the tail to keep. ffmpeg's diagnosis is the last thing it writes, +# and ten lines is enough to carry the error plus the input analysis that +# explains it without letting a chatty decoder rotate the log file. +_MAX_LINES = 10 + +# And a ceiling on the whole thing. Ten lines is only a bound on the log record +# if the lines are a sane length, and ffmpeg quotes what the peer sent it back +# at us -- a printer's RTSP response is not something Bambuddy controls. Well +# above any real diagnosis, so this only ever trims a line that was already not +# going to be read. +_MAX_CHARACTERS = 2000 + +# What to log when the summary is empty. A failure whose stderr held nothing but +# the banner still deserves a line saying so -- ``failed: `` with an empty tail +# reads like a truncation bug rather than a printer that closed the connection. +NO_FFMPEG_OUTPUT = "no diagnostic output" + + +def summarize_ffmpeg_stderr(text: str | bytes | None) -> str: + """Strip ffmpeg's boilerplate banner and keep the last lines that matter. + + Accepts raw ``bytes`` as well as ``str`` and decodes with ``errors= + "replace"``: ffmpeg copies fragments of the stream into its error messages, + so a bare ``.decode()`` at the call site can raise ``UnicodeDecodeError`` + while reporting an unrelated failure. Losing the diagnosis to a second + exception is the one outcome worse than logging the banner. + + Returns ``""`` when there is nothing left after the banner, which is the + signal the streaming endpoint uses to stay quiet. One-shot callers that log + unconditionally should fall back to :data:`NO_FFMPEG_OUTPUT`. + """ + if not text: + return "" + if isinstance(text, (bytes, bytearray)): + text = text.decode(errors="replace") + # Redaction runs on the whole string before anything is dropped: a + # credentialed URL that straddles the cut would otherwise leave its tail in + # the log with no ``@`` left for the pattern to anchor on. + text = redact_url_credentials(text) or "" + meaningful = [line for line in text.splitlines() if line.strip() and not line.startswith(_BANNER_PREFIXES)] + summary = "\n".join(meaningful[-_MAX_LINES:]) + if len(summary) > _MAX_CHARACTERS: + # From the end, for the same reason the whole module exists. + summary = "..." + summary[-_MAX_CHARACTERS:] + return summary diff --git a/backend/tests/unit/test_archive_delete_no_3mf_dirs_2968.py b/backend/tests/unit/test_archive_delete_no_3mf_dirs_2968.py new file mode 100644 index 000000000..8a4ae0d16 --- /dev/null +++ b/backend/tests/unit/test_archive_delete_no_3mf_dirs_2968.py @@ -0,0 +1,336 @@ +"""Deleting a no-3MF archive used to leave every file it owned on disk (#2968). + +An archive created without a 3MF carries ``file_path == ""``. Both delete paths +derived the directory to remove from that path, found nothing, and logged + + SECURITY: Refusing to delete files for archive 7 - file_path is empty or invalid: '' + +at ERROR. That was accurate once, when such an archive really was an empty row. +It stopped being accurate when a no-3MF archive gained places to put things: +``//`` for its timelapse and finish photos (the shared helper +in ``utils.archive_paths``, #1820), and ``archive/no_source//`` for a source +3MF uploaded onto it afterwards (#1531). Neither was ever removed, so deleting +the archive freed the row and kept the video -- on an H2-series or P2S printer, +where a print sent from Bambu Studio always archives without a 3MF, that is most +of the library. + +Reported by @ceasley, whose log carries three of those ERROR lines from a single +afternoon of deleting no-3MF archives. + +**The trap this file exists to hold shut.** ``/`` shares a +namespace with the per-printer folders: a normal archive lives at +``//_/``, so ``archive/1`` is printer +1's folder *and* the directory ``resolve_archive_dir`` hands archive id 1. +Archive ids and printer ids are small integers from unrelated sequences, so on +every install the first few archives collide with the printers. An ``rmtree`` +there deletes every print that printer ever made. The first draft of this fix +did exactly that, and passed a full suite before the collision was found by +reading the archive layout rather than the tests. Nothing in the delete path may +remove a directory one level under ``archive_dir``. +""" + +from __future__ import annotations + +import logging +from pathlib import Path + +import pytest + +from backend.app.models.archive import PrintArchive +from backend.app.models.printer import Printer +from backend.app.services.archive import ArchiveService + + +@pytest.fixture +def archive_root(tmp_path, monkeypatch): + """A data directory both settings bindings agree on. + + ``services.archive`` and ``utils.archive_paths`` each hold their own + module-level ``settings``; patching one and not the other is how an earlier + change to this code wrote outside tmp_path and littered a working tree. + """ + from backend.app.services import archive as archive_module + from backend.app.utils import archive_paths + + for module in (archive_module, archive_paths): + monkeypatch.setattr(module.settings, "base_dir", tmp_path, raising=False) + monkeypatch.setattr(module.settings, "archive_dir", tmp_path / "archive", raising=False) + (tmp_path / "archive").mkdir(parents=True, exist_ok=True) + return tmp_path + + +def _service() -> ArchiveService: + """The resolvers need no database; ``None`` keeps the test to one subject.""" + return ArchiveService(None) # type: ignore[arg-type] + + +def _archive(archive_id: int, file_path: str = "") -> PrintArchive: + return PrintArchive(id=archive_id, file_path=file_path) + + +def _printer_folder_with_a_print(archive_root, printer_id: int) -> Path: + """A printer folder laid out exactly as ``_create_archive`` builds it.""" + directory = archive_root / "archive" / str(printer_id) / "20260828_193000_Benchy" + directory.mkdir(parents=True) + (directory / "Benchy.3mf").write_bytes(b"a real archived print") + return directory + + +class TestItCannotDeleteAPrinterFolder: + """The collision above. Every one of these would have destroyed real data.""" + + def test_a_no_3mf_archive_whose_id_matches_a_printer(self, archive_root): + real = _printer_folder_with_a_print(archive_root, 1) + + assert _service()._resolve_archive_dirs_for_delete(_archive(1)) == [] + assert real.exists() + + def test_and_the_purge_leaves_it_standing(self, archive_root): + """The id-named directory is cleaned in place rather than removed, so + the purge has to survive the folder being somebody else's.""" + real = _printer_folder_with_a_print(archive_root, 1) + + _service()._purge_id_named_dir(1, (None, None)) + + assert (real / "Benchy.3mf").exists() + assert (archive_root / "archive" / "1").is_dir() + + @pytest.mark.asyncio + async def test_end_to_end_through_delete_archive(self, archive_root, db_session): + """Not just the resolver: the whole delete, against real rows.""" + printer = Printer(name="H2D", ip_address="192.0.2.9", access_code="12345678", serial_number="COLLIDE") + db_session.add(printer) + await db_session.flush() + + real = archive_root / "archive" / str(printer.id) / "20260828_193000_Benchy" + real.mkdir(parents=True) + (real / "Benchy.3mf").write_bytes(b"a real archived print") + + archive = PrintArchive( + printer_id=printer.id, filename="Cleaner_PRO", file_path="", file_size=0, status="completed" + ) + db_session.add(archive) + await db_session.commit() + if archive.id != printer.id: + pytest.skip(f"ids did not collide in this fixture (archive {archive.id}, printer {printer.id})") + + assert await ArchiveService(db_session).delete_archive(archive.id) is True + assert (real / "Benchy.3mf").exists(), "deleting the archive took the printer's whole folder" + + def test_a_corrupted_row_pointing_at_a_printer_folder(self, archive_root, caplog): + """``archive/1/Benchy.3mf`` -- a file_path that lost a path component. + Its parent is the printer folder. Refused on depth, and said out loud.""" + real = _printer_folder_with_a_print(archive_root, 1) + (archive_root / "archive" / "1" / "Benchy.3mf").write_bytes(b"x") + + with caplog.at_level(logging.ERROR): + dirs = _service()._resolve_archive_dirs_for_delete(_archive(7, "archive/1/Benchy.3mf")) + + assert dirs == [] + assert (real / "Benchy.3mf").exists() + assert any("not deep enough" in r.getMessage() for r in caplog.records) + + def test_the_archive_root_itself_is_refused(self, archive_root, caplog): + (archive_root / "archive" / "Benchy.3mf").write_bytes(b"x") + + with caplog.at_level(logging.ERROR): + dirs = _service()._resolve_archive_dirs_for_delete(_archive(7, "archive/Benchy.3mf")) + + assert dirs == [] + assert (archive_root / "archive").exists() + + def test_nothing_it_returns_is_ever_one_level_deep(self, archive_root): + """The invariant, stated once against every shape a row can take.""" + _printer_folder_with_a_print(archive_root, 1) + (archive_root / "archive" / "no_source" / "1").mkdir(parents=True) + + for file_path in ("", "archive/1/x.3mf", "archive/x.3mf", "../escape/x.3mf", "/absolute/x.3mf"): + for directory in _service()._resolve_archive_dirs_for_delete(_archive(1, file_path)): + relative = Path(directory).resolve().relative_to((archive_root / "archive").resolve()) + assert len(relative.parts) >= 2, f"{file_path} resolved to {relative}" + + +class TestANo3mfArchivesFiles: + def test_its_timelapse_and_photos_are_removed(self, archive_root): + """The files it really owns, taken by name rather than by rmtree.""" + directory = archive_root / "archive" / "7" + (directory / "photos").mkdir(parents=True) + (directory / "photos" / "finish.jpg").write_bytes(b"p") + (directory / "video_2026-08-27_08-35-49.mp4").write_bytes(b"v") + + _service()._purge_id_named_dir(7, ("archive/7/video_2026-08-27_08-35-49.mp4", None)) + + assert not directory.exists() + + def test_its_uploaded_source_directory_is_removed(self, archive_root): + """``archive/no_source//`` is two levels down and nested under a + name no printer id can take, so it is safe to remove whole.""" + source_dir = archive_root / "archive" / "no_source" / "7" + source_dir.mkdir(parents=True) + (source_dir / "Cleaner_PRO.3mf").write_bytes(b"x") + + assert _service()._resolve_archive_dirs_for_delete(_archive(7)) == [source_dir] + + def test_no_error_is_logged_for_an_ordinary_empty_path(self, archive_root, caplog): + """It is the normal shape of a Studio-sent print, not a security event. + Three of these were the only ERRORs in the reporter's whole log.""" + with caplog.at_level(logging.ERROR): + _service()._resolve_archive_dirs_for_delete(_archive(7)) + _service()._purge_id_named_dir(7, (None, None)) + + assert not [r for r in caplog.records if "SECURITY" in r.getMessage()] + + def test_an_unrecognised_file_keeps_the_directory(self, archive_root): + """Leaking beats guessing: something this archive did not record stops + the rmdir, and nothing is removed on a hunch.""" + directory = archive_root / "archive" / "7" + directory.mkdir(parents=True) + (directory / "something_else.bin").write_bytes(b"?") + + _service()._purge_id_named_dir(7, (None, None)) + + assert (directory / "something_else.bin").exists() + + def test_a_recorded_path_outside_the_directory_is_not_followed(self, archive_root): + """A row whose timelapse_path names another archive's file must not + take it with this delete.""" + elsewhere = archive_root / "archive" / "1" / "20260828_193000_Benchy" + elsewhere.mkdir(parents=True) + (elsewhere / "video.mp4").write_bytes(b"v") + (archive_root / "archive" / "7").mkdir(parents=True) + + _service()._purge_id_named_dir(7, ("archive/1/20260828_193000_Benchy/video.mp4", None)) + + assert (elsewhere / "video.mp4").exists() + + def test_the_shared_legacy_photo_directory_is_never_touched(self, archive_root): + """``/photos`` was written to by *every* no-3MF archive at + once. Removing it on one delete would take the others' photos too.""" + shared = archive_root / "photos" + shared.mkdir(parents=True) + (shared / "finish_7.jpg").write_bytes(b"x") + + assert shared not in _service()._resolve_archive_dirs_for_delete(_archive(7)) + _service()._purge_id_named_dir(7, (None, None)) + + assert (shared / "finish_7.jpg").exists() + + +class TestAnArchiveWithA3mf: + def test_its_own_directory_is_removed(self, archive_root): + archive_dir = archive_root / "archive" / "1" / "20260828_193000_Benchy" + archive_dir.mkdir(parents=True) + (archive_dir / "Benchy.3mf").write_bytes(b"x") + + dirs = _service()._resolve_archive_dirs_for_delete(_archive(7, "archive/1/20260828_193000_Benchy/Benchy.3mf")) + + assert dirs == [archive_dir] + + def test_a_missing_3mf_no_longer_strands_the_directory(self, archive_root): + """The old code keyed on the 3MF still being there, so an archive whose + 3MF had gone kept its thumbnail and timelapse forever.""" + archive_dir = archive_root / "archive" / "1" / "20260828_193000_Benchy" + archive_dir.mkdir(parents=True) + (archive_dir / "thumbnail.png").write_bytes(b"x") + + dirs = _service()._resolve_archive_dirs_for_delete(_archive(7, "archive/1/20260828_193000_Benchy/Benchy.3mf")) + + assert dirs == [archive_dir] + + def test_a_directory_that_does_not_exist_is_not_offered(self, archive_root): + assert _service()._resolve_archive_dirs_for_delete(_archive(7, "archive/1/gone/Benchy.3mf")) == [] + + def test_a_file_where_a_directory_should_be_is_not_offered(self, archive_root): + """``is_dir()`` rather than ``exists()``: rmtree on a file raises, and + the delete would take the whole request down with it.""" + (archive_root / "archive" / "no_source").mkdir(parents=True) + (archive_root / "archive" / "no_source" / "7").write_bytes(b"not a directory") + + assert _service()._resolve_archive_dirs_for_delete(_archive(7)) == [] + + def test_a_path_outside_the_archive_tree_is_refused_and_logged(self, archive_root, caplog): + """Only a corrupted import or hand-edited SQL produces this.""" + outside = archive_root / "elsewhere" / "deep" + outside.mkdir(parents=True) + (outside / "Benchy.3mf").write_bytes(b"x") + + with caplog.at_level(logging.ERROR): + dirs = _service()._resolve_archive_dirs_for_delete(_archive(7, "elsewhere/deep/Benchy.3mf")) + + assert dirs == [] + assert (outside / "Benchy.3mf").exists() + assert any("outside archive directory" in r.getMessage() for r in caplog.records) + + +class TestBothDeletePathsUseIt: + """Hard delete kept its own copy of these rules and had already diverged + from the helper whose docstring said it was extracted to prevent that.""" + + async def _no_3mf_archive_with_files(self, archive_root, db_session, serial: str, ip: str): + printer = Printer(name="H2D", ip_address=ip, access_code="12345678", serial_number=serial) + db_session.add(printer) + await db_session.flush() + archive = PrintArchive( + printer_id=printer.id, filename="Cleaner_PRO", file_path="", file_size=0, status="completed" + ) + db_session.add(archive) + await db_session.commit() + + video_dir = archive_root / "archive" / str(archive.id) + video_dir.mkdir(parents=True, exist_ok=True) + (video_dir / "video.mp4").write_bytes(b"v") + archive.timelapse_path = f"archive/{archive.id}/video.mp4" + source_dir = archive_root / "archive" / "no_source" / str(archive.id) + source_dir.mkdir(parents=True, exist_ok=True) + (source_dir / "Cleaner_PRO.3mf").write_bytes(b"x") + await db_session.commit() + return archive, video_dir, source_dir + + @pytest.mark.asyncio + async def test_soft_delete_removes_the_video_and_the_upload(self, archive_root, db_session): + archive, video_dir, source_dir = await self._no_3mf_archive_with_files( + archive_root, db_session, "SOFT1", "192.0.2.1" + ) + + assert await ArchiveService(db_session).soft_delete_archive(archive.id) is True + + assert not video_dir.exists() + assert not source_dir.exists() + + @pytest.mark.asyncio + async def test_hard_delete_removes_the_video_and_the_upload(self, archive_root, db_session): + archive, video_dir, source_dir = await self._no_3mf_archive_with_files( + archive_root, db_session, "HARD1", "192.0.2.2" + ) + + assert await ArchiveService(db_session).delete_archive(archive.id) is True + + assert not video_dir.exists() + assert not source_dir.exists() + + @pytest.mark.asyncio + async def test_hard_delete_still_removes_the_row_when_a_guard_trips(self, archive_root, db_session): + """A row pointing outside the tree must still be deletable, or the + archive becomes permanently stuck in the UI.""" + printer = Printer(name="H2D", ip_address="192.0.2.3", access_code="12345678", serial_number="GUARD1") + db_session.add(printer) + await db_session.flush() + + outside = archive_root / "elsewhere" / "deep" + outside.mkdir(parents=True) + (outside / "Benchy.3mf").write_bytes(b"x") + + archive = PrintArchive( + printer_id=printer.id, + filename="Benchy", + file_path="elsewhere/deep/Benchy.3mf", + file_size=0, + status="completed", + ) + db_session.add(archive) + await db_session.commit() + archive_id = archive.id + + assert await ArchiveService(db_session).delete_archive(archive_id) is True + assert await ArchiveService(db_session).get_archive(archive_id) is None + assert (outside / "Benchy.3mf").exists() diff --git a/backend/tests/unit/test_ffmpeg_output_summary.py b/backend/tests/unit/test_ffmpeg_output_summary.py new file mode 100644 index 000000000..0b145102d --- /dev/null +++ b/backend/tests/unit/test_ffmpeg_output_summary.py @@ -0,0 +1,185 @@ +"""ffmpeg's diagnosis survives the log line, and its banner does not (#2968). + +ffmpeg prints ~20 lines of version and build banner first and its actual error +last, so the ``stderr[:200]`` most call sites used kept the banner and dropped +the error. The reporter's H2D logged twelve capture failures that way; every +one of them was the same 200 characters of ``--prefix=/usr --extra-version=`` +and none of them said why the capture failed. + +#925 already solved this for the camera streaming endpoint. These tests cover +the shared module the other seven call sites now go through, and the two things +that were only ever true of the private copy: it takes bytes, and it masks +credentials for callers that never did. +""" + +import inspect + +import pytest + +from backend.app.utils.ffmpeg_output import NO_FFMPEG_OUTPUT, summarize_ffmpeg_stderr + +# Verbatim from the reporter's log, trimmed to the width the old truncation +# allowed through. The point of the fixture is that 200 characters of it carry +# no information at all. +_REAL_BANNER = """ffmpeg version 7.1.4-0+deb13u1 Copyright (c) 2000-2026 the FFmpeg developers + built with gcc 14 (Debian 14.2.0-19) + configuration: --prefix=/usr --extra-version=0+deb13u1 --toolchain=hardened --enable-gpl + libavutil 59. 39.100 / 59. 39.100 + libavcodec 61. 19.101 / 61. 19.101 + libavformat 61. 7.100 / 61. 7.100 + libavdevice 61. 3.100 / 61. 3.100 + libavfilter 10. 4.100 / 10. 4.100 + libswscale 8. 3.100 / 8. 3.100 + libswresample 5. 3.100 / 5. 3.100 + libpostproc 58. 3.100 / 58. 3.100 +""" + + +class TestTheDiagnosisSurvives: + def test_the_error_is_kept_and_the_banner_is_not(self): + """The whole point: the last line, not the first 200 characters.""" + stderr = _REAL_BANNER + "[rtsp @ 0x5f] method DESCRIBE failed: 401 Unauthorized\n" + + result = summarize_ffmpeg_stderr(stderr) + + assert "method DESCRIBE failed: 401 Unauthorized" in result + assert "ffmpeg version" not in result + assert "--prefix=/usr" not in result + + def test_the_old_truncation_would_have_kept_none_of_it(self): + """Guards the claim the fix rests on rather than asserting it in prose: + 200 characters from the front of a real failure is banner only.""" + stderr = _REAL_BANNER + "[rtsp @ 0x5f] method DESCRIBE failed: 401 Unauthorized\n" + + assert "DESCRIBE" not in stderr[:200] + + def test_input_analysis_is_kept(self): + """Indented, but not banner. ``Duration:`` and ``Stream #0:0`` explain + the error above them and are the reason the match is on exact prefixes + rather than on leading whitespace.""" + stderr = _REAL_BANNER + ( + "Input #0, rtsp, from 'rtsp://192.0.2.1:322/streaming/live/1':\n" + " Duration: N/A, start: 0.000000, bitrate: N/A\n" + " Stream #0:0: Video: h264, yuv420p, 1920x1080\n" + "Output file is empty, nothing was encoded\n" + ) + + result = summarize_ffmpeg_stderr(stderr) + + assert "Duration: N/A" in result + assert "Stream #0:0: Video: h264" in result + assert "Output file is empty" in result + + def test_only_the_last_lines_are_kept(self): + """A chatty decoder must not rotate the log file on one failure.""" + stderr = _REAL_BANNER + "\n".join(f"error line {i}" for i in range(40)) + + lines = summarize_ffmpeg_stderr(stderr).splitlines() + + assert len(lines) == 10 + assert lines[-1] == "error line 39" + + def test_a_banner_only_failure_says_so(self): + """Empty, so the caller substitutes a phrase. ``failed: `` with nothing + after it reads like a truncation bug rather than a silent printer.""" + assert summarize_ffmpeg_stderr(_REAL_BANNER) == "" + assert (summarize_ffmpeg_stderr(_REAL_BANNER) or NO_FFMPEG_OUTPUT) == NO_FFMPEG_OUTPUT + + +class TestWhatTheCallSitesUsedToGetWrong: + def test_bytes_are_accepted(self): + """Every call site held bytes and decoded them itself.""" + assert "Connection refused" in summarize_ffmpeg_stderr(b"rtsp://192.0.2.1: Connection refused\n") + + def test_undecodable_bytes_do_not_raise(self): + """ffmpeg copies stream fragments into its messages, so a bare + ``.decode()`` could raise UnicodeDecodeError while reporting an + unrelated failure -- losing the diagnosis to a second exception.""" + result = summarize_ffmpeg_stderr(b"\xff\xfe broken input\nInvalid data found\n") + + assert "Invalid data found" in result + + def test_the_access_code_is_masked(self): + """ffmpeg echoes its input URL back, and four of the call sites logged + it unmasked. The mask is part of the summary so it cannot be skipped.""" + stderr = b"Error opening input file rtsp://bblp:12345678@192.0.2.1:322/streaming/live/1.\n" + + result = summarize_ffmpeg_stderr(stderr) + + assert "12345678" not in result + assert "[REDACTED]" in result + # Host and user survive, or the line stops being useful for diagnosis. + assert "192.0.2.1:322" in result + assert "bblp" in result + + def test_a_credential_masked_before_the_cut_not_after(self): + """Truncating first would leave a URL with no ``@`` for the pattern to + anchor on, and the secret in the log.""" + stderr = "\n".join(f"noise {i}" for i in range(30)) + stderr += "\nOpening rtsp://user:hunter2@192.0.2.1:322/live and 40 more characters of tail\n" + + result = summarize_ffmpeg_stderr(stderr) + + assert "hunter2" not in result + + @pytest.mark.parametrize("empty", ["", None, b""]) + def test_nothing_in_nothing_out(self, empty): + assert summarize_ffmpeg_stderr(empty) == "" + + def test_a_single_enormous_line_is_bounded(self): + """Ten lines only bounds the record if the lines are sane, and ffmpeg + quotes back what the peer sent it. The tail is what is kept.""" + stderr = _REAL_BANNER + "x" * 50_000 + " Connection refused\n" + + result = summarize_ffmpeg_stderr(stderr) + + assert len(result) < 2_100 + assert result.endswith("Connection refused") + assert result.startswith("...") + + def test_an_ordinary_diagnosis_is_never_trimmed(self): + """The ceiling must not be reachable by real ffmpeg output.""" + stderr = _REAL_BANNER + "\n".join(f"[rtsp @ 0x5f] error line {i}" for i in range(10)) + + assert not summarize_ffmpeg_stderr(stderr).startswith("...") + + +class TestEveryCallSiteGoesThroughIt: + """The defect was seven copies of the same truncation, not one bad line. + + Asserted against the source because the alternative -- driving all seven + subprocesses -- tests ffmpeg, and because the failure mode being guarded is + somebody adding an eighth. + """ + + @pytest.mark.parametrize( + "module_path", + [ + "backend.app.services.camera", + "backend.app.services.external_camera", + "backend.app.services.layer_timelapse", + "backend.app.services.timelapse_processor", + "backend.app.services.archive", + "backend.app.api.routes.camera", + ], + ) + def test_no_module_truncates_stderr_by_hand(self, module_path): + import importlib + + source = inspect.getsource(importlib.import_module(module_path)) + + for lineno, raw in enumerate(source.splitlines(), 1): + # Comments discuss the defect by name -- this file's own fix notes + # do -- so only what executes is checked. + line = raw.split("#", 1)[0] + if "stderr" not in line: + continue + assert "stderr.decode()[:" not in line, f"{module_path}:{lineno} truncates stderr from the front" + assert "stderr_text[:" not in line, f"{module_path}:{lineno} truncates stderr from the front" + assert 'stderr.decode(errors="replace")[:' not in line, ( + f"{module_path}:{lineno} truncates stderr from the front" + ) + # A bare decode is the other half of the defect: it can raise + # UnicodeDecodeError while reporting an unrelated failure, and it + # leaves the input URL's credentials unmasked. + assert "stderr.decode()" not in line, f"{module_path}:{lineno} decodes stderr by hand" diff --git a/backend/tests/unit/test_tls_proxy_teardown_2968.py b/backend/tests/unit/test_tls_proxy_teardown_2968.py new file mode 100644 index 000000000..c164633c7 --- /dev/null +++ b/backend/tests/unit/test_tls_proxy_teardown_2968.py @@ -0,0 +1,263 @@ +"""The RTSPS proxy must not leave a handler running past its server (#2968). + +The reporter's log carries three of these, one per camera snapshot, at ERROR +with a traceback pointing into ``camera.py``: + + ERROR [asyncio] Task was destroyed but it is pending! + task: ._handle() done, defined at camera.py:243> + wait_for=<_GatheringFuture pending ...>> + +``asyncio.start_server`` wraps the connection callback in a task and keeps only +a weak reference to it, so a handler still awaiting its two forwarders can be +collected while pending -- which is exactly what that message is. Nothing was +broken by it (the snapshot on either side of each one succeeded), but it reads +like a camera fault in a log people attach to bug reports, and the shape behind +it is real: teardown closed the listener and then waited on handlers that only +finish when the *peer* drops the socket. + +Two things fix it. The handlers are strongly referenced for as long as they run, +and ``close_tls_proxy`` cancels them rather than hoping ffmpeg has already gone. + +The upstream here is a real TLS listener rather than a bare socket, because the +proxy spends its first ten seconds inside ``open_connection``: a stand-in that +never completes a handshake never reaches the forwarding state these tests are +about. The proxy sets ``CERT_NONE`` (Bambu printers are self-signed), so a +throwaway certificate is all it takes. +""" + +from __future__ import annotations + +import asyncio +import datetime +import gc +import logging +import ssl + +import pytest + +from backend.app.services.camera import close_tls_proxy, create_tls_proxy + + +@pytest.fixture(scope="module") +def self_signed_cert(tmp_path_factory): + """Certificate and key for the stand-in printer, generated once.""" + from cryptography import x509 + from cryptography.hazmat.primitives import hashes, serialization + from cryptography.hazmat.primitives.asymmetric import rsa + from cryptography.x509.oid import NameOID + + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "127.0.0.1")]) + now = datetime.datetime.now(datetime.timezone.utc) + cert = ( + x509.CertificateBuilder() + .subject_name(name) + .issuer_name(name) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - datetime.timedelta(days=1)) + .not_valid_after(now + datetime.timedelta(days=1)) + .sign(key, hashes.SHA256()) + ) + + directory = tmp_path_factory.mktemp("tls") + cert_file = directory / "cert.pem" + key_file = directory / "key.pem" + cert_file.write_bytes(cert.public_bytes(serialization.Encoding.PEM)) + key_file.write_bytes( + key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.TraditionalOpenSSL, + encryption_algorithm=serialization.NoEncryption(), + ) + ) + return cert_file, key_file + + +async def _printer(self_signed_cert, on_data=None) -> tuple[asyncio.Server, int]: + """A TLS listener standing in for the printer's RTSPS port. + + Accepts, hands anything it receives to *on_data*, and otherwise waits -- + which is the state the upstream is in while ffmpeg is being reaped. + """ + + async def _accept(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + try: + while True: + data = await reader.read(4096) + if not data: + break + if on_data is not None: + on_data(data) + except (ConnectionError, OSError, asyncio.CancelledError): + pass + finally: + if not writer.is_closing(): + writer.close() + + cert_file, key_file = self_signed_cert + context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + context.load_cert_chain(str(cert_file), str(key_file)) + server = await asyncio.start_server(_accept, "127.0.0.1", 0, ssl=context) + return server, server.sockets[0].getsockname()[1] + + +async def _close(proxy) -> None: + """Teardown, bounded. + + Every close in this file goes through the timeout, including the ones in + ``finally`` blocks that are only there to tidy up. Losing the cancellation + or the handler tracking makes ``close_tls_proxy`` wait on a peer that is + not going to drop, and an unbounded await turns that regression into a + hung suite instead of a failing test. + """ + await asyncio.wait_for(close_tls_proxy(proxy), timeout=5.0) + + +async def _shutdown(server: asyncio.Server) -> None: + """Bounded teardown for the stand-in printer. + + ``wait_closed`` waits for the listener's own handlers, and one of those is + reading a socket the proxy still holds. Left unbounded it inherits any + regression in the proxy's teardown and hangs the suite in a second place. + """ + server.close() + try: + await asyncio.wait_for(server.wait_closed(), timeout=5.0) + except asyncio.TimeoutError: + pass + + +async def _wait_for(predicate, timeout: float = 5.0) -> bool: + """Poll rather than sleep a fixed amount: these are real sockets.""" + deadline = asyncio.get_running_loop().time() + timeout + while asyncio.get_running_loop().time() < deadline: + if predicate(): + return True + await asyncio.sleep(0.02) + return predicate() + + +class TestTheHandlerIsHeldWhileItRuns: + @pytest.mark.asyncio + async def test_an_open_connection_is_tracked(self, self_signed_cert): + """The set is the strong reference asyncio does not keep.""" + upstream, upstream_port = await _printer(self_signed_cert) + try: + port, proxy = await create_tls_proxy("127.0.0.1", upstream_port) + try: + _, writer = await asyncio.open_connection("127.0.0.1", port) + assert await _wait_for(lambda: len(proxy._bambuddy_proxy_handlers) == 1) + assert not next(iter(proxy._bambuddy_proxy_handlers)).done() + + writer.close() + finally: + await _close(proxy) + finally: + await _shutdown(upstream) + + @pytest.mark.asyncio + async def test_a_finished_handler_is_released(self, self_signed_cert): + """Tracked for the connection's life, not the process's -- a long + stream must not accumulate one entry per reconnect.""" + upstream, upstream_port = await _printer(self_signed_cert) + try: + port, proxy = await create_tls_proxy("127.0.0.1", upstream_port) + try: + _, writer = await asyncio.open_connection("127.0.0.1", port) + assert await _wait_for(lambda: len(proxy._bambuddy_proxy_handlers) == 1) + + writer.close() + + assert await _wait_for(lambda: proxy._bambuddy_proxy_handlers == set()) + finally: + await _close(proxy) + finally: + await _shutdown(upstream) + + +class TestCloseDoesNotDependOnThePeer: + @pytest.mark.asyncio + async def test_a_live_connection_does_not_stall_the_close(self, self_signed_cert): + """``server.close()`` leaves established connections running, so the + old close/wait pair finished only when the client happened to drop. + Here the client is still attached and close still returns.""" + upstream, upstream_port = await _printer(self_signed_cert) + try: + port, proxy = await create_tls_proxy("127.0.0.1", upstream_port) + _, writer = await asyncio.open_connection("127.0.0.1", port) + assert await _wait_for(lambda: len(proxy._bambuddy_proxy_handlers) == 1) + + await _close(proxy) + + assert proxy._bambuddy_proxy_handlers == set() + writer.close() + finally: + await _shutdown(upstream) + + @pytest.mark.asyncio + async def test_no_handler_survives_the_close(self, self_signed_cert, caplog): + """The actual complaint: nothing is left pending for the garbage + collector to shout about afterwards.""" + upstream, upstream_port = await _printer(self_signed_cert) + try: + port, proxy = await create_tls_proxy("127.0.0.1", upstream_port) + _, writer = await asyncio.open_connection("127.0.0.1", port) + assert await _wait_for(lambda: len(proxy._bambuddy_proxy_handlers) == 1) + handler = next(iter(proxy._bambuddy_proxy_handlers)) + + with caplog.at_level(logging.ERROR, logger="asyncio"): + await _close(proxy) + writer.close() + await asyncio.sleep(0.05) + gc.collect() + await asyncio.sleep(0.05) + + assert handler.done() + assert not [r for r in caplog.records if "Task was destroyed" in r.getMessage()] + finally: + await _shutdown(upstream) + + @pytest.mark.asyncio + async def test_closing_twice_is_harmless(self, self_signed_cert): + """Both callers reach their finally block on the error paths too.""" + upstream, upstream_port = await _printer(self_signed_cert) + try: + _, proxy = await create_tls_proxy("127.0.0.1", upstream_port) + await _close(proxy) + await _close(proxy) + finally: + await _shutdown(upstream) + + @pytest.mark.asyncio + async def test_it_works_on_a_server_it_did_not_create(self): + """Degrades to the close/wait it replaces rather than raising.""" + plain = await asyncio.start_server(lambda r, w: None, "127.0.0.1", 0) + + await _close(plain) + + assert not plain.is_serving() + + +@pytest.mark.asyncio +async def test_the_proxy_still_forwards(self_signed_cert): + """The teardown changes must not cost the proxy its job: plain TCP in one + end, TLS to the printer out the other.""" + received: list[bytes] = [] + upstream, upstream_port = await _printer(self_signed_cert, on_data=received.append) + try: + port, proxy = await create_tls_proxy("127.0.0.1", upstream_port) + try: + _, writer = await asyncio.open_connection("127.0.0.1", port) + writer.write(b"OPTIONS rtsp://127.0.0.1/streaming/live/1 RTSP/1.0\r\n\r\n") + await writer.drain() + + assert await _wait_for(lambda: bool(received)) + assert b"OPTIONS" in received[0] + + writer.close() + finally: + await _close(proxy) + finally: + await _shutdown(upstream)