From 865793b6022d1cca83bf3872f2a98c85b88528c3 Mon Sep 17 00:00:00 2001 From: DLANSAMA <258674612+DLANSAMA@users.noreply.github.com> Date: Wed, 12 Aug 2026 17:54:13 -0400 Subject: [PATCH] refactor: split MQTT into tls/cmd/print/monitor and pin via SSLContext MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 860-line mqtt.py is now a facade. TLS construction, command/status, print-ack, and the monitor loop live in sibling modules. Pinning uses PinningSSLContext.wrap_socket, which handshakes then calls verify_cert_fingerprint ā€” no instance wrap_socket monkeypatch. --- CHANGELOG.md | 5 + bambu_cli/protocols/mqtt.py | 905 ++------------------------ bambu_cli/protocols/mqtt_cmd.py | 322 +++++++++ bambu_cli/protocols/mqtt_monitor.py | 161 +++++ bambu_cli/protocols/mqtt_print.py | 205 ++++++ bambu_cli/protocols/mqtt_tls.py | 152 +++++ pyproject.toml | 4 +- tests/test_coverage_platform_paths.py | 10 +- tests/test_doctor_and_safety.py | 2 +- tests/test_netsafety_handlers.py | 10 +- tests/test_protocol_clients.py | 12 +- tests/test_sim_transport_setup.py | 2 +- tests/test_tls_pinning.py | 155 ++--- 13 files changed, 969 insertions(+), 976 deletions(-) create mode 100644 bambu_cli/protocols/mqtt_cmd.py create mode 100644 bambu_cli/protocols/mqtt_monitor.py create mode 100644 bambu_cli/protocols/mqtt_print.py create mode 100644 bambu_cli/protocols/mqtt_tls.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 009ff1b..2d2668d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,11 @@ Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); version ### Changed +- MQTT client construction, command/status, print-ack, and the monitor loop + live in separate `protocols/mqtt_*.py` modules. TLS pinning uses a real + `SSLContext` subclass (`PinningSSLContext`) instead of patching + `wrap_socket` on a stock context. + - **`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 diff --git a/bambu_cli/protocols/mqtt.py b/bambu_cli/protocols/mqtt.py index 61ae365..63a27b0 100644 --- a/bambu_cli/protocols/mqtt.py +++ b/bambu_cli/protocols/mqtt.py @@ -1,860 +1,53 @@ -import json -import socket -import ssl -import sys -import threading -import time -from typing import Optional - -from bambu_cli.config import get_command_timeout -from bambu_cli.errors import BambuError, PrinterStatusIncomplete, abort -from bambu_cli.logging_utils import logger -from bambu_cli.utils import _resolve_ip, get_sequence_id - -# Lazily import mqtt or load at module level if available -try: - import paho.mqtt.client as mqtt -except ImportError: - mqtt = None - - -def _require_mqtt(): - """Ensure paho-mqtt is importable; abort with config exit if missing.""" - global mqtt - if mqtt is not None: - return - # The import/abort paths only run when the optional dep is absent at import - # time (not exercised in CI where paho-mqtt is installed). - try: - import paho.mqtt.client as paho_mqtt - - mqtt = paho_mqtt - except ImportError: - logger.error( - "Missing dependency: paho-mqtt. Reinstall the package " - "(e.g. `uv pip install -e .` from a source checkout, or `pip install platecli`)." - ) - from bambu_cli.constants import EXIT_CONFIG_ERROR - from bambu_cli.errors import abort - - abort("", exit_code=EXIT_CONFIG_ERROR) - - -class _SimMqttClient: - """Small MQTT stand-in for --sim without importing test-only mocks.""" - - def __init__(self): - self.on_connect = None - self.on_message = None - self.on_publish = None - - def username_pw_set(self, username, password): - pass - - def tls_set(self, *args, **kwargs): - pass - - def tls_insecure_set(self, *args, **kwargs): - pass - - def connect(self, host, port, keepalive=10): - if self.on_connect: - self.on_connect(self, None, None, 0) - - def subscribe(self, topic): - pass - - def publish(self, topic, payload): - if self.on_publish: - self.on_publish(self, None, 1) - - def loop_start(self): - pass - - def loop_stop(self): - pass - - def disconnect(self): - pass - - def socket(self): - return None - - -# _resolve_ip is imported from bambu_cli.utils - - -def probe_cert_fingerprint(host, port=990, timeout=5): - """Open a TLS connection purely to read the server cert's SHA-256 fingerprint.""" - from bambu_cli.config import fingerprint_sha256 - - ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) - ctx.check_hostname = False - ctx.verify_mode = ssl.CERT_NONE - with socket.create_connection((host, port), timeout) as raw, ctx.wrap_socket(raw, server_hostname=host) as tls: - return fingerprint_sha256(tls.getpeercert(binary_form=True)) - - -def create_mqtt_client(printer, client_id=""): - if printer.simulation_mode: - return _SimMqttClient() - - _require_mqtt() - try: - client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id) - except AttributeError: - client = mqtt.Client(client_id) - client.username_pw_set("bblp", printer.access_code) - - if printer.insecure_tls: - client.tls_set(cert_reqs=ssl.CERT_NONE) - client.tls_insecure_set(True) - elif printer.cert_fingerprint: - expected_fp = printer.cert_fingerprint - ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) - ctx.check_hostname = False - ctx.verify_mode = ssl.CERT_NONE - - orig_wrap = ctx.wrap_socket - - def wrap_socket_with_pinning(*args, **kwargs): - tls_sock = orig_wrap(*args, **kwargs) - from bambu_cli.tlspin import verify_cert_fingerprint - - def _verify_pin(): - der = tls_sock.getpeercert(binary_form=True) - verify_cert_fingerprint(der, expected_fp) - - # paho wraps with do_handshake_on_connect=False, so the peer cert - # is not available yet; defer verification to handshake completion. - try: - tls_sock.getpeercert(binary_form=True) - handshake_done = True - except ValueError: - handshake_done = False - if handshake_done: - _verify_pin() - else: - orig_handshake = tls_sock.do_handshake - - def do_handshake_with_pinning(*hs_args, **hs_kwargs): - orig_handshake(*hs_args, **hs_kwargs) - _verify_pin() - - tls_sock.do_handshake = do_handshake_with_pinning # type: ignore[method-assign] - return tls_sock - - ctx.wrap_socket = wrap_socket_with_pinning # type: ignore[method-assign] - client.tls_set_context(ctx) - client.tls_insecure_set(True) - else: - client.tls_set(cert_reqs=ssl.CERT_REQUIRED) - return client - - -def _mqtt_connect(printer, client): - resolved_ip = _resolve_ip(printer.ip) - old_timeout = socket.getdefaulttimeout() - try: - # Bound the connect phase by the printer's configured timeout. The socket - # default covers blocking name/socket ops, while paho caps its own TCP/TLS - # connect via client._connect_timeout (default 5s) independent of the - # socket default ā€” so set both to actually honor the configured value. - socket.setdefaulttimeout(printer.mqtt_timeout) - if hasattr(client, "_connect_timeout"): - client._connect_timeout = printer.mqtt_timeout - client.connect(resolved_ip, 8883, keepalive=10) - finally: - socket.setdefaulttimeout(old_timeout) - - -def send_command( - printer, - payload, - timeout=None, - retries=2, - *, - client_factory=None, - sleep=None, -): - """Send a command to the printer with retries. - - ``client_factory`` and ``sleep`` default to create_mqtt_client / time.sleep; - tests inject fakes instead of patching module globals. - """ - _client_factory = client_factory if client_factory is not None else create_mqtt_client - _sleep = sleep if sleep is not None else time.sleep - if timeout is None: - timeout = printer.mqtt_timeout - - if printer.simulation_mode: - logger.info(f"šŸ¤– [SIM] Sending command: {payload}") - return True - - for attempt in range(retries + 1): - client = _client_factory(printer) - client.user_data_set({}) - publish_done = threading.Event() - success = [False] - # Once-flag so paho's auto-reconnect (loop_start -> loop_forever with - # reconnect_on_failure=True) can never fire on_connect a second time and - # re-publish this state-changing command (pause/stop/gcode_line) to the - # printer. Bound as a default arg so a stale callback from a previous - # attempt's client cannot re-publish either. - published = [False] - - def on_connect(client, userdata, flags, rc, properties=None, published=published): - if rc == 0: - if not published[0]: - published[0] = True - # QoS 1: pairs on_publish with a broker PUBACK rather than a - # bare local socket write. The Bambu broker is the printer, - # so a PUBACK is real receipt, not "left our OS buffer". - client.publish(f"device/{printer.serial}/request", payload, qos=1) - else: - logger.error(f"Connection failed: rc={rc}") - publish_done.set() - - def on_publish(client, userdata, mid, reason_code=None, properties=None): - success[0] = True - publish_done.set() - - client.on_connect = on_connect - client.on_publish = on_publish - - try: - _mqtt_connect(printer, client) - client.loop_start() - try: - if publish_done.wait(timeout): - return success[0] - finally: - try: - client.loop_stop() - except Exception: - pass - try: - client.disconnect() - except Exception: - pass - - if attempt < retries: - logger.warning(f"MQTT command timeout on attempt {attempt + 1}. Retrying...") - _sleep(2**attempt) - except (OSError, ssl.SSLError) as e: - if attempt < retries: - logger.warning(f"MQTT command attempt {attempt + 1} failed: {e}. Retrying...") - _sleep(2**attempt) - else: - logger.error(f"MQTT command error: {e}") - - return False - - -# Keys a full state snapshot always carries. The printer publishes incremental -# deltas on the report topic and only answers `pushall` with the complete state, -# so "a message arrived" is not the same as "we have the state" ā€” mid-print the -# first thing to land is often a lone nozzle_temper reading. These four are what -# both docs/schemas/status.json and the human renderer treat as always-present, -# so they are the gate for telling a snapshot from a delta. -_REQUIRED_STATUS_KEYS = ("gcode_state", "mc_percent", "bed_temper", "nozzle_temper") - -# The gcode_state values that end a print watch. Module-level (rather than a -# local in monitor_status) so every follower of a print ā€” the CLI monitor loop -# and the TUI monitor screen ā€” stops on exactly the same set instead of keeping -# a second copy that can drift. -TERMINAL_GCODE_STATES = frozenset({"FINISH", "FAILED", "STOP", "IDLE"}) - - -def status_is_complete(data): - """True when ``data`` is a full snapshot rather than an incremental delta.""" - return isinstance(data, dict) and all(key in data for key in _REQUIRED_STATUS_KEYS) - - -def get_status(printer, timeout=None, retries=2, *, require_complete=True): - """Get printer status via MQTT with retries. +"""MQTT transport facade. - Report-topic messages are merged into one accumulated state (later values - win) and we keep waiting ā€” re-issuing ``pushall`` on each retry ā€” until every - key in ``_REQUIRED_STATUS_KEYS`` is present, so callers never receive a - delta dressed up as a snapshot. +Implementations live in ``mqtt_tls``, ``mqtt_cmd``, ``mqtt_print``, and +``mqtt_monitor``. This module re-exports the public names so existing imports +and test patches on ``bambu_cli.protocols.mqtt`` keep working. +""" - ``require_complete=False`` returns the first payload that arrives, for - callers using the reply only as a liveness probe (``doctor``, print - ``--dry-run``). With the default, a connection that yields nothing but - deltas raises ``PrinterStatusIncomplete`` rather than returning a partial; - a connection that yields nothing at all still returns ``None``. - """ - if timeout is None: - timeout = printer.mqtt_timeout +from __future__ import annotations - if printer.simulation_mode: - logger.info("šŸ¤– [SIM] Fetching simulated printer status...") - return { - "gcode_state": "IDLE", - "mc_percent": 0, - "hw_ver": "P1P-SIM", - "sw_ver": "01.XX.XX.XX", - "bed_temper": 25, - "bed_target_temper": 0, - "nozzle_temper": 25, - "nozzle_target_temper": 0, - "cooling_fan_speed": 0, - "wifi_signal": "-42dBm", - # A representative AMS so agents can exercise `status --json` AMS - # parsing (and --ams-mapping decisions) without hardware. - "ams": { - "tray_now": "0", - "ams": [ - { - "id": "0", - "humidity": "5", - "temp": "26.0", - "tray": [ - {"id": "0", "tray_type": "PLA", "tray_color": "F2F2F2FF", "remain": 90}, - {"id": "1", "tray_type": "PETG", "tray_color": "0A0AC8FF", "remain": 60}, - {"id": "2"}, - {"id": "3", "tray_type": "TPU", "tray_color": "000000FF", "remain": 45}, - ], - } - ], - }, - } - - # Accumulated across attempts: a retry re-issues pushall, and keys collected - # before the timeout are still the freshest we have. - merged: dict = {} - merged_lock = threading.Lock() - - for attempt in range(retries + 1): - status_received = threading.Event() - connect_failed = [False] - client = create_mqtt_client(printer) - client.user_data_set({}) - - # status_received / connect_failed are bound per attempt so a late - # callback from a previous attempt's client cannot wake this one. - def on_connect( - client, - userdata, - flags, - rc, - properties=None, - connect_failed=connect_failed, - status_received=status_received, - ): - if rc == 0: - client.subscribe(f"device/{printer.serial}/report") - push = json.dumps({"pushing": {"sequence_id": get_sequence_id(), "command": "pushall"}}) - client.publish(f"device/{printer.serial}/request", push) - else: - logger.error(f"Connection failed: rc={rc}") - connect_failed[0] = True - status_received.set() - - def on_message(client, userdata, msg, status_received=status_received): - try: - data = json.loads(msg.payload.decode("utf-8")) - except (json.JSONDecodeError, UnicodeDecodeError) as e: - logger.debug(f"MQTT decode error: {e}") - return - if not isinstance(data, dict) or not isinstance(data.get("print"), dict): - return - with merged_lock: - merged.update(data["print"]) - complete = status_is_complete(merged) - if complete or not require_complete: - status_received.set() - - client.on_connect = on_connect - client.on_message = on_message - - try: - _mqtt_connect(printer, client) - client.loop_start() - try: - if status_received.wait(timeout): - if connect_failed[0]: - return None - with merged_lock: - snapshot = dict(merged) - # Never hand back a falsy-but-not-None {}: callers treat the - # return value as "reachable or not". - if snapshot: - return snapshot - finally: - try: - client.loop_stop() - except Exception: - pass - try: - client.disconnect() - except Exception: - pass - if attempt < retries: - with merged_lock: - saw_partial = bool(merged) - if saw_partial: - logger.warning( - f"Printer sent only partial status on attempt {attempt + 1}. Re-requesting full state..." - ) - else: - logger.warning(f"MQTT status timeout on attempt {attempt + 1}. Retrying...") - time.sleep(2**attempt) - except (OSError, ssl.SSLError) as e: - if attempt < retries: - logger.warning(f"MQTT status attempt {attempt + 1} failed: {e}. Retrying...") - time.sleep(2**attempt) - else: - logger.error(f"MQTT status error: {e}") - - with merged_lock: - partial = dict(merged) - if partial and require_complete: - # We reached the printer, it just never answered pushall with a whole - # state. Emitting `partial` here is what hands agents a KeyError later. - missing = [key for key in _REQUIRED_STATUS_KEYS if key not in partial] - raise PrinterStatusIncomplete( - "Printer returned only partial status updates, never a full snapshot " - f"(missing {', '.join(missing)}). It may be busy mid-print; retry the command.", - detail={"missing_keys": missing, "received_keys": sorted(partial)}, - next_command="plate status", - ) - return None - - -def get_version(printer, timeout=5, retries=1): - """Fetch printer module versions via the MQTT get_version command.""" - if printer.simulation_mode: - return [{"name": "ota", "sw_ver": "01.00.00.00", "hw_ver": "P1P-SIM"}] - - for attempt in range(retries + 1): - result = {"modules": None} - received = threading.Event() - client = create_mqtt_client(printer) - client.user_data_set({}) - - def on_connect(client, userdata, flags, rc, properties=None): - if rc == 0: - client.subscribe(f"device/{printer.serial}/report") - client.publish( - f"device/{printer.serial}/request", - json.dumps({"info": {"sequence_id": get_sequence_id(), "command": "get_version"}}), - ) - else: - logger.error(f"Connection failed: rc={rc}") - received.set() - - def on_message(client, userdata, msg): - try: - data = json.loads(msg.payload.decode("utf-8")) - except (json.JSONDecodeError, UnicodeDecodeError): - return - info = data.get("info") - if isinstance(info, dict) and info.get("command") == "get_version" and "module" in info: - result["modules"] = info["module"] - received.set() - - client.on_connect = on_connect - client.on_message = on_message - - try: - _mqtt_connect(printer, client) - client.loop_start() - try: - if received.wait(timeout): - return result["modules"] - finally: - try: - client.loop_stop() - except Exception: - pass - try: - client.disconnect() - except Exception: - pass - if attempt < retries: - time.sleep(2**attempt) - except (OSError, ssl.SSLError): - if attempt < retries: - time.sleep(2**attempt) - - return None - - -def _status_event(p, event): - """Build a compact, agent-friendly status event from a raw MQTT print payload. - - ``event`` is ``"update"`` for an in-progress change or ``"terminal"`` for the - final state. Only the fields agents care about for print progress are kept, - so a streamed line stays small. - """ - - def _int(value, default=0): - try: - return int(value) - except (TypeError, ValueError): - return default - - return { - "event": event, - "command": "status", - "gcode_state": p.get("gcode_state", "UNKNOWN"), - "mc_percent": _int(p.get("mc_percent", 0)), - "layer_num": _int(p.get("layer_num", 0)), - "total_layer_num": _int(p.get("total_layer_num", 0)), - "mc_remaining_time": _int(p.get("mc_remaining_time", 0)), - "nozzle_temper": p.get("nozzle_temper"), - "nozzle_target_temper": p.get("nozzle_target_temper"), - "bed_temper": p.get("bed_temper"), - "bed_target_temper": p.get("bed_target_temper"), - "gcode_file": p.get("gcode_file", ""), - } - - -def monitor_status(args, printer): - """Subscribe to the printer's report topic and stream updates until a terminal state. - - In ``--json`` mode each change is emitted as one compact NDJSON line (an - ``event: "update"`` object, then a final ``event: "terminal"``) so an agent - can follow a print in real time. Otherwise a live human-readable progress - bar is shown. - - ``printer`` is injected by the caller ā€” this module must not reach up to - ``bambu_cli.printer`` for an ambient one (see scripts/check_layers.py). - """ - from bambu_cli.argutils import namespace_get as _namespace_get - from bambu_cli.utils import emit_json_line - - json_mode = bool(_namespace_get(args, "json", False)) - logger.info("šŸ“” Starting status monitor loop. Press Ctrl+C to stop.") - if printer.simulation_mode: - # Stream the same shape of events a real print would, so agents can - # exercise the --monitor --json contract without hardware. - for state, pct, event in (("PREPARE", 0, "update"), ("RUNNING", 50, "update"), ("FINISH", 100, "terminal")): - if json_mode: - emit_json_line(_status_event({"gcode_state": state, "mc_percent": pct}, event)) - else: - logger.info(f"šŸ¤– [SIM] Simulated status: State={state}, Progress={pct}%") - if event != "terminal": - time.sleep(0.5) - if not json_mode: - logger.info("šŸ Reached terminal state: FINISH") - return - - terminal_states = TERMINAL_GCODE_STATES - received_terminal = threading.Event() - show_progress_bar = not json_mode and sys.stdout.isatty() - client = create_mqtt_client(printer) - userdata: dict = {} - client.user_data_set(userdata) - - def on_connect(client, userdata, flags, rc, properties=None): - if rc == 0: - client.subscribe(f"device/{printer.serial}/report") - push = json.dumps({"pushing": {"sequence_id": get_sequence_id(), "command": "pushall"}}) - client.publish(f"device/{printer.serial}/request", push) - else: - logger.error(f"Connection failed: rc={rc}") - received_terminal.set() - - last_state = [None] - last_pct = [None] - # Same accumulation as get_status: report-topic messages are deltas, so a - # lone temperature update must not read as gcode_state=UNKNOWN at 0%. - merged: dict = {} - - def on_message(client, userdata, msg): - try: - data = json.loads(msg.payload.decode("utf-8")) - if isinstance(data, dict) and isinstance(data.get("print"), dict): - merged.update(data["print"]) - p = merged - state = p.get("gcode_state", "UNKNOWN") - pct = p.get("mc_percent", 0) - - if state != last_state[0] or pct != last_pct[0]: - if "progress" not in userdata and not show_progress_bar: - userdata["progress"] = None - if "progress" not in userdata: - try: # pragma: no cover -- rich TTY progress UI - from rich.progress import BarColumn, Progress, TextColumn, TimeElapsedColumn - - progress = Progress( - TextColumn("[bold blue]Print Status"), - BarColumn(), - "[progress.percentage]{task.percentage:>3.1f}%", - "ā€¢", - TextColumn("{task.description}"), - "ā€¢", - TimeElapsedColumn(), - transient=True, - ) - progress.start() - userdata["progress"] = progress - userdata["task_id"] = progress.add_task(f"State: {state}", total=100, completed=pct) - except ImportError: - userdata["progress"] = None - - if userdata.get("progress"): - userdata["progress"].update(userdata["task_id"], completed=pct, description=f"State: {state}") - elif json_mode: - emit_json_line(_status_event(p, "update")) - else: - logger.info(f"ā³ Status: State={state}, Progress={pct}%") - - last_state[0] = state - last_pct[0] = pct - - if state in terminal_states: - if json_mode: - emit_json_line(_status_event(p, "terminal")) - else: - logger.info(f"šŸ Reached terminal state: {state}") - received_terminal.set() - except (UnicodeDecodeError, json.JSONDecodeError) as e: - logger.debug(f"MQTT decode error: {e}") - except Exception as e: - logger.warning(f"MQTT message handling error: {e}") - - client.on_connect = on_connect - client.on_message = on_message - - try: - _mqtt_connect(printer, client) - client.loop_start() - while not received_terminal.is_set(): - received_terminal.wait(1.0) - except KeyboardInterrupt: - logger.info("\nšŸ›‘ Monitor loop stopped by user.") - finally: - if userdata.get("progress"): - try: - userdata["progress"].stop() - except Exception: - pass - try: - client.loop_stop() - client.disconnect() - except Exception: - pass - - -def _printer_error_hex(code: object) -> Optional[str]: - """Render a printer error code as the hex form Bambu documents (e.g. 0x0500C010). - - Returns None when the code is not an integer we can render. - """ - if isinstance(code, bool) or not isinstance(code, int): - return None - return f"0x{code & 0xFFFFFFFF:08X}" - - -def execute_print_command( - printer, - payload, - basename, - dry_run=False, - *, - command_timeout=None, - client_factory=None, -): - """Send the print payload via MQTT and monitor for errors. - - ``command_timeout`` and ``client_factory`` are injectable; defaults are - get_command_timeout() / create_mqtt_client. - """ - from bambu_cli.constants import EXIT_FILE_ERROR, EXIT_NETWORK_ERROR, EXIT_PRINTER_ERROR, EXIT_TIMEOUT - from bambu_cli.utils import record_error_detail - - _client_factory = client_factory if client_factory is not None else create_mqtt_client - - if dry_run: - logger.info(f"šŸ” Dry Run: Checking if {basename} exists on printer...") - try: - with printer.get_ftp_client(timeout=5) as ftp: - files = ftp.nlst("/model/") - if basename in files or f"/model/{basename}" in files: - logger.info(f" āœ… File {basename} found on printer.") - else: - message = f"File {basename} was not found on printer. Upload it first." - logger.error(f" āŒ File {basename} NOT found on printer. Upload it first.") - record_error_detail( - "print", EXIT_FILE_ERROR, message, failed_step="dry_run", file=basename, printed=False - ) - abort("", exit_code=EXIT_FILE_ERROR) - logger.info(" āœ… Printer reachable via MQTT (status check)...") - if printer.status(timeout=5, require_complete=False): - logger.info(" āœ… MQTT connection verified.") - else: - message = "MQTT connection failed." - logger.error(f" āŒ {message}") - record_error_detail( - "print", EXIT_NETWORK_ERROR, message, failed_step="dry_run", file=basename, printed=False - ) - abort("", exit_code=EXIT_NETWORK_ERROR) - return - except BambuError: - raise - except Exception as e: - message = f"Dry run failed: {e}" - logger.error(message) - record_error_detail( - "print", EXIT_NETWORK_ERROR, message, failed_step="dry_run", file=basename, printed=False - ) - abort("", exit_code=EXIT_NETWORK_ERROR) - - if printer.simulation_mode: - from bambu_cli.protocols.ftps import _SIM_FTP_FILES - - if basename not in _SIM_FTP_FILES: - message = f"File {basename} not found on simulated printer. Upload it first." - logger.error(message) - record_error_detail("print", EXIT_FILE_ERROR, message, failed_step="print", file=basename, printed=False) - abort("", exit_code=EXIT_FILE_ERROR) - logger.info(f"šŸ¤– [SIM] Print started: {basename}") - return - - client = _client_factory(printer, "bambu_print") - - print_error = [None] - reject_reason: list[Optional[str]] = [None] - command_accepted = threading.Event() - # rc != 0 (bad CONNACK, e.g. wrong/rotated LAN access code) is delivered - # asynchronously via on_connect after loop_start, so _mqtt_connect cannot - # raise on it. Track it like get_status does and fail closed after the wait - # instead of setting command_accepted and falling through to "Print started". - connect_failed = [False] - # Once-flag so paho auto-reconnect can never re-fire on_connect and re-issue - # the print-start command to a printer that may already be running it. - published = [False] - # A stale, latched print_error from a *prior* job rides in on the printer's - # first periodic *full-state snapshot* (P1-series push the complete state on - # the report topic). Before our project_file ack lands, such a snapshot must - # not be blamed on this print. A genuine error for our command arrives either - # with/after the ack, or as a lone print_error delta (not a full snapshot). - ack_seen = [False] - - def on_connect(client, userdata, flags, rc, properties=None): - if rc == 0: - # Resubscribe on every (re)connect: with clean_session=True a - # mid-window reconnect drops the subscription, so a printer ack after - # reconnect would otherwise be invisible and time out a print that - # was actually accepted. Only the state-changing publish is guarded. - client.subscribe(f"device/{printer.serial}/report") - if not published[0]: - published[0] = True - client.publish(f"device/{printer.serial}/request", payload) - else: - logger.error(f"Connection failed: rc={rc}") - connect_failed[0] = True - command_accepted.set() - - def on_message(client, userdata, msg): - try: - # Consistent decode prior to json.loads (A0530-ERR-05) - data = json.loads(msg.payload.decode("utf-8")) - if "print" in data: - p = data["print"] - is_ack = p.get("command") == "project_file" - if is_ack: - ack_seen[0] = True - pe = p.get("print_error", 0) - # Blame the error on this print unless it is a pre-existing value - # latched into a full-state snapshot that arrived before our ack - # (the classic stale-latch vector). Errors in/after the ack, or in - # a lone delta, are ours. - stale_snapshot = status_is_complete(p) and not ack_seen[0] and not is_ack - if pe and pe != 0 and not stale_snapshot: - print_error[0] = pe - command_accepted.set() - if is_ack: - # A firmware rejection carries result=fail (+ optional - # reason); do not report success for a rejected job. - result = p.get("result") - if isinstance(result, str) and result.strip().lower() not in ("", "success"): - reject_reason[0] = str(p.get("reason") or result) - command_accepted.set() - except (UnicodeDecodeError, json.JSONDecodeError) as e: - logger.debug(f"MQTT decode error: {e}") - except Exception as e: - logger.warning(f"MQTT message handling error: {e}") - - client.on_connect = on_connect - client.on_message = on_message - - # Dynamically get timeouts (A0530-NET-07) - base_timeout = command_timeout if command_timeout is not None else get_command_timeout() - print_ack_timeout = base_timeout + 5 # default historical: 10 - - try: - _mqtt_connect(printer, client) - client.loop_start() - try: - accepted = command_accepted.wait(print_ack_timeout) - if not accepted: - message = f"Timed out waiting for printer to acknowledge print start for {basename}" - logger.error(message) - record_error_detail("print", EXIT_TIMEOUT, message, failed_step="print", file=basename, printed=False) - abort("", exit_code=EXIT_TIMEOUT) - finally: - try: - client.loop_stop() - except Exception: - pass - try: - client.disconnect() - except Exception: - pass - except BambuError: - raise - except Exception as e: - message = f"Error: {e}" - logger.error(message) - record_error_detail("print", EXIT_NETWORK_ERROR, message, failed_step="print", file=basename, printed=False) - abort("", exit_code=EXIT_NETWORK_ERROR) - - if connect_failed[0]: - message = f"Failed to connect to printer to start print for {basename} (check LAN access code)" - logger.error(message) - record_error_detail("print", EXIT_NETWORK_ERROR, message, failed_step="print", file=basename, printed=False) - abort("", exit_code=EXIT_NETWORK_ERROR) - - if reject_reason[0]: - message = f"Printer rejected print of {basename}: {reject_reason[0]}" - logger.error(message) - record_error_detail("print", EXIT_PRINTER_ERROR, message, failed_step="print", file=basename, printed=False) - abort("", exit_code=EXIT_PRINTER_ERROR) +import time - if print_error[0]: - error_hex = _printer_error_hex(print_error[0]) - message = f"Print failed with error code {print_error[0]}" - if error_hex: - message += f" (hex {error_hex})" - logger.error(message) - if print_error[0] == 83935248: - logger.info(" File not found on printer SD card. Check filename with 'files' command.") - record_error_detail( - "print", - EXIT_FILE_ERROR, - "File not found on printer SD card. Check filename with 'files' command.", - failed_step="print", - file=basename, - printer_error_code=print_error[0], - printer_error_code_hex=error_hex, - printed=False, - ) - abort("", exit_code=EXIT_FILE_ERROR) - record_error_detail( - "print", - EXIT_PRINTER_ERROR, - message, - failed_step="print", - file=basename, - printer_error_code=print_error[0], - printer_error_code_hex=error_hex, - printed=False, - ) - abort("", exit_code=EXIT_PRINTER_ERROR) - else: - logger.info(f"šŸ–Øļø Print started: {basename}") +from bambu_cli.protocols.mqtt_cmd import ( + _REQUIRED_STATUS_KEYS, + TERMINAL_GCODE_STATES, + get_status, + get_version, + send_command, + status_is_complete, +) +from bambu_cli.protocols.mqtt_monitor import _status_event, monitor_status +from bambu_cli.protocols.mqtt_print import _printer_error_hex, execute_print_command +from bambu_cli.protocols.mqtt_tls import ( + PinningSSLContext, + _mqtt_connect, + _require_mqtt, + _SimMqttClient, + create_mqtt_client, + mqtt, + pinning_ssl_context, + probe_cert_fingerprint, +) + +__all__ = [ + "TERMINAL_GCODE_STATES", + "PinningSSLContext", + "_REQUIRED_STATUS_KEYS", + "_SimMqttClient", + "_mqtt_connect", + "_printer_error_hex", + "_require_mqtt", + "_status_event", + "create_mqtt_client", + "execute_print_command", + "get_status", + "get_version", + "monitor_status", + "mqtt", + "pinning_ssl_context", + "probe_cert_fingerprint", + "send_command", + "status_is_complete", + "time", +] diff --git a/bambu_cli/protocols/mqtt_cmd.py b/bambu_cli/protocols/mqtt_cmd.py new file mode 100644 index 0000000..f9a4683 --- /dev/null +++ b/bambu_cli/protocols/mqtt_cmd.py @@ -0,0 +1,322 @@ +"""MQTT command and status request/response.""" + +from __future__ import annotations + +import json +import ssl +import threading + +from bambu_cli.errors import PrinterStatusIncomplete +from bambu_cli.logging_utils import logger +from bambu_cli.utils import get_sequence_id + +_REQUIRED_STATUS_KEYS = ("gcode_state", "mc_percent", "bed_temper", "nozzle_temper") +TERMINAL_GCODE_STATES = frozenset({"FINISH", "FAILED", "STOP", "IDLE"}) + + +def _client_factory(client_factory): + if client_factory is not None: + return client_factory + from bambu_cli.protocols import mqtt as mqtt_mod + + return mqtt_mod.create_mqtt_client + + +def _connect(printer, client): + from bambu_cli.protocols import mqtt as mqtt_mod + + return mqtt_mod._mqtt_connect(printer, client) + + +def _sleep(sleep): + if sleep is not None: + return sleep + from bambu_cli.protocols import mqtt as mqtt_mod + + return mqtt_mod.time.sleep + + +def send_command( + printer, + payload, + timeout=None, + retries=2, + *, + client_factory=None, + sleep=None, +): + """Send a command to the printer with retries. + + ``client_factory`` and ``sleep`` default to create_mqtt_client / time.sleep; + tests inject fakes instead of patching module globals. + """ + _factory = _client_factory(client_factory) + _sleep_fn = _sleep(sleep) + if timeout is None: + timeout = printer.mqtt_timeout + + if printer.simulation_mode: + logger.info(f"šŸ¤– [SIM] Sending command: {payload}") + return True + + for attempt in range(retries + 1): + client = _factory(printer) + client.user_data_set({}) + publish_done = threading.Event() + success = [False] + published = [False] + + def on_connect(client, userdata, flags, rc, properties=None, published=published): + if rc == 0: + if not published[0]: + published[0] = True + client.publish(f"device/{printer.serial}/request", payload, qos=1) + else: + logger.error(f"Connection failed: rc={rc}") + publish_done.set() + + def on_publish(client, userdata, mid, reason_code=None, properties=None): + success[0] = True + publish_done.set() + + client.on_connect = on_connect + client.on_publish = on_publish + + try: + _connect(printer, client) + client.loop_start() + try: + if publish_done.wait(timeout): + return success[0] + finally: + try: + client.loop_stop() + except Exception: + pass + try: + client.disconnect() + except Exception: + pass + + if attempt < retries: + logger.warning(f"MQTT command timeout on attempt {attempt + 1}. Retrying...") + _sleep_fn(2**attempt) + except (OSError, ssl.SSLError) as e: + if attempt < retries: + logger.warning(f"MQTT command attempt {attempt + 1} failed: {e}. Retrying...") + _sleep_fn(2**attempt) + else: + logger.error(f"MQTT command error: {e}") + + return False + + +def status_is_complete(data): + """True when ``data`` is a full snapshot rather than an incremental delta.""" + return isinstance(data, dict) and all(key in data for key in _REQUIRED_STATUS_KEYS) + + +def get_status(printer, timeout=None, retries=2, *, require_complete=True): + """Get printer status via MQTT with retries. + + Report-topic messages are merged into one accumulated state (later values + win) and we keep waiting ā€” re-issuing ``pushall`` on each retry ā€” until every + key in ``_REQUIRED_STATUS_KEYS`` is present, so callers never receive a + delta dressed up as a snapshot. + + ``require_complete=False`` returns the first payload that arrives, for + callers using the reply only as a liveness probe (``doctor``, print + ``--dry-run``). With the default, a connection that yields nothing but + deltas raises ``PrinterStatusIncomplete`` rather than returning a partial; + a connection that yields nothing at all still returns ``None``. + """ + if timeout is None: + timeout = printer.mqtt_timeout + + if printer.simulation_mode: + logger.info("šŸ¤– [SIM] Fetching simulated printer status...") + return { + "gcode_state": "IDLE", + "mc_percent": 0, + "hw_ver": "P1P-SIM", + "sw_ver": "01.XX.XX.XX", + "bed_temper": 25, + "bed_target_temper": 0, + "nozzle_temper": 25, + "nozzle_target_temper": 0, + "cooling_fan_speed": 0, + "wifi_signal": "-42dBm", + "ams": { + "tray_now": "0", + "ams": [ + { + "id": "0", + "humidity": "5", + "temp": "26.0", + "tray": [ + {"id": "0", "tray_type": "PLA", "tray_color": "F2F2F2FF", "remain": 90}, + {"id": "1", "tray_type": "PETG", "tray_color": "0A0AC8FF", "remain": 60}, + {"id": "2"}, + {"id": "3", "tray_type": "TPU", "tray_color": "000000FF", "remain": 45}, + ], + } + ], + }, + } + + merged: dict = {} + merged_lock = threading.Lock() + _factory = _client_factory(None) + _sleep_fn = _sleep(None) + + for attempt in range(retries + 1): + status_received = threading.Event() + connect_failed = [False] + client = _factory(printer) + client.user_data_set({}) + + def on_connect( + client, + userdata, + flags, + rc, + properties=None, + connect_failed=connect_failed, + status_received=status_received, + ): + if rc == 0: + client.subscribe(f"device/{printer.serial}/report") + push = json.dumps({"pushing": {"sequence_id": get_sequence_id(), "command": "pushall"}}) + client.publish(f"device/{printer.serial}/request", push) + else: + logger.error(f"Connection failed: rc={rc}") + connect_failed[0] = True + status_received.set() + + def on_message(client, userdata, msg, status_received=status_received): + try: + data = json.loads(msg.payload.decode("utf-8")) + except (json.JSONDecodeError, UnicodeDecodeError) as e: + logger.debug(f"MQTT decode error: {e}") + return + if not isinstance(data, dict) or not isinstance(data.get("print"), dict): + return + with merged_lock: + merged.update(data["print"]) + complete = status_is_complete(merged) + if complete or not require_complete: + status_received.set() + + client.on_connect = on_connect + client.on_message = on_message + + try: + _connect(printer, client) + client.loop_start() + try: + if status_received.wait(timeout): + if connect_failed[0]: + return None + with merged_lock: + snapshot = dict(merged) + if snapshot: + return snapshot + finally: + try: + client.loop_stop() + except Exception: + pass + try: + client.disconnect() + except Exception: + pass + if attempt < retries: + with merged_lock: + saw_partial = bool(merged) + if saw_partial: + logger.warning( + f"Printer sent only partial status on attempt {attempt + 1}. Re-requesting full state..." + ) + else: + logger.warning(f"MQTT status timeout on attempt {attempt + 1}. Retrying...") + _sleep_fn(2**attempt) + except (OSError, ssl.SSLError) as e: + if attempt < retries: + logger.warning(f"MQTT status attempt {attempt + 1} failed: {e}. Retrying...") + _sleep_fn(2**attempt) + else: + logger.error(f"MQTT status error: {e}") + + with merged_lock: + partial = dict(merged) + if partial and require_complete: + missing = [key for key in _REQUIRED_STATUS_KEYS if key not in partial] + raise PrinterStatusIncomplete( + "Printer returned only partial status updates, never a full snapshot " + f"(missing {', '.join(missing)}). It may be busy mid-print; retry the command.", + detail={"missing_keys": missing, "received_keys": sorted(partial)}, + next_command="plate status", + ) + return None + + +def get_version(printer, timeout=5, retries=1): + """Fetch printer module versions via the MQTT get_version command.""" + if printer.simulation_mode: + return [{"name": "ota", "sw_ver": "01.00.00.00", "hw_ver": "P1P-SIM"}] + + _factory = _client_factory(None) + _sleep_fn = _sleep(None) + + for attempt in range(retries + 1): + result = {"modules": None} + received = threading.Event() + client = _factory(printer) + client.user_data_set({}) + + def on_connect(client, userdata, flags, rc, properties=None): + if rc == 0: + client.subscribe(f"device/{printer.serial}/report") + client.publish( + f"device/{printer.serial}/request", + json.dumps({"info": {"sequence_id": get_sequence_id(), "command": "get_version"}}), + ) + else: + logger.error(f"Connection failed: rc={rc}") + received.set() + + def on_message(client, userdata, msg): + try: + data = json.loads(msg.payload.decode("utf-8")) + except (json.JSONDecodeError, UnicodeDecodeError): + return + info = data.get("info") + if isinstance(info, dict) and info.get("command") == "get_version" and "module" in info: + result["modules"] = info["module"] + received.set() + + client.on_connect = on_connect + client.on_message = on_message + + try: + _connect(printer, client) + client.loop_start() + try: + if received.wait(timeout): + return result["modules"] + finally: + try: + client.loop_stop() + except Exception: + pass + try: + client.disconnect() + except Exception: + pass + if attempt < retries: + _sleep_fn(2**attempt) + except (OSError, ssl.SSLError): + if attempt < retries: + _sleep_fn(2**attempt) + + return None diff --git a/bambu_cli/protocols/mqtt_monitor.py b/bambu_cli/protocols/mqtt_monitor.py new file mode 100644 index 0000000..62ee66c --- /dev/null +++ b/bambu_cli/protocols/mqtt_monitor.py @@ -0,0 +1,161 @@ +"""MQTT live status monitor (human progress bar or NDJSON events).""" + +from __future__ import annotations + +import json +import sys +import threading + +from bambu_cli.logging_utils import logger +from bambu_cli.protocols.mqtt_cmd import ( + TERMINAL_GCODE_STATES, + _client_factory, + _connect, + _sleep, +) +from bambu_cli.utils import get_sequence_id + + +def _status_event(p, event): + """Build a compact, agent-friendly status event from a raw MQTT print payload.""" + + def _int(value, default=0): + try: + return int(value) + except (TypeError, ValueError): + return default + + return { + "event": event, + "command": "status", + "gcode_state": p.get("gcode_state", "UNKNOWN"), + "mc_percent": _int(p.get("mc_percent", 0)), + "layer_num": _int(p.get("layer_num", 0)), + "total_layer_num": _int(p.get("total_layer_num", 0)), + "mc_remaining_time": _int(p.get("mc_remaining_time", 0)), + "nozzle_temper": p.get("nozzle_temper"), + "nozzle_target_temper": p.get("nozzle_target_temper"), + "bed_temper": p.get("bed_temper"), + "bed_target_temper": p.get("bed_target_temper"), + "gcode_file": p.get("gcode_file", ""), + } + + +def monitor_status(args, printer): + """Subscribe to the printer's report topic and stream updates until a terminal state. + + ``printer`` is injected by the caller ā€” this module must not reach up to + ``bambu_cli.printer`` for an ambient one (see scripts/check_layers.py). + """ + from bambu_cli.argutils import namespace_get as _namespace_get + from bambu_cli.utils import emit_json_line + + json_mode = bool(_namespace_get(args, "json", False)) + logger.info("šŸ“” Starting status monitor loop. Press Ctrl+C to stop.") + if printer.simulation_mode: + _sleep_fn = _sleep(None) + for state, pct, event in (("PREPARE", 0, "update"), ("RUNNING", 50, "update"), ("FINISH", 100, "terminal")): + if json_mode: + emit_json_line(_status_event({"gcode_state": state, "mc_percent": pct}, event)) + else: + logger.info(f"šŸ¤– [SIM] Simulated status: State={state}, Progress={pct}%") + if event != "terminal": + _sleep_fn(0.5) + if not json_mode: + logger.info("šŸ Reached terminal state: FINISH") + return + + terminal_states = TERMINAL_GCODE_STATES + received_terminal = threading.Event() + show_progress_bar = not json_mode and sys.stdout.isatty() + client = _client_factory(None)(printer) + userdata: dict = {} + client.user_data_set(userdata) + + def on_connect(client, userdata, flags, rc, properties=None): + if rc == 0: + client.subscribe(f"device/{printer.serial}/report") + push = json.dumps({"pushing": {"sequence_id": get_sequence_id(), "command": "pushall"}}) + client.publish(f"device/{printer.serial}/request", push) + else: + logger.error(f"Connection failed: rc={rc}") + received_terminal.set() + + last_state = [None] + last_pct = [None] + merged: dict = {} + + def on_message(client, userdata, msg): + try: + data = json.loads(msg.payload.decode("utf-8")) + if isinstance(data, dict) and isinstance(data.get("print"), dict): + merged.update(data["print"]) + p = merged + state = p.get("gcode_state", "UNKNOWN") + pct = p.get("mc_percent", 0) + + if state != last_state[0] or pct != last_pct[0]: + if "progress" not in userdata and not show_progress_bar: + userdata["progress"] = None + if "progress" not in userdata: + try: # pragma: no cover -- rich TTY progress UI + from rich.progress import BarColumn, Progress, TextColumn, TimeElapsedColumn + + progress = Progress( + TextColumn("[bold blue]Print Status"), + BarColumn(), + "[progress.percentage]{task.percentage:>3.1f}%", + "ā€¢", + TextColumn("{task.description}"), + "ā€¢", + TimeElapsedColumn(), + transient=True, + ) + progress.start() + userdata["progress"] = progress + userdata["task_id"] = progress.add_task(f"State: {state}", total=100, completed=pct) + except ImportError: + userdata["progress"] = None + + if userdata.get("progress"): + userdata["progress"].update(userdata["task_id"], completed=pct, description=f"State: {state}") + elif json_mode: + emit_json_line(_status_event(p, "update")) + else: + logger.info(f"ā³ Status: State={state}, Progress={pct}%") + + last_state[0] = state + last_pct[0] = pct + + if state in terminal_states: + if json_mode: + emit_json_line(_status_event(p, "terminal")) + else: + logger.info(f"šŸ Reached terminal state: {state}") + received_terminal.set() + except (UnicodeDecodeError, json.JSONDecodeError) as e: + logger.debug(f"MQTT decode error: {e}") + except Exception as e: + logger.warning(f"MQTT message handling error: {e}") + + client.on_connect = on_connect + client.on_message = on_message + + try: + _connect(printer, client) + client.loop_start() + while not received_terminal.is_set(): + received_terminal.wait(1.0) + except KeyboardInterrupt: + logger.info("\nšŸ›‘ Monitor loop stopped by user.") + finally: + if userdata.get("progress"): + try: + userdata["progress"].stop() + except Exception: + pass + try: + client.loop_stop() + client.disconnect() + except Exception: + pass diff --git a/bambu_cli/protocols/mqtt_print.py b/bambu_cli/protocols/mqtt_print.py new file mode 100644 index 0000000..f6b804f --- /dev/null +++ b/bambu_cli/protocols/mqtt_print.py @@ -0,0 +1,205 @@ +"""MQTT print-start: publish project_file and wait for ack / error.""" + +from __future__ import annotations + +import json +import threading + +from bambu_cli.config import get_command_timeout +from bambu_cli.errors import BambuError, abort +from bambu_cli.logging_utils import logger +from bambu_cli.protocols.mqtt_cmd import _client_factory, _connect, status_is_complete + + +def _printer_error_hex(code: object) -> str | None: + """Render a printer error code as the hex form Bambu documents (e.g. 0x0500C010).""" + if isinstance(code, bool) or not isinstance(code, int): + return None + return f"0x{code & 0xFFFFFFFF:08X}" + + +def execute_print_command( + printer, + payload, + basename, + dry_run=False, + *, + command_timeout=None, + client_factory=None, +): + """Send the print payload via MQTT and monitor for errors. + + ``command_timeout`` and ``client_factory`` are injectable; defaults are + get_command_timeout() / create_mqtt_client. + """ + from bambu_cli.constants import EXIT_FILE_ERROR, EXIT_NETWORK_ERROR, EXIT_PRINTER_ERROR, EXIT_TIMEOUT + from bambu_cli.utils import record_error_detail + + _factory = _client_factory(client_factory) + + if dry_run: + logger.info(f"šŸ” Dry Run: Checking if {basename} exists on printer...") + try: + with printer.get_ftp_client(timeout=5) as ftp: + files = ftp.nlst("/model/") + if basename in files or f"/model/{basename}" in files: + logger.info(f" āœ… File {basename} found on printer.") + else: + message = f"File {basename} was not found on printer. Upload it first." + logger.error(f" āŒ File {basename} NOT found on printer. Upload it first.") + record_error_detail( + "print", EXIT_FILE_ERROR, message, failed_step="dry_run", file=basename, printed=False + ) + abort("", exit_code=EXIT_FILE_ERROR) + logger.info(" āœ… Printer reachable via MQTT (status check)...") + if printer.status(timeout=5, require_complete=False): + logger.info(" āœ… MQTT connection verified.") + else: + message = "MQTT connection failed." + logger.error(f" āŒ {message}") + record_error_detail( + "print", EXIT_NETWORK_ERROR, message, failed_step="dry_run", file=basename, printed=False + ) + abort("", exit_code=EXIT_NETWORK_ERROR) + return + except BambuError: + raise + except Exception as e: + message = f"Dry run failed: {e}" + logger.error(message) + record_error_detail( + "print", EXIT_NETWORK_ERROR, message, failed_step="dry_run", file=basename, printed=False + ) + abort("", exit_code=EXIT_NETWORK_ERROR) + + if printer.simulation_mode: + from bambu_cli.protocols.ftps import _SIM_FTP_FILES + + if basename not in _SIM_FTP_FILES: + message = f"File {basename} not found on simulated printer. Upload it first." + logger.error(message) + record_error_detail("print", EXIT_FILE_ERROR, message, failed_step="print", file=basename, printed=False) + abort("", exit_code=EXIT_FILE_ERROR) + logger.info(f"šŸ¤– [SIM] Print started: {basename}") + return + + client = _factory(printer, "bambu_print") + + print_error = [None] + reject_reason: list[str | None] = [None] + command_accepted = threading.Event() + connect_failed = [False] + published = [False] + ack_seen = [False] + + def on_connect(client, userdata, flags, rc, properties=None): + if rc == 0: + client.subscribe(f"device/{printer.serial}/report") + if not published[0]: + published[0] = True + client.publish(f"device/{printer.serial}/request", payload) + else: + logger.error(f"Connection failed: rc={rc}") + connect_failed[0] = True + command_accepted.set() + + def on_message(client, userdata, msg): + try: + data = json.loads(msg.payload.decode("utf-8")) + if "print" in data: + p = data["print"] + is_ack = p.get("command") == "project_file" + if is_ack: + ack_seen[0] = True + pe = p.get("print_error", 0) + stale_snapshot = status_is_complete(p) and not ack_seen[0] and not is_ack + if pe and pe != 0 and not stale_snapshot: + print_error[0] = pe + command_accepted.set() + if is_ack: + result = p.get("result") + if isinstance(result, str) and result.strip().lower() not in ("", "success"): + reject_reason[0] = str(p.get("reason") or result) + command_accepted.set() + except (UnicodeDecodeError, json.JSONDecodeError) as e: + logger.debug(f"MQTT decode error: {e}") + except Exception as e: + logger.warning(f"MQTT message handling error: {e}") + + client.on_connect = on_connect + client.on_message = on_message + + base_timeout = command_timeout if command_timeout is not None else get_command_timeout() + print_ack_timeout = base_timeout + 5 + + try: + _connect(printer, client) + client.loop_start() + try: + accepted = command_accepted.wait(print_ack_timeout) + if not accepted: + message = f"Timed out waiting for printer to acknowledge print start for {basename}" + logger.error(message) + record_error_detail("print", EXIT_TIMEOUT, message, failed_step="print", file=basename, printed=False) + abort("", exit_code=EXIT_TIMEOUT) + finally: + try: + client.loop_stop() + except Exception: + pass + try: + client.disconnect() + except Exception: + pass + except BambuError: + raise + except Exception as e: + message = f"Error: {e}" + logger.error(message) + record_error_detail("print", EXIT_NETWORK_ERROR, message, failed_step="print", file=basename, printed=False) + abort("", exit_code=EXIT_NETWORK_ERROR) + + if connect_failed[0]: + message = f"Failed to connect to printer to start print for {basename} (check LAN access code)" + logger.error(message) + record_error_detail("print", EXIT_NETWORK_ERROR, message, failed_step="print", file=basename, printed=False) + abort("", exit_code=EXIT_NETWORK_ERROR) + + if reject_reason[0]: + message = f"Printer rejected print of {basename}: {reject_reason[0]}" + logger.error(message) + record_error_detail("print", EXIT_PRINTER_ERROR, message, failed_step="print", file=basename, printed=False) + abort("", exit_code=EXIT_PRINTER_ERROR) + + if print_error[0]: + error_hex = _printer_error_hex(print_error[0]) + message = f"Print failed with error code {print_error[0]}" + if error_hex: + message += f" (hex {error_hex})" + logger.error(message) + if print_error[0] == 83935248: + logger.info(" File not found on printer SD card. Check filename with 'files' command.") + record_error_detail( + "print", + EXIT_FILE_ERROR, + "File not found on printer SD card. Check filename with 'files' command.", + failed_step="print", + file=basename, + printer_error_code=print_error[0], + printer_error_code_hex=error_hex, + printed=False, + ) + abort("", exit_code=EXIT_FILE_ERROR) + record_error_detail( + "print", + EXIT_PRINTER_ERROR, + message, + failed_step="print", + file=basename, + printer_error_code=print_error[0], + printer_error_code_hex=error_hex, + printed=False, + ) + abort("", exit_code=EXIT_PRINTER_ERROR) + else: + logger.info(f"šŸ–Øļø Print started: {basename}") diff --git a/bambu_cli/protocols/mqtt_tls.py b/bambu_cli/protocols/mqtt_tls.py new file mode 100644 index 0000000..24a1f54 --- /dev/null +++ b/bambu_cli/protocols/mqtt_tls.py @@ -0,0 +1,152 @@ +"""MQTT TLS client construction and connect. + +Pinning uses :class:`PinningSSLContext` ā€” a real ``ssl.SSLContext`` subclass +whose ``wrap_socket`` completes the handshake and calls +``verify_cert_fingerprint``. paho 2.x wraps with +``do_handshake_on_connect=False`` and then calls ``do_handshake()`` itself; +we handshake and pin inside ``wrap_socket`` so a later ``do_handshake`` is a +no-op and the pin is a straight ``tlspin`` call. +""" + +from __future__ import annotations + +import socket +import ssl + +from bambu_cli.logging_utils import logger +from bambu_cli.utils import _resolve_ip + +try: + import paho.mqtt.client as mqtt +except ImportError: + mqtt = None + + +def _require_mqtt(): + """Ensure paho-mqtt is importable; abort with config exit if missing.""" + global mqtt + if mqtt is not None: + return + try: + import paho.mqtt.client as paho_mqtt + + mqtt = paho_mqtt + except ImportError: + logger.error( + "Missing dependency: paho-mqtt. Reinstall the package " + "(e.g. `uv pip install -e .` from a source checkout, or `pip install platecli`)." + ) + from bambu_cli.constants import EXIT_CONFIG_ERROR + from bambu_cli.errors import abort + + abort("", exit_code=EXIT_CONFIG_ERROR) + + +class PinningSSLContext(ssl.SSLContext): + """TLS client context that pins the peer cert after handshake.""" + + expected_fingerprint: str + + def wrap_socket(self, sock, *args, **kwargs): + kwargs = dict(kwargs) + kwargs["do_handshake_on_connect"] = False + tls_sock = super().wrap_socket(sock, *args, **kwargs) + tls_sock.do_handshake() + from bambu_cli.tlspin import verify_cert_fingerprint + + verify_cert_fingerprint(tls_sock.getpeercert(binary_form=True), self.expected_fingerprint) + return tls_sock + + +def pinning_ssl_context(expected_fingerprint: str) -> PinningSSLContext: + ctx = PinningSSLContext(ssl.PROTOCOL_TLS_CLIENT) + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + ctx.expected_fingerprint = expected_fingerprint + return ctx + + +class _SimMqttClient: + """Small MQTT stand-in for --sim without importing test-only mocks.""" + + def __init__(self): + self.on_connect = None + self.on_message = None + self.on_publish = None + + def username_pw_set(self, username, password): + pass + + def tls_set(self, *args, **kwargs): + pass + + def tls_insecure_set(self, *args, **kwargs): + pass + + def connect(self, host, port, keepalive=10): + if self.on_connect: + self.on_connect(self, None, None, 0) + + def subscribe(self, topic): + pass + + def publish(self, topic, payload): + if self.on_publish: + self.on_publish(self, None, 1) + + def loop_start(self): + pass + + def loop_stop(self): + pass + + def disconnect(self): + pass + + def socket(self): + return None + + +def probe_cert_fingerprint(host, port=990, timeout=5): + """Open a TLS connection purely to read the server cert's SHA-256 fingerprint.""" + from bambu_cli.config import fingerprint_sha256 + + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + with socket.create_connection((host, port), timeout) as raw, ctx.wrap_socket(raw, server_hostname=host) as tls: + return fingerprint_sha256(tls.getpeercert(binary_form=True)) + + +def create_mqtt_client(printer, client_id=""): + if printer.simulation_mode: + return _SimMqttClient() + + _require_mqtt() + try: + client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id) + except AttributeError: + client = mqtt.Client(client_id) + client.username_pw_set("bblp", printer.access_code) + + if printer.insecure_tls: + client.tls_set(cert_reqs=ssl.CERT_NONE) + client.tls_insecure_set(True) + elif printer.cert_fingerprint: + client.tls_set_context(pinning_ssl_context(printer.cert_fingerprint)) + client.tls_insecure_set(True) + else: + client.tls_set(cert_reqs=ssl.CERT_REQUIRED) + return client + + +def _mqtt_connect(printer, client): + resolved_ip = _resolve_ip(printer.ip) + old_timeout = socket.getdefaulttimeout() + try: + socket.setdefaulttimeout(printer.mqtt_timeout) + if hasattr(client, "_connect_timeout"): + client._connect_timeout = printer.mqtt_timeout + client.connect(resolved_ip, 8883, keepalive=10) + finally: + socket.setdefaulttimeout(old_timeout) diff --git a/pyproject.toml b/pyproject.toml index 1a6ca72..01492bd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -140,7 +140,9 @@ ignore = ["SIM105"] [tool.ruff.lint.per-file-ignores] # MQTT callbacks close over per-iteration loop variables deliberately; each # callback's lifetime ends with its iteration. -"bambu_cli/protocols/mqtt.py" = ["B023"] +"bambu_cli/protocols/mqtt_cmd.py" = ["B023"] +"bambu_cli/protocols/mqtt_print.py" = ["B023"] +"bambu_cli/protocols/mqtt_monitor.py" = ["B023"] [tool.mypy] # Rot-proof gate: check the whole bambu_cli package by default (`uvx mypy -p bambu_cli`). diff --git a/tests/test_coverage_platform_paths.py b/tests/test_coverage_platform_paths.py index 81257e9..b0dd942 100644 --- a/tests/test_coverage_platform_paths.py +++ b/tests/test_coverage_platform_paths.py @@ -110,13 +110,15 @@ def test_ftps_connection_error_path_cleanup(): def test_mqtt_require_missing_dependency(): - prev = mqtt_mod.mqtt + from bambu_cli.protocols import mqtt_tls + + prev = mqtt_tls.mqtt try: - mqtt_mod.mqtt = None + mqtt_tls.mqtt = None with patch.dict("sys.modules", {"paho.mqtt.client": None}), pytest.raises(BambuError): - mqtt_mod._require_mqtt() + mqtt_tls._require_mqtt() finally: - mqtt_mod.mqtt = prev + mqtt_tls.mqtt = prev def test_cmd_gcode_success(): diff --git a/tests/test_doctor_and_safety.py b/tests/test_doctor_and_safety.py index 368b1dc..d891bdc 100644 --- a/tests/test_doctor_and_safety.py +++ b/tests/test_doctor_and_safety.py @@ -238,7 +238,7 @@ def test_simulation_mode_status(self): with ( settings_ctx(simulation=True), patch("bambu_cli.commands.status.logger", mock_logger), - patch("bambu_cli.protocols.mqtt.logger", mock_logger), + patch("bambu_cli.protocols.mqtt_cmd.logger", mock_logger), ): cmd_status(args) self.assertTrue( diff --git a/tests/test_netsafety_handlers.py b/tests/test_netsafety_handlers.py index d98870b..074169b 100644 --- a/tests/test_netsafety_handlers.py +++ b/tests/test_netsafety_handlers.py @@ -69,9 +69,11 @@ def test_safe_https_handler_open_with_context_attrs(): def test_require_mqtt_import_error_aborts(): - prev = mqtt_mod.mqtt + from bambu_cli.protocols import mqtt_tls + + prev = mqtt_tls.mqtt try: - mqtt_mod.mqtt = None + mqtt_tls.mqtt = None with patch.dict("sys.modules", {"paho": None, "paho.mqtt": None, "paho.mqtt.client": None}): import builtins @@ -83,10 +85,10 @@ def _boom(name, *a, **k): return real_import(name, *a, **k) with patch("builtins.__import__", side_effect=_boom), pytest.raises(BambuError) as ei: - mqtt_mod._require_mqtt() + mqtt_tls._require_mqtt() assert ei.value.exit_code != 0 finally: - mqtt_mod.mqtt = prev + mqtt_tls.mqtt = prev def test_preflight_permission_win32_skips(monkeypatch): diff --git a/tests/test_protocol_clients.py b/tests/test_protocol_clients.py index 16514fc..7a19b6b 100644 --- a/tests/test_protocol_clients.py +++ b/tests/test_protocol_clients.py @@ -120,7 +120,7 @@ def side_effect_connect(host, port, keepalive): mock_client.connect.side_effect = side_effect_connect - with patch("bambu_cli.protocols.mqtt.logger", mock_logger): + with patch("bambu_cli.protocols.mqtt_cmd.logger", mock_logger): result = send_command( _test_printer(ip="192.168.1.1"), '{"test": "payload"}', @@ -180,7 +180,7 @@ def test_create_mqtt_client_simulation(self): self.assertIsInstance(client, _SimMqttClient) - @patch("bambu_cli.protocols.mqtt.mqtt.Client") + @patch("bambu_cli.protocols.mqtt_tls.mqtt.Client") def test_create_mqtt_client_secure(self, mock_mqtt_client): mock_client_instance = MagicMock() mock_mqtt_client.return_value = mock_client_instance @@ -195,7 +195,7 @@ def test_create_mqtt_client_secure(self, mock_mqtt_client): mock_client_instance.tls_insecure_set.assert_not_called() self.assertEqual(client, mock_client_instance) - @patch("bambu_cli.protocols.mqtt.mqtt.Client") + @patch("bambu_cli.protocols.mqtt_tls.mqtt.Client") def test_create_mqtt_client_insecure(self, mock_mqtt_client): mock_client_instance = MagicMock() mock_mqtt_client.return_value = mock_client_instance @@ -214,7 +214,7 @@ class TestMqttConnectTimeout(unittest.TestCase): def test_mqtt_connect_honors_configured_timeout_and_restores_socket_default(self): import socket as socket_mod - from bambu_cli.protocols import mqtt + from bambu_cli.protocols import mqtt_tls client = MagicMock() client._connect_timeout = 5.0 @@ -223,8 +223,8 @@ def test_mqtt_connect_honors_configured_timeout_and_restores_socket_default(self printer.mqtt_timeout = 30.0 before = socket_mod.getdefaulttimeout() - with patch.object(mqtt, "_resolve_ip", return_value="192.168.1.5"): - mqtt._mqtt_connect(printer, client) + with patch.object(mqtt_tls, "_resolve_ip", return_value="192.168.1.5"): + mqtt_tls._mqtt_connect(printer, client) # paho's own connect cap is raised to the configured timeout... self.assertEqual(client._connect_timeout, 30.0) diff --git a/tests/test_sim_transport_setup.py b/tests/test_sim_transport_setup.py index 717d567..0b63d19 100644 --- a/tests/test_sim_transport_setup.py +++ b/tests/test_sim_transport_setup.py @@ -108,7 +108,7 @@ def test_probe_cert_fingerprint_reads_der(): ctx = MagicMock() ctx.wrap_socket.return_value = tls with ( - patch("bambu_cli.protocols.mqtt.socket.create_connection", return_value=raw_cm), + patch("bambu_cli.protocols.mqtt_tls.socket.create_connection", return_value=raw_cm), patch("ssl.SSLContext", return_value=ctx), ): fp = mqtt_mod.probe_cert_fingerprint("10.0.0.1", 990, timeout=1) diff --git a/tests/test_tls_pinning.py b/tests/test_tls_pinning.py index 0fc4f21..0840c6d 100644 --- a/tests/test_tls_pinning.py +++ b/tests/test_tls_pinning.py @@ -18,7 +18,7 @@ sys.modules.setdefault("paho.mqtt.client", _mock_mqtt) from bambu_cli.protocols import ftps as ftps_mod # noqa: E402 -from bambu_cli.protocols import mqtt as mqtt_mod # noqa: E402 +from bambu_cli.protocols import mqtt_tls as mqtt_tls # noqa: E402 from tests.bambu_test_base import _test_printer # noqa: E402 pytestmark = pytest.mark.security @@ -28,129 +28,78 @@ _FP_OTHER = "ab" * 32 -def _mqtt_client_with_pin_context(cert_fingerprint: str): - """Build an MQTT client under a fake SSLContext; return (client, ctx, base_wrap).""" - tls_sock = MagicMock(name="tls_sock") - tls_sock.getpeercert.return_value = _DER - - ctx_inst = MagicMock(name="ssl_context") - base_wrap = MagicMock(return_value=tls_sock) - ctx_inst.wrap_socket = base_wrap - - mock_client = MagicMock(name="mqtt_client") - with ( - patch.object(mqtt_mod, "mqtt") as mock_mqtt_mod, - patch("ssl.SSLContext", return_value=ctx_inst), - ): - mock_mqtt_mod.Client.return_value = mock_client - mock_mqtt_mod.CallbackAPIVersion.VERSION2 = "v2" - printer = _test_printer(cert_fingerprint=cert_fingerprint, insecure_tls=False) - client = mqtt_mod.create_mqtt_client(printer) - return client, ctx_inst, base_wrap, tls_sock, mock_client - - -def test_mqtt_create_client_with_pin_uses_context_and_insecure_flag(): - client, ctx_inst, _base, _tls, mock_client = _mqtt_client_with_pin_context(_FP) - assert client is mock_client - mock_client.tls_set_context.assert_called_once_with(ctx_inst) - mock_client.tls_insecure_set.assert_called_once_with(True) - mock_client.tls_set.assert_not_called() - - -def test_mqtt_pin_match_allows_wrap(): - _client, ctx_inst, base_wrap, tls_sock, _mc = _mqtt_client_with_pin_context(_FP) - pinned_wrap = ctx_inst.wrap_socket - assert pinned_wrap is not base_wrap - assert pinned_wrap(object(), server_hostname="printer.local") is tls_sock - base_wrap.assert_called() - - -def test_mqtt_pin_mismatch_raises_sslerror(): - _client, ctx_inst, base_wrap, tls_sock, _mc = _mqtt_client_with_pin_context(_FP) - tls_sock.getpeercert.return_value = b"\x00wrong-cert" - with pytest.raises(ssl.SSLError, match="fingerprint mismatch"): - ctx_inst.wrap_socket(object()) - - -def test_mqtt_pin_missing_peer_cert_raises(): - _client, ctx_inst, _base, tls_sock, _mc = _mqtt_client_with_pin_context(_FP) - tls_sock.getpeercert.return_value = None - with pytest.raises(ssl.SSLError, match="No peer certificate"): - ctx_inst.wrap_socket(object()) - - -def test_mqtt_pin_deferred_until_handshake(): - """paho often connects with handshake deferred; pin must run on do_handshake.""" - der = _DER - fp = _FP - tls_sock = MagicMock() - # First getpeercert (handshake probe) raises; after do_handshake it returns der. +def _tls_sock(der=_DER): + tls = MagicMock(name="tls_sock") state = {"ready": False} + def do_handshake(*a, **k): + state["ready"] = True + def getpeercert(binary_form=False): if not state["ready"]: raise ValueError("handshake not done") return der - tls_sock.getpeercert.side_effect = getpeercert + tls.do_handshake.side_effect = do_handshake + tls.getpeercert.side_effect = getpeercert + tls._pin_state = state + return tls - def do_handshake(*a, **k): - state["ready"] = True - return None - tls_sock.do_handshake = do_handshake - - ctx_inst = MagicMock() - base_wrap = MagicMock(return_value=tls_sock) - ctx_inst.wrap_socket = base_wrap - mock_client = MagicMock() - - with ( - patch.object(mqtt_mod, "mqtt") as mock_mqtt_mod, - patch("ssl.SSLContext", return_value=ctx_inst), - ): +def test_mqtt_create_client_with_pin_uses_pinning_context(): + mock_client = MagicMock(name="mqtt_client") + with patch.object(mqtt_tls, "mqtt") as mock_mqtt_mod: mock_mqtt_mod.Client.return_value = mock_client mock_mqtt_mod.CallbackAPIVersion.VERSION2 = "v2" - mqtt_mod.create_mqtt_client(_test_printer(cert_fingerprint=fp)) - - out = ctx_inst.wrap_socket(object()) - assert out is tls_sock - # Handshake wrapper installed; invoking it must pin successfully. - out.do_handshake() - assert state["ready"] is True + client = mqtt_tls.create_mqtt_client(_test_printer(cert_fingerprint=_FP, insecure_tls=False)) + assert client is mock_client + mock_client.tls_set_context.assert_called_once() + ctx = mock_client.tls_set_context.call_args[0][0] + assert isinstance(ctx, mqtt_tls.PinningSSLContext) + assert ctx.expected_fingerprint == _FP + mock_client.tls_insecure_set.assert_called_once_with(True) + mock_client.tls_set.assert_not_called() -def test_mqtt_pin_deferred_mismatch_on_handshake(): - tls_sock = MagicMock() - state = {"ready": False} +def test_pinning_context_match_handshakes_then_verifies(): + tls = _tls_sock(_DER) + ctx = mqtt_tls.pinning_ssl_context(_FP) + with patch.object(ssl.SSLContext, "wrap_socket", return_value=tls) as super_wrap: + assert ctx.wrap_socket(object(), server_hostname="printer.local") is tls + super_wrap.assert_called() + tls.do_handshake.assert_called_once() + assert tls._pin_state["ready"] is True - def getpeercert(binary_form=False): - if not state["ready"]: - raise ValueError("handshake not done") - return b"\xffnot-the-pinned-cert" - tls_sock.getpeercert.side_effect = getpeercert +def test_pinning_context_mismatch_raises_sslerror(): + tls = _tls_sock(b"\x00wrong-cert") + ctx = mqtt_tls.pinning_ssl_context(_FP) + with ( + patch.object(ssl.SSLContext, "wrap_socket", return_value=tls), + pytest.raises(ssl.SSLError, match="fingerprint mismatch"), + ): + ctx.wrap_socket(object()) + tls.do_handshake.assert_called_once() - def do_handshake(*a, **k): - state["ready"] = True - tls_sock.do_handshake = do_handshake +def test_pinning_context_missing_peer_cert_raises(): + tls = _tls_sock(None) + ctx = mqtt_tls.pinning_ssl_context(_FP) + with ( + patch.object(ssl.SSLContext, "wrap_socket", return_value=tls), + pytest.raises(ssl.SSLError, match="No peer certificate"), + ): + ctx.wrap_socket(object()) - ctx_inst = MagicMock() - ctx_inst.wrap_socket = MagicMock(return_value=tls_sock) - mock_client = MagicMock() +def test_pinning_context_malformed_pin_raises(): + tls = _tls_sock(_DER) + ctx = mqtt_tls.pinning_ssl_context("Š°" + "b" * 63) with ( - patch.object(mqtt_mod, "mqtt") as mock_mqtt_mod, - patch("ssl.SSLContext", return_value=ctx_inst), + patch.object(ssl.SSLContext, "wrap_socket", return_value=tls), + pytest.raises(ssl.SSLError, match="[Mm]alformed"), ): - mock_mqtt_mod.Client.return_value = mock_client - mock_mqtt_mod.CallbackAPIVersion.VERSION2 = "v2" - mqtt_mod.create_mqtt_client(_test_printer(cert_fingerprint=_FP)) - - sock = ctx_inst.wrap_socket(object()) - with pytest.raises(ssl.SSLError, match="fingerprint mismatch"): - sock.do_handshake() + ctx.wrap_socket(object()) def test_ftps_pin_match_on_connect():