Security hardening (maziggy/bambuddy-security #23)

Security hardening (maziggy/bambuddy-security #24)

.basedpyright/baseline.json

@@ -2865,14 +2865,6 @@
                     "lineCount": 1
                 }
             },
-            {
-                "code": "reportOptionalMemberAccess",
-                "range": {
-                    "startColumn": 42,
-                    "endColumn": 46,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportOptionalMemberAccess",
                 "range": {

backend/app/api/routes/camera.py

@@ -54,7 +54,12 @@ from backend.app.services.camera_fanout import (
 )
 from backend.app.services.camera_light import camera_light
 from backend.app.services.camera_profiles import get_camera_profile
-from backend.app.utils.ffmpeg_output import summarize_ffmpeg_stderr
+from backend.app.utils.ffmpeg_output import (
+    FFMPEG_RTSP_LOGLEVEL,
+    refuse_rtsp_redirect,
+    rtsp_log_event,
+    summarize_ffmpeg_stderr,
+)
 
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/printers", tags=["camera"])
@@ -149,6 +154,12 @@ _FFMPEG_STDERR_TAIL_BYTES = 16384
 # Live stderr collectors by pid — see _FfmpegStderrTail. Present means "this
 # process's stderr already has a reader; do not open a second one".
 _stderr_tails: dict[int, "_FfmpegStderrTail"] = {}
+# How long a frame waits for the stderr pump to reach ffmpeg's input-open line,
+# which ffmpeg writes before any frame.
+_FFMPEG_INPUT_OPEN_WAIT = 10.0
+# How long a session that ended without a frame waits for the pump to read the
+# rest of ffmpeg's log, to tell a refused redirect from a routine failure.
+_FFMPEG_REDIRECT_SETTLE = 2.0
 
 
 def get_buffered_frame(printer_id: int) -> bytes | None:
@@ -484,6 +495,11 @@ class _FfmpegStderrTail:
         self._process = process
         self._buffer = bytearray()
         self._task: asyncio.Task | None = None
+        # Whether ffmpeg reported its input open ("open") or a redirect it
+        # followed ("redirected") first, or ended first ("ended").
+        self.input_state: str | None = None
+        self._input_decided = asyncio.Event()
+        self._line = b""
         if process.stderr is None:
             return
         self._task = asyncio.create_task(self._pump())
@@ -496,6 +512,7 @@ class _FfmpegStderrTail:
                 chunk = await reader.read(8192)
                 if not chunk:
                     return  # EOF — ffmpeg has exited
+                self._watch(chunk)
                 self._buffer.extend(chunk)
                 excess = len(self._buffer) - _FFMPEG_STDERR_TAIL_BYTES
                 if excess > 0:
@@ -504,6 +521,43 @@ class _FfmpegStderrTail:
             raise
         except Exception:  # noqa: BLE001 — a broken pipe just ends the tail
             return
+        finally:
+            self._decide("ended")
+
+    def _decide(self, state: str) -> None:
+        if self.input_state is None:
+            self.input_state = state
+            self._input_decided.set()
+
+    def _watch(self, chunk: bytes) -> None:
+        """Follow ffmpeg's log line by line for the redirect and input-open
+        lines (see utils.ffmpeg_output), in the order ffmpeg wrote them."""
+        data = self._line + chunk
+        lines = data.split(b"\n")
+        self._line = lines.pop()[-_FFMPEG_STDERR_TAIL_BYTES:]
+        for line in lines:
+            event = rtsp_log_event(line)
+            if event == "redirected":
+                self.input_state = "redirected"
+                self._input_decided.set()
+                refuse_rtsp_redirect(self._process, line)
+                return
+            if event == "open":
+                self._decide("open")
+
+    async def wait_input_open(self, timeout: float) -> bool:
+        """True once ffmpeg reported its input open without a redirect first.
+
+        ffmpeg reports it before writing any frame, so a caller holding a
+        frame waits only for this pump to catch up.
+        """
+        if self._task is None:
+            return True  # no log pipe to read: spawned without one
+        try:
+            await asyncio.wait_for(self._input_decided.wait(), timeout=timeout)
+        except TimeoutError:
+            return False
+        return self.input_state == "open"
 
     def text(self) -> str | None:
         """The retained tail, summarised. None when nothing was captured.
@@ -612,6 +666,7 @@ async def generate_rtsp_mjpeg_stream(
     # ffmpeg command to output MJPEG stream to stdout
     cmd = [
         ffmpeg,
+        *FFMPEG_RTSP_LOGLEVEL,
         "-rtsp_transport",
         "tcp",
         "-rtsp_flags",
@@ -760,6 +815,7 @@ async def generate_rtsp_mjpeg_stream(
             last_change = time.time()
             identical_frames = 0
             frozen = False
+            refused = False
 
             while True:
                 if disconnect_event and disconnect_event.is_set():
@@ -796,6 +852,13 @@ async def generate_rtsp_mjpeg_stream(
 
                         frame = buffer[: end_idx + 2]
                         buffer = buffer[end_idx + 2 :]
+                        # No frame before ffmpeg reported its input open: a
+                        # redirect it followed on the way ends the stream.
+                        if not session_got_frames and (
+                            stderr_tail is None or not await stderr_tail.wait_input_open(_FFMPEG_INPUT_OPEN_WAIT)
+                        ):
+                            refused = True
+                            break
                         got_any_frames = True
                         session_got_frames = True
 
@@ -844,6 +907,14 @@ async def generate_rtsp_mjpeg_stream(
                             frozen_crc = last_frame_crc
                             break
 
+                    if refused:
+                        logger.error(
+                            "RTSP input not confirmed open for %s (stream_id=%s), stopping the stream",
+                            ip_address,
+                            stream_id,
+                        )
+                        break
+
                     if frozen:
                         if still_scene:
                             logger.info(
@@ -882,6 +953,13 @@ async def generate_rtsp_mjpeg_stream(
                     client_gone = True
                     break
 
+            # A session that ended before its first frame may have ended on a
+            # redirect (ffmpeg is killed for one): that is not redialled.
+            if not session_got_frames and not client_gone and stderr_tail is not None:
+                if stderr_tail.input_state is None:
+                    await stderr_tail.wait_input_open(_FFMPEG_REDIRECT_SETTLE)
+                refused = refused or stderr_tail.input_state == "redirected"
+
             # Clean up this ffmpeg process before reconnecting or exiting
             await _terminate_ffmpeg(process, stream_id)
             # Released after teardown, not before: _terminate_ffmpeg deliberately
@@ -892,7 +970,7 @@ async def generate_rtsp_mjpeg_stream(
                 stderr_tail = None
             process = None
 
-            if client_gone:
+            if client_gone or refused:
                 break
 
             # Check if stream was explicitly stopped (e.g., by stop endpoint)

backend/app/core/logging_filters.py

@@ -134,6 +134,7 @@ _MASKED_TAILS = re.compile(r"(invalid URL: |headers JSON: |returned error: |REST
 # An HTTP client refusing a header quotes the header's value, commonly a token,
 # whichever service sent it: masked to the end of the line.
 _REFUSED_HEADER = re.compile(r"(Illegal header (?:value|name) ).*")
+_UNMASKED_REFUSED_HEADER = re.compile(r"Illegal header (?:value|name) (?!\[REDACTED\])")
 
 
 def redact_refused_headers(text: str | None) -> str | None:
@@ -143,6 +144,58 @@ def redact_refused_headers(text: str | None) -> str | None:
     return _REFUSED_HEADER.sub(r"\1[REDACTED]", text)
 
 
+def _redact_refused_header_arg(value: object) -> object:
+    """*value* as masked text when its text quotes a refused header (an
+    exception, a string), else *value* itself."""
+    text = value if isinstance(value, str) else str(value)
+    return redact_refused_headers(text) if "Illegal header" in text else value
+
+
+class RefusedHeaderRedactFilter(logging.Filter):
+    """Mask a refused header's value in a record before a handler writes it.
+
+    The value reaches a record as an argument (``"...: %s", e``), in the
+    message itself, or in the traceback of ``logger.exception``, so the
+    rendered message and traceback are checked, not the format string. A
+    record with nothing to mask is left untouched.
+
+    Attach to the handlers that write lines (a handler filter sees records
+    from every logger that propagates to it) and to uvicorn's loggers, which
+    do not propagate to root.
+    """
+
+    _formatter = logging.Formatter()
+
+    def filter(self, record: logging.LogRecord) -> bool:  # noqa: A003 — stdlib API name
+        try:
+            message = record.getMessage()
+        except Exception:  # noqa: BLE001 — a bad format string is the handler's error to report, not the caller's
+            message = ""
+        if "Illegal header" in message:
+            # Masked in the arguments, keeping their shape: uvicorn's access
+            # formatter unpacks them.
+            if isinstance(record.args, tuple):
+                record.args = tuple(_redact_refused_header_arg(a) for a in record.args)
+            elif isinstance(record.args, dict):
+                record.args = {k: _redact_refused_header_arg(v) for k, v in record.args.items()}
+            try:
+                masked = record.getMessage()
+            except Exception:  # noqa: BLE001 — the masked format string no longer fits its arguments
+                masked = message
+            if _UNMASKED_REFUSED_HEADER.search(masked):
+                # The quote is (also) in the format string: the rendered
+                # line, masked, replaces both.
+                record.msg = redact_refused_headers(message)
+                record.args = None
+        if record.exc_info and not record.exc_text:
+            # Formatters reuse exc_text once set, so the masked text is what
+            # every handler then writes.
+            record.exc_text = self._formatter.formatException(record.exc_info)
+        if record.exc_text and "Illegal header" in record.exc_text:
+            record.exc_text = redact_refused_headers(record.exc_text)
+        return True
+
+
 def redact_device_urls(text: str | None) -> str | None:
     """Every URL in *text* reduced to :func:`url_host_only`, and the tails in
     ``_MASKED_TAILS`` masked."""

backend/app/core/response_redaction.py

@@ -0,0 +1,84 @@
+"""A refused header's value masked in every JSON response.
+
+An HTTP client refusing a header quotes the header's value, and the value is
+usually a credential: a stored token saved with a stray space or line break.
+That error text reaches the API from many places -- a provider's stored error,
+a backup or restore result, a connection test, a status endpoint, rows stored
+before any of them masked it -- so it is masked once, where every JSON body
+leaves, rather than at each place that may hand one on. Logs are masked by
+``logging_filters.RefusedHeaderRedactFilter``.
+"""
+
+from __future__ import annotations
+
+import re
+
+from starlette.datastructures import MutableHeaders
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
+
+_MARKER = b"Illegal header "
+# Inside a JSON string: masked up to the string's closing quote, so the body
+# stays valid JSON.
+_REFUSED_HEADER_IN_JSON = re.compile(rb'(Illegal header (?:value|name) )(?:[^"\\]|\\.)*')
+
+
+def redact_refused_headers_json(body: bytes) -> bytes:
+    """*body*, a JSON document, with every quoted refused header masked."""
+    if _MARKER not in body:
+        return body
+    return _REFUSED_HEADER_IN_JSON.sub(rb"\1[REDACTED]", body)
+
+
+class RefusedHeaderResponseMask:
+    """ASGI middleware masking refused headers in JSON response bodies.
+
+    Install it innermost (added before any other middleware): outer
+    ``@app.middleware`` layers re-stream a body in chunks. A body with a
+    Content-Length is collected and masked whole, with the length fixed up; a
+    streamed one is masked chunk by chunk.
+    """
+
+    def __init__(self, app: ASGIApp) -> None:
+        self.app = app
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        start: Message | None = None
+        sized = False
+        chunks: list[bytes] = []
+
+        async def masking_send(message: Message) -> None:
+            nonlocal start, sized
+            if message["type"] == "http.response.start":
+                headers = MutableHeaders(raw=message["headers"])
+                media_type = headers.get("content-type", "").split(";", 1)[0].strip().lower()
+                if not (media_type == "application/json" or media_type.endswith("+json")):
+                    await send(message)
+                    return
+                start = message
+                sized = "content-length" in headers
+                if not sized:
+                    await send(start)
+                return
+            if start is None or message["type"] != "http.response.body":
+                await send(message)
+                return
+            if not sized:
+                await send({**message, "body": redact_refused_headers_json(message.get("body", b""))})
+                return
+            chunks.append(message.get("body", b""))
+            if message.get("more_body", False):
+                return
+            original = b"".join(chunks)
+            body = redact_refused_headers_json(original)
+            if body is not original:
+                # Only a masked body changes length; a HEAD response's empty
+                # body keeps the length of the GET body it stands for.
+                MutableHeaders(raw=start["headers"])["content-length"] = str(len(body))
+            await send(start)
+            await send({"type": "http.response.body", "body": body, "more_body": False})
+
+        await self.app(scope, receive, masking_send)

backend/app/main.py

@@ -323,6 +323,8 @@ root_logger.setLevel(log_level)
 # dropped — which is exactly the "logs/bambuddy.log only shows logs
 # partially" bug we hit. See backend/app/core/trace.py for the
 # ContextVar the filter reads.
+from backend.app.core.logging_filters import RefusedHeaderRedactFilter
+from backend.app.core.response_redaction import RefusedHeaderResponseMask
 from backend.app.core.trace import TraceIDFilter
 
 _trace_id_filter = TraceIDFilter()
@@ -332,6 +334,9 @@ console_handler = logging.StreamHandler()
 console_handler.setLevel(log_level)
 console_handler.setFormatter(logging.Formatter(log_format))
 console_handler.addFilter(_trace_id_filter)
+# A refused header's value (often a token) masked before the line is written.
+_refused_header_filter = RefusedHeaderRedactFilter()
+console_handler.addFilter(_refused_header_filter)
 root_logger.addHandler(console_handler)
 
 
@@ -359,6 +364,7 @@ if app_settings.log_to_file:
     file_handler.setLevel(log_level)
     file_handler.setFormatter(logging.Formatter(log_format))
     file_handler.addFilter(_trace_id_filter)
+    file_handler.addFilter(_refused_header_filter)
     root_logger.addHandler(file_handler)
     logging.info("Logging to file: %s", log_file)
 
@@ -396,6 +402,7 @@ from backend.app.core.logging_filters import QueryTokenRedactFilter  # noqa: E40
 
 for _uvicorn_logger in ("uvicorn.access", "uvicorn.error"):
     logging.getLogger(_uvicorn_logger).addFilter(QueryTokenRedactFilter())
+    logging.getLogger(_uvicorn_logger).addFilter(_refused_header_filter)
 
 # These stay at WARNING even in debug mode: at INFO/DEBUG httpx/httpcore log
 # full request URLs, and some carry credentials (bot tokens, webhook URLs, API
@@ -10797,6 +10804,10 @@ app = FastAPI(
     lifespan=lifespan,
 )
 
+# Added first so it sits innermost, next to the routes: a refused header's value
+# (often a stored token) masked in every JSON body. See core.response_redaction.
+app.add_middleware(RefusedHeaderResponseMask)
+
 
 # =============================================================================
 # Authentication Middleware - Secures ALL API routes by default

backend/app/services/camera.py

@@ -18,7 +18,12 @@ import weakref
 from datetime import datetime
 from pathlib import Path
 
-from backend.app.utils.ffmpeg_output import NO_FFMPEG_OUTPUT, summarize_ffmpeg_stderr
+from backend.app.utils.ffmpeg_output import (
+    FFMPEG_RTSP_LOGLEVEL,
+    NO_FFMPEG_OUTPUT,
+    read_ffmpeg_log_until_open,
+    summarize_ffmpeg_stderr,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -772,6 +777,7 @@ async def _capture_camera_frame_bytes_uncoalesced(
     cmd = [
         ffmpeg,
         "-y",
+        *FFMPEG_RTSP_LOGLEVEL,
         "-rtsp_transport",
         "tcp",
         "-rtsp_flags",
@@ -801,14 +807,21 @@ async def _capture_camera_frame_bytes_uncoalesced(
 
         _active_capture_pids.add(process.pid)
         try:
-            stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)
+            # The printer's answers reach ffmpeg through the proxy unchanged,
+            # so a redirect is refused here as for an external camera.
+            async with asyncio.timeout(timeout):
+                state, log = await read_ffmpeg_log_until_open(process)
+                stdout, stderr = await process.communicate()
         except TimeoutError:
             process.kill()
             await process.wait()
             logger.error("Camera frame bytes capture timed out after %ss", timeout)
             return None
+        stderr = log + stderr
 
-        if process.returncode == 0 and stdout and len(stdout) >= 100:
+        if state == "redirected":
+            return None
+        if state == "open" and process.returncode == 0 and stdout and len(stdout) >= 100:
             logger.info("Successfully captured camera frame bytes: %s bytes", len(stdout))
             return stdout
         else:

backend/app/services/external_camera.py

@@ -22,7 +22,13 @@ from urllib.parse import urljoin, 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
+from backend.app.utils.ffmpeg_output import (
+    FFMPEG_RTSP_LOGLEVEL,
+    NO_FFMPEG_OUTPUT,
+    read_ffmpeg_log_until_open,
+    summarize_ffmpeg_stderr,
+    watch_ffmpeg_log,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -712,6 +718,7 @@ async def _capture_rtsp_frame(url: str, timeout: int) -> bytes | None:
 
     cmd = [
         ffmpeg,
+        *FFMPEG_RTSP_LOGLEVEL,
         "-rtsp_transport",
         "tcp",
         # Belt and braces on the scheme check above: a demuxer that follows a
@@ -739,7 +746,10 @@ async def _capture_rtsp_frame(url: str, timeout: int) -> bytes | None:
             stderr=asyncio.subprocess.PIPE,
         )
 
-        stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)
+        async with asyncio.timeout(timeout):
+            state, log = await read_ffmpeg_log_until_open(process)
+            stdout, stderr = await process.communicate()
+        stderr = log + stderr
         logger.debug(
             "ffmpeg returned: code=%s, stdout=%s bytes, stderr=%s bytes",
             process.returncode,
@@ -747,7 +757,9 @@ async def _capture_rtsp_frame(url: str, timeout: int) -> bytes | None:
             len(stderr),
         )
 
-        if process.returncode != 0:
+        if state == "redirected":
+            return None
+        if process.returncode != 0 or state != "open":
             # 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
@@ -1167,6 +1179,7 @@ async def _stream_rtsp(
 
     cmd = [
         ffmpeg,
+        *FFMPEG_RTSP_LOGLEVEL,
         "-rtsp_transport",
         "tcp",
         "-rtsp_flags",
@@ -1210,6 +1223,7 @@ async def _stream_rtsp(
     ]
 
     process = None
+    log_watch = None
     try:
         process = await asyncio.create_subprocess_exec(
             *cmd,
@@ -1222,15 +1236,22 @@ async def _stream_rtsp(
         if on_process is not None:
             on_process(process)
 
-        # Brief check for immediate startup failures
-        await asyncio.sleep(0.1)
-        if process.returncode is not None:
-            stderr = await process.stderr.read()
+        # No frame is read before ffmpeg reports the input open: a redirect it
+        # followed on the way would be in its log by then.
+        try:
+            state, log = await asyncio.wait_for(read_ffmpeg_log_until_open(process), timeout=30.0)
+        except TimeoutError:
+            logger.warning("RTSP stream did not open within 30s")
+            return
+        if state == "ended":
+            assert process.stderr is not None
+            log += await process.stderr.read()
             # 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
-            )
+            logger.error("ffmpeg RTSP stream failed: %s", summarize_ffmpeg_stderr(log) or NO_FFMPEG_OUTPUT)
+            return
+        if state != "open":
             return
+        log_watch = asyncio.create_task(watch_ffmpeg_log(process))
 
         buffer = b""
         jpeg_start = b"\xff\xd8"
@@ -1272,6 +1293,8 @@ async def _stream_rtsp(
     except OSError as e:
         logger.error("RTSP stream error: %s", e)
     finally:
+        if log_watch is not None:
+            log_watch.cancel()
         if process and process.returncode is None:
             process.terminate()
             try:

backend/app/services/github_backup.py

@@ -12,7 +12,7 @@ from sqlalchemy import desc, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.app.core.database import async_session
-from backend.app.core.logging_filters import URL_CREDENTIALS_PATTERN
+from backend.app.core.logging_filters import URL_CREDENTIALS_PATTERN, redact_refused_headers
 from backend.app.models.archive import PrintArchive
 from backend.app.models.github_backup import GitHubBackupConfig, GitHubBackupLog
 from backend.app.models.printer import Printer
@@ -93,6 +93,12 @@ _PROVIDER_DISPLAY_NAMES = {
 }
 
 
+def _without_refused_headers(result: dict) -> dict:
+    """*result* with its texts masked where an HTTP client refusing a header
+    quoted the value: the stored token, which the API never returns."""
+    return {key: redact_refused_headers(value) if isinstance(value, str) else value for key, value in result.items()}
+
+
 class GitHubBackupService:
     """Service for backing up profiles to GitHub."""
 
@@ -166,7 +172,7 @@ class GitHubBackupService:
         """Test connection and permissions for the given provider."""
         backend = get_provider_backend(provider)
         client = await self._get_client()
-        return await backend.test_connection(repo_url, token, client)
+        return _without_refused_headers(await backend.test_connection(repo_url, token, client))
 
     async def run_backup(self, config_id: int, trigger: str = "manual") -> dict:
         """Run a backup operation.
@@ -297,7 +303,7 @@ class GitHubBackupService:
 
                     provider_name = _PROVIDER_DISPLAY_NAMES.get(config.provider, config.provider)
                     self._backup_progress = f"Pushing to {provider_name}..."
-                    push_result = await self._push_to_provider(config, backup_data)
+                    push_result = _without_refused_headers(await self._push_to_provider(config, backup_data))
 
                     # Update log and config
                     log.status = push_result["status"]
@@ -326,13 +332,14 @@ class GitHubBackupService:
 
                 except Exception as e:
                     logger.exception("Backup failed")
+                    error = redact_refused_headers(str(e)) or ""
                     log.status = "failed"
                     log.completed_at = datetime.now(timezone.utc)
-                    log.error_message = str(e)
+                    log.error_message = error
 
                     config.last_backup_at = datetime.now(timezone.utc)
                     config.last_backup_status = "failed"
-                    config.last_backup_message = str(e)
+                    config.last_backup_message = error
 
                     if config.schedule_enabled:
                         config.next_scheduled_run = self.calculate_next_run(config.schedule_type)
@@ -340,7 +347,7 @@ class GitHubBackupService:
                     await db.commit()
                     return {
                         "success": False,
-                        "message": str(e),
+                        "message": error,
                         "log_id": log_id,
                         "commit_sha": None,
                         "files_changed": 0,

backend/app/services/homeassistant.py

@@ -7,6 +7,8 @@ from urllib.parse import urlparse
 
 import httpx
 
+from backend.app.core.logging_filters import redact_refused_headers
+
 if TYPE_CHECKING:
     from backend.app.models.smart_plug import SmartPlug
 
@@ -256,7 +258,8 @@ class HomeAssistantService:
         except httpx.ConnectError:
             return {"success": False, "message": None, "error": "Could not connect to Home Assistant"}
         except Exception as e:
-            return {"success": False, "message": None, "error": str(e)}
+            # May be the saved token's test: a refused header's value stays out.
+            return {"success": False, "message": None, "error": redact_refused_headers(str(e))}
 
     async def list_entities(self, url: str, token: str, search: str | None = None) -> list[dict]:
         """List available entities from HA.

backend/app/services/notification_service.py

@@ -17,6 +17,7 @@ import httpx
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from backend.app.core.logging_filters import redact_refused_headers
 from backend.app.models.archive import PrintArchive
 from backend.app.models.notification import (
     NotificationDigestQueue,
@@ -362,6 +363,18 @@ class NotificationService:
         attach_photo mirrors the provider's own toggle (see _send_to_provider)
         — when set, a bundled sample image stands in for a real snapshot.
         """
+        success, message = await self._send_test_notification(provider_type, config, db, attach_photo)
+        # The text goes back to the caller: an HTTP client refusing a header
+        # quotes its value, which may be a credential the caller never sees.
+        return success, redact_refused_headers(message) or message
+
+    async def _send_test_notification(
+        self,
+        provider_type: str,
+        config: dict[str, Any],
+        db: AsyncSession | None,
+        attach_photo: bool,
+    ) -> tuple[bool, str]:
         if db:
             title, message = await self._build_message_from_template(db, "test", {})
         else:
@@ -1547,7 +1560,8 @@ class NotificationService:
             if success:
                 provider.last_success = datetime.now(timezone.utc)
             else:
-                provider.last_error = error
+                # Read by every notification reader: no refused header's value.
+                provider.last_error = redact_refused_headers(error)
                 provider.last_error_at = datetime.now(timezone.utc)
             await db.commit()
 
@@ -1605,7 +1619,7 @@ class NotificationService:
                 title=title,
                 message=message,
                 success=success,
-                error_message=error_message,
+                error_message=redact_refused_headers(error_message),
                 printer_id=printer_id,
                 printer_name=printer_name,
             )

backend/app/utils/ffmpeg_output.py

@@ -33,8 +33,14 @@ pulling a startup graph behind them.
 
 from __future__ import annotations
 
+import asyncio
+import contextlib
+import logging
+
 from backend.app.core.logging_filters import redact_url_credentials
 
+logger = logging.getLogger(__name__)
+
 # 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
@@ -100,3 +106,76 @@ def summarize_ffmpeg_stderr(text: str | bytes | None) -> str:
         # From the end, for the same reason the whole module exists.
         summary = "..." + summary[-_MAX_CHARACTERS:]
     return summary
+
+
+# ffmpeg's RTSP demuxer follows a server's 3xx Location by itself, with no
+# option to refuse, and logs the first line below before it connects there.
+# Only the configured address was ever checked, so a redirect ends the fetch:
+# nothing fetched from the new destination is used. av_dump_format prints the
+# second once the input is open, before ffmpeg writes any output.
+_RTSP_REDIRECT_LOG = b"Redirecting to "
+_INPUT_OPEN_LOG = b"Input #0, "
+# Pinned: both lines are logged at "info".
+FFMPEG_RTSP_LOGLEVEL = ("-loglevel", "info")
+# Longest log line read before the input is open; ffmpeg's own are far shorter.
+_MAX_LOG_LINE = 65536
+
+
+def rtsp_log_event(line: bytes) -> str | None:
+    """``"redirected"`` or ``"open"`` for the ffmpeg log line that says so."""
+    if _RTSP_REDIRECT_LOG in line:
+        return "redirected"
+    if line.startswith(_INPUT_OPEN_LOG):
+        return "open"
+    return None
+
+
+def refuse_rtsp_redirect(process: asyncio.subprocess.Process, line: bytes) -> None:
+    """Stop an ffmpeg that reported following a redirect, saying where to."""
+    target = line.split(_RTSP_REDIRECT_LOG, 1)[-1].split(b"\n", 1)[0].decode("utf-8", "replace").strip()
+    logger.error(
+        "RTSP camera redirected to %s; redirects are not followed, configure that address instead",
+        (redact_url_credentials(target) or "")[:80],
+    )
+    with contextlib.suppress(ProcessLookupError):
+        process.kill()
+
+
+async def read_ffmpeg_log_until_open(process: asyncio.subprocess.Process) -> tuple[str, bytes]:
+    """Read an RTSP ffmpeg's log up to the line saying its input is open.
+
+    Returns ``("open", log)``; ``("redirected", log)`` after a redirect, with
+    ffmpeg killed; or ``("ended", log)`` when ffmpeg stopped first.
+    """
+    assert process.stderr is not None
+    log = b""
+    while True:
+        try:
+            line = await process.stderr.readline()
+        except ValueError:
+            # A line past the reader's limit is no line of ffmpeg's own.
+            with contextlib.suppress(ProcessLookupError):
+                process.kill()
+            return "ended", log
+        if not line:
+            return "ended", log
+        log = (log + line)[-_MAX_LOG_LINE:]
+        event = rtsp_log_event(line)
+        if event == "redirected":
+            refuse_rtsp_redirect(process, line)
+            return event, log
+        if event == "open":
+            return event, log
+
+
+async def watch_ffmpeg_log(process: asyncio.subprocess.Process) -> None:
+    """Drain a streaming ffmpeg's log, so a full pipe never stalls it, and
+    stop ffmpeg should it report a redirect after all."""
+    assert process.stderr is not None
+    tail = b""
+    while chunk := await process.stderr.read(4096):
+        window = tail + chunk
+        if _RTSP_REDIRECT_LOG in window:
+            refuse_rtsp_redirect(process, window[window.index(_RTSP_REDIRECT_LOG) :])
+            return
+        tail = window[-len(_RTSP_REDIRECT_LOG) :]

backend/tests/_fixtures/external_camera.py

@@ -7,15 +7,27 @@ argv without an ffmpeg binary being involved, so these patch the lookup and the
 spawn and record the call.
 """
 
+import asyncio
 from unittest.mock import AsyncMock, MagicMock, patch
 
+# What ffmpeg logs once an RTSP input is open; the RTSP paths read no frame
+# before it.
+INPUT_OPEN_LOG = b"Input #0, rtsp, from 'rtsp://192.168.1.50:554/live':\n"
+
+
+def stream_of(data: bytes) -> asyncio.StreamReader:
+    reader = asyncio.StreamReader()
+    reader.feed_data(data)
+    reader.feed_eof()
+    return reader
+
 
 def fake_ffmpeg():
     """Pretend ffmpeg is installed, so the paths get as far as building argv."""
     return patch("backend.app.services.external_camera.get_ffmpeg_path", return_value="/usr/bin/ffmpeg")
 
 
-def spawn_spy(returncode: int | None = 0, stdout: bytes = b"\xff\xd8" + b"\x00" * 200):
+def spawn_spy(returncode: int | None = 0, stdout: bytes = b"\xff\xd8" + b"\x00" * 200, stderr: bytes = INPUT_OPEN_LOG):
     """Stand in for the ffmpeg subprocess, recording the argv it was handed.
 
     The streaming path reads until EOF, so stdout.read returns b"" and the
@@ -26,7 +38,7 @@ def spawn_spy(returncode: int | None = 0, stdout: bytes = b"\xff\xd8" + b"\x00"
     process.returncode = returncode
     process.communicate = AsyncMock(return_value=(stdout, b""))
     process.stdout.read = AsyncMock(return_value=b"")
-    process.stderr.read = AsyncMock(return_value=b"")
+    process.stderr = stream_of(stderr)
     process.wait = AsyncMock(return_value=returncode)
     process.kill = MagicMock()
     process.terminate = MagicMock()

backend/tests/integration/test_refused_header_error_texts.py

@@ -0,0 +1,174 @@
+"""A stored credential must not come back in an error's text.
+
+An HTTP client refusing a header (a stored token with a stray space or line
+break) quotes the header's value. The Git backup's results and logs and the
+Home Assistant connection test returned that text, so a saved credential, which
+the API otherwise never hands back, could be read in an error message.
+"""
+
+from __future__ import annotations
+
+import socket
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from backend.app.services.homeassistant import HomeAssistantService
+
+# Made up. Never paste a value from a real log here.
+SECRET = "git-refused-value-3"
+REFUSED = f"Illegal header value b'token {SECRET} '"
+
+_CONNECTED = {
+    "success": True,
+    "message": "Connection successful",
+    "repo_name": "test/repo",
+    "permissions": {"push": True},
+    "is_private": True,
+}
+
+
+async def _configure_backup(async_client: AsyncClient) -> None:
+    with patch(
+        "backend.app.services.github_backup.github_backup_service.test_connection",
+        new=AsyncMock(return_value=_CONNECTED),
+    ):
+        response = await async_client.post(
+            "/api/v1/github-backup/config",
+            json={
+                "repository_url": "https://github.com/test/repo",
+                "access_token": "ghp_token",
+                "branch": "main",
+                "schedule_enabled": False,
+                "schedule_type": "daily",
+                "backup_kprofiles": True,
+                "backup_cloud_profiles": False,
+                "backup_settings": False,
+                "enabled": True,
+            },
+        )
+    assert response.status_code == 200, response.text
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+@pytest.mark.parametrize(
+    "push",
+    [
+        AsyncMock(side_effect=RuntimeError(REFUSED)),
+        AsyncMock(return_value={"status": "failed", "message": REFUSED, "error": REFUSED}),
+    ],
+    ids=["raised", "returned"],
+)
+async def test_a_failed_backup_reports_no_header_value(async_client: AsyncClient, db_session, push):
+    await _configure_backup(async_client)
+    # The backup opens its own sessions.
+    sessions = async_sessionmaker(db_session.bind, class_=AsyncSession, expire_on_commit=False)
+
+    with (
+        patch("backend.app.services.github_backup.github_backup_service._push_to_provider", new=push),
+        patch("backend.app.services.github_backup.async_session", sessions),
+        patch(
+            "backend.app.services.github_backup.github_backup_service.test_connection",
+            new=AsyncMock(return_value=_CONNECTED),
+        ),
+    ):
+        run = await async_client.post("/api/v1/github-backup/run")
+    config = await async_client.get("/api/v1/github-backup/config")
+    logs = await async_client.get("/api/v1/github-backup/logs")
+
+    for response in (run, config, logs):
+        assert response.status_code == 200, response.text
+        assert SECRET not in response.text
+    assert "Illegal header value [REDACTED]" in run.text
+    assert "Illegal header value [REDACTED]" in logs.text
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_the_stored_token_test_reports_no_header_value(async_client: AsyncClient):
+    await _configure_backup(async_client)
+    backend = MagicMock()
+    backend.test_connection = AsyncMock(return_value={"success": False, "message": REFUSED})
+
+    with patch("backend.app.services.github_backup.get_provider_backend", return_value=backend):
+        response = await async_client.post("/api/v1/github-backup/test-stored")
+
+    assert response.status_code == 200, response.text
+    assert SECRET not in response.text
+    assert "Illegal header value [REDACTED]" in response.text
+
+
+@pytest.mark.asyncio
+async def test_the_home_assistant_test_reports_no_header_value():
+    with socket.socket() as server:
+        server.bind(("127.0.0.1", 0))
+        server.listen()
+        url = f"http://127.0.0.1:{server.getsockname()[1]}"
+        with patch.object(HomeAssistantService, "_validate_url", side_effect=lambda u: u):
+            result = await HomeAssistantService(timeout=2).test_connection(url, f"{SECRET} ")
+
+    assert result["success"] is False
+    assert SECRET not in result["error"]
+    assert "Illegal header value [REDACTED]" in result["error"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_errors_stored_before_masking_are_served_masked(async_client: AsyncClient, db_session):
+    """Rows written by an earlier version still hold the quoted value."""
+    from backend.app.models.github_backup import GitHubBackupLog
+    from backend.app.models.notification import NotificationLog, NotificationProvider
+
+    provider = NotificationProvider(
+        name="HA", provider_type="homeassistant", enabled=True, config="{}", last_error=REFUSED
+    )
+    db_session.add(provider)
+    await db_session.flush()
+    db_session.add(
+        NotificationLog(
+            provider_id=provider.id, event_type="test", title="t", message="m", success=False, error_message=REFUSED
+        )
+    )
+    await db_session.commit()
+    await _configure_backup(async_client)
+    config = (await async_client.get("/api/v1/github-backup/config")).json()
+    db_session.add(GitHubBackupLog(config_id=config["id"], status="failed", trigger="manual", error_message=REFUSED))
+    await db_session.commit()
+
+    for path in ("/api/v1/notifications/", "/api/v1/notifications/logs", "/api/v1/github-backup/logs"):
+        response = await async_client.get(path)
+        assert response.status_code == 200, (path, response.text)
+        assert SECRET not in response.text, path
+        assert "Illegal header value [REDACTED]" in response.text, path
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_the_restore_commit_list_reports_no_header_value(async_client: AsyncClient):
+    await _configure_backup(async_client)
+    backend = MagicMock()
+    backend.list_commits = AsyncMock(
+        return_value={"success": False, "message": f"LocalProtocolError: {REFUSED}", "commits": []}
+    )
+
+    with patch("backend.app.services.github_restore.get_provider_backend", return_value=backend):
+        response = await async_client.get("/api/v1/github-backup/commits")
+
+    assert SECRET not in response.text
+    assert "Illegal header value [REDACTED]" in response.text
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_the_obico_status_reports_no_header_value(async_client: AsyncClient):
+    from backend.app.services.obico_detection import obico_detection_service
+
+    with patch.object(obico_detection_service, "_last_error", f"ML API call failed for printer 1: {REFUSED}"):
+        response = await async_client.get("/api/v1/obico/status")
+
+    assert response.status_code == 200, response.text
+    assert SECRET not in response.text
+    assert "Illegal header value [REDACTED]" in response.text

backend/tests/unit/services/test_notification_refused_header.py

@@ -0,0 +1,81 @@
+"""A refused header's value must not reach a notification's stored error.
+
+A Home Assistant token with a stray space makes httpx refuse the header and
+quote its value. The send's error text went into the provider's last_error and
+the notification log, which notification readers (Viewers, by default) read
+through the API, and back to whoever ran a test notification. The token is an
+admin-only setting.
+"""
+
+from __future__ import annotations
+
+import socket
+
+import pytest
+from sqlalchemy import select
+
+from backend.app.models.notification import NotificationLog, NotificationProvider
+from backend.app.services.notification_service import NotificationService
+
+# Made up. Never paste a value from a real log here.
+SECRET = "ha-notify-value-7"
+
+
+@pytest.fixture
+def home_assistant(monkeypatch):
+    """HA settings from the environment, pointed at a socket that accepts."""
+    server = socket.socket()
+    server.bind(("127.0.0.1", 0))
+    server.listen()
+    monkeypatch.setenv("HA_URL", f"http://127.0.0.1:{server.getsockname()[1]}")
+    monkeypatch.setenv("HA_TOKEN", f"{SECRET} ")
+    yield
+    server.close()
+
+
+@pytest.mark.asyncio
+async def test_a_failed_send_stores_no_header_value(db_session, home_assistant):
+    provider = NotificationProvider(
+        name="HA", provider_type="homeassistant", enabled=True, config="{}", attach_photo=False
+    )
+    db_session.add(provider)
+    await db_session.commit()
+
+    await NotificationService()._send_to_providers([provider], "Title", "Body", db_session, event_type="test")
+
+    await db_session.refresh(provider)
+    log = (await db_session.execute(select(NotificationLog))).scalar_one()
+    assert provider.last_error and "Illegal header value [REDACTED]" in provider.last_error
+    assert SECRET not in provider.last_error
+    assert log.error_message and SECRET not in log.error_message
+
+
+@pytest.mark.asyncio
+async def test_a_test_notification_returns_no_header_value(db_session, home_assistant):
+    success, message = await NotificationService().send_test_notification(
+        "homeassistant", {}, db=db_session, attach_photo=False
+    )
+
+    assert success is False
+    assert "Illegal header value [REDACTED]" in message
+    assert SECRET not in message
+
+
+@pytest.mark.asyncio
+async def test_an_error_raised_out_of_a_send_is_stored_masked(db_session):
+    """The dispatcher's own except stores str(e) as well."""
+    provider = NotificationProvider(name="X", provider_type="webhook", enabled=True, config="{}")
+    db_session.add(provider)
+    await db_session.commit()
+    service = NotificationService()
+
+    async def _raise(*_args, **_kwargs):
+        raise RuntimeError(f"Illegal header value b'Bearer {SECRET} '")
+
+    service._send_to_provider = _raise
+    await service._send_to_providers([provider], "Title", "Body", db_session, event_type="test")
+
+    await db_session.refresh(provider)
+    log = (await db_session.execute(select(NotificationLog))).scalar_one()
+    assert SECRET not in (provider.last_error or "") and "[REDACTED]" in provider.last_error
+    assert SECRET not in (log.error_message or "")

backend/tests/unit/test_camera_stderr_tail.py

@@ -199,7 +199,7 @@ async def test_the_stream_generator_owns_then_releases_the_collector(monkeypatch
             self.pid = 88042
             self.returncode = None
             self.stdout = _FrameThenBlock()
-            self.stderr = _Reader([b"Stream #0:0: Video: h264\n"])
+            self.stderr = _Reader([b"Input #0, rtsp, from 'rtsp://x':\n", b"Stream #0:0: Video: h264\n"])
 
         def terminate(self) -> None:
             self.returncode = 0

backend/tests/unit/test_capture_pid_tracking.py

@@ -15,6 +15,7 @@ from backend.app.services.camera import (
     _active_capture_pids,
     capture_camera_frame_bytes,
 )
+from backend.tests._fixtures.external_camera import INPUT_OPEN_LOG, stream_of
 
 
 @pytest.fixture(autouse=True)
@@ -36,6 +37,7 @@ class TestCapturePidRegistration:
         fake_process = MagicMock()
         fake_process.pid = 99999
         fake_process.returncode = 0
+        fake_process.stderr = stream_of(INPUT_OPEN_LOG)
 
         async def fake_communicate():
             # Snapshot what's in the set while "ffmpeg is running"
@@ -69,6 +71,7 @@ class TestCapturePidRegistration:
         fake_process = MagicMock()
         fake_process.pid = 88888
         fake_process.returncode = 1
+        fake_process.stderr = stream_of(b"some error\n")
 
         async def fake_communicate():
             return (b"", b"some error")
@@ -96,6 +99,7 @@ class TestCapturePidRegistration:
         fake_process = MagicMock()
         fake_process.pid = 77777
         fake_process.returncode = None
+        fake_process.stderr = asyncio.StreamReader()  # ffmpeg still connecting
         fake_process.kill = MagicMock()
 
         async def fake_communicate():

backend/tests/unit/test_external_camera_rtsp_redirects.py

@@ -0,0 +1,231 @@
+"""An RTSP camera's redirect is not followed into a frame.
+
+The camera guard checks the configured RTSP URL, but ffmpeg's RTSP demuxer
+follows a server's 3xx Location by itself and has no option to refuse it, so
+a camera host could steer ffmpeg to an RTSP service the guard refuses (one on
+the Bambuddy host itself) and its frames would reach the viewers. ffmpeg logs
+the redirect before it connects there, and reports the input open before it
+writes any frame: a redirect ends the fetch, and no frame is read before the
+input is open.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app.services import external_camera
+from backend.tests._fixtures.external_camera import INPUT_OPEN_LOG, fake_ffmpeg, stream_of
+
+CAMERA = "rtsp://admin:pw@192.168.1.50:554/live"
+JPEG = b"\xff\xd8" + b"\x00" * 200 + b"\xff\xd9"
+REDIRECT_LOG = b"[rtsp @ 0x5eb63792b240] Status 302: Redirecting to rtsp://u:pw@127.0.0.1:8554/secret\n"
+BANNER = b"ffmpeg version 7.1.5 Copyright (c) 2000-2026 the FFmpeg developers\n"
+
+
+def _ffmpeg(stderr: bytes, stdout: bytes = JPEG, returncode: int = 0):
+    """An ffmpeg process whose pipes really stream; kill is recorded."""
+    process = MagicMock()
+    process.returncode = None
+    process.stderr = stream_of(stderr)
+    process.stdout = stream_of(stdout)
+
+    def _finish(code: int) -> None:
+        if process.returncode is None:
+            process.returncode = code
+
+    async def _communicate():
+        out, err = await process.stdout.read(), await process.stderr.read()
+        _finish(returncode)
+        return out, err
+
+    async def _wait():
+        _finish(returncode)
+        return process.returncode
+
+    process.communicate = _communicate
+    process.wait = _wait
+    process.kill = MagicMock(side_effect=lambda: _finish(-9))
+    process.terminate = MagicMock(side_effect=lambda: _finish(-15))
+    spawn = patch.object(external_camera.asyncio, "create_subprocess_exec", new=AsyncMock(return_value=process))
+    return process, spawn
+
+
+@pytest.mark.asyncio
+async def test_a_redirected_capture_gives_no_frame(caplog):
+    process, spawn = _ffmpeg(BANNER + REDIRECT_LOG + INPUT_OPEN_LOG)
+
+    with fake_ffmpeg(), spawn:
+        frame = await external_camera._capture_rtsp_frame(CAMERA, timeout=5)
+
+    assert frame is None
+    process.kill.assert_called()
+    # Reported as the redirect it was, not as ffmpeg failing.
+    assert "redirects are not followed" in caplog.text
+    assert "capture failed" not in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_a_log_line_past_the_reader_limit_ends_the_capture():
+    process, spawn = _ffmpeg(b"x" * 70000 + b"\n" + INPUT_OPEN_LOG)
+
+    with fake_ffmpeg(), spawn:
+        frame = await external_camera._capture_rtsp_frame(CAMERA, timeout=5)
+
+    assert frame is None
+    process.kill.assert_called()
+
+
+@pytest.mark.asyncio
+async def test_a_redirected_stream_gives_no_frame():
+    process, spawn = _ffmpeg(BANNER + REDIRECT_LOG + INPUT_OPEN_LOG, stdout=JPEG * 3)
+
+    with fake_ffmpeg(), spawn:
+        frames = [frame async for frame in external_camera._stream_rtsp(CAMERA, fps=5)]
+
+    assert frames == []
+    process.kill.assert_called()
+
+
+@pytest.mark.asyncio
+async def test_the_redirect_is_logged_without_its_login(caplog):
+    _process, spawn = _ffmpeg(REDIRECT_LOG + INPUT_OPEN_LOG)
+
+    with fake_ffmpeg(), spawn:
+        await external_camera._capture_rtsp_frame(CAMERA, timeout=5)
+
+    assert "127.0.0.1:8554" in caplog.text
+    assert "u:pw@" not in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_a_stream_reads_no_frame_before_the_input_is_open():
+    """ffmpeg ending without opening the input: whatever is on stdout is not a frame."""
+    _process, spawn = _ffmpeg(BANNER + b"Connection refused\n", stdout=JPEG * 3, returncode=1)
+
+    with fake_ffmpeg(), spawn:
+        frames = [frame async for frame in external_camera._stream_rtsp(CAMERA, fps=5)]
+
+    assert frames == []
+
+
+@pytest.mark.asyncio
+async def test_a_capture_that_never_opened_gives_no_frame(caplog):
+    _process, spawn = _ffmpeg(BANNER + b"Connection refused\n", stdout=JPEG, returncode=0)
+
+    with fake_ffmpeg(), spawn:
+        frame = await external_camera._capture_rtsp_frame(CAMERA, timeout=5)
+
+    assert frame is None
+    assert "Connection refused" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_a_camera_without_a_redirect_still_captures_and_streams():
+    _process, spawn = _ffmpeg(BANNER + INPUT_OPEN_LOG)
+    with fake_ffmpeg(), spawn:
+        assert await external_camera._capture_rtsp_frame(CAMERA, timeout=5) == JPEG
+
+    process, spawn = _ffmpeg(BANNER + INPUT_OPEN_LOG, stdout=JPEG * 3)
+    with fake_ffmpeg(), spawn:
+        frames = [frame async for frame in external_camera._stream_rtsp(CAMERA, fps=5)]
+    assert frames == [JPEG] * 3
+    process.kill.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_a_redirect_reported_while_streaming_stops_ffmpeg():
+    """Belt and braces: ffmpeg follows no redirect once the input is open, but
+    should it, the log watch stops it."""
+    process = MagicMock()
+    process.returncode = None
+    process.stderr = asyncio.StreamReader()
+    process.stderr.feed_data(INPUT_OPEN_LOG)
+    process.stdout = asyncio.StreamReader()
+    process.stdout.feed_data(JPEG)
+
+    def _kill():
+        process.returncode = -9
+        process.stdout.feed_eof()
+
+    process.kill = MagicMock(side_effect=_kill)
+    process.wait = AsyncMock(return_value=-9)
+
+    async def _late_redirect():
+        await asyncio.sleep(0.05)
+        process.stderr.feed_data(b"frame=  1 fps=0.0\r" + REDIRECT_LOG)
+
+    with (
+        fake_ffmpeg(),
+        patch.object(external_camera.asyncio, "create_subprocess_exec", new=AsyncMock(return_value=process)),
+    ):
+        feeder = asyncio.create_task(_late_redirect())
+        frames = [frame async for frame in external_camera._stream_rtsp(CAMERA, fps=5)]
+        await feeder
+
+    assert frames == [JPEG]
+    process.kill.assert_called()
+
+
+@pytest.mark.asyncio
+async def test_a_streaming_ffmpeg_log_is_drained():
+    """Nothing read ffmpeg's log while streaming, so a long session could fill
+    the pipe and stall ffmpeg on its next log line."""
+    process = MagicMock()
+    process.returncode = None
+    process.stderr = asyncio.StreamReader()
+    process.stderr.feed_data(INPUT_OPEN_LOG + b"frame=  1 fps=0.0\r" * 10000)
+    process.stdout = stream_of(JPEG)
+    process.terminate = MagicMock()
+    process.wait = AsyncMock(return_value=0)
+
+    with (
+        fake_ffmpeg(),
+        patch.object(external_camera.asyncio, "create_subprocess_exec", new=AsyncMock(return_value=process)),
+    ):
+        async for _frame in external_camera._stream_rtsp(CAMERA, fps=5):
+            await asyncio.sleep(0.05)
+
+    assert len(process.stderr._buffer) == 0
+
+
+@pytest.mark.asyncio
+async def test_ffmpeg_logs_at_the_level_that_carries_both_lines():
+    for run in ("capture", "stream"):
+        _process, spawn = _ffmpeg(INPUT_OPEN_LOG)
+        with fake_ffmpeg(), spawn as spawned:
+            if run == "capture":
+                await external_camera._capture_rtsp_frame(CAMERA, timeout=5)
+            else:
+                [frame async for frame in external_camera._stream_rtsp(CAMERA, fps=5)]
+        argv = spawned.await_args.args
+        assert argv[argv.index("-loglevel") + 1] == "info", run
+
+
+@pytest.mark.asyncio
+async def test_closing_the_stream_stops_the_log_watch():
+    process = MagicMock()
+    process.returncode = None
+    process.stderr = asyncio.StreamReader()  # never ends on its own
+    process.stderr.feed_data(INPUT_OPEN_LOG)
+    process.stdout = asyncio.StreamReader()
+    process.stdout.feed_data(JPEG * 2)
+    process.terminate = MagicMock()
+    process.wait = AsyncMock(return_value=0)
+
+    def _watches():
+        return [t for t in asyncio.all_tasks() if t.get_coro().__name__ == "watch_ffmpeg_log" and not t.done()]
+
+    with (
+        fake_ffmpeg(),
+        patch.object(external_camera.asyncio, "create_subprocess_exec", new=AsyncMock(return_value=process)),
+    ):
+        stream = external_camera._stream_rtsp(CAMERA, fps=5)
+        assert await anext(stream) == JPEG
+        assert _watches()
+        await stream.aclose()
+        await asyncio.sleep(0)
+
+    assert _watches() == []

backend/tests/unit/test_printer_camera_rtsp_redirects.py

@@ -0,0 +1,249 @@
+"""A printer camera's RTSP redirect is not followed into a frame either.
+
+The built-in camera paths run ffmpeg on rtsp://bblp:<code>@127.0.0.1:<proxy>,
+a local TLS proxy to the printer's address. The proxy forwards the printer's
+answers unchanged, so whatever answers at that address could redirect ffmpeg
+to an RTSP service on the Bambuddy host or the link-local range, which the
+external-camera guard refuses for the same editors. Same rule as there: a
+redirect ends the fetch, and no frame is used before the input is open.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app.api.routes import camera as camera_routes
+from backend.app.services import camera as camera_service
+from backend.tests._fixtures.external_camera import INPUT_OPEN_LOG, stream_of
+
+JPEG = b"\xff\xd8" + b"\x00" * 200 + b"\xff\xd9"
+REDIRECT_LOG = b"[rtsp @ 0x5eb63792b240] Status 302: Redirecting to rtsp://127.0.0.1:8554/secret\n"
+BANNER = b"ffmpeg version 7.1.5 Copyright (c) 2000-2026 the FFmpeg developers\n"
+
+
+class _FakeServer:
+    def close(self) -> None:
+        pass
+
+    async def wait_closed(self) -> None:
+        pass
+
+
+class _Proc:
+    """An ffmpeg process whose pipes really stream."""
+
+    _next_pid = 79000
+
+    def __init__(self, stderr: bytes, stdout: bytes, returncode: int = 0) -> None:
+        _Proc._next_pid += 1
+        self.pid = _Proc._next_pid
+        self.returncode = None
+        self._exit = returncode
+        self.stderr = stream_of(stderr)
+        self.stdout = stream_of(stdout)
+        self.kill = MagicMock(side_effect=lambda: self._finish(-9))
+        self.terminate = MagicMock(side_effect=lambda: self._finish(-15))
+
+    def _finish(self, code: int) -> None:
+        if self.returncode is None:
+            self.returncode = code
+
+    async def communicate(self):
+        out, err = await self.stdout.read(), await self.stderr.read()
+        self._finish(self._exit)
+        return out, err
+
+    async def wait(self) -> int:
+        self._finish(self._exit)
+        return self.returncode
+
+
+def _capture(process: _Proc):
+    async def _proxy(_ip, _port):
+        return 48998, _FakeServer()
+
+    return (
+        patch.object(camera_service, "is_chamber_image_model", return_value=False),
+        patch.object(camera_service, "create_tls_proxy", new=_proxy),
+        patch.object(camera_service, "close_tls_proxy", new=AsyncMock()),
+        patch.object(camera_service, "get_ffmpeg_path", return_value="/usr/bin/ffmpeg"),
+        patch.object(camera_service.asyncio, "create_subprocess_exec", new=AsyncMock(return_value=process)),
+    )
+
+
+async def _capture_frame(process: _Proc):
+    p1, p2, p3, p4, p5 = _capture(process)
+    with p1, p2, p3, p4, p5 as spawn:
+        frame = await camera_service._capture_camera_frame_bytes_uncoalesced("192.168.1.20", "code", "X1C", 5)
+    return frame, spawn
+
+
+@pytest.mark.asyncio
+async def test_a_redirected_printer_capture_gives_no_frame(caplog):
+    process = _Proc(BANNER + REDIRECT_LOG + INPUT_OPEN_LOG, JPEG)
+
+    frame, _spawn = await _capture_frame(process)
+
+    assert frame is None
+    process.kill.assert_called()
+    assert "redirects are not followed" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_a_printer_capture_that_never_opened_gives_no_frame():
+    frame, _spawn = await _capture_frame(_Proc(BANNER + b"Connection refused\n", JPEG))
+
+    assert frame is None
+
+
+@pytest.mark.asyncio
+async def test_a_printer_capture_without_a_redirect_still_works():
+    process = _Proc(BANNER + INPUT_OPEN_LOG, JPEG)
+
+    frame, spawn = await _capture_frame(process)
+
+    assert frame == JPEG
+    process.kill.assert_not_called()
+    argv = spawn.await_args.args
+    assert argv[argv.index("-loglevel") + 1] == "info"
+
+
+@pytest.fixture
+def printer_stream(monkeypatch):
+    """Each spawn returns the next queued _Proc."""
+    queue: list[_Proc] = []
+    spawned: list[tuple] = []
+
+    async def _exec(*args, **_kwargs):
+        spawned.append(args)
+        return queue.pop(0) if queue else _Proc(b"", b"", returncode=1)
+
+    async def _proxy(_ip, _port):
+        return 48998, _FakeServer()
+
+    monkeypatch.setattr(camera_routes, "get_ffmpeg_path", lambda: "/usr/bin/ffmpeg")
+    monkeypatch.setattr(camera_routes, "create_tls_proxy", _proxy)
+    monkeypatch.setattr(camera_routes, "close_tls_proxy", AsyncMock())
+    monkeypatch.setattr(camera_routes.asyncio, "create_subprocess_exec", _exec)
+    return queue, spawned
+
+
+async def _frames(limit: int = 10) -> int:
+    count = 0
+    stream = camera_routes.generate_rtsp_mjpeg_stream(
+        ip_address="192.168.1.20", access_code="code", model="X1C", fps=10, stream_id="7-redirect-test"
+    )
+    try:
+        async for chunk in stream:
+            if b"image/jpeg" in chunk:
+                count += 1
+            if count >= limit:
+                break
+    finally:
+        await stream.aclose()
+    return count
+
+
+@pytest.mark.asyncio
+async def test_a_redirected_printer_stream_gives_no_frame_and_does_not_redial(printer_stream):
+    queue, spawned = printer_stream
+    process = _Proc(BANNER + REDIRECT_LOG + INPUT_OPEN_LOG, JPEG * 3)
+    queue.append(process)
+
+    assert await _frames() == 0
+    process.kill.assert_called()
+    assert len(spawned) == 1
+
+
+@pytest.mark.asyncio
+async def test_a_printer_stream_without_a_redirect_still_streams(printer_stream):
+    queue, spawned = printer_stream
+    queue.append(_Proc(BANNER + INPUT_OPEN_LOG, JPEG * 3))
+
+    assert await _frames(limit=3) == 3
+    argv = spawned[0]
+    assert argv[argv.index("-loglevel") + 1] == "info"
+
+
+@pytest.mark.asyncio
+async def test_a_redirect_reported_while_the_printer_streams_stops_ffmpeg(printer_stream):
+    queue, _spawned = printer_stream
+    process = _Proc(b"", b"")
+    process.stderr = asyncio.StreamReader()
+    process.stderr.feed_data(INPUT_OPEN_LOG)
+    process.stdout = asyncio.StreamReader()
+    process.stdout.feed_data(JPEG)
+
+    def _kill():
+        process._finish(-9)
+        process.stdout.feed_eof()
+        process.stderr.feed_eof()
+
+    process.kill = MagicMock(side_effect=_kill)
+    queue.append(process)
+
+    async def _late_redirect():
+        await asyncio.sleep(0.05)
+        process.stderr.feed_data(b"frame=  1 fps=0.0\r" + REDIRECT_LOG)
+
+    feeder = asyncio.create_task(_late_redirect())
+    await _frames(limit=1)
+    await asyncio.sleep(0.1)
+    await feeder
+
+    process.kill.assert_called()
+
+
+@pytest.mark.asyncio
+async def test_a_printer_stream_that_ended_before_opening_stops_at_once(printer_stream):
+    """Its log ends without the input-open line: no wait for a line that cannot come."""
+    queue, _spawned = printer_stream
+    queue.append(_Proc(BANNER + b"Connection refused\n", JPEG * 3, returncode=1))
+
+    started = asyncio.get_running_loop().time()
+    assert await _frames() == 0
+    assert asyncio.get_running_loop().time() - started < 5
+
+
+@pytest.mark.asyncio
+async def test_a_printer_stream_killed_for_a_redirect_does_not_redial(printer_stream):
+    """As with a real ffmpeg: killed before its first frame, its stdout just ends.
+    That is not a routine drop to reconnect after."""
+    queue, spawned = printer_stream
+    process = _Proc(b"", b"")
+    process.stderr = stream_of(BANNER + REDIRECT_LOG)
+    process.stdout = asyncio.StreamReader()
+
+    def _kill():
+        process._finish(-9)
+        process.stdout.feed_eof()
+
+    process.kill = MagicMock(side_effect=_kill)
+    queue.append(process)
+
+    assert await _frames() == 0
+    process.kill.assert_called()
+    assert len(spawned) == 1
+
+
+@pytest.mark.asyncio
+async def test_a_redirect_logged_after_stdout_ended_still_stops_the_stream(printer_stream):
+    """The two pipes are read independently: stdout's end can be seen before
+    the log pump reaches the redirect line."""
+    queue, spawned = printer_stream
+    process = _Proc(b"", b"")
+    process.stderr = asyncio.StreamReader()
+    queue.append(process)
+
+    async def _late_log():
+        await asyncio.sleep(0.2)
+        process.stderr.feed_data(BANNER + REDIRECT_LOG)
+        process.stderr.feed_eof()
+
+    feeder = asyncio.create_task(_late_log())
+    assert await _frames() == 0
+    await feeder
+    assert len(spawned) == 1

backend/tests/unit/test_refused_header_redaction.py

@@ -0,0 +1,155 @@
+"""A refused header's value must not reach the live logs.
+
+An HTTP client that refuses a header quotes its value in the error, and that
+value is commonly a token (Home Assistant's ``Authorization: Bearer ...`` with
+a stray space or newline). Services log such errors with ``%s`` or with their
+traceback, so the text went to the console, ``docker logs`` and bambuddy.log;
+only the support download masked it. ``RefusedHeaderRedactFilter`` masks it
+before any handler writes the line.
+"""
+
+from __future__ import annotations
+
+import io
+import logging
+import socket
+
+import httpx
+import pytest
+
+from backend.app.core.logging_filters import RefusedHeaderRedactFilter
+
+# Made up. Never paste a value from a real log here.
+SECRET = "hdr-live-value-1"
+
+
+@pytest.fixture
+def logged():
+    """A logger writing through a handler that carries the filter, as main.py's do."""
+    logger = logging.getLogger("test.refused_header_redaction")
+    stream = io.StringIO()
+    handler = logging.StreamHandler(stream)
+    handler.setFormatter(logging.Formatter("%(levelname)s %(message)s"))
+    handler.addFilter(RefusedHeaderRedactFilter())
+    logger.addHandler(handler)
+    logger.setLevel(logging.DEBUG)
+    logger.propagate = False
+    try:
+        yield logger, stream
+    finally:
+        logger.removeHandler(handler)
+
+
+def _refused_header_error() -> Exception:
+    """The error httpx really raises for a header value with a trailing space."""
+    with socket.socket() as server:
+        server.bind(("127.0.0.1", 0))
+        server.listen()
+        url = f"http://127.0.0.1:{server.getsockname()[1]}/api/states"
+        try:
+            with httpx.Client(timeout=2) as client:
+                client.get(url, headers={"Authorization": f"Bearer {SECRET} "})
+        except Exception as e:  # noqa: BLE001 -- whatever httpx raises is the point
+            return e
+    raise AssertionError("httpx accepted the header")
+
+
+def test_the_real_httpx_error_quotes_the_value():
+    """Guards the premise: if httpx stops quoting the value, this filter is moot."""
+    assert SECRET in str(_refused_header_error())
+
+
+def test_an_error_logged_as_an_argument_is_masked(logged):
+    logger, stream = logged
+
+    logger.warning("Failed to get HA entity state for %s: %s", "switch.x", _refused_header_error())
+
+    text = stream.getvalue()
+    assert SECRET not in text
+    assert "Failed to get HA entity state for switch.x: Illegal header value [REDACTED]" in text
+
+
+def test_an_error_logged_with_its_traceback_is_masked(logged):
+    logger, stream = logged
+
+    try:
+        raise _refused_header_error()
+    except Exception:
+        logger.exception("Failed to list HA entities")
+
+    text = stream.getvalue()
+    assert SECRET not in text
+    assert "Failed to list HA entities" in text
+    assert "Traceback" in text
+    assert "Illegal header value [REDACTED]" in text
+
+
+def test_a_refused_header_name_is_masked(logged):
+    logger, stream = logged
+
+    logger.error("Send failed: %s", f"Illegal header name b'X-{SECRET}'")
+
+    assert SECRET not in stream.getvalue()
+    assert "Send failed: Illegal header name [REDACTED]" in stream.getvalue()
+
+
+def test_other_records_are_left_as_they_are():
+    record = logging.LogRecord("x", logging.INFO, "", 0, "%d items in %s", (3, "queue"), None)
+
+    assert RefusedHeaderRedactFilter().filter(record) is True
+    assert record.msg == "%d items in %s" and record.args == (3, "queue")
+
+
+def test_main_attaches_the_filter_where_lines_are_written():
+    """Root's handlers (console, bambuddy.log) and uvicorn's loggers, which do
+    not hand records to root."""
+    import backend.app.main as main
+
+    handlers = [main.console_handler]
+    if getattr(main, "file_handler", None) is not None:
+        handlers.append(main.file_handler)
+    for handler in handlers:
+        assert any(isinstance(f, RefusedHeaderRedactFilter) for f in handler.filters), handler
+    for name in ("uvicorn.access", "uvicorn.error"):
+        assert any(isinstance(f, RefusedHeaderRedactFilter) for f in logging.getLogger(name).filters), name
+
+
+def test_a_record_whose_arguments_do_not_fit_does_not_raise_from_the_filter():
+    """A bad format string is reported by the handler as before; the filter
+    must not turn it into an exception at the caller's log call."""
+    record = logging.LogRecord("x", logging.INFO, "", 0, "%d items", ("three",), None)
+
+    assert RefusedHeaderRedactFilter().filter(record) is True
+
+
+def test_uvicorns_access_line_keeps_its_argument_tuple():
+    """uvicorn's console AccessFormatter unpacks record.args into five fields,
+    and a request path is the client's to choose."""
+    from uvicorn.logging import AccessFormatter
+
+    logger = logging.getLogger("test.refused_header_redaction.access")
+    stream = io.StringIO()
+    handler = logging.StreamHandler(stream)
+    handler.setFormatter(AccessFormatter('%(client_addr)s - "%(request_line)s" %(status_code)s', use_colors=False))
+    logger.addHandler(handler)
+    logger.addFilter(RefusedHeaderRedactFilter())
+    logger.propagate = False
+    try:
+        logger.warning('%s - "%s %s HTTP/%s" %d', "1.2.3.4:5", "POST", f"/Illegal header value b'{SECRET}'", "1.1", 404)
+    finally:
+        logger.removeHandler(handler)
+
+    text = stream.getvalue()
+    assert SECRET not in text
+    assert text.startswith('1.2.3.4:5 - "POST /Illegal header value [REDACTED]')
+    assert "404" in text
+
+
+def test_a_quote_split_between_format_string_and_argument_is_masked(logged):
+    logger, stream = logged
+
+    logger.warning("Illegal header value %s", f"b'Bearer {SECRET} '")
+
+    assert SECRET not in stream.getvalue()
+    assert "Illegal header value [REDACTED]" in stream.getvalue()
+    assert "Logging error" not in stream.getvalue()

backend/tests/unit/test_refused_header_response_mask.py

@@ -0,0 +1,130 @@
+"""A refused header's value never leaves the API in a JSON response.
+
+An HTTP client refusing a header (a stored token with a stray space or line
+break) quotes the value in its error, and that text reaches responses from many
+places: stored provider and backup errors (including rows written before this
+was masked anywhere), restore and connection-test results, status endpoints.
+``RefusedHeaderResponseMask`` masks it in every JSON body on the way out.
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+from fastapi import FastAPI, HTTPException
+from fastapi.responses import FileResponse, JSONResponse, PlainTextResponse, StreamingResponse
+from httpx import ASGITransport, AsyncClient
+
+from backend.app.core.response_redaction import RefusedHeaderResponseMask, redact_refused_headers_json
+
+# Made up. Never paste a value from a real log here.
+SECRET = "resp-value-5"
+QUOTE = f"Illegal header value b'Bearer {SECRET} '"
+
+
+def _app(manifest=None) -> FastAPI:
+    app = FastAPI()
+
+    @app.api_route("/manifest.json", methods=["GET", "HEAD"])
+    async def manifest_file():
+        return FileResponse(manifest, media_type="application/manifest+json")
+
+    @app.get("/model")
+    async def model():
+        return {"last_error": QUOTE, "other": "kept", "n": 3}
+
+    @app.get("/nested")
+    async def nested():
+        return [{"message": f"LocalProtocolError: {QUOTE}"}, {"message": f'with "quotes" {QUOTE}'}]
+
+    @app.get("/raised")
+    async def raised():
+        raise HTTPException(status_code=502, detail=f"Could not reach it: {QUOTE}")
+
+    @app.get("/explicit")
+    async def explicit():
+        return JSONResponse({"error": QUOTE}, status_code=400)
+
+    @app.get("/text")
+    async def text():
+        return PlainTextResponse(QUOTE)
+
+    @app.get("/stream")
+    async def stream():
+        async def body():
+            yield b'{"error": "'
+            yield QUOTE.encode()
+            yield b'"}'
+
+        return StreamingResponse(body(), media_type="application/json")
+
+    app.add_middleware(RefusedHeaderResponseMask)
+    return app
+
+
+async def _get(path: str):
+    async with AsyncClient(transport=ASGITransport(app=_app()), base_url="http://test") as client:
+        return await client.get(path)
+
+
+@pytest.mark.asyncio
+async def test_a_head_request_keeps_the_length_of_the_body_it_omits(tmp_path):
+    """HEAD sends no body; the length still describes the GET body."""
+    manifest = tmp_path / "manifest.json"
+    manifest.write_text('{"name": "Bambuddy"}')
+    async with AsyncClient(transport=ASGITransport(app=_app(manifest)), base_url="http://test") as client:
+        head = await client.head("/manifest.json")
+        get = await client.get("/manifest.json")
+
+    assert head.headers["content-length"] == get.headers["content-length"] == str(len('{"name": "Bambuddy"}'))
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("path", ["/model", "/nested", "/raised", "/explicit"])
+async def test_json_responses_carry_no_header_value(path):
+    response = await _get(path)
+
+    assert SECRET not in response.text
+    assert "Illegal header value [REDACTED]" in response.text
+    json.loads(response.text)  # still valid JSON
+    assert int(response.headers["content-length"]) == len(response.content)
+
+
+@pytest.mark.asyncio
+async def test_the_rest_of_the_body_is_kept():
+    body = (await _get("/model")).json()
+
+    assert body == {"last_error": "Illegal header value [REDACTED]", "other": "kept", "n": 3}
+
+
+@pytest.mark.asyncio
+async def test_a_streamed_json_body_is_masked_too():
+    response = await _get("/stream")
+
+    assert SECRET not in response.text
+
+
+@pytest.mark.asyncio
+async def test_other_content_types_pass_untouched():
+    """Only JSON is the API's; a text body is left as its route wrote it."""
+    assert (await _get("/text")).text == QUOTE
+
+
+@pytest.mark.parametrize(
+    ("body", "expected"),
+    [
+        (b'{"a": "x Illegal header name b\'X-1\' y", "b": 1}', b'{"a": "x Illegal header name [REDACTED]", "b": 1}'),
+        (b'{"a": "Illegal header value b\'\\"q\\" s\'"}', b'{"a": "Illegal header value [REDACTED]"}'),
+        (b'{"a": "nothing here"}', b'{"a": "nothing here"}'),
+    ],
+)
+def test_the_mask_ends_with_the_json_string(body, expected):
+    assert redact_refused_headers_json(body) == expected
+
+
+def test_main_installs_the_mask_next_to_the_routes():
+    """Innermost: outer middleware re-stream the body in chunks."""
+    from backend.app.main import app
+
+    assert app.user_middleware[-1].cls is RefusedHeaderResponseMask