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

Log ffmpeg's error instead of its build banner

Browse Source 
ffmpeg opens every run with ~20 lines of version and build banner and prints
    its diagnosis last, so the stderr[:200] eight of the nine call sites used kept
    the banner and threw the error away. The reporter's twelve capture failures all
    read "ffmpeg version 7.1.4 ... configuration: --prefix=/usr --extra-version=",
    identical on every install; the exit code was the only usable byte.

    The banner-stripping summariser written for #925 lived private to the camera
    route. It now lives in backend/app/utils/ffmpeg_output.py and every ffmpeg and
    ffprobe stderr goes through it. Two things the scattered copies also got wrong:
    four logged the input URL unmasked, publishing a printer access code or camera
    password, and four called a bare .decode() on bytes ffmpeg copies stream
    fragments into.

    -----

    Delete the files a no-3MF archive owns, without taking a printer folder

    Both delete paths derived the directory from file_path, which such an archive
    does not have, so they removed nothing and logged it at ERROR under a SECURITY
    banner. That was true when the archive was an empty row and stopped being true
    once one could hold a timelapse and finish photos in <archive_dir>/<id>/ and an
    uploaded source in archive/no_source/<id>/.

    The two are cleaned up by different means, because <archive_dir>/<id> shares a
    namespace with the per-printer folders: a normal archive lives at
    <archive_dir>/<printer_id>/<timestamp>_<name>/, so archive/1 is printer 1's
    folder and also the directory the shared helper hands archive id 1. Ids come
    from unrelated sequences, so the first few archives collide with the printers
    on every install, and an rmtree there takes every print that printer made --
    measured on a scratch tree. no_source/<id> is a level deeper under a name no
    printer id can take and is removed whole; the id-named directory gives up only
    its photos subdirectory and the video the row records, then goes only if that
    left it empty. The depth guard moves from one to two for the same reason: a
    file_path that lost a path component could point the delete at a printer
    folder, and no archive directory has been one level deep since the first
    commit.

    Hard delete had its own copy of these rules, which the helper's docstring says
    it exists to prevent, and it had diverged -- it skipped the print-log thumbnail
    cleanup whenever a guard tripped.

    -----

    Stop the RTSPS proxy leaving a handler behind at shutdown

    asyncio.start_server keeps only a weak reference to the connection callback's
    task, so a handler still awaiting its forwarders could be collected while
    pending -- "Task was destroyed but it is pending!", at ERROR with a traceback
    into camera.py, once every few hundred snapshots. Teardown had the matching
    gap: server.close() leaves established connections running, so the close waited
    on a handler that only finishes when the peer drops, and ffmpeg has already
    been reaped by then.

    Handlers are held for as long as they run and cancelled at shutdown, which is
    Server.close_clients() by hand -- that landed in 3.13 and Bambuddy supports
    3.10. Both the snapshot path and the streaming endpoint share the shutdown.
This commit is contained in:
maziggy
2026-08-29 14:19:22 +02:00
parent ff9b956156 d4477e9b71
commit 20627b2188
10 changed files with 1109 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/routes/camera.py
+8 -34
View File
@@ -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")
backend/app/services/archive.py
+132 -76
View File
@@ -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,
``<archive_dir>/<printer_id>/<timestamp>_<name>/``. Any archive may also
own ``archive/no_source/<id>/``, 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. ``<base_dir>/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
``<archive_dir>/<id>`` — 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
# ``<archive_dir>/<printer_id>/<timestamp>_<name>/`` 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 ``<archive_dir>/<id>``, 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
``<archive_dir>/<printer_id>/<timestamp>_<name>/``, 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 ``<timestamp>_<name>``, 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()
backend/app/services/camera.py
+62 -10
View File
@@ -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:<proxy_port>`` → ``<target_host>:<target_port>`` 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:
backend/app/services/external_camera.py
+11 -6
View File
@@ -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""
backend/app/services/layer_timelapse.py
+2 -1
View File
@@ -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)
backend/app/services/timelapse_processor.py
+8 -3
View File
@@ -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()
backend/app/utils/ffmpeg_output.py
+102
View File
@@ -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
backend/tests/unit/test_archive_delete_no_3mf_dirs_2968.py
+336
View File
@@ -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:
``<archive_dir>/<id>/`` for its timelapse and finish photos (the shared helper
in ``utils.archive_paths``, #1820), and ``archive/no_source/<id>/`` 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.** ``<archive_dir>/<id>`` shares a
namespace with the per-printer folders: a normal archive lives at
``<archive_dir>/<printer_id>/<timestamp>_<name>/``, 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/<id>/`` 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):
"""``<base_dir>/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()
backend/tests/unit/test_ffmpeg_output_summary.py
+185
View File
@@ -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"
backend/tests/unit/test_tls_proxy_teardown_2968.py
+263
View File
@@ -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: <Task pending name='Task-1889625'
coro=<create_tls_proxy.<locals>._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)