viewit
ebb48aac82
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.
76 lines
3.1 KiB
Python
76 lines
3.1 KiB
Python
"""
|
|
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"),
|
|
}
|