From ebb48aac828473b7051c07ae3c741321dac78f00 Mon Sep 17 00:00:00 2001 From: viewit Date: Tue, 4 Aug 2026 19:43:25 +0200 Subject: [PATCH] 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 | 75 ++++++++++++++++++++++++++++++++++++++ kobrax_moonraker_bridge.py | 55 +--------------------------- 2 files changed, 76 insertions(+), 54 deletions(-) create mode 100644 credentials.py diff --git a/credentials.py b/credentials.py new file mode 100644 index 0000000..e9d8970 --- /dev/null +++ b/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"), + } diff --git a/kobrax_moonraker_bridge.py b/kobrax_moonraker_bridge.py index 5f7f371..272b597 100644 --- a/kobrax_moonraker_bridge.py +++ b/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")