refactor(bridge): extract printer credential fetch/decrypt into credentials.py (stage 2)

Moves the coherent printer-credential unit (_kx_generate_signature,
_kx_decrypt_info, _kx_fetch_credentials plus the pycryptodome import guard)
out of the facade into credentials.py, re-exported for the "add printer"
flow. The remaining free functions (build_app, main, _build_per_printer_args,
_default_data_dir, _mqtt_error_msg) stay in the facade - they're entrypoint
logic or travel more naturally with the core mixin in the next stage.

All 184 tests green, no behavior change.

credentials.py

@@ -0,0 +1,75 @@
+"""
+credentials.py - fetch + decrypt Anycubic Kobra printer credentials.
+
+Talks to the printer's HTTP /info + /ctrl endpoints (port 18910) and decrypts
+the AES-256-CBC response to recover username/password/device_id/mode_id.
+Algorithm reverse-engineered in tools/fetch_credentials.py. Extracted from
+kobrax_moonraker_bridge.py; _kx_fetch_credentials is re-imported there for the
+"add printer" flow.
+
+────────────────────────────────────────────────────────────────────────────
+Copyright (C) 2026 viewit (KX-Bridge contributors)
+
+Licensed under GPLv3 — see LICENSE in the project root.
+Protocol reverse-engineered for interoperability (§69e UrhG). See NOTICE.md.
+"""
+
+import json
+import time
+import hashlib
+
+import aiohttp
+
+try:
+    import base64 as _base64
+    from Crypto.Cipher import AES as _AES
+    from Crypto.Util.Padding import unpad as _unpad
+    _HAS_CRYPTO = True
+except ImportError:
+    _HAS_CRYPTO = False
+
+
+def _kx_generate_signature(token: str, ts: int, nonce: str) -> str:
+    first = hashlib.md5(token[:16].encode()).hexdigest()
+    return hashlib.md5((first + str(ts) + nonce).encode()).hexdigest()
+
+
+def _kx_decrypt_info(encrypted_b64: str, key: str, iv: str) -> dict:
+    cipher = _AES.new(key.encode(), _AES.MODE_CBC, iv.encode())
+    raw = _base64.b64decode(encrypted_b64)
+    return json.loads(_unpad(cipher.decrypt(raw), _AES.block_size).decode())
+
+
+async def _kx_fetch_credentials(ip: str, port: int = 18910) -> dict:
+    """Fetches + decrypts printer credentials via HTTP /info + /ctrl.
+
+    Raises an exception on network/decrypt errors. Algorithm from
+    tools/fetch_credentials.py (AES-256-CBC, Key=token[16:32], IV=ctrl-token).
+    """
+    if not _HAS_CRYPTO:
+        raise RuntimeError("pycryptodome is not installed")
+    import random, string
+    nonce = "".join(random.choice(string.ascii_letters + string.digits) for _ in range(6))
+    timeout = aiohttp.ClientTimeout(total=10)
+    async with aiohttp.ClientSession() as s:
+        async with s.get(f"http://{ip}:{port}/info", timeout=timeout) as r:
+            r.raise_for_status()
+            info = await r.json()
+        token = info["token"]
+        ts = int(time.time() * 1000)
+        sign = _kx_generate_signature(token, ts, nonce)
+        params = {"ts": ts, "nonce": nonce, "sign": sign, "did": "random"}
+        async with s.post(f"http://{ip}:{port}/ctrl", params=params, timeout=timeout) as r:
+            r.raise_for_status()
+            data = await r.json()
+    result = _kx_decrypt_info(data["data"]["info"], token[16:32], data["data"]["token"])
+    if "error" in result:
+        raise RuntimeError(result.get("error", "decrypt failed"))
+    return {
+        "printer_ip": result.get("ip", ip),
+        "username":   result.get("username", ""),
+        "password":   result.get("password", ""),
+        "device_id":  result.get("deviceId", ""),
+        "mode_id":    str(result.get("modeId", "20030")),
+        "model":      result.get("modelName", "Anycubic Kobra"),
+    }

kobrax_moonraker_bridge.py

@@ -66,6 +66,7 @@ from gcode_meta import (
     _extract_filament_info,
 )
 from camera import CameraCache, _find_ffmpeg
+from credentials import _kx_fetch_credentials, _kx_generate_signature, _kx_decrypt_info
 
 
 try:
@@ -75,60 +76,6 @@ except ImportError:
     print("Error: aiohttp is not installed. Run: pip install aiohttp")
     sys.exit(1)
 
-try:
-    import base64 as _base64
-    from Crypto.Cipher import AES as _AES
-    from Crypto.Util.Padding import unpad as _unpad
-    _HAS_CRYPTO = True
-except ImportError:
-    _HAS_CRYPTO = False
-
-
-def _kx_generate_signature(token: str, ts: int, nonce: str) -> str:
-    first = hashlib.md5(token[:16].encode()).hexdigest()
-    return hashlib.md5((first + str(ts) + nonce).encode()).hexdigest()
-
-
-def _kx_decrypt_info(encrypted_b64: str, key: str, iv: str) -> dict:
-    cipher = _AES.new(key.encode(), _AES.MODE_CBC, iv.encode())
-    raw = _base64.b64decode(encrypted_b64)
-    return json.loads(_unpad(cipher.decrypt(raw), _AES.block_size).decode())
-
-
-async def _kx_fetch_credentials(ip: str, port: int = 18910) -> dict:
-    """Fetches + decrypts printer credentials via HTTP /info + /ctrl.
-
-    Raises an exception on network/decrypt errors. Algorithm from
-    tools/fetch_credentials.py (AES-256-CBC, Key=token[16:32], IV=ctrl-token).
-    """
-    if not _HAS_CRYPTO:
-        raise RuntimeError("pycryptodome is not installed")
-    import random, string
-    nonce = "".join(random.choice(string.ascii_letters + string.digits) for _ in range(6))
-    timeout = aiohttp.ClientTimeout(total=10)
-    async with aiohttp.ClientSession() as s:
-        async with s.get(f"http://{ip}:{port}/info", timeout=timeout) as r:
-            r.raise_for_status()
-            info = await r.json()
-        token = info["token"]
-        ts = int(time.time() * 1000)
-        sign = _kx_generate_signature(token, ts, nonce)
-        params = {"ts": ts, "nonce": nonce, "sign": sign, "did": "random"}
-        async with s.post(f"http://{ip}:{port}/ctrl", params=params, timeout=timeout) as r:
-            r.raise_for_status()
-            data = await r.json()
-    result = _kx_decrypt_info(data["data"]["info"], token[16:32], data["data"]["token"])
-    if "error" in result:
-        raise RuntimeError(result.get("error", "decrypt failed"))
-    return {
-        "printer_ip": result.get("ip", ip),
-        "username":   result.get("username", ""),
-        "password":   result.get("password", ""),
-        "device_id":  result.get("deviceId", ""),
-        "mode_id":    str(result.get("modeId", "20030")),
-        "model":      result.get("modelName", "Anycubic Kobra"),
-    }
-
 logging.basicConfig(level=logging.INFO,
                     format="[%(asctime)s] %(levelname)-5s %(name)s: %(message)s",
                     datefmt="%H:%M:%S")