diff --git a/AGENTS.md b/AGENTS.md index 321a17d..4b55e30 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -25,7 +25,7 @@ Agent-facing JSON path fields compact paths under the current home directory to Core printer interaction is `BambuPrinter` in `bambu_cli/printer.py`. Agents and library users should instantiate it via the `get_printer()` factory (or `RuntimeContext.printer()`), not by manipulating globals. - `BambuPrinter` handles FTPS and MQTT. -- Set `insecure_tls = False` and supply `cert_fingerprint` for MITM protection. Camera TLS (port 6000): the **direct grab** fails closed on a pin mismatch or on an `ssl.SSLError` during the handshake whenever a pin is configured, and without a pin it still refuses to send the access code over the direct connection — but `snapshot` then falls through to the Docker/RTSP streamer path, which has **no TLS verification** of its own; see [SECURITY.md](SECURITY.md) for the residual. MQTT/FTPS without a pin use system CA verification (`CERT_REQUIRED`), which fails for typical Bambu self-signed certs — still effectively fail-closed, but prefer an explicit pin. Pin match/mismatch is enforced when a fingerprint is configured. +- Set `insecure_tls = False` and supply `cert_fingerprint` for MITM protection. Camera TLS (port 6000): the **direct grab** fails closed on a pin mismatch or on an `ssl.SSLError` during the handshake whenever a pin is configured, and without a pin it still refuses to send the access code over the direct connection. The Docker/RTSP streamer (no TLS verification of its own) is **opt-in** via `camera_allow_streamer` or `--allow-camera-streamer`; the default is to abort. See [SECURITY.md](SECURITY.md). MQTT/FTPS without a pin use system CA verification (`CERT_REQUIRED`), which fails for typical Bambu self-signed certs — still effectively fail-closed, but prefer an explicit pin. Pin match/mismatch is enforced when a fingerprint is configured. - `doctor` prints the live certificate fingerprint only when it is not yet pinned (or with `-v`); once pinned it prints a hex-free match confirmation, and the printer's LAN IP is redacted from human output unless `-v` is passed. `--json` always carries `certificate_fingerprint`. In an interactive TTY with no pin, doctor may offer to write `cert_fingerprint` into config.json. It never prompts in `--json` mode or non-interactive runs. - Secret-bearing files are tightened to `0600` automatically on POSIX: config.json on load, and the `access_code_file` when `load_access_code()` reads it. Windows relies on NTFS ACLs (see [SECURITY.md](SECURITY.md)). - Network operations support `timeout` and `retries` through `printer.send_command()` and `printer.status()`. @@ -109,7 +109,7 @@ When adding tests, follow [docs/test-backlog.md](docs/test-backlog.md) and the q ## Camera snapshots for agents -`plate snapshot` captures a JPEG from the printer camera. To avoid stale-photo mistakes — where an agent re-sends a cached file instead of a fresh capture — always pass a fresh `--output` name or use `--unique` (generates `printer_snapshot_Z.jpg`). Every successful `--json` response includes `sha256` (hex digest of the JPEG bytes) and `captured_at` (ISO-8601 UTC); compare these fields before sending the image to a user to verify the capture is genuinely new. +`plate snapshot` captures a JPEG from the printer camera. To avoid stale-photo mistakes — where an agent re-sends a cached file instead of a fresh capture — always pass a fresh `--output` name or use `--unique` (generates `printer_snapshot_Z.jpg`). Every successful `--json` response includes `sha256` (hex digest of the JPEG bytes) and `captured_at` (ISO-8601 UTC); compare these fields before sending the image to a user to verify the capture is genuinely new. Do not pass `--allow-camera-streamer` unless the user asked: that path has no TLS pin. ## Agent usage diff --git a/CHANGELOG.md b/CHANGELOG.md index 4aad5a3..009ff1b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,16 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); version ## [Unreleased] +### Changed + +- **`plate snapshot` no longer falls back to the Docker streamer by default.** + The streamer does not honour `cert_fingerprint`, so a failed direct grab now + aborts unless you set `camera_allow_streamer` in config or pass + `--allow-camera-streamer`. X1-series printers need that opt-in. + `camera_direct_only` still forbids the streamer even when the opt-in is set. + The snapshot command moved out of `protocols/camera.py` into + `bambu_cli.commands.snapshot`. + ## [0.5.0] - 2026-08-05 ### Added diff --git a/README.md b/README.md index 9cf0f67..b737c93 100644 --- a/README.md +++ b/README.md @@ -28,7 +28,7 @@ model URL or file → download → slice (OrcaSlicer) → upload → pri one command: plate job --confirm ``` -**Supports:** any Bambu Lab printer with LAN mode — P1P, P1S, X1C, X1E, A1, A1 Mini. **Hardware-tested on the P1 series (P1P/P1S) only.** The rest speak the same LAN protocols and are expected to work, but are unverified on real hardware — treat them as best-effort and please [open an issue](https://github.com/DLANSAMA/platecli/issues) with what you hit. One caveat: `plate snapshot` grabs the camera directly (no extra software) on P1/A1-class printers, but X1-series cameras need a locally-running Docker streamer container. +**Supports:** any Bambu Lab printer with LAN mode — P1P, P1S, X1C, X1E, A1, A1 Mini. **Hardware-tested on the P1 series (P1P/P1S) only.** The rest speak the same LAN protocols and are expected to work, but are unverified on real hardware — treat them as best-effort and please [open an issue](https://github.com/DLANSAMA/platecli/issues) with what you hit. One caveat: `plate snapshot` grabs the camera directly (no extra software) on P1/A1-class printers. X1-series cameras need a locally-running Docker streamer, and that path is opt-in (`camera_allow_streamer` or `--allow-camera-streamer`) because the streamer does not honour `cert_fingerprint`. ## Install diff --git a/SECURITY.md b/SECURITY.md index fc0cbc3..c76b80a 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -56,10 +56,10 @@ and (b) explicit model downloads you request. Key properties: exists only as a last resort — it is never the default and the CLI warns when it is used. - **Camera (port 6000):** the *direct* grab refuses to proceed if neither pin nor - `insecure_tls` is set. Note that `snapshot` then falls back to the Docker streamer, - which does not honour the pin, so the command as a whole is not fail-closed by - default — set `camera_direct_only: true` to refuse that fallback (see - [Known limitations](#known-limitations)). + `insecure_tls` is set. The Docker streamer is **opt-in** (`camera_allow_streamer` + or `--allow-camera-streamer`) because it does not honour the pin. The default + is fail-closed: a failed direct grab aborts. `camera_direct_only: true` still + forbids the streamer even if the opt-in is also set. - **MQTT / FTPS without a pin:** use system CA verification (`CERT_REQUIRED`), which fails for typical Bambu self-signed certs (effective fail-closed). Prefer an explicit pin for clear errors and uniform policy. @@ -103,7 +103,7 @@ Tracked for hardening; not all are “bugs” in the sense of broken claims. | Topic | Detail | Status | |-------|--------|--------| | **Camera Docker port bind** | Default `camera_port` is now `127.0.0.1:1985:1984`, so the streamer publishes the (unauthenticated) camera feed on **loopback only**. Set `camera_port` to `0.0.0.0:1985:1984` to deliberately expose it on the LAN. Host-qualified specs now parse correctly, `camera_port` is validated, and the CLI warns if a *pre-existing* container is still bound to a non-loopback interface (recreate with `docker rm -f bambu_camera`). | Fixed | -| **Camera pin fallback** | A pinned-fingerprint **mismatch**, and any `ssl.SSLError` from the direct grab — handshake or post-handshake — hard-abort the snapshot when a pin is configured. The `camera_direct_only` config key (default `false`) closes the remaining fallback routes: when set, any failure of the direct port-6000 grab — including no-pin SSLError, non-TLS connection failures (refused/reset/timeout), or a silent no-frame return — refuses to fall back to the Docker streamer and aborts with `EXIT_NETWORK_ERROR`. **What it does not cover:** (a) `camera_direct_only=true` with `insecure_tls=true` is direct-only but **completely unverified** — direct-only ≠ verified; verification requires a `cert_fingerprint`. (b) It stops platecli from using or starting the streamer, but an **already-running `bambu_camera` container keeps serving the unauthenticated feed** — run `docker rm -f bambu_camera` to stop it. (Re-running `plate setup` used to drop this hand-added key, silently disabling the control; setup now preserves every key it does not manage, and reports which ones it kept.) | Fixed | +| **Camera pin fallback** | Default is fail-closed: a failed direct grab does **not** start or use the Docker streamer. X1-series users must set `camera_allow_streamer: true` or pass `--allow-camera-streamer`. A pinned-fingerprint **mismatch**, and any `ssl.SSLError` from the direct grab when a pin is configured, still hard-abort (no streamer even if opted in). `camera_direct_only` remains a forbid switch that wins over the opt-in. **Residuals:** (a) `insecure_tls=true` is unverified whether or not the streamer is used. (b) An already-running `bambu_camera` container keeps serving the unauthenticated feed — run `docker rm -f bambu_camera` to stop it. Setup still preserves unmanaged keys. | Fixed | | **HTTP downloads** | `http://` and `https://` are both accepted. SSRF controls apply; **content integrity** over cleartext HTTP does not (a network attacker can substitute a model). Prefer HTTPS sources. | Residual | | **pause / resume** | Required `--confirm` as of 0.3.0, matching stop/print/delete/gcode. | Fixed | | **Windows secret ACLs** | POSIX `0600` enforcement does not apply on Windows; protect the config directory with NTFS ACLs on shared machines. | Platform residual | diff --git a/bambu_cli/cliparse.py b/bambu_cli/cliparse.py index 5968015..dcbcf22 100644 --- a/bambu_cli/cliparse.py +++ b/bambu_cli/cliparse.py @@ -404,6 +404,16 @@ def build_parser(): "With --output: inserts the timestamp before the file extension." ), ) + p_snap.add_argument( + "--allow-camera-streamer", + action="store_true", + dest="allow_camera_streamer", + help=( + "Allow the unpinned Docker/RTSP camera streamer when the direct grab fails. " + "Required for X1-series. Also settable as camera_allow_streamer in config.json. " + "Ignored when camera_direct_only is set." + ), + ) p_doc = sub.add_parser( "doctor", parents=[get_global_parser()], help="Run health check and discover printer capabilities" diff --git a/bambu_cli/commands/__init__.py b/bambu_cli/commands/__init__.py index 04d5441..23d36cf 100644 --- a/bambu_cli/commands/__init__.py +++ b/bambu_cli/commands/__init__.py @@ -30,7 +30,7 @@ cmd_preflight, cmd_setup, cmd_slice, - cmd_snapshot, ) +from bambu_cli.commands.snapshot import cmd_snapshot # noqa: F401 from bambu_cli.commands.status import cmd_status # noqa: F401 from bambu_cli.commands.tui_cmd import cmd_tui # noqa: F401 diff --git a/bambu_cli/commands/doctor.py b/bambu_cli/commands/doctor.py index b9a2b39..1e93894 100644 --- a/bambu_cli/commands/doctor.py +++ b/bambu_cli/commands/doctor.py @@ -203,13 +203,20 @@ def shown_ip(): "chamber_light": True, "camera_snapshot": ctx.settings.printer_model in _DIRECT_CAMERA_MODELS, "camera_snapshot_note": ( - "P1P/P1S/A1/A1M capture directly from the printer camera and need no Docker; " - "X1-series fall back to the optional BambuP1Streamer container" - if not ctx.settings.camera_direct_only + "camera_direct_only is set: only the direct printer-camera grab is used " + "(P1P/P1S/A1/A1M). The BambuP1Streamer is refused, so X1-series " + "snapshots are unavailable until the option is unset and " + "camera_allow_streamer is set" + if ctx.settings.camera_direct_only else ( - "camera_direct_only is set: only the direct printer-camera grab is used " - "(P1P/P1S/A1/A1M). The BambuP1Streamer fallback is refused, so X1-series " - "snapshots are unavailable until the option is unset" + "P1P/P1S/A1/A1M capture directly from the printer camera and need no Docker; " + "X1-series need camera_allow_streamer (or --allow-camera-streamer) to use " + "the optional BambuP1Streamer container" + if not ctx.settings.camera_allow_streamer + else ( + "P1P/P1S/A1/A1M capture directly; camera_allow_streamer is set so " + "X1-series can use the unpinned BambuP1Streamer if the direct grab fails" + ) ) ), }, diff --git a/bambu_cli/commands/setup_wrappers.py b/bambu_cli/commands/setup_wrappers.py index fccddb1..ddf522d 100644 --- a/bambu_cli/commands/setup_wrappers.py +++ b/bambu_cli/commands/setup_wrappers.py @@ -1,7 +1,5 @@ """Thin command wrappers that delegate to focused packages.""" -from bambu_cli.context import RuntimeContext - def cmd_setup(args): """Interactive or non-interactive printer configuration setup.""" @@ -34,18 +32,6 @@ def cmd_slice(args, **collaborators): return _cmd_slice(args) -def cmd_snapshot(args, ctx=None, **collaborators): - """Capture a camera snapshot using the RTSP Streamer Docker container. - - Extra keyword args are forwarded to ``camera._cmd_snapshot`` (injectable - collaborators: grab_frame, which, subprocess_run, access_code_loader, …). - """ - from bambu_cli.protocols.camera import _cmd_snapshot - - ctx = ctx or RuntimeContext.for_request(args) - _cmd_snapshot(args, ctx=ctx, **collaborators) - - def cmd_preflight(args): """Check local install/config readiness without contacting printer.""" from bambu_cli.setup_cmd import _cmd_preflight diff --git a/bambu_cli/commands/snapshot.py b/bambu_cli/commands/snapshot.py new file mode 100644 index 0000000..d0d085a --- /dev/null +++ b/bambu_cli/commands/snapshot.py @@ -0,0 +1,466 @@ +"""``plate snapshot``: save a JPEG from the printer camera. + +Direct P1/A1 grab lives in ``bambu_cli.protocols.camera``. The Docker/RTSP +streamer is opt-in (``camera_allow_streamer`` / ``--allow-camera-streamer``) +because it does not honour ``cert_fingerprint``. +""" + +from __future__ import annotations + +import datetime +import hashlib +import json +import os +import re +import shutil +import ssl +import subprocess +import tempfile +import time +import urllib.error +import urllib.request +from urllib.parse import urlparse + +from bambu_cli.argutils import namespace_get as _namespace_get +from bambu_cli.config import load_access_code +from bambu_cli.constants import ( + DEFAULT_NETWORK_TIMEOUT, + EXIT_COMMAND_ERROR, + EXIT_CONFIG_ERROR, + EXIT_FILE_ERROR, + EXIT_NETWORK_ERROR, +) +from bambu_cli.context import RuntimeContext +from bambu_cli.errors import BambuError, abort +from bambu_cli.logging_utils import logger, safe_log_error +from bambu_cli.paths import exception_for_message as _exception_for_message +from bambu_cli.paths import expand_path as _expand_path +from bambu_cli.paths import path_for_message as _path_for_message +from bambu_cli.protocols.camera import ( + _SNAPSHOT_SKIP_FRAMES, + _CameraPinMismatch, + _grab_camera_frame_direct, +) +from bambu_cli.utils import _ensure_parent_dir, emit_json, emit_json_error + +_CONTAINER_PORT_RE = re.compile(r"^\d{1,5}(-\d{1,5})?(/(tcp|udp|sctp))?$", re.IGNORECASE) + + +def _utc_stamp(now: datetime.datetime | None = None) -> str: + """Return a compact UTC timestamp string ``YYYYMMDDTHHMMSSz``.""" + if now is None: + now = datetime.datetime.now(datetime.timezone.utc) + return now.strftime("%Y%m%dT%H%M%SZ") + + +def _is_valid_port_number(token): + try: + return 1 <= int(token) <= 65535 + except ValueError: + return False + + +def _camera_port_is_valid(camera_port): + """True if ``camera_port`` is a usable docker ``-p`` value.""" + if not camera_port: + return False + container = camera_port.split(":")[-1] + match = _CONTAINER_PORT_RE.match(container) + if not match: + return False + port_spec = container.split("/", 1)[0] + return all(_is_valid_port_number(p) for p in port_spec.split("-")) + + +def _camera_bind_host(camera_port): + """Host/IP a docker ``-p`` spec binds to; ``""`` means all interfaces.""" + parts = camera_port.split(":") + if len(parts) >= 3: + return ":".join(parts[:-2]).strip("[]") + return "" + + +def _bind_is_loopback(host): + return host.startswith("127.") or host in ("localhost", "::1") + + +def streamer_is_allowed(settings, args=None) -> bool: + """True only when the user opted into the unpinned Docker streamer. + + ``camera_direct_only`` still forbids the streamer even if + ``camera_allow_streamer`` or ``--allow-camera-streamer`` is set. + """ + if getattr(settings, "camera_direct_only", False): + return False + if args is not None and _namespace_get(args, "allow_camera_streamer", False): + return True + return bool(getattr(settings, "camera_allow_streamer", False)) + + +def _streamer_refused_message(settings, fallback_reason: str) -> str: + reason = f" ({fallback_reason})" if fallback_reason else "" + if getattr(settings, "camera_direct_only", False): + return ( + f"Direct camera grab produced no frame{reason} and camera_direct_only is set, so " + "the Docker streamer was refused (the streamer ignores cert_fingerprint). " + "X1-series printers require the streamer: remove camera_direct_only and set " + "camera_allow_streamer in config (or pass --allow-camera-streamer)." + ) + return ( + f"Direct camera grab produced no frame{reason}. The Docker streamer is opt-in " + "because it does not honour cert_fingerprint. X1-series printers need it: set " + "camera_allow_streamer in config.json, or pass --allow-camera-streamer." + ) + + +def _warn_if_running_bind_exposed(ctx, run): + """Warn if an already-running streamer publishes on a non-loopback interface.""" + try: + out = run( + ["docker", "inspect", "-f", "{{json .NetworkSettings.Ports}}", ctx.settings.camera_container_name], + capture_output=True, + text=True, + timeout=5, + ) + if out.returncode != 0: + return + ports = json.loads((out.stdout or "").strip() or "null") + except (FileNotFoundError, subprocess.SubprocessError, ValueError, TypeError): + return + if not isinstance(ports, dict): + return + exposed = set() + for binds in ports.values(): + for bind in binds or []: + host_ip = (bind or {}).get("HostIp", "") if isinstance(bind, dict) else "" + if not _bind_is_loopback(host_ip.strip("[]")): + exposed.add(host_ip or "0.0.0.0") + if exposed: + name = ctx.settings.camera_container_name + logger.warning( + f"The running '{name}' container publishes the camera on non-loopback " + f"interface(s) {', '.join(sorted(exposed))}; anyone on the network can view it. " + f"Run 'docker rm -f {name}' to recreate it with the loopback-only camera_port default." + ) + + +def _require_localhost_streamer_url(args, streamer_url, outpath): + """Fail closed unless the configured camera streamer URL targets localhost.""" + parsed = urlparse(streamer_url) + if parsed.scheme not in ("http", "https") or parsed.hostname not in ("localhost", "127.0.0.1", "::1"): + message = "Security Error: camera_stream_url must point to localhost." + emit_json_error(args, "snapshot", EXIT_CONFIG_ERROR, message, failed_step="validate", output=outpath) + safe_log_error(message) + abort("", exit_code=EXIT_CONFIG_ERROR) + + +def _write_snapshot_atomic(outpath, data): + outdir = os.path.dirname(outpath) or "." + fd, temp_path = tempfile.mkstemp(dir=outdir, suffix=".jpg") + try: + with os.fdopen(fd, "wb") as f: + f.write(data) + os.replace(temp_path, outpath) + except Exception: + if os.path.exists(temp_path): + try: + os.unlink(temp_path) + except OSError: + pass + raise + + +def cmd_snapshot( + args, + ctx=None, + *, + grab_frame=None, + which=None, + subprocess_run=None, + access_code_loader=None, + urlopen=None, + sleep=None, + now=None, +): + """Capture a snapshot from the printer camera. + + P1/A1-class printers are captured directly over the native TLS port-6000 + protocol. The Docker/RTSP streamer is used only when the user opted in. + + Collaborators are injectable so tests pass fakes instead of patching + module globals. + """ + _grab = ( + grab_frame + if grab_frame is not None + else (lambda printer: _grab_camera_frame_direct(printer, skip_frames=_SNAPSHOT_SKIP_FRAMES)) + ) + _which = which if which is not None else shutil.which + _run = subprocess_run if subprocess_run is not None else subprocess.run + _load_code = access_code_loader if access_code_loader is not None else load_access_code + _urlopen = urlopen if urlopen is not None else urllib.request.urlopen + _sleep = sleep if sleep is not None else time.sleep + + ctx = ctx or RuntimeContext.for_request(args) + + unique = bool(_namespace_get(args, "unique", False)) + user_output = args.output if hasattr(args, "output") else None + if unique: + stamp = _utc_stamp(now) + if user_output: + base, ext = os.path.splitext(user_output) + resolved_output = f"{base}_{stamp}{ext}" + else: + resolved_output = f"printer_snapshot_{stamp}.jpg" + else: + resolved_output = user_output or "printer_snapshot.jpg" + outpath = _expand_path(resolved_output) + if outpath.startswith("-"): + message = f"Invalid output path: {_path_for_message(outpath)}" + emit_json_error(args, "snapshot", EXIT_FILE_ERROR, message, failed_step="validate", output=outpath) + safe_log_error(message) + abort("", exit_code=EXIT_FILE_ERROR) + try: + _ensure_parent_dir(outpath) + except BambuError as e: + message = f"Could not prepare output path: {_path_for_message(outpath)}" + emit_json_error( + args, + "snapshot", + (getattr(e, "exit_code", None) or EXIT_FILE_ERROR), + message, + failed_step="validate", + output=outpath, + ) + raise + + _fallback_reason = "" + try: + printer = ctx.printer() + _frame = _grab(printer) + except _CameraPinMismatch as _exc: + message = f"Camera TLS certificate does not match pinned fingerprint: {_exc}" + emit_json_error(args, "snapshot", EXIT_NETWORK_ERROR, message, failed_step="grab", output=outpath) + safe_log_error(message) + abort("", exit_code=EXIT_NETWORK_ERROR) + except ssl.SSLError as _exc: + if not printer.insecure_tls and printer.cert_fingerprint: + message = ( + "Camera TLS error with a cert pin configured " + f"(refusing to fall back to the unverified Docker streamer): {_exc}" + ) + emit_json_error(args, "snapshot", EXIT_NETWORK_ERROR, message, failed_step="grab", output=outpath) + safe_log_error(message) + abort("", exit_code=EXIT_NETWORK_ERROR) + _frame = None + _fallback_reason = str(_exc) + logger.debug(f"Direct camera grab unavailable ({_exc}).") + except BambuError: + raise + except Exception as _exc: + _frame = None + _fallback_reason = str(_exc) + logger.debug(f"Direct camera grab unavailable ({_exc}).") + if _frame: + try: + _write_snapshot_atomic(outpath, _frame) + size = os.path.getsize(outpath) + except OSError as _exc: + message = f"Could not write snapshot: {_path_for_message(outpath)}: {_exception_for_message(_exc)}" + emit_json_error(args, "snapshot", EXIT_FILE_ERROR, message, failed_step="capture", output=outpath) + safe_log_error(message) + abort("", exit_code=EXIT_FILE_ERROR) + captured_at = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + sha256 = hashlib.sha256(_frame).hexdigest() + logger.info(f"\U0001f4f8 Snapshot saved: {_path_for_message(outpath)} ({size // 1024}KB)") + if bool(_namespace_get(args, "json", False)): + emit_json( + { + "status": "saved", + "command": "snapshot", + "output": outpath, + "size_bytes": size, + "captured_at": captured_at, + "sha256": sha256, + "method": "direct", + } + ) + return + + if not streamer_is_allowed(ctx.settings, args): + message = _streamer_refused_message(ctx.settings, _fallback_reason) + emit_json_error(args, "snapshot", EXIT_NETWORK_ERROR, message, failed_step="grab", output=outpath) + safe_log_error(message) + abort("", exit_code=EXIT_NETWORK_ERROR) + + streamer_url = ctx.settings.camera_stream_url + camera_image = ctx.settings.camera_image + + _require_localhost_streamer_url(args, streamer_url, outpath) + + if not _which("docker"): + message = "Docker not found in PATH. Install Docker Desktop (Windows/macOS) or docker-ce (Linux) and retry." + emit_json_error(args, "snapshot", EXIT_CONFIG_ERROR, message, failed_step="docker", output=outpath) + safe_log_error(message) + abort("", exit_code=EXIT_CONFIG_ERROR) + + camera_port = ctx.settings.camera_port + if not _camera_port_is_valid(camera_port): + message = ( + f"Invalid camera_port {camera_port!r}: expected docker port form " + "[HOST:]HOSTPORT:CONTAINERPORT (e.g. 127.0.0.1:1985:1984)." + ) + emit_json_error(args, "snapshot", EXIT_CONFIG_ERROR, message, failed_step="docker", output=outpath) + safe_log_error(message) + abort("", exit_code=EXIT_CONFIG_ERROR) + config_exposed = not _bind_is_loopback(_camera_bind_host(camera_port)) + if config_exposed: + logger.warning( + f"camera_port {camera_port!r} publishes the printer camera on a non-loopback " + f"interface ({_camera_bind_host(camera_port) or 'all interfaces (0.0.0.0)'}); " + "anyone on the network can view it. Set camera_port to '127.0.0.1:1985:1984' " + "to restrict it to this machine." + ) + try: + check = _run( + ["docker", "inspect", "-f", "{{.State.Running}}", ctx.settings.camera_container_name], + capture_output=True, + text=True, + timeout=5, + ) + except (FileNotFoundError, subprocess.SubprocessError) as e: + message = f"Docker not reachable (is the daemon running?): {e}" + emit_json_error(args, "snapshot", EXIT_CONFIG_ERROR, message, failed_step="docker", output=outpath) + safe_log_error(message) + abort("", exit_code=EXIT_CONFIG_ERROR) + if check.returncode != 0 or "true" not in check.stdout: + logger.info("🔄 Starting camera streamer...") + access_code = _load_code() + docker_env = {**os.environ, "PRINTER_ACCESS_CODE": access_code} + try: + _run(["docker", "rm", "-f", ctx.settings.camera_container_name], capture_output=True, timeout=5) + run = _run( + [ + "docker", + "run", + "-d", + "--name", + ctx.settings.camera_container_name, + "-p", + ctx.settings.camera_port, + "-e", + f"PRINTER_ADDRESS={ctx.settings.printer_ip}", + "-e", + "PRINTER_ACCESS_CODE", + camera_image, + ], + capture_output=True, + timeout=10, + env=docker_env, + ) + except (FileNotFoundError, subprocess.SubprocessError) as e: + message = f"Docker not reachable (is the daemon running?): {e}" + emit_json_error( + args, + "snapshot", + EXIT_CONFIG_ERROR, + message, + failed_step="docker", + output=outpath, + camera_image=camera_image, + ) + safe_log_error(message) + abort("", exit_code=EXIT_CONFIG_ERROR) + if run.returncode != 0: + detail = run.stderr or run.stdout or "unknown Docker error" + if isinstance(detail, bytes): + detail = detail.decode(errors="replace") + if access_code: + detail = detail.replace(access_code, "") + if ctx.settings.printer_ip: + detail = detail.replace(ctx.settings.printer_ip, "") + message = f"Could not start camera streamer Docker container using image {camera_image}: {detail.strip()}" + emit_json_error( + args, + "snapshot", + EXIT_CONFIG_ERROR, + message, + failed_step="docker", + output=outpath, + camera_image=camera_image, + ) + safe_log_error(message) + logger.info(" Build the BambuP1Streamer image locally or set `camera_image` in config.json.") + abort("", exit_code=EXIT_CONFIG_ERROR) + + req = urllib.request.Request(streamer_url, headers={"User-Agent": "Mozilla/5.0"}) + for _ in range(30): + try: + with _urlopen(req, timeout=1) as resp: + if resp.status == 200: + break + except urllib.error.URLError: + pass + _sleep(0.5) + elif not config_exposed: + _warn_if_running_bind_exposed(ctx, _run) + + logger.info("📸 Capturing snapshot...") + try: + req = urllib.request.Request(streamer_url, headers={"User-Agent": "Mozilla/5.0"}) + with _urlopen(req, timeout=DEFAULT_NETWORK_TIMEOUT) as resp: + data = resp.read() + _write_snapshot_atomic(outpath, data) + size = os.path.getsize(outpath) + captured_at = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + sha256 = hashlib.sha256(data).hexdigest() + logger.info(f"✅ Snapshot saved: {_path_for_message(outpath)} ({size // 1024}KB)") + if bool(_namespace_get(args, "json", False)): + emit_json( + { + "status": "saved", + "command": "snapshot", + "output": outpath, + "size_bytes": size, + "captured_at": captured_at, + "sha256": sha256, + "camera_image": camera_image, + "docker_container": "bambu_camera", + } + ) + except urllib.error.URLError as e: + message = f"Snapshot network error: {e}" + emit_json_error( + args, + "snapshot", + EXIT_NETWORK_ERROR, + message, + failed_step="streamer", + output=outpath, + camera_image=camera_image, + ) + safe_log_error(message) + logger.info(f" Make sure the {camera_image} Docker container is running and reachable.") + abort("", exit_code=EXIT_NETWORK_ERROR) + except OSError as e: + message = f"Snapshot file error: {_exception_for_message(e)}" + emit_json_error( + args, "snapshot", EXIT_FILE_ERROR, message, failed_step="capture", output=outpath, camera_image=camera_image + ) + safe_log_error(message) + abort("", exit_code=EXIT_FILE_ERROR) + except BambuError: + raise + except Exception as e: + message = f"Snapshot failed: {_exception_for_message(e)}" + emit_json_error( + args, + "snapshot", + EXIT_COMMAND_ERROR, + message, + failed_step="capture", + output=outpath, + camera_image=camera_image, + ) + safe_log_error(message) + abort("", exit_code=EXIT_COMMAND_ERROR) diff --git a/bambu_cli/context.py b/bambu_cli/context.py index aabccb4..fcb2788 100644 --- a/bambu_cli/context.py +++ b/bambu_cli/context.py @@ -100,6 +100,7 @@ class Settings: camera_port: str = "127.0.0.1:1985:1984" # mirrors config.DEFAULT_CAMERA_PORT camera_stream_url: str = "" camera_direct_only: bool = False + camera_allow_streamer: bool = False allow_private_ips: bool = False @classmethod @@ -153,6 +154,9 @@ def from_config(cls, cfg: dict[str, Any]) -> Settings: # security opt-in, so it must survive across invocations. bool() so a JSON # string cannot sneak through truthy. camera_direct_only=bool(cfg.get("camera_direct_only", False)), + # Sticky opt-in for the unpinned Docker streamer. Default false: snapshot + # no longer falls through to Docker unless the user asked. + camera_allow_streamer=bool(cfg.get("camera_allow_streamer", False)), # Always false from config: private-IP downloads are a per-invocation # CLI override only (``--allow-private-ips``), never a sticky config key. allow_private_ips=False, diff --git a/bambu_cli/protocols/camera.py b/bambu_cli/protocols/camera.py index aebc011..ca3951c 100644 --- a/bambu_cli/protocols/camera.py +++ b/bambu_cli/protocols/camera.py @@ -1,160 +1,36 @@ -"""Camera snapshot capture: direct P1/A1 port-6000 TLS grab with a -BambuP1Streamer Docker fallback for X1-series printers. +"""Direct P1/A1 port-6000 TLS camera grab. -Collaborators (socket connect, SSL context factory, frame grabber, docker -runner, docker-which, access-code loader) are injectable so tests pass fakes -instead of patching module globals. +This module is the transport: open a pinned (or explicitly insecure) TLS +socket, authenticate, and return JPEG bytes. Docker, JSON envelopes, and the +``snapshot`` command live in ``bambu_cli.commands.snapshot``. """ -import datetime -import hashlib -import json -import os -import re -import shutil +from __future__ import annotations + import socket import ssl import struct -import subprocess -import tempfile -import time -import urllib.error -import urllib.request -from urllib.parse import urlparse -from bambu_cli.argutils import namespace_get as _namespace_get -from bambu_cli.config import load_access_code -from bambu_cli.constants import ( - DEFAULT_NETWORK_TIMEOUT, - EXIT_COMMAND_ERROR, - EXIT_CONFIG_ERROR, - EXIT_FILE_ERROR, - EXIT_NETWORK_ERROR, -) -from bambu_cli.context import RuntimeContext -from bambu_cli.errors import BambuError, abort -from bambu_cli.logging_utils import logger, safe_log_error -from bambu_cli.paths import exception_for_message as _exception_for_message -from bambu_cli.paths import expand_path as _expand_path -from bambu_cli.paths import path_for_message as _path_for_message -from bambu_cli.utils import _ensure_parent_dir, emit_json, emit_json_error +from bambu_cli.constants import EXIT_NETWORK_ERROR +from bambu_cli.errors import BambuError # The port-6000 camera stream's first frames can be stale (buffered from a # previous connection); skip a few so the snapshot reflects the current scene. _SNAPSHOT_SKIP_FRAMES = 5 -def _utc_stamp(now: "datetime.datetime | None" = None) -> str: - """Return a compact UTC timestamp string ``YYYYMMDDTHHMMSSz``. - - ``now`` is injectable so tests can supply a fixed datetime instead of - reading the wall clock (avoids fragile time-dependent assertions). - """ - if now is None: - now = datetime.datetime.now(datetime.timezone.utc) - return now.strftime("%Y%m%dT%H%M%SZ") - - class _CameraPinMismatch(BambuError): """The camera TLS cert does not match the pinned ``cert_fingerprint``. A pinned fingerprint is an explicit security control, so a mismatch must - hard-abort rather than fall back to the Docker streamer path (which would - connect to the printer without honoring the pin — a silent downgrade). This - is distinct from a missing pin or an ordinary connection failure, both of - which legitimately fall through to the streamer. + hard-abort rather than fall back to the Docker streamer (which would + connect to the printer without honoring the pin). """ exit_code = EXIT_NETWORK_ERROR failed_step = "grab" -# Container-port token of a docker ``-p`` spec: a port (or range) plus optional -# protocol suffix, e.g. ``1984``, ``1984/tcp``, ``1984-1989/udp``. The digit -# groups are only bounded to 1-5 characters here; the actual 1-65535 range -# check happens in `_camera_port_is_valid` since a regex can't express it -# cleanly (and \d{1,5} alone lets 99999 through). -_CONTAINER_PORT_RE = re.compile(r"^\d{1,5}(-\d{1,5})?(/(tcp|udp|sctp))?$", re.IGNORECASE) - - -def _is_valid_port_number(token): - """True if ``token`` is a decimal integer in the valid TCP/UDP port range - (1-65535).""" - try: - return 1 <= int(token) <= 65535 - except ValueError: - return False - - -def _camera_port_is_valid(camera_port): - """True if ``camera_port`` is a usable docker ``-p`` value. Only the - container port (the last colon field) is strictly checked; the optional host - IP/port fields are left for docker to validate (and go list-form into argv, - so there is no injection risk). Turns a confusing docker error into a clear - config error for the common typo cases (empty value, missing container port, - or a port number outside 1-65535). - """ - if not camera_port: - return False - container = camera_port.split(":")[-1] - match = _CONTAINER_PORT_RE.match(container) - if not match: - return False - port_spec = container.split("/", 1)[0] - return all(_is_valid_port_number(p) for p in port_spec.split("-")) - - -def _camera_bind_host(camera_port): - """Host/IP a docker ``-p`` spec binds to; ``""`` means all interfaces. - - Form is ``[HOST:]HOSTPORT:CONTAINERPORT``, so the host is everything before - the last two colon fields (handles bracketed IPv6, which is then unbracketed). - """ - parts = camera_port.split(":") - if len(parts) >= 3: - return ":".join(parts[:-2]).strip("[]") - return "" - - -def _bind_is_loopback(host): - return host.startswith("127.") or host in ("localhost", "::1") - - -def _warn_if_running_bind_exposed(ctx, run): - """Warn if an already-running streamer container publishes the camera on a - non-loopback interface — e.g. a container created before the loopback-only - default, whose binding only changes when the container is recreated. Purely - best-effort: any inspect/parse failure is ignored (it is only a warning). - """ - try: - out = run( - ["docker", "inspect", "-f", "{{json .NetworkSettings.Ports}}", ctx.settings.camera_container_name], - capture_output=True, - text=True, - timeout=5, - ) - if out.returncode != 0: - return - ports = json.loads((out.stdout or "").strip() or "null") - except (FileNotFoundError, subprocess.SubprocessError, ValueError, TypeError): - return - if not isinstance(ports, dict): - return - exposed = set() - for binds in ports.values(): - for bind in binds or []: - host_ip = (bind or {}).get("HostIp", "") if isinstance(bind, dict) else "" - if not _bind_is_loopback(host_ip.strip("[]")): - exposed.add(host_ip or "0.0.0.0") - if exposed: - name = ctx.settings.camera_container_name - logger.warning( - f"The running '{name}' container publishes the camera on non-loopback " - f"interface(s) {', '.join(sorted(exposed))}; anyone on the network can view it. " - f"Run 'docker rm -f {name}' to recreate it with the loopback-only camera_port default." - ) - - def _grab_camera_frame_direct( printer, timeout=12, @@ -165,8 +41,7 @@ def _grab_camera_frame_direct( ): """Grab one JPEG frame from a P1/A1 printer camera using Bambu's native TLS port-6000 protocol (the same one Bambu Studio uses). Returns JPEG bytes, or - None if no frame is obtained. Requires no Docker. X1-series use RTSP instead, - so callers should fall back to the Docker/RTSP streamer when this returns None. + None if no frame is obtained. Requires no Docker. X1-series use RTSP instead. ``create_connection`` and ``ssl_context_factory`` default to the real ``socket.create_connection`` / ``ssl.create_default_context``; tests inject @@ -202,14 +77,10 @@ def _recv_exact(sock_, n): tls = ctx.wrap_socket(sock, server_hostname=printer.ip) tls.settimeout(timeout) - # TLS Verification if not printer.insecure_tls and printer.cert_fingerprint: from bambu_cli.tlspin import verify_cert_fingerprint der = tls.getpeercert(binary_form=True) - # A mismatch (or unobtainable peer cert) is a security failure, not a - # transient error: raise _CameraPinMismatch so the caller hard-aborts - # instead of silently falling back to the unpinned Docker streamer. verify_cert_fingerprint(der, printer.cert_fingerprint, exc_factory=_CameraPinMismatch) elif not printer.insecure_tls and not printer.cert_fingerprint: raise ssl.SSLError( @@ -223,13 +94,8 @@ def _recv_exact(sock_, n): hdr = _recv_exact(tls, 16) size = int.from_bytes(hdr[0:4], "little") if size <= 0: - # Empty/keepalive frame: nothing to drain, just read the next header. continue if size > 12_000_000: - # Implausible frame length means the stream is desynced — the body - # we'd skip would be misread as the next header, so every later - # iteration reads garbage. Abandon the direct grab and let the - # caller fall back to the Docker streamer instead. break data = _recv_exact(tls, size) if data[:2] == b"\xff\xd8" and data[-2:] == b"\xff\xd9": @@ -239,397 +105,8 @@ def _recv_exact(sock_, n): return last_frame return last_frame finally: - # wrap_socket() detaches the underlying fd into the SSLSocket, so on the - # success path closing `sock` is a no-op and the real fd would leak (a - # ResourceWarning under GC, an fd leak in a long-lived process). Close - # whichever object still owns the fd: `tls` once wrapped, else `sock` - # (wrap_socket raised before detaching). closer = tls if tls is not None else sock try: closer.close() except Exception: pass - - -def _require_localhost_streamer_url(args, streamer_url, outpath): - """Fail closed unless the configured camera streamer URL targets localhost. - - Called before any request is issued to the URL (readiness polling included) - so a misconfigured non-local ``camera_stream_url`` can never trigger - outbound requests. - """ - parsed = urlparse(streamer_url) - if parsed.scheme not in ("http", "https") or parsed.hostname not in ("localhost", "127.0.0.1", "::1"): - message = "Security Error: camera_stream_url must point to localhost." - emit_json_error(args, "snapshot", EXIT_CONFIG_ERROR, message, failed_step="validate", output=outpath) - safe_log_error(message) - abort("", exit_code=EXIT_CONFIG_ERROR) - - -def _write_snapshot_atomic(outpath, data): - outdir = os.path.dirname(outpath) or "." - fd, temp_path = tempfile.mkstemp(dir=outdir, suffix=".jpg") - try: - with os.fdopen(fd, "wb") as f: - f.write(data) - os.replace(temp_path, outpath) - except Exception: - if os.path.exists(temp_path): - try: - os.unlink(temp_path) - except OSError: - pass - raise - - -def _cmd_snapshot( - args, - ctx=None, - *, - grab_frame=None, - which=None, - subprocess_run=None, - access_code_loader=None, - urlopen=None, - sleep=None, - now=None, -): - """Capture a snapshot from the printer camera. - - P1/A1-class printers are captured directly over the native TLS port-6000 - protocol and need no Docker; X1-series (or any printer where the direct grab - yields no frame) fall back to the optional BambuP1Streamer container. - - Collaborators are injectable so tests pass fakes instead of patching - module globals. Defaults are the real production implementations. - """ - _grab = ( - grab_frame - if grab_frame is not None - else (lambda printer: _grab_camera_frame_direct(printer, skip_frames=_SNAPSHOT_SKIP_FRAMES)) - ) - _which = which if which is not None else shutil.which - _run = subprocess_run if subprocess_run is not None else subprocess.run - _load_code = access_code_loader if access_code_loader is not None else load_access_code - _urlopen = urlopen if urlopen is not None else urllib.request.urlopen - _sleep = sleep if sleep is not None else time.sleep - - ctx = ctx or RuntimeContext.for_request(args) - - # Resolve output path, honouring --unique for agent-safe timestamped names. - unique = bool(_namespace_get(args, "unique", False)) - user_output = args.output if hasattr(args, "output") else None - if unique: - stamp = _utc_stamp(now) - if user_output: - base, ext = os.path.splitext(user_output) - resolved_output = f"{base}_{stamp}{ext}" - else: - resolved_output = f"printer_snapshot_{stamp}.jpg" - else: - resolved_output = user_output or "printer_snapshot.jpg" - outpath = _expand_path(resolved_output) - if outpath.startswith("-"): - message = f"Invalid output path: {_path_for_message(outpath)}" - emit_json_error(args, "snapshot", EXIT_FILE_ERROR, message, failed_step="validate", output=outpath) - safe_log_error(message) - abort("", exit_code=EXIT_FILE_ERROR) - try: - _ensure_parent_dir(outpath) - except BambuError as e: - message = f"Could not prepare output path: {_path_for_message(outpath)}" - emit_json_error( - args, - "snapshot", - (getattr(e, "exit_code", None) or EXIT_FILE_ERROR), - message, - failed_step="validate", - output=outpath, - ) - raise - - # --- Primary path: direct P1/A1 camera grab (no Docker). Falls through to the - # Docker/RTSP streamer below for X1-series or if no frame is obtained. --- - # Why the direct grab gave up, when it raised ("" when it simply returned no - # frame). Reported by the camera_direct_only gate below so the refusal names an - # actual cause. Held as str rather than the exception: it is only ever - # interpolated, and a plain str needs no annotation -- an ``Exception | None`` one - # would force a choice between ruff's UP037 and python_compat_smoke's PEP 604 - # check, since this module has no ``from __future__ import annotations``. - _fallback_reason = "" - try: - printer = ctx.printer() - _frame = _grab(printer) - except _CameraPinMismatch as _exc: - # A pinned fingerprint that does not match is a security failure, not a - # "this printer needs Docker" signal: fail closed instead of silently - # falling back to the streamer (which would ignore the pin). - message = f"Camera TLS certificate does not match pinned fingerprint: {_exc}" - emit_json_error(args, "snapshot", EXIT_NETWORK_ERROR, message, failed_step="grab", output=outpath) - safe_log_error(message) - abort("", exit_code=EXIT_NETWORK_ERROR) - except ssl.SSLError as _exc: - # A TLS handshake failure (e.g. from wrap_socket()) is a normal signal to - # fall back to Docker when no pin is configured -- but when a pin *is* - # configured, an attacker able to interfere with the port-6000 handshake - # could otherwise defeat the pin simply by breaking TLS instead of - # presenting a mismatched cert. Treat that case the same as a pin - # mismatch: fail closed, no Docker fallthrough. This deliberately covers - # post-handshake SSLErrors too (a truncation attack is indistinguishable - # from a flaky read, and the streamer would be unpinned). - if not printer.insecure_tls and printer.cert_fingerprint: - message = f"Camera TLS error with a cert pin configured (refusing to fall back to the unverified Docker streamer): {_exc}" - emit_json_error(args, "snapshot", EXIT_NETWORK_ERROR, message, failed_step="grab", output=outpath) - safe_log_error(message) - abort("", exit_code=EXIT_NETWORK_ERROR) - _frame = None - _fallback_reason = str(_exc) - logger.debug(f"Direct camera grab unavailable ({_exc}); trying Docker streamer.") - except BambuError: - # A domain abort (e.g. ctx.printer() -> load_access_code() rejecting a - # malformed access_code with EXIT_CONFIG_ERROR) is a hard config error, - # NOT a "this printer needs Docker" signal. Let it propagate to cli.py - # with its own exit code instead of demoting it to a debug log and - # silently falling through to the Docker path (mirrors the Docker path's - # own `except BambuError: raise` guard). - raise - except Exception as _exc: - _frame = None - _fallback_reason = str(_exc) - logger.debug(f"Direct camera grab unavailable ({_exc}); trying Docker streamer.") - if _frame: - try: - _write_snapshot_atomic(outpath, _frame) - size = os.path.getsize(outpath) - except OSError as _exc: - # Disk-full / permission-denied / a directory removed after - # _ensure_parent_dir succeeded: emit a structured file error and exit - # EXIT_FILE_ERROR, matching the Docker path — rather than letting the - # OSError escape to cli.py's generic handler (traceback + wrong - # EXIT_COMMAND_ERROR, no JSON error object for --json consumers). - message = f"Could not write snapshot: {_path_for_message(outpath)}: {_exception_for_message(_exc)}" - emit_json_error(args, "snapshot", EXIT_FILE_ERROR, message, failed_step="capture", output=outpath) - safe_log_error(message) - abort("", exit_code=EXIT_FILE_ERROR) - captured_at = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - sha256 = hashlib.sha256(_frame).hexdigest() - logger.info(f"\U0001f4f8 Snapshot saved: {_path_for_message(outpath)} ({size // 1024}KB)") - if bool(_namespace_get(args, "json", False)): - emit_json( - { - "status": "saved", - "command": "snapshot", - "output": outpath, - "size_bytes": size, - "captured_at": captured_at, - "sha256": sha256, - "method": "direct", - } - ) - return - - # --- Single fail-closed choke point for camera_direct_only. --- - # Reached only when the direct grab produced no frame. Deliberately placed here - # rather than inside the except arms above, because the direct grab gives up via - # many routes that all land here: no cert_fingerprint configured, insecure_tls - # set (which skips verification entirely), a non-TLS network error such as a - # refused/reset port 6000, a desynced stream, or 30 headers with no valid JPEG on - # an otherwise verified connection. Gating once means a future route cannot - # silently reopen the downgrade to the unpinned streamer. - if ctx.settings.camera_direct_only: - _reason = f" ({_fallback_reason})" if _fallback_reason else "" - message = ( - f"Direct camera grab produced no frame{_reason} and camera_direct_only is set, so " - "falling back to the Docker streamer was refused (the streamer ignores " - "cert_fingerprint). X1-series printers require the streamer: unset " - "camera_direct_only in your config to restore snapshots on those models." - ) - emit_json_error(args, "snapshot", EXIT_NETWORK_ERROR, message, failed_step="grab", output=outpath) - safe_log_error(message) - abort("", exit_code=EXIT_NETWORK_ERROR) - - streamer_url = ctx.settings.camera_stream_url - camera_image = ctx.settings.camera_image - - # Fail closed before ANY request is issued to the streamer URL — including - # the readiness-polling loop below — so a misconfigured non-local - # camera_stream_url can never trigger outbound (SSRF-shaped) requests. - _require_localhost_streamer_url(args, streamer_url, outpath) - - # Check if streamer container is running, start if needed - if not _which("docker"): - message = "Docker not found in PATH. Install Docker Desktop (Windows/macOS) or docker-ce (Linux) and retry." - emit_json_error(args, "snapshot", EXIT_CONFIG_ERROR, message, failed_step="docker", output=outpath) - safe_log_error(message) - abort("", exit_code=EXIT_CONFIG_ERROR) - - camera_port = ctx.settings.camera_port - if not _camera_port_is_valid(camera_port): - message = ( - f"Invalid camera_port {camera_port!r}: expected docker port form " - "[HOST:]HOSTPORT:CONTAINERPORT (e.g. 127.0.0.1:1985:1984)." - ) - emit_json_error(args, "snapshot", EXIT_CONFIG_ERROR, message, failed_step="docker", output=outpath) - safe_log_error(message) - abort("", exit_code=EXIT_CONFIG_ERROR) - config_exposed = not _bind_is_loopback(_camera_bind_host(camera_port)) - if config_exposed: - logger.warning( - f"camera_port {camera_port!r} publishes the printer camera on a non-loopback " - f"interface ({_camera_bind_host(camera_port) or 'all interfaces (0.0.0.0)'}); " - "anyone on the network can view it. Set camera_port to '127.0.0.1:1985:1984' " - "to restrict it to this machine." - ) - try: - check = _run( - ["docker", "inspect", "-f", "{{.State.Running}}", ctx.settings.camera_container_name], - capture_output=True, - text=True, - timeout=5, - ) - except (FileNotFoundError, subprocess.SubprocessError) as e: - message = f"Docker not reachable (is the daemon running?): {e}" - emit_json_error(args, "snapshot", EXIT_CONFIG_ERROR, message, failed_step="docker", output=outpath) - safe_log_error(message) - abort("", exit_code=EXIT_CONFIG_ERROR) - if check.returncode != 0 or "true" not in check.stdout: - logger.info("🔄 Starting camera streamer...") - access_code = _load_code() - # Pass the access code via the child environment (the `-e NAME` form with - # no value tells docker to read it from our env) rather than embedding it - # in argv, so the secret never appears in the process list (`ps`). - docker_env = {**os.environ, "PRINTER_ACCESS_CODE": access_code} - try: - _run(["docker", "rm", "-f", ctx.settings.camera_container_name], capture_output=True, timeout=5) - run = _run( - [ - "docker", - "run", - "-d", - "--name", - ctx.settings.camera_container_name, - "-p", - ctx.settings.camera_port, - "-e", - f"PRINTER_ADDRESS={ctx.settings.printer_ip}", - "-e", - "PRINTER_ACCESS_CODE", - camera_image, - ], - capture_output=True, - timeout=10, - env=docker_env, - ) - except (FileNotFoundError, subprocess.SubprocessError) as e: - message = f"Docker not reachable (is the daemon running?): {e}" - emit_json_error( - args, - "snapshot", - EXIT_CONFIG_ERROR, - message, - failed_step="docker", - output=outpath, - camera_image=camera_image, - ) - safe_log_error(message) - abort("", exit_code=EXIT_CONFIG_ERROR) - if run.returncode != 0: - detail = run.stderr or run.stdout or "unknown Docker error" - if isinstance(detail, bytes): - detail = detail.decode(errors="replace") - if access_code: - detail = detail.replace(access_code, "") - if ctx.settings.printer_ip: - detail = detail.replace(ctx.settings.printer_ip, "") - message = f"Could not start camera streamer Docker container using image {camera_image}: {detail.strip()}" - emit_json_error( - args, - "snapshot", - EXIT_CONFIG_ERROR, - message, - failed_step="docker", - output=outpath, - camera_image=camera_image, - ) - safe_log_error(message) - logger.info(" Build the BambuP1Streamer image locally or set `camera_image` in config.json.") - abort("", exit_code=EXIT_CONFIG_ERROR) - - # Polling to wait for stream to connect (up to 15 seconds) - req = urllib.request.Request(streamer_url, headers={"User-Agent": "Mozilla/5.0"}) - for _ in range(30): - try: - with _urlopen(req, timeout=1) as resp: - if resp.status == 200: - break - except urllib.error.URLError: - pass - _sleep(0.5) - elif not config_exposed: - # Container already running: its published port was fixed at creation - # time, so the loopback-only default only takes effect on recreation. - # Warn if a pre-existing container is still exposed. Skipped when the - # configured value already warned above (avoid duplicate noise). - _warn_if_running_bind_exposed(ctx, _run) - - logger.info("📸 Capturing snapshot...") - try: - # streamer_url was already validated as localhost-only before polling. - req = urllib.request.Request(streamer_url, headers={"User-Agent": "Mozilla/5.0"}) - # Use standard urlopen for localhost streamer to bypass SSRF protections - with _urlopen(req, timeout=DEFAULT_NETWORK_TIMEOUT) as resp: - data = resp.read() - _write_snapshot_atomic(outpath, data) - size = os.path.getsize(outpath) - captured_at = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - sha256 = hashlib.sha256(data).hexdigest() - logger.info(f"✅ Snapshot saved: {_path_for_message(outpath)} ({size // 1024}KB)") - if bool(_namespace_get(args, "json", False)): - emit_json( - { - "status": "saved", - "command": "snapshot", - "output": outpath, - "size_bytes": size, - "captured_at": captured_at, - "sha256": sha256, - "camera_image": camera_image, - "docker_container": "bambu_camera", - } - ) - except urllib.error.URLError as e: - message = f"Snapshot network error: {e}" - emit_json_error( - args, - "snapshot", - EXIT_NETWORK_ERROR, - message, - failed_step="streamer", - output=outpath, - camera_image=camera_image, - ) - safe_log_error(message) - logger.info(f" Make sure the {camera_image} Docker container is running and reachable.") - abort("", exit_code=EXIT_NETWORK_ERROR) - except OSError as e: - message = f"Snapshot file error: {_exception_for_message(e)}" - emit_json_error( - args, "snapshot", EXIT_FILE_ERROR, message, failed_step="capture", output=outpath, camera_image=camera_image - ) - safe_log_error(message) - abort("", exit_code=EXIT_FILE_ERROR) - except BambuError: - raise - except Exception as e: - message = f"Snapshot failed: {_exception_for_message(e)}" - emit_json_error( - args, - "snapshot", - EXIT_COMMAND_ERROR, - message, - failed_step="capture", - output=outpath, - camera_image=camera_image, - ) - safe_log_error(message) - abort("", exit_code=EXIT_COMMAND_ERROR) diff --git a/docs/api.md b/docs/api.md index d6cfedb..24f3209 100644 --- a/docs/api.md +++ b/docs/api.md @@ -391,6 +391,8 @@ Schema: [`snapshot.json`](schemas/snapshot.json) — `"status": "saved"`, `"outp `"size_bytes"`, `"captured_at"` (ISO-8601 UTC), `"sha256"` (hex digest of JPEG bytes), plus `method` (`direct`) or Docker-related fields when the streamer path is used. Use `--unique` to get a timestamped filename so repeated captures never overwrite/confuse. +The Docker streamer path is opt-in (`--allow-camera-streamer` or +`camera_allow_streamer` in config); without it a failed direct grab is an error. Agents should compare `sha256` / `captured_at` to verify a new frame was captured before sending the image to a user. diff --git a/docs/manual.md b/docs/manual.md index 5738915..33b6853 100644 --- a/docs/manual.md +++ b/docs/manual.md @@ -378,11 +378,12 @@ or manually. | `nozzle` / `nozzle_size` | no | `0.4` | Nozzle diameter string | | `orca_slicer` | for slice | auto-detect | Path to OrcaSlicer binary | | `profiles_dir` | for slice | auto-detect | Path to OrcaSlicer `profiles/BBL` directory | -| `camera_image` | no | `bambu_p1_streamer` | Docker image for X1-style streamer fallback | +| `camera_image` | no | `bambu_p1_streamer` | Docker image for the opt-in X1-style streamer | | `camera_container_name` | no | `bambu_camera` | Docker container name | | `camera_port` | no | `127.0.0.1:1985:1984` | Docker publish mapping; loopback-only by default. Set to `0.0.0.0:1985:1984` to expose on the LAN (see [SECURITY.md](https://github.com/DLANSAMA/platecli/blob/main/SECURITY.md)) | | `camera_stream_url` | no | derived | Must be localhost if set; used for Docker frame fetch | -| `camera_direct_only` | no | `false` | When `true`, disables the Docker/RTSP streamer fallback — if the direct port-6000 grab fails for any reason, `snapshot` aborts instead of falling through. X1-series printers need the streamer; unset this to restore snapshots. Does not imply TLS verification — combine with `cert_fingerprint` for a verified direct-only camera. | +| `camera_allow_streamer` | no | `false` | Opt in to the unpinned Docker/RTSP streamer when the direct grab fails. Required for X1-series. Also available as `--allow-camera-streamer`. | +| `camera_direct_only` | no | `false` | When `true`, forbids the Docker streamer even if `camera_allow_streamer` is set. Does not imply TLS verification — combine with `cert_fingerprint`. | | Timeouts | no | package defaults | Optional `network_timeout`, `slicer_timeout`, `command_timeout`, `upload_timeout` (seconds) | \* Either `access_code_file` or `access_code` is required. Inline `access_code` is deprecated and will be removed in a future release. diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index c760361..7bd7657 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -25,7 +25,7 @@ the `plate doctor` output — but check first that no access code is visible. - [zeroconf is not installed, so auto-discovery is disabled](#zeroconf-is-not-installed-so-auto-discovery-is-disabled) - [Docker not found in PATH when taking a snapshot](#docker-not-found-in-path-when-taking-a-snapshot) - [The camera TLS certificate does not match the pinned fingerprint](#the-camera-tls-certificate-does-not-match-the-pinned-fingerprint) -- [Snapshot fails with "falling back to the Docker streamer was refused"](#snapshot-fails-with-falling-back-to-the-docker-streamer-was-refused) +- [Snapshot fails and mentions `camera_allow_streamer`](#snapshot-fails-and-mentions-camera_allow_streamer) - [Printer returned only partial status updates](#printer-returned-only-partial-status-updates) - [The file was not found on the printer](#the-file-was-not-found-on-the-printer) - [Timed out waiting for the printer to acknowledge print start](#timed-out-waiting-for-the-printer-to-acknowledge-print-start) @@ -336,14 +336,18 @@ If you have no pin at all, `plate` warns that no `cert_fingerprint` is pinned for the camera connection and tells you to run `plate setup` to pin one. Pin it — don't set `insecure_tls`. -## Snapshot fails with "falling back to the Docker streamer was refused" +## Snapshot fails and mentions `camera_allow_streamer` -If you see an error about `camera_direct_only` being set, the direct port-6000 -camera grab failed and `plate` refused to fall back to the Docker streamer because -`camera_direct_only: true` is in your `config.json`. Remove or set it to `false` -to restore the Docker streamer fallback, or diagnose and fix the direct grab -failure (check that port 6000 is reachable with `nc -vz 6000` and -that a `cert_fingerprint` is pinned). +The Docker streamer is opt-in because it does not honour `cert_fingerprint`. +If the direct port-6000 grab failed, `snapshot` aborts unless you set +`camera_allow_streamer: true` in `config.json` or pass `--allow-camera-streamer`. +X1-series printers have no port-6000 camera and need that opt-in. + +If the message names `camera_direct_only`, that key is also set and **forbids** +the streamer even when the opt-in is present. Remove `camera_direct_only` first. + +To diagnose a failed direct grab on P1/A1: check that port 6000 is reachable +with `nc -vz 6000` and that a `cert_fingerprint` is pinned. ## Printer returned only partial status updates diff --git a/tests/test_audit_fixes_pr4_integration.py b/tests/test_audit_fixes_pr4_integration.py index d236d03..25436bc 100644 --- a/tests/test_audit_fixes_pr4_integration.py +++ b/tests/test_audit_fixes_pr4_integration.py @@ -86,7 +86,7 @@ def _snapshot_args(self, tmpdir): @patch("bambu_cli.logging_utils._BACKEND") def test_bambu_error_from_printer_propagates(self, _mock_logger): - from bambu_cli.protocols import camera + from bambu_cli.commands import snapshot as camera tmpdir = tempfile.mkdtemp() args = self._snapshot_args(tmpdir) @@ -98,12 +98,12 @@ def test_bambu_error_from_printer_propagates(self, _mock_logger): fake_ctx.printer.side_effect = BambuError("bad access code", exit_code=13) with self.assertRaises(BambuError) as cm: - camera._cmd_snapshot(args, ctx=fake_ctx) + camera.cmd_snapshot(args, ctx=fake_ctx) self.assertEqual(cm.exception.exit_code, 13) @patch("bambu_cli.logging_utils._BACKEND") def test_direct_write_oserror_becomes_file_error(self, _mock_logger): - from bambu_cli.protocols import camera + from bambu_cli.commands import snapshot as camera from bambu_cli.constants import EXIT_FILE_ERROR tmpdir = tempfile.mkdtemp() @@ -115,7 +115,7 @@ def test_direct_write_oserror_becomes_file_error(self, _mock_logger): with patch.object(camera, "_write_snapshot_atomic", side_effect=OSError("No space left on device")): with self.assertRaises(BambuError) as cm: - camera._cmd_snapshot(args, ctx=fake_ctx, grab_frame=lambda printer: b"\xff\xd8jpegbytes") + camera.cmd_snapshot(args, ctx=fake_ctx, grab_frame=lambda printer: b"\xff\xd8jpegbytes") # A file write failure exits EXIT_FILE_ERROR, not the generic command # error the uncaught OSError would have produced. self.assertEqual(cm.exception.exit_code, EXIT_FILE_ERROR) diff --git a/tests/test_bambu_cli_regressions.py b/tests/test_bambu_cli_regressions.py index 7a744a9..e1ddf89 100644 --- a/tests/test_bambu_cli_regressions.py +++ b/tests/test_bambu_cli_regressions.py @@ -11,7 +11,8 @@ (c) The download success path must be able to resolve `_record_download_success` (it was a NameError) -- bambu_cli/download/downloader.py `_cmd_download`. (d) snapshot must prefer the direct camera grab and NOT shell out to Docker when - a frame is obtained -- bambu_cli/protocols/camera.py `_cmd_snapshot` + `_grab_camera_frame_direct`. + a frame is obtained -- bambu_cli/commands/snapshot.py ``cmd_snapshot`` + + bambu_cli/protocols/camera.py `_grab_camera_frame_direct`. """ import os @@ -323,7 +324,7 @@ def test_c_cmd_download_references_record_download_success_without_nameerror(): def test_d_snapshot_uses_direct_grab_not_docker(): - from bambu_cli.protocols.camera import _cmd_snapshot + from bambu_cli.commands.snapshot import cmd_snapshot as _cmd_snapshot jpeg = b"\xff\xd8" + b"\x00" * 256 + b"\xff\xd9" with tempfile.TemporaryDirectory() as tmpdir: diff --git a/tests/test_camera.py b/tests/test_camera.py index 119fc6f..3756d6a 100644 --- a/tests/test_camera.py +++ b/tests/test_camera.py @@ -3,7 +3,7 @@ import tempfile from unittest.mock import MagicMock, patch -from bambu_cli.protocols.camera import _require_localhost_streamer_url, _write_snapshot_atomic +from bambu_cli.commands.snapshot import _require_localhost_streamer_url, _write_snapshot_atomic class TestCameraBase(unittest.TestCase): @@ -15,7 +15,7 @@ def test_require_localhost_streamer_url_valid(self): _require_localhost_streamer_url(args, "http://[::1]:8080/stream", "out.jpg") # Should not raise any error - @patch("bambu_cli.protocols.camera.abort") + @patch("bambu_cli.commands.snapshot.abort") def test_require_localhost_streamer_url_invalid(self, mock_abort): mock_abort.side_effect = SystemExit(3) args = MagicMock() diff --git a/tests/test_camera_capture.py b/tests/test_camera_capture.py index 7bde198..c890435 100644 --- a/tests/test_camera_capture.py +++ b/tests/test_camera_capture.py @@ -7,24 +7,26 @@ from tests.bambu_test_base import * # noqa: F401,F403 from bambu_cli.errors import BambuError + class TestCameraPortIsValid(unittest.TestCase): def test_rejects_out_of_range_container_port(self): """A container port above 65535 must be rejected: \\d{1,5} alone lets '99999' match the regex even though it is not a valid port number.""" - from bambu_cli.protocols.camera import _camera_port_is_valid + from bambu_cli.commands.snapshot import _camera_port_is_valid self.assertFalse(_camera_port_is_valid("1985:99999")) self.assertFalse(_camera_port_is_valid("0")) self.assertFalse(_camera_port_is_valid("70000-70005")) def test_accepts_valid_container_ports(self): - from bambu_cli.protocols.camera import _camera_port_is_valid + from bambu_cli.commands.snapshot import _camera_port_is_valid self.assertTrue(_camera_port_is_valid("127.0.0.1:1985:1984")) self.assertTrue(_camera_port_is_valid("1984")) self.assertTrue(_camera_port_is_valid("1984/tcp")) self.assertTrue(_camera_port_is_valid("1984-1989/udp")) + class TestGrabCameraFrameDirect(unittest.TestCase): def _mock_net(self): mock_sock = MagicMock() diff --git a/tests/test_cmd_snapshot.py b/tests/test_cmd_snapshot.py index 5d57f07..da5ce46 100644 --- a/tests/test_cmd_snapshot.py +++ b/tests/test_cmd_snapshot.py @@ -5,6 +5,7 @@ from tests.bambu_test_base import * # noqa: F401,F403 from bambu_cli.errors import BambuError + class TestBambuCmdSnapshot(unittest.TestCase): def _logger_patch(self): return patch("bambu_cli.logging_utils.logger", new=MagicMock()) @@ -21,7 +22,10 @@ def test_cmd_snapshot_non_localhost_url_blocked_before_any_request(self): args.output = "snap.jpg" with ( patch("bambu_cli.logging_utils._BACKEND", mock_logger), - settings_ctx(camera_stream_url="http://evil.example.com:8080/frame.jpeg"), + settings_ctx( + camera_stream_url="http://evil.example.com:8080/frame.jpeg", + camera_allow_streamer=True, + ), self.assertRaises((SystemExit, BambuError)) as cm, ): cmd_snapshot( @@ -71,6 +75,7 @@ def test_cmd_snapshot_url_error(self): with ( patch("bambu_cli.logging_utils._BACKEND", mock_logger), + settings_ctx(camera_allow_streamer=True), patch("sys.exit", side_effect=SystemExit(2)), self.assertRaises((SystemExit, BambuError)) as cm, ): @@ -101,6 +106,7 @@ def test_cmd_snapshot_generic_error(self): with ( patch("bambu_cli.logging_utils._BACKEND", mock_logger), + settings_ctx(camera_allow_streamer=True), patch("sys.exit", side_effect=SystemExit(5)), self.assertRaises((SystemExit, BambuError)) as cm, ): @@ -194,10 +200,44 @@ def _grab(printer): any("TLS error with a cert pin configured" in c[0][0] for c in mock_logger.error.call_args_list) ) - def test_cmd_snapshot_ssl_error_without_pin_falls_back_to_docker(self): - """The same ssl.SSLError, but with no pin configured, must still fall - through to the Docker streamer -- this preserves the existing - no-pin-configured fallback behavior.""" + def test_cmd_snapshot_ssl_error_without_pin_fails_closed_by_default(self): + """No pin and no streamer opt-in: an ssl.SSLError must abort, not Docker.""" + import ssl as ssl_mod + + from bambu_cli.commands import cmd_snapshot + from bambu_cli.context import Settings as _Settings + + mock_logger, mock_run, mock_urlopen = MagicMock(), MagicMock(), MagicMock() + args = MagicMock() + args.output = "snap.jpg" + + def _grab(printer): + raise ssl_mod.SSLError("no pin configured") + + with ( + patch("bambu_cli.logging_utils._BACKEND", mock_logger), + settings_ctx(cert_fingerprint=None, insecure_tls=False), + self.assertRaises((SystemExit, BambuError)) as cm, + ): + cmd_snapshot( + args, + grab_frame=_grab, + which=lambda name: "/usr/bin/docker", + subprocess_run=mock_run, + urlopen=mock_urlopen, + ) + + self.assertEqual(getattr(cm.exception, "exit_code", getattr(cm.exception, "code", None)), 2) + mock_run.assert_not_called() + mock_urlopen.assert_not_called() + self.assertTrue( + any("camera_allow_streamer" in c[0][0] for c in mock_logger.error.call_args_list), + "refusal must name the opt-in so X1 users can find it", + ) + self.assertFalse(_Settings().camera_allow_streamer) + + def test_cmd_snapshot_ssl_error_without_pin_uses_streamer_when_opted_in(self): + """The same ssl.SSLError falls through to Docker only when opted in.""" import ssl as ssl_mod from bambu_cli.commands import cmd_snapshot @@ -205,9 +245,9 @@ def test_cmd_snapshot_ssl_error_without_pin_falls_back_to_docker(self): mock_logger = MagicMock() mock_subproc = MagicMock( side_effect=[ - MagicMock(returncode=1), # inspect fails - MagicMock(returncode=0), # rm - MagicMock(returncode=0), # run + MagicMock(returncode=1), + MagicMock(returncode=0), + MagicMock(returncode=0), ] ) mock_response = MagicMock() @@ -225,12 +265,12 @@ def _grab(printer): with ( patch("bambu_cli.logging_utils._BACKEND", mock_logger), - settings_ctx(cert_fingerprint=None, insecure_tls=False), + settings_ctx(cert_fingerprint=None, insecure_tls=False, camera_allow_streamer=True), patch("os.path.exists", return_value=True), patch("os.fdopen", mock_open()), patch("os.unlink"), patch("os.path.getsize", return_value=2048), - patch("bambu_cli.protocols.camera._write_snapshot_atomic"), + patch("bambu_cli.commands.snapshot._write_snapshot_atomic"), patch("builtins.open", new_callable=mock_open), ): cmd_snapshot( @@ -244,11 +284,6 @@ def _grab(printer): ) mock_subproc.assert_called() - # Guards the default: camera_direct_only must be OFF unless opted in, or this - # existing fallback (and every X1-series user) would break. - from bambu_cli.context import Settings as _Settings - - self.assertFalse(_Settings().camera_direct_only) # --- camera_direct_only: the streamer fallback is refused entirely. --- # SECURITY.md promises this option closes the two accepted camera residuals (the @@ -398,7 +433,7 @@ def test_cmd_snapshot_direct_only_direct_success_unaffected(self): with ( patch("bambu_cli.logging_utils._BACKEND", mock_logger), - patch("bambu_cli.protocols.camera._write_snapshot_atomic"), + patch("bambu_cli.commands.snapshot._write_snapshot_atomic"), patch("os.path.getsize", return_value=2048), ): cmd_snapshot( @@ -441,11 +476,12 @@ def test_cmd_snapshot_start_container(self): with ( patch("bambu_cli.logging_utils._BACKEND", mock_logger), + settings_ctx(camera_allow_streamer=True), patch("os.path.exists", return_value=True), patch("os.fdopen", mock_open()), patch("os.unlink"), patch("os.path.getsize", return_value=2048), - patch("bambu_cli.protocols.camera._write_snapshot_atomic"), + patch("bambu_cli.commands.snapshot._write_snapshot_atomic"), patch("builtins.open", new_callable=mock_open), ): cmd_snapshot( @@ -480,7 +516,7 @@ def test_cmd_snapshot_invalid_camera_port_aborts(self): args.output = "snap.jpg" with ( patch("bambu_cli.logging_utils._BACKEND", mock_logger), - settings_ctx(camera_port="not-a-port"), + settings_ctx(camera_port="not-a-port", camera_allow_streamer=True), self.assertRaises((SystemExit, BambuError)) as cm, ): cmd_snapshot( @@ -509,9 +545,9 @@ def test_cmd_snapshot_non_loopback_bind_warns(self): args.output = "snap.jpg" with ( patch("bambu_cli.logging_utils._BACKEND", mock_logger), - patch("bambu_cli.protocols.camera._write_snapshot_atomic"), + patch("bambu_cli.commands.snapshot._write_snapshot_atomic"), patch("os.path.getsize", return_value=1024), - settings_ctx(camera_port="0.0.0.0:1985:1984"), + settings_ctx(camera_port="0.0.0.0:1985:1984", camera_allow_streamer=True), ): cmd_snapshot( args, @@ -546,9 +582,9 @@ def test_cmd_snapshot_running_container_exposed_warns(self): args.output = "snap.jpg" with ( patch("bambu_cli.logging_utils._BACKEND", mock_logger), - patch("bambu_cli.protocols.camera._write_snapshot_atomic"), + patch("bambu_cli.commands.snapshot._write_snapshot_atomic"), patch("os.path.getsize", return_value=1024), - settings_ctx(camera_port="127.0.0.1:1985:1984"), # config is safe + settings_ctx(camera_port="127.0.0.1:1985:1984", camera_allow_streamer=True), ): cmd_snapshot( args, diff --git a/tests/test_context.py b/tests/test_context.py index 89cd69b..39f4ba6 100644 --- a/tests/test_context.py +++ b/tests/test_context.py @@ -72,9 +72,8 @@ def test_settings_from_config_defaults_for_missing_keys(): assert settings.camera_container_name == "bambu_camera" assert settings.camera_port == "127.0.0.1:1985:1984" assert settings.camera_stream_url == "http://localhost:1985/api/frame.jpeg?src=p1s" - # Off unless opted in: enabling it by default would break every X1-series user, - # whose snapshots legitimately require the Docker streamer. assert settings.camera_direct_only is False + assert settings.camera_allow_streamer is False def test_settings_from_config_camera_direct_only_is_a_sticky_config_key(): @@ -89,6 +88,8 @@ def test_settings_from_config_camera_direct_only_is_a_sticky_config_key(): assert context.Settings.from_config({"camera_direct_only": False}).camera_direct_only is False # Coerced, so a JSON string cannot arrive as a non-bool. assert context.Settings.from_config({"camera_direct_only": "yes"}).camera_direct_only is True + assert context.Settings.from_config({"camera_allow_streamer": True}).camera_allow_streamer is True + assert context.Settings.from_config({}).camera_allow_streamer is False # Contrast with the forced-False key, which must keep ignoring config. assert context.Settings.from_config({"allow_private_ips": True}).allow_private_ips is False diff --git a/tests/test_coverage_platform_paths.py b/tests/test_coverage_platform_paths.py index c10df71..81257e9 100644 --- a/tests/test_coverage_platform_paths.py +++ b/tests/test_coverage_platform_paths.py @@ -18,7 +18,7 @@ sys.modules.setdefault("paho.mqtt", _mock_mqtt) sys.modules.setdefault("paho.mqtt.client", _mock_mqtt) -from bambu_cli.protocols import camera as camera_mod # noqa: E402 +from bambu_cli.commands import snapshot as snapshot_mod # noqa: E402 from bambu_cli import commands as commands_mod # noqa: E402 from bambu_cli import config as config_mod # noqa: E402 from bambu_cli import slicer as slicer_mod # noqa: E402 @@ -63,13 +63,8 @@ def _no_gmsh(*_a, **_k): def test_camera_simulation_snapshot(tmp_path, capsys): out = tmp_path / "snap.jpg" args = Namespace(output=str(out), json=True, direct=True) - printer = _test_printer(simulation_mode=True, insecure_tls=True) - with ( - patch("bambu_cli.protocols.camera.get_printer", return_value=printer, create=True), - patch("bambu_cli.printer.get_printer", return_value=printer), - patch.object(camera_mod, "_grab_camera_frame_direct", return_value=b"\xff\xd8\xfffakejpeg"), - ): - camera_mod._cmd_snapshot(args) + with patch.object(snapshot_mod, "_grab_camera_frame_direct", return_value=b"\xff\xd8\xfffakejpeg"): + snapshot_mod.cmd_snapshot(args) assert out.is_file() assert out.read_bytes().startswith(b"\xff\xd8\xff") payload = json.loads(capsys.readouterr().out) diff --git a/tests/test_json_envelope_ordering.py b/tests/test_json_envelope_ordering.py index 5a104e8..03b07dc 100644 --- a/tests/test_json_envelope_ordering.py +++ b/tests/test_json_envelope_ordering.py @@ -37,9 +37,9 @@ def _envelope(capsys): def _camera_localhost_guard(args): - from bambu_cli.protocols import camera + from bambu_cli.commands import snapshot - camera._require_localhost_streamer_url(args, "http://camera.example.com:1984/x", "/tmp/snap.jpg") + snapshot._require_localhost_streamer_url(args, "http://camera.example.com:1984/x", "/tmp/snap.jpg") def _downloader_bad_output_dir(args): diff --git a/tests/test_setup_helpers.py b/tests/test_setup_helpers.py index 28f4eac..11adf78 100644 --- a/tests/test_setup_helpers.py +++ b/tests/test_setup_helpers.py @@ -116,6 +116,7 @@ def test_setup_rerun_preserves_unmanaged_keys(tmp_path, monkeypatch): "serial": "OLDSERIAL", "access_code": "99998888", "camera_direct_only": True, + "camera_allow_streamer": True, "camera_port": "0.0.0.0:1985:1984", "network_timeout": 42, } @@ -126,6 +127,7 @@ def test_setup_rerun_preserves_unmanaged_keys(tmp_path, monkeypatch): # Unmanaged keys survive, values intact. assert data["camera_direct_only"] is True + assert data["camera_allow_streamer"] is True assert data["camera_port"] == "0.0.0.0:1985:1984" assert data["network_timeout"] == 42 # Wizard-owned keys are still updated to the new answers. diff --git a/tests/test_snapshot_output.py b/tests/test_snapshot_output.py index 14b6bc5..8c93ef0 100644 --- a/tests/test_snapshot_output.py +++ b/tests/test_snapshot_output.py @@ -19,7 +19,7 @@ def _snap_args(self, output=None, unique=False): def test_unique_flag_no_output_uses_timestamp(self): """With --unique and no --output, filename is printer_snapshot_.jpg.""" import datetime - from bambu_cli.protocols.camera import _utc_stamp + from bambu_cli.commands.snapshot import _utc_stamp fixed_dt = datetime.datetime(2026, 7, 24, 19, 15, 30, tzinfo=datetime.timezone.utc) stamp = _utc_stamp(fixed_dt) @@ -35,10 +35,10 @@ def _fake_write(path, data): args = self._snap_args(output=None, unique=True) with ( - patch("bambu_cli.protocols.camera._write_snapshot_atomic", side_effect=_fake_write), + patch("bambu_cli.commands.snapshot._write_snapshot_atomic", side_effect=_fake_write), patch("bambu_cli.logging_utils._BACKEND", MagicMock()), patch("os.path.getsize", return_value=1024), - patch("bambu_cli.protocols.camera._ensure_parent_dir"), + patch("bambu_cli.commands.snapshot._ensure_parent_dir"), ): cmd_snapshot( args, @@ -65,10 +65,10 @@ def _fake_write(path, data): args = self._snap_args(output="cam.jpg", unique=True) with ( - patch("bambu_cli.protocols.camera._write_snapshot_atomic", side_effect=_fake_write), + patch("bambu_cli.commands.snapshot._write_snapshot_atomic", side_effect=_fake_write), patch("bambu_cli.logging_utils._BACKEND", MagicMock()), patch("os.path.getsize", return_value=1024), - patch("bambu_cli.protocols.camera._ensure_parent_dir"), + patch("bambu_cli.commands.snapshot._ensure_parent_dir"), ): cmd_snapshot( args, @@ -92,10 +92,10 @@ def _fake_write(path, data): args = self._snap_args(output="myshot.jpg", unique=False) with ( - patch("bambu_cli.protocols.camera._write_snapshot_atomic", side_effect=_fake_write), + patch("bambu_cli.commands.snapshot._write_snapshot_atomic", side_effect=_fake_write), patch("bambu_cli.logging_utils._BACKEND", MagicMock()), patch("os.path.getsize", return_value=1024), - patch("bambu_cli.protocols.camera._ensure_parent_dir"), + patch("bambu_cli.commands.snapshot._ensure_parent_dir"), ): cmd_snapshot( args, @@ -106,6 +106,7 @@ def _fake_write(path, data): self.assertTrue(saved_paths[0].endswith("myshot.jpg")) self.assertNotIn("Z.jpg", saved_paths[0]) + class TestSnapshotJsonMetadata(unittest.TestCase): """captured_at and sha256 appear in --json output on every successful capture.""" @@ -129,11 +130,11 @@ def test_direct_path_json_includes_captured_at_and_sha256(self, capsys=None): buf = io.StringIO() with ( - patch("bambu_cli.protocols.camera._write_snapshot_atomic"), + patch("bambu_cli.commands.snapshot._write_snapshot_atomic"), patch("bambu_cli.logging_utils._BACKEND", MagicMock()), patch("os.path.getsize", return_value=len(frame_data)), - patch("bambu_cli.protocols.camera._ensure_parent_dir"), - patch("bambu_cli.protocols.camera.emit_json", side_effect=lambda d: buf.write(json.dumps(d))), + patch("bambu_cli.commands.snapshot._ensure_parent_dir"), + patch("bambu_cli.commands.snapshot.emit_json", side_effect=lambda d: buf.write(json.dumps(d))), ): cmd_snapshot( args, @@ -165,11 +166,12 @@ def test_docker_path_json_includes_captured_at_and_sha256(self): buf = io.StringIO() with ( - patch("bambu_cli.protocols.camera._write_snapshot_atomic"), + patch("bambu_cli.commands.snapshot._write_snapshot_atomic"), patch("bambu_cli.logging_utils._BACKEND", MagicMock()), patch("os.path.getsize", return_value=len(frame_data)), - patch("bambu_cli.protocols.camera._ensure_parent_dir"), - patch("bambu_cli.protocols.camera.emit_json", side_effect=lambda d: buf.write(json.dumps(d))), + patch("bambu_cli.commands.snapshot._ensure_parent_dir"), + patch("bambu_cli.commands.snapshot.emit_json", side_effect=lambda d: buf.write(json.dumps(d))), + settings_ctx(camera_allow_streamer=True), ): cmd_snapshot( args,