Compare commits

..

7 Commits

9 changed files with 478 additions and 23 deletions

View File

@@ -1,21 +1,28 @@
# Changelog
## [0.9.6.1] 2026-05-02
## [0.9.7] 2026-05-08
### Neu
- **fetch_credentials-Tool:** Ruft MQTT-Credentials direkt vom Drucker per HTTP ab — kein laufender Anycubic Slicer nötig, nur die Drucker-IP. Linux-Binary und Windows-EXE im Release enthalten. (Beitrag von bebu, PR #19)
### Fixes
- **Upload-Banner:** Banner wird nach Stopp/Abbruch nicht mehr erneut angezeigt — `file_ready` und Thumbnail werden jetzt gecleared wenn der Drucker `stoped` oder `canceled` meldet
- **Upload großer GCode-Dateien:** Dateien >1 MB wurden mit HTTP 413 abgelehnt — aiohttp `client_max_size` auf 256 MB erhöht
- **Upload-Timeout:** Socket-Timeout nach GCode-Upload von 10s auf 120s erhöht — große Dateien führten zu einem Absturz der Bridge mit leerer Antwort während der Drucker noch verarbeitete
---
## [0.9.6] 2026-05-02
### Neu
- **Fortschritts-Karte:** Verstrichen / Slicer-Schätzung / Restzeit als Mini-Cards (gleicher Stil wie Temperaturkarten)
- **Layer-Mini-Card:** Layerzahl als Mini-Card neben der Fortschrittsleiste
- **Licht-Status-Synchronisierung:** Ein/Aus-Zustand und Helligkeit des Druckerlichts werden jetzt live über `light/report` MQTT gelesen — der Licht-Toggle in der UI spiegelt den echten Druckerstatus wider
- **Zeit-Minicards:** Fortschritts-Panel zeigt jetzt drei Karten — Verstrichen, Restzeit und Slicer-Schätzung — sowie einen Layer-Badge neben dem Fortschrittsbalken
- **Slicer-Schätzzeit aus GCode:** Geschätzte Druckzeit wird direkt aus der hochgeladenen GCode-Datei gelesen (OrcaSlicer: `; total estimated time:` am Dateiende, PrusaSlicer: `; estimated printing time` im Header)
- **Erweiterte Druckerstatus-Strings:** `pausing`, `paused`, `resuming`, `resumed`, `stopping`, `stopped` hinzugefügt — fehlten bisher und ließen die UI rohe Status-Codes bei Pause/Fortsetzen/Stopp anzeigen
### Fixes
- **Slicer-Schätzzeit:** OrcaSlicer schreibt die geschätzte Zeit ans Ende der GCode-Datei — Bridge liest jetzt auch die letzten 64 KB (vorher nur die ersten 16 KB)
- **start.sh:** `config/`-Verzeichnis wird jetzt automatisch erstellt und `config.ini.example` wird beim ersten Start hineinkopiert (Issue #15)
- **file_ready-Banner:** Upload-Banner wird nach Stopp oder Abbruch eines Drucks nicht mehr angezeigt
- **Zeitanzeige bei Stopp/Abbruch:** Verstrichen-, Restzeit- und Slicer-Schätzung werden auf null zurückgesetzt wenn ein Druck gestoppt oder abgebrochen wird
- **start.sh:** `config/`-Verzeichnis und `config.ini.example` werden beim ersten Start automatisch angelegt wenn sie fehlen (Issue #15)
---

View File

@@ -1,21 +1,28 @@
# Changelog
## [0.9.6.1] 2026-05-02
## [0.9.7] 2026-05-08
### New
- **fetch_credentials tool:** Fetches and decrypts MQTT credentials directly from the printer via HTTP — no running Anycubic Slicer required, only the printer IP needed. Linux binary and Windows EXE included in release. (Contributed by bebu, PR #19)
### Fixes
- **Upload banner:** Banner is no longer shown again after print stop/cancel — `file_ready` and thumbnail are now cleared when the printer reports `stoped` or `canceled`
- **Large GCode upload:** Files >1 MB were rejected with HTTP 413 — aiohttp `client_max_size` raised to 256 MB
- **Upload timeout:** Socket timeout after GCode upload raised from 10s to 120s — large files caused the bridge to crash with an empty response while the printer was still processing
---
## [0.9.6] 2026-05-02
### New
- **Progress card:** Elapsed / Slicer estimate / Remaining time shown as mini-cards (same style as temperature cards)
- **Layer mini-card:** Layer count displayed as mini-card next to the progress bar
- **Light status sync:** Light on/off state and brightness are now read live from the printer via `light/report` MQTT message — the light toggle in the UI reflects the actual printer state
- **Time mini-cards:** Progress panel now shows three cards — Elapsed, Remaining and Slicer estimate — with a layer counter badge next to the progress bar
- **Slicer estimate from GCode:** Estimated print time is parsed directly from the uploaded GCode file (OrcaSlicer: `; total estimated time:` at end of file, PrusaSlicer: `; estimated printing time` in header)
- **Extended printer status strings:** Added `pausing`, `paused`, `resuming`, `resumed`, `stopping`, `stopped` states — previously missing, causing the UI to show raw status codes during pause/resume/stop transitions
### Fixes
- **Slicer estimate time:** OrcaSlicer writes the estimated time at the end of the GCode file — bridge now also scans the last 64 KB (previously only the first 16 KB were checked)
- **start.sh:** `config/` directory is now created automatically and `config.ini.example` is copied into it on first run (Issue #15)
- **file_ready banner:** Upload banner is no longer shown after print stop or cancel
- **Timers on stop/cancel:** Elapsed, remaining and slicer estimate times are reset to zero when a print is stopped or cancelled
- **start.sh:** `config/` directory and `config.ini.example` are now created automatically on first run if missing (Issue #15)
---

View File

@@ -2,7 +2,7 @@
# KX-Bridge Anycubic Kobra X
**Version:** 0.9.6.1
**Version:** 0.9.7
Steuere deinen **Anycubic Kobra X** mit OrcaSlicer — ohne Klipper, ohne Raspberry Pi.
KX-Bridge ist eine Moonraker-kompatible Bridge die direkt mit dem Drucker kommuniziert.
@@ -18,13 +18,23 @@ Den Kobra X in den LAN-Modus versetzen:
### Schritt 2 Credentials holen
Die MQTT-Zugangsdaten sind druckerspezifisch. So holst du sie:
Die MQTT-Zugangsdaten sind druckerspezifisch und an die Hardware gebunden.
**Option A fetch_credentials (empfohlen):**
```bash
fetch_credentials --ip 192.168.x.x --write-config
```
Holt die Credentials direkt per HTTP vom Drucker und schreibt sie automatisch in `config/config.ini`. Benötigt nur die Drucker-IP — kein Slicer nötig.
**Option B extract_credentials (wenn Drucker-IP unbekannt):**
1. **AnycubicSlicerNext** öffnen und Drucker verbinden (bis Status angezeigt wird)
2. **`extract_credentials.exe`** (Windows) oder **`extract_credentials`** (Linux) ausführen — gibt Username, Password, Device-ID und Drucker-IP aus
3. Werte merken / kopieren
2. **`extract_credentials`** ausführen — gibt Username, Password, Device-ID und Drucker-IP aus
3. Werte im Web-UI eintragen (⚙-Menü)
> **Download:** [gitea.it-drui.de/viewit/KX-Bridge-Release/releases](https://gitea.it-drui.de/viewit/KX-Bridge-Release/releases) → `extract_credentials.exe` (Windows) / `extract_credentials` (Linux) im jeweiligen Release-Asset
> **Download:** [gitea.it-drui.de/viewit/KX-Bridge-Release/releases](https://gitea.it-drui.de/viewit/KX-Bridge-Release/releases) → `fetch_credentials` / `extract_credentials` (Linux & Windows) im jeweiligen Release-Asset
### Schritt 3 Bridge starten
@@ -36,7 +46,7 @@ Das Skript baut das Docker-Image automatisch beim ersten Aufruf.
**Web-UI öffnen:** `http://BRIDGE-IP:7125`
→ Das ⚙-Menü öffnet sich beim ersten Start automatisch
→ Credentials aus Schritt 2 eintragen → **Speichern & Neustart**
Bei Option B: Credentials aus Schritt 2 eintragen → **Speichern & Neustart**
**OrcaSlicer verbinden:**
Drucker → Verbindungstyp **Moonraker** → Host: `http://BRIDGE-IP:7125`
@@ -46,6 +56,12 @@ Drucker → Verbindungstyp **Moonraker** → Host: `http://BRIDGE-IP:7125`
---
## 📺 Video Tutorial
[![KX-Bridge Setup & Bedienung](https://img.youtube.com/vi/1Ql4wfH27fM/hqdefault.jpg)](https://www.youtube.com/watch?v=1Ql4wfH27fM)
---
## ⚠️ Update von 0.9.1 oder älter
Ab **0.9.2** speichert KX-Bridge Einstellungen in `config/config.ini` statt in `.env`.
@@ -110,7 +126,8 @@ docker-compose down
## Fehlerbehebung
**„Falsche MQTT-Zugangsdaten"** beim Start:
- AnycubicSlicerNext neu starten, Drucker verbinden, `extract_credentials` erneut ausführen
- `fetch_credentials --ip <Drucker-IP> --write-config` erneut ausführen und Bridge neu starten
- Wenn IP unbekannt: AnycubicSlicerNext neu starten, Drucker verbinden, `extract_credentials` erneut ausführen
- Nur die IP-Adresse ins Feld eintragen, keinen Port (✗ `192.168.1.102:9883` → ✓ `192.168.1.102`)
**Drucker nicht gefunden / kein LAN-Modus:**
@@ -135,3 +152,9 @@ sudo usermod -aG docker $USER # dann neu einloggen
## Lizenz & Rechtliches
Interoperabilitätsforschung gem. §69e UrhG — ausschließlich private, nicht-kommerzielle Nutzung.
<p align="center">
<a href="https://ko-fi.com/viewitde">
<img src="https://ko-fi.com/img/githubbutton_sm.svg" alt="Ko-fi Support"/>
</a>
</p>

View File

@@ -2,7 +2,7 @@
# KX-Bridge Anycubic Kobra X
**Version:** 0.9.6.1
**Version:** 0.9.7
Control your **Anycubic Kobra X** with OrcaSlicer — no Klipper, no Raspberry Pi.
KX-Bridge is a Moonraker-compatible bridge that communicates directly with the printer.
@@ -46,6 +46,12 @@ Printer → Connection type **Moonraker** → Host: `http://BRIDGE-IP:7125`
---
## 📺 Video Tutorial
[![KX-Bridge Setup & Usage](https://img.youtube.com/vi/1Ql4wfH27fM/hqdefault.jpg)](https://www.youtube.com/watch?v=1Ql4wfH27fM)
---
## ⚠️ Upgrading from 0.9.1 or earlier
Starting with **0.9.2**, KX-Bridge stores settings in `config/config.ini` instead of `.env`.
@@ -135,3 +141,9 @@ sudo usermod -aG docker $USER # then log out and back in
## License
Interoperability research under §69e UrhG — private, non-commercial use only.
<p align="center">
<a href="https://ko-fi.com/viewitde">
<img src="https://ko-fi.com/img/githubbutton_sm.svg" alt="Ko-fi Support"/>
</a>
</p>

View File

@@ -1 +1 @@
0.9.6.1
0.9.7

397
fetch_credentials.py Normal file
View File

@@ -0,0 +1,397 @@
#!/usr/bin/env python3
"""
fetch_credentials.py Fetches and decrypts Anycubic Kobra X credentials via HTTP API.
Original approach by bebu (PR #19, KX-Bridge-Release).
Reverse engineered from the Vue project embedded in libWorkbench.so (Anycubic Slicer Next).
No running slicer required — only the printer IP in LAN.
Algorithm: AES-256-CBC
Key: token[16:32] from /info response
IV: response token from /ctrl response
Usage:
python3 fetch_credentials.py --ip 192.168.x.x
python3 fetch_credentials.py --ip 192.168.x.x --write-config
python3 fetch_credentials.py --ip 192.168.x.x --write-config --config-file ../config/config.ini
python3 fetch_credentials.py --ctrl ctrl.json --info info.json
"""
import json
import sys
import base64
import hashlib
import argparse
import os
import time
import random
import string
import requests
from pathlib import Path
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
def evp_bytes_to_key(password, salt, key_len, iv_len):
"""
Derive key and IV from password and salt using OpenSSL EVP_BytesToKey
This mimics the CryptoJS default key derivation
"""
m = []
i = 0
while len(b''.join(m)) < (key_len + iv_len):
md5 = hashlib.md5()
data = password + salt
if i > 0:
data = m[i - 1] + password + salt
md5.update(data)
m.append(md5.digest())
i += 1
ms = b''.join(m)
return ms[:key_len], ms[key_len:key_len + iv_len]
def generate_signature(token, ts, nonce):
"""
Generate MD5 signature for /ctrl endpoint
Signature = md5(md5(token[0:16]) + ts + nonce)
"""
# First MD5: token.slice(0, 16)
first_md5 = hashlib.md5(token[:16].encode('utf-8')).hexdigest()
# Second MD5: first_md5 + ts + nonce
signature_data = first_md5 + str(ts) + nonce
signature = hashlib.md5(signature_data.encode('utf-8')).hexdigest()
return signature
def generate_nonce(length=6):
"""Generate a random alphanumeric nonce"""
chars = string.ascii_letters + string.digits
return ''.join(random.choice(chars) for _ in range(length))
def fetch_from_http(ip, port, endpoint, token=None, did="random", verbose=False):
"""
Fetch data from HTTP endpoint on the printer
Args:
ip (str): IP address of the printer
port (int): Port number (default 18910)
endpoint (str): Either 'info' or 'ctrl'
token (str): Device token (required for /ctrl endpoint)
did (str): Device ID (required for /ctrl endpoint)
verbose (bool): Print debug information
Returns:
dict: JSON response data
"""
try:
if endpoint == 'info':
url = f"http://{ip}:{port}/info"
if verbose:
print(f"Fetching: {url}")
response = requests.get(url, timeout=10)
response.raise_for_status()
return response.json()
elif endpoint == 'ctrl':
if not token:
raise ValueError("Token is required for /ctrl endpoint")
# Generate signature parameters
ts = int(time.time() * 1000) # Current timestamp in ms
nonce = generate_nonce(6)
signature = generate_signature(token, ts, nonce)
url = f"http://{ip}:{port}/ctrl"
params = {
'ts': ts,
'nonce': nonce,
'sign': signature,
'did': did
}
if verbose:
print(f"Fetching: {url}")
print(f" Parameters:")
print(f" ts: {ts}")
print(f" nonce: {nonce}")
print(f" sign: {signature}")
print(f" did: {did}")
response = requests.post(url, params=params, timeout=10)
response.raise_for_status()
return response.json()
else:
raise ValueError(f"Unknown endpoint: {endpoint}")
except requests.exceptions.RequestException as e:
raise Exception(f"HTTP request failed for {endpoint}: {e}")
except json.JSONDecodeError as e:
raise Exception(f"Invalid JSON response from {endpoint}: {e}")
def decrypt_text(encrypted_data, key, iv):
"""
Decrypt data using AES-256-CBC
Handles CryptoJS-style encrypted data (OpenSSL format with salt)
Args:
encrypted_data (str): Encrypted data string (CryptoJS format)
key (str): Decryption key string
iv (str): Initialization vector string
Returns:
dict: Decrypted JSON data
"""
try:
# Convert key and IV to bytes
key_bytes = key.encode('utf-8')
iv_bytes = iv.encode('utf-8')
# Decrypt using direct key and IV (as per the original code)
cipher = AES.new(key_bytes, AES.MODE_CBC, iv_bytes)
# The encrypted_data might be base64 or hex encoded
# Try base64 first
try:
encrypted_bytes = base64.b64decode(encrypted_data)
except:
try:
# Try as hex
encrypted_bytes = bytes.fromhex(encrypted_data)
except:
# If all else fails, encode as UTF-8
encrypted_bytes = encrypted_data.encode('utf-8')
# Decrypt
decrypted = cipher.decrypt(encrypted_bytes)
# Try to unpad
try:
unpadded = unpad(decrypted, AES.block_size)
except ValueError:
# If unpadding fails, use as-is
unpadded = decrypted
plaintext = unpadded.decode('utf-8')
# Parse JSON
return json.loads(plaintext)
except Exception as e:
return {"error": str(e), "error_type": type(e).__name__}
def main():
"""Main function to decrypt printer data"""
# Parse command-line arguments
parser = argparse.ArgumentParser(
description='Fetch and decrypt Anycubic Kobra X credentials via HTTP API',
)
# HTTP mode
parser.add_argument('--ip', help='IP address of the printer')
parser.add_argument('--port', type=int, default=18910, help='Printer HTTP port (default: 18910)')
# File mode
parser.add_argument('--ctrl', default='ctrl.json', help='Path to ctrl.json (default: ctrl.json)')
parser.add_argument('--info', default='info.json', help='Path to info.json (default: info.json)')
# Output
parser.add_argument('--output', default=None, help='Save raw decrypted JSON to file (optional)')
parser.add_argument('--write-config', action='store_true',
help='Write credentials to config.ini')
parser.add_argument('--config-file', default=None,
help='Path to config.ini (default: ../config/config.ini relative to this script)')
parser.add_argument('--verbose', '-v', action='store_true', help='Verbose output')
args = parser.parse_args()
# Determine mode: HTTP or file
if args.ip:
# HTTP mode: fetch from printer
if args.verbose:
print("=" * 70)
print("Fetching configuration from printer via HTTP")
print("=" * 70)
print(f"Printer IP: {args.ip}:{args.port}")
print()
try:
# Fetch info.json
if args.verbose:
print("Step 1: Fetching device info...")
info = fetch_from_http(args.ip, args.port, 'info', verbose=args.verbose)
# Get token from info
token = info.get('token')
if not token:
print("Error: No token found in /info response", file=sys.stderr)
return 1
# Fetch data.json (encrypted config) from /ctrl endpoint
if args.verbose:
print("\nStep 2: Fetching encrypted configuration from /ctrl...")
data = fetch_from_http(args.ip, args.port, 'ctrl', token=token, verbose=args.verbose)
if args.verbose:
print("\nData fetched successfully!")
except Exception as e:
print(f"Error: {e}", file=sys.stderr)
return 1
else:
# File mode: load from disk
if args.verbose:
print("=" * 70)
print("Loading configuration from files")
print("=" * 70)
# Check if input files exist
if not Path(args.ctrl).exists():
print(f"Error: {args.ctrl} not found", file=sys.stderr)
return 1
if not Path(args.info).exists():
print(f"Error: {args.info} not found", file=sys.stderr)
return 1
# Read ctrl.json
try:
with open(args.ctrl, 'r') as f:
data = json.load(f)
except json.JSONDecodeError as e:
print(f"Error reading {args.ctrl}: {e}", file=sys.stderr)
return 1
except Exception as e:
print(f"Error reading {args.ctrl}: {e}", file=sys.stderr)
return 1
# Read info.json
try:
with open(args.info, 'r') as f:
info = json.load(f)
except json.JSONDecodeError as e:
print(f"Error reading {args.info}: {e}", file=sys.stderr)
return 1
except Exception as e:
print(f"Error reading {args.info}: {e}", file=sys.stderr)
return 1
# Extract values
try:
encrypted_info = data['data']['info']
response_token = data['data']['token']
full_token = info['token']
except KeyError as e:
print(f"Error: Missing required key {e}", file=sys.stderr)
return 1
# Generate decryption key and IV
key_part = full_token[16:32]
if args.verbose:
print("=" * 70)
print("Printer Configuration Decryption")
print("=" * 70)
print(f"Input data file: {args.ctrl}")
print(f"Input info file: {args.info}")
print(f"Output file: {args.output}")
print()
print("Decryption Parameters:")
print(f" Encrypted data length: {len(encrypted_info)} bytes")
print(f" Full token: {full_token}")
print(f" Full token length: {len(full_token)} characters")
print(f" Response token (IV): {response_token}")
print(f" Decryption key: {key_part}")
print(f" Key length: {len(key_part)} characters")
print(f" IV length: {len(response_token)} characters")
print()
# Decrypt
if args.verbose:
print("Decrypting...")
result = decrypt_text(encrypted_info, key_part, response_token)
if 'error' in result:
print(f"Error during decryption: {result.get('error')}", file=sys.stderr)
return 1
# Show result
print()
print("=" * 55)
print(" CREDENTIALS")
print("=" * 55)
print(f" {'Printer IP':12s} {result.get('ip', 'n/a')}")
print(f" {'Username':12s} {result.get('username', 'n/a')}")
print(f" {'Password':12s} {result.get('password', 'n/a')}")
print(f" {'Device-ID':12s} {result.get('deviceId', 'n/a')}")
print(f" {'Mode-ID':12s} {result.get('modeId', 'n/a')}")
print(f" {'Model':12s} {result.get('modelName', 'n/a')}")
print(f" {'Broker':12s} {result.get('broker', 'n/a')}")
print("=" * 55)
if args.verbose:
print()
print("Full decrypted config:")
# Strip certs/keys from verbose output to avoid cluttering terminal
display = {k: v for k, v in result.items() if k not in ('devicecrt', 'devicepk')}
print(json.dumps(display, indent=2))
# Optionally save raw JSON
if args.output:
try:
with open(args.output, 'w') as f:
json.dump(result, f, indent=2)
print(f"\nRaw config saved to: {args.output}")
except Exception as e:
print(f"Error writing to {args.output}: {e}", file=sys.stderr)
return 1
# Write config.ini
if args.write_config:
if args.config_file:
config_path = args.config_file
else:
config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)),
'..', 'config', 'config.ini')
config_path = os.path.normpath(config_path)
_write_config_ini(result, config_path)
else:
print(f"\nTip: pass --write-config to write credentials directly to config.ini")
return 0
def _write_config_ini(result: dict, config_path: str):
"""Write fetched credentials into config.ini, preserving existing non-credential keys."""
import configparser
cfg = configparser.ConfigParser()
if os.path.isfile(config_path):
cfg.read(config_path)
if not cfg.has_section('connection'):
cfg.add_section('connection')
cfg.set('connection', 'printer_ip', result.get('ip', ''))
cfg.set('connection', 'mqtt_port', '9883')
cfg.set('connection', 'username', result.get('username', ''))
cfg.set('connection', 'password', result.get('password', ''))
cfg.set('connection', 'device_id', result.get('deviceId', ''))
cfg.set('connection', 'mode_id', result.get('modeId', '20030'))
os.makedirs(os.path.dirname(config_path), exist_ok=True)
with open(config_path, 'w') as f:
cfg.write(f)
print(f"\n✓ Credentials written to '{config_path}'.")
if __name__ == '__main__':
sys.exit(main())

View File

@@ -523,7 +523,7 @@ class KobraXClient:
sock = socket.create_connection((self.host, 18910), timeout=30)
sock.sendall(headers + body)
sock.settimeout(10)
sock.settimeout(120) # große GCode-Dateien brauchen Zeit bis der Drucker antwortet
response = b""
try:
while True:

View File

@@ -178,6 +178,7 @@ class KobraXBridge:
client.callbacks["info/report"] = self._on_info
client.callbacks["file/report"] = self._on_file
client.callbacks["multiColorBox/report"] = self._on_multicolor_box
client.callbacks["light/report"] = self._on_light
# -------------------------------------------------------------------------
# MQTT callbacks (called from reader thread)
@@ -289,6 +290,12 @@ class KobraXBridge:
log.info(f"AMS-Slots empfangen: {len(slots)}, loaded_slot={self._ams_loaded_slot}")
self._push_status_update()
def _on_light(self, payload: dict):
d = payload.get("data") or {}
self._state["light_on"] = bool(d.get("status", 0))
self._state["light_brightness"] = int(d.get("brightness", 80))
self._push_status_update()
# OrcaSlicer filament preset IDs (MoonrakerPrinterAgent.cpp mapping)
_TRAY_INFO_IDX = {
"PLA": "OGFL99", "PLA-CF": "OGFL98", "PLA SILK": "OGFL96",
@@ -3077,7 +3084,7 @@ def _mqtt_error_msg(exc: Exception) -> str:
def build_app(bridge: KobraXBridge) -> web.Application:
app = web.Application()
app = web.Application(client_max_size=256 * 1024 * 1024) # 256 MB für große GCode-Dateien
r = app.router
# Moonraker API

View File

@@ -1,2 +1,4 @@
aiohttp>=3.9
imageio-ffmpeg>=0.4.9
requests>=2.30.0
pycryptodome>=3.20.0