refactor: drop dead MQTT/FTPS leftovers and unify pin + redaction

bambu_cli/commands/doctor.py

@@ -66,7 +66,6 @@ def _offer_pin_fingerprint(
 
 def cmd_doctor(args, ctx=None):
     """Health-check: auto-discover printer capabilities and verify configuration."""
-    from bambu_cli.protocols.ftps import get_ftp
     from bambu_cli.protocols.mqtt import probe_cert_fingerprint
     from bambu_cli.utils import emit_json
 
@@ -161,7 +160,7 @@ def shown_ip():
 
     logger.info(f"   [3/3] Verifying FTPS connectivity to {shown_ip()}:990...")
     try:
-        with get_ftp(printer, timeout=net_timeout):
+        with printer.get_ftp_client(timeout=net_timeout):
             logger.info("   ✅ FTPS connection established.")
     except Exception as e:
         message = f"FTPS connection failed: {e}"

bambu_cli/config.py

@@ -480,11 +480,9 @@ def load_username():
 def _expected_fingerprint():
     """Return the normalized (lowercase, separator-free) pinned SHA-256, or None."""
     from bambu_cli.context import current_config
+    from bambu_cli.tlspin import normalize_fingerprint
 
-    fp = current_config().get("cert_fingerprint")
-    if not fp:
-        return None
-    return fp.lower().replace(":", "").replace(" ", "")
+    return normalize_fingerprint(current_config().get("cert_fingerprint"))
 
 
 def fingerprint_sha256(der_cert):

bambu_cli/context.py

@@ -72,13 +72,10 @@ def _coerce_insecure_tls(value: Any) -> bool:
 
 
 def _normalize_fingerprint(fp: str | None) -> str | None:
-    """Normalize a pinned SHA-256 fingerprint (lowercase, separator-free).
+    """Normalize a pinned SHA-256 fingerprint (lowercase, separator-free)."""
+    from bambu_cli.tlspin import normalize_fingerprint
 
-    Mirrors ``bambu_cli.config._expected_fingerprint``.
-    """
-    if not fp:
-        return None
-    return fp.lower().replace(":", "").replace(" ", "")
+    return normalize_fingerprint(fp)
 
 
 @dataclass
@@ -190,14 +187,15 @@ def printer(self) -> BambuPrinter:
 
         from bambu_cli.config import load_access_code
         from bambu_cli.printer import BambuPrinter
+        from bambu_cli.tlspin import normalize_fingerprint
 
         access_code = "" if self.simulation else load_access_code()
         self._printer = BambuPrinter(
             ip=self.settings.printer_ip,
             serial=self.settings.serial,
             access_code=access_code,
             insecure_tls=self.settings.insecure_tls,
-            cert_fingerprint=_normalize_fingerprint(self.settings.cert_fingerprint),
+            cert_fingerprint=normalize_fingerprint(self.settings.cert_fingerprint),
             simulation_mode=self.simulation,
         )
         return self._printer

bambu_cli/jsonio.py

@@ -5,10 +5,9 @@
 before they reach a log line or a JSON envelope, without importing from the CLI
 entrypoint. These helpers never terminate the process.
 
-Note: ``bambu_cli.utils`` carries its own credential-redaction pass applied
-uniformly across every emitted JSON payload; the ``redact_url_credentials``
-here is the eager, single-value variant callers use when building the strings
-that go into those payloads and log messages.
+This is the single redactor: command code calls it when building log lines
+and payload fields, and ``bambu_cli.utils.emit_json`` runs the same function
+over every emitted string.
 """
 
 from urllib.parse import urlparse, urlunparse

bambu_cli/printer.py

@@ -67,15 +67,12 @@ def get_ftp_client(self, timeout: Optional[float] = None):
         """Context manager to get a connected FTP client."""
         if timeout is None:
             timeout = self.ftps_timeout
-        # We can implement pooling here in the future
         client = ftps_protocol._create_raw_ftp(self, timeout=timeout)
         try:
             yield client
         finally:
-            try:
-                client.quit()
-            except _FTP_SSL_ERRORS:
-                pass
+            # Bambu's FTPS control channel can hang in quit(); close() tears
+            # the socket down without waiting for the 221 reply.
             try:
                 client.close()
             except _FTP_SSL_ERRORS:
@@ -288,7 +285,8 @@ def get_printer(*, access_code_loader=None) -> BambuPrinter:
     inject a fake instead of patching module globals.
     """
     from bambu_cli.config import load_access_code
-    from bambu_cli.context import _normalize_fingerprint, current_settings, current_simulation
+    from bambu_cli.context import current_settings, current_simulation
+    from bambu_cli.tlspin import normalize_fingerprint
 
     _load = access_code_loader if access_code_loader is not None else load_access_code
     settings = current_settings()
@@ -300,6 +298,6 @@ def get_printer(*, access_code_loader=None) -> BambuPrinter:
         # require credentials (load_access_code exits when unconfigured).
         access_code="" if simulation_mode else _load(),
         insecure_tls=settings.insecure_tls,
-        cert_fingerprint=_normalize_fingerprint(settings.cert_fingerprint),
+        cert_fingerprint=normalize_fingerprint(settings.cert_fingerprint),
         simulation_mode=simulation_mode,
     )

bambu_cli/protocols/ftps.py

@@ -1,25 +1,13 @@
-import atexit
 import ftplib
 import os
 import socket
 import ssl
-import threading
 from typing import Any
 
 from bambu_cli.utils import _resolve_ip
 
 _SIM_FTP_FILES = {"simulated_file.3mf": 1000}
 
-# Module-level so mypy accepts these in `except` clauses (star-unpack of
-# ftplib.all_errors inline is rejected as non-exception-type).
-_FTP_ERRORS: tuple[type[BaseException], ...] = ftplib.all_errors
-_FTP_SSL_ERRORS: tuple[type[BaseException], ...] = ftplib.all_errors + (ssl.SSLError,)
-_FTP_POOL_ERRORS: tuple[type[BaseException], ...] = ftplib.all_errors + (
-    ssl.SSLError,
-    OSError,
-    AttributeError,
-)
-
 
 class _SimFtp:
     """Small FTPS stand-in for --sim without importing test-only mocks."""
@@ -56,7 +44,6 @@ def delete(self, path):
         _SIM_FTP_FILES.pop(os.path.basename(path), None)
 
     def voidcmd(self, cmd):
-        """Pool health-check (NOOP) used by ConnectionManager.get_ftp."""
         return "200 OK"
 
     def quit(self):
@@ -159,86 +146,6 @@ def ntransfercmd(self, cmd, rest=None):
         return conn, size
 
 
-class PooledFTPWrapper:
-    def __init__(self, ftp, manager):
-        self._ftp = ftp
-        self._manager = manager
-
-    def __getattr__(self, name):
-        return getattr(self._ftp, name)
-
-    def __enter__(self):
-        self._manager._ftp_usage_lock.acquire()
-        return self
-
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        try:
-            if exc_type is not None:
-                with self._manager._lock:
-                    if self._manager._ftp_client is self._ftp:
-                        self._manager._ftp_client = None
-                try:
-                    self._ftp.close()
-                except Exception:
-                    pass
-        finally:
-            self._manager._ftp_usage_lock.release()
-
-
-class ConnectionManager:
-    """Manages reusable MQTT and FTPS connections to reduce socket churn."""
-
-    def __init__(self):
-        self._mqtt_client = None
-        self._ftp_client = None
-        self._lock = threading.Lock()
-        self._ftp_usage_lock = threading.Lock()
-
-    def get_ftp(self, printer, timeout=60):
-        with self._lock:
-            client = self._ftp_client
-        if client is not None:
-            try:
-                with self._ftp_usage_lock:
-                    client.voidcmd("NOOP")
-                return PooledFTPWrapper(client, self)
-            except _FTP_POOL_ERRORS:
-                with self._lock:
-                    if self._ftp_client is client and client is not None:
-                        try:
-                            client.close()
-                        except Exception:
-                            pass
-                        self._ftp_client = None
-
-        ftp = _create_raw_ftp(printer, timeout=timeout)
-        with self._lock:
-            self._ftp_client = ftp
-            return PooledFTPWrapper(ftp, self)
-
-    def close_all(self):
-        self.clear()
-
-    def clear(self):
-        with self._lock:
-            if self._mqtt_client is not None:
-                try:
-                    self._mqtt_client.disconnect()
-                except Exception:
-                    pass
-                self._mqtt_client = None
-            if self._ftp_client is not None:
-                try:
-                    self._ftp_client.close()
-                except Exception:
-                    pass
-                self._ftp_client = None
-
-
-connection_manager = ConnectionManager()
-atexit.register(connection_manager.close_all)
-
-
 def _create_raw_ftp(printer, timeout=60):
     """Connect to printer's FTPS server."""
     if printer.simulation_mode:
@@ -255,13 +162,3 @@ def _create_raw_ftp(printer, timeout=60):
     ftp.login("bblp", printer.access_code)
     ftp.prot_p()
     return ftp
-
-
-def get_ftp(printer, timeout=60):
-    """Borrow a pooled FTPS client for *printer*.
-
-    ``printer`` is required: this module must not reach up to
-    ``bambu_cli.printer`` for an ambient one (see scripts/check_layers.py).
-    Callers own the lookup.
-    """
-    return connection_manager.get_ftp(printer, timeout=timeout)

bambu_cli/protocols/mqtt.py

@@ -96,7 +96,6 @@ def probe_cert_fingerprint(host, port=990, timeout=5):
 
 
 def create_mqtt_client(printer, client_id=""):
-    global _TRUSTED_CERT_FILE
     if printer.simulation_mode:
         return _SimMqttClient()
 
@@ -641,31 +640,6 @@ def on_message(client, userdata, msg):
             pass
 
 
-import base64
-
-_TRUSTED_CERT_FILE = None
-
-# probe_cert_fingerprint is defined above
-
-
-def _get_and_verify_cert_pem(host, port, expected_fingerprint, timeout=5):
-    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:
-        der = tls.getpeercert(binary_form=True)
-        from bambu_cli.tlspin import verify_cert_fingerprint
-
-        verify_cert_fingerprint(der, expected_fingerprint)
-        assert der is not None  # verify_cert_fingerprint raises on a missing cert
-        pem = "-----BEGIN CERTIFICATE-----\n"
-        b64 = base64.b64encode(der).decode("ascii")
-        for i in range(0, len(b64), 64):
-            pem += b64[i : i + 64] + "\n"
-        pem += "-----END CERTIFICATE-----\n"
-        return pem
-
-
 def _printer_error_hex(code: object) -> Optional[str]:
     """Render a printer error code as the hex form Bambu documents (e.g. 0x0500C010).
 

bambu_cli/tlspin.py

@@ -10,9 +10,8 @@
 * No pin configured where a pin is required, a mismatched pin, or an
   unobtainable peer cert all raise; none of them return normally.
 * The compare is constant-time (``hmac.compare_digest``) on normalized hex.
-* Inputs are normalized identically to :func:`bambu_cli.config._expected_fingerprint`
-  and :func:`bambu_cli.context._normalize_fingerprint`: lowercased with ``:`` and
-  ASCII spaces stripped. Accepted pin formats therefore include
+* Inputs are normalized by :func:`normalize_fingerprint`: lowercased with ``:``
+  and ASCII spaces stripped. Accepted pin formats therefore include
   ``AA:BB:CC...`` (colon-separated, any case), ``aa bb cc ...`` (space-separated),
   and the bare 64-hex-char digest.
 

bambu_cli/utils.py

@@ -60,26 +60,10 @@ def _ensure_parent_dir(path):
 
 
 def _redact_url_credentials(url):
-    # Performance optimization: Fast-path for non-strings and strings that clearly
-    # cannot contain credentials (missing '@'). This avoids the overhead of
-    # lazy importing urllib and running the relatively expensive urlparse
-    # on every string value in large JSON responses.
-    if not isinstance(url, str) or "@" not in url:
-        return url
+    """Strip URL userinfo. Delegates to ``jsonio.redact_url_credentials``."""
+    from bambu_cli.jsonio import redact_url_credentials
 
-    from urllib.parse import urlparse, urlunparse
-
-    try:
-        parsed = urlparse(url)
-        if parsed.username or parsed.password:
-            netloc = f"***@{parsed.hostname}"
-            if parsed.port:
-                netloc += f":{parsed.port}"
-            parsed = parsed._replace(netloc=netloc)
-            return urlunparse(parsed)
-    except Exception:
-        pass
-    return url
+    return redact_url_credentials(url)
 
 
 _HOME_DIR = os.path.expanduser("~")

tests/bambu_test_base.py

@@ -122,7 +122,6 @@ def cleanup_mock_config():
     from bambu_cli.cli import setup_logging
     from bambu_cli.commands import cmd_stop, cmd_light
     from bambu_cli.config import load_config
-    from bambu_cli.protocols.ftps import get_ftp
     from bambu_cli.protocols.mqtt import create_mqtt_client, execute_print_command
     import ssl
     import urllib.error
@@ -179,7 +178,6 @@ def _setup_slice_proc(mock_proc, returncode=0, stdout=b"", stderr=b""):
         "mock_config_path",
         "bambu",
         "cmd_stop",
-        "get_ftp",
         "load_config",
         "create_mqtt_client",
         "cmd_light",

tests/json_contract_base.py

@@ -50,8 +50,6 @@ def assert_shape(payload, spec, path="$"):
       - "enum": iterable of allowed values for this exact node.
       - "items": subspec applied to every element when type is list.
     """
-    assert isinstance(payload, dict) or "type" in spec or True, path
-
     if "type" in spec:
         expected_type = spec["type"]
         assert isinstance(payload, expected_type), (
@@ -99,7 +97,6 @@ def base_error_spec(command=None, require_failed_step=True):
     return {"type": dict, "required": required}
 
 
-
 # ---------------------------------------------------------------------------
 # Harness
 # ---------------------------------------------------------------------------
@@ -142,5 +139,3 @@ def make_ready_file(tmp_path, name="ready.3mf", content="simulated 3mf content")
     path = tmp_path / name
     path.write_text(content, encoding="utf-8")
     return path
-
-

tests/test_audit_fixes_pr4.py

@@ -56,6 +56,19 @@ def test_redact_preserves_existing_schemeless_and_full_url_behavior():
     assert redact_url_credentials("no-at-sign") == "no-at-sign"
 
 
+def test_emit_json_uses_jsonio_redactor(capsys):
+    """emit_json must strip userinfo, not the weaker ***@ placeholder."""
+    from bambu_cli import utils
+
+    at = "@"
+    utils._JSON_EMITTED = False
+    utils.emit_json({"source": "https://user:pass" + at + "host.com/x.stl"})
+    payload = capsys.readouterr().out
+    assert "user:pass" not in payload
+    assert "***@" not in payload
+    assert "https://host.com/x.stl" in payload
+
+
 # ---------------------------------------------------------------------------
 # utils._display_path — home-prefix separator boundary
 # ---------------------------------------------------------------------------