p2l/KX-Bridge-Release
1
0
Fork 0
forked from viewit/KX-Bridge-Release
Code Pull Requests Activity

feat: KX-Bridge 0.9.0-beta1 – initiales Release-Repo

Browse Source
This commit is contained in:
thysson2701
2026-04-20 16:07:43 +02:00
commit 2c379b915f
14 changed files with 3411 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
.env.example Normal file
View File

@@ -0,0 +1,23 @@
# KX-Bridge – Verbindungsparameter
# Kopiere diese Datei nach .env und trage deine Werte ein:
# cp .env.example .env
# .env wird NICHT ins Repository committed.
#
# Credentials mit extract_credentials.exe (Windows) oder
# extract_credentials (Linux) aus dem laufenden AnycubicSlicerNext auslesen.
# IP-Adresse des Druckers im lokalen Netzwerk
PRINTER_IP=192.168.x.x
# MQTT-Port (Anycubic Kobra X Standard: 9883)
MQTT_PORT=9883
# MQTT-Zugangsdaten (druckerspezifisch, beginnt mit "user")
MQTT_USERNAME=userXXXXXXXXXX
MQTT_PASSWORD=XXXXXXXXXXXXXXX
# Geräte-ID (32-stelliger Hex-String, druckerspezifisch)
DEVICE_ID=xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
# Modell-ID (Kobra X Standard: 20030)
MODE_ID=20030

6
.gitignore vendored Normal file
View File

@@ -0,0 +1,6 @@
.env
__pycache__/
*.pyc
build/
dist/
*.spec

14
Dockerfile Normal file
View File

@@ -0,0 +1,14 @@
FROM python:3.11-slim
WORKDIR /app
COPY 05_scripts/requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY 05_scripts/kobrax_moonraker_bridge.py .
COPY 05_scripts/env_loader.py .
COPY 05_scripts/kobrax_client.py .
EXPOSE 7125
ENTRYPOINT ["python", "kobrax_moonraker_bridge.py"]

187
README.md Normal file
View File

@@ -0,0 +1,187 @@
# KX-Bridge
Verbindet den Anycubic Kobra X mit OrcaSlicer – ohne Klipper, ohne Raspberry Pi.
KX-Bridge läuft auf deinem PC oder NAS und stellt eine Schnittstelle bereit, über die OrcaSlicer den Drucker direkt steuern kann: Druckstart, Temperatur, Fortschritt, AMS-Farbwechsel.
---
## Enthaltene Dateien
| Datei | Beschreibung |
|-------|-------------|
| `kobrax_moonraker_bridge.py` | Bridge-Hauptprogramm |
| `extract_credentials.exe` | Zugangsdaten aus AnycubicSlicerNext auslesen (Windows) |
| `extract_credentials` | Zugangsdaten aus AnycubicSlicerNext auslesen (Linux) |
| `kobra_x_orcaslicer_preset.zip` | OrcaSlicer-Druckerprofil für den Kobra X |
| `bridge.sh` | Service-Manager für Linux |
| `Dockerfile` / `docker-compose.yml` | Docker-Deployment |
| `.env.example` | Konfigurationsvorlage |
---
## Voraussetzungen
- Anycubic Kobra X im LAN-Modus (Drucker muss über LAN erreichbar sein, nicht nur über Anycubic-Cloud)
- PC, NAS oder Server im gleichen Netzwerk (Windows oder Linux)
- Docker oder Python 3.9+
- MQTT-Zugangsdaten des Druckers → [Schritt 1](#schritt-1-zugangsdaten-ermitteln)
---
## Schnellstart
### Schritt 1: Zugangsdaten ermitteln
Die Bridge benötigt druckerspezifische MQTT-Zugangsdaten.
> Wichtig: Der Drucker muss sich im LAN-Modus befinden. Nur wenn der Drucker direkt über LAN (nicht ausschließlich über die Anycubic-Cloud) erreichbar ist, können die Zugangsdaten ermittelt und die Bridge genutzt werden.
Die Zugangsdaten werden mit `extract_credentials` aus dem laufenden AnycubicSlicerNext ausgelesen.
Vorbereitung: AnycubicSlicerNext starten und mit dem Drucker verbinden (bis der Drucker-Status angezeigt wird).
Windows:
```
extract_credentials.exe --write-env
```
Linux:
```bash
chmod +x extract_credentials
./extract_credentials --write-env
```
Die Zugangsdaten werden automatisch in `.env` gespeichert.
> Falls das Ergebnis unsicher wirkt: `--verbose` zeigt alle gefundenen Kandidaten. Den richtigen Wert manuell in `.env` eintragen.
---
### Schritt 2: Konfiguration prüfen
```bash
cp .env.example .env
# .env öffnen und Werte kontrollieren
```
---
### Schritt 3: Bridge starten
Option A – Docker (empfohlen):
```bash
docker compose up -d
```
Läuft im Hintergrund, startet automatisch nach Systemneustart.
Option B – Linux Binary:
```bash
chmod +x kx-bridge
./kx-bridge
# Oder mit Service-Manager:
./bridge.sh start
```
Option C – Python direkt:
```bash
pip install aiohttp
python kobrax_moonraker_bridge.py
```
---
### Schritt 4: OrcaSlicer-Profil installieren
1. `kobra_x_orcaslicer_preset.zip` in OrcaSlicer importieren:
Datei → Konfigurationen importieren → ZIP auswählen
2. Anycubic Kobra X als Drucker auswählen
---
### Schritt 5: OrcaSlicer verbinden
1. Drucker-Einstellungen öffnen
2. Verbindungstyp: Moonraker
3. Adresse: `http://IP-DES-BRIDGE-PC:7125` eintragen
4. Auf "Test" klicken – bei erfolgreicher Verbindung erscheint eine Bestätigungsmeldung
---
## bridge.sh – Service-Manager (Linux)
```bash
./bridge.sh start # Im Hintergrund starten
./bridge.sh stop # Beenden
./bridge.sh restart # Neustarten
./bridge.sh status # Status anzeigen
./bridge.sh log 50 # Letzte 50 Log-Zeilen
```
---
## Docker – Nützliche Befehle
```bash
docker compose up -d # Starten
docker compose down # Stoppen
docker compose logs -f # Logs verfolgen
docker compose pull && docker compose up -d # Update
```
---
## Fehlerbehebung
Port 7125 belegt:
```bash
./bridge.sh stop
./bridge.sh start
```
Verbindungstest in OrcaSlicer schlägt fehl:
- Firewall prüfen: Port 7125 muss erreichbar sein
- Bridge-Log prüfen: `./bridge.sh log` oder `docker compose logs`
- Drucker-IP in `.env` korrekt?
Zugangsdaten werden abgelehnt:
- AnycubicSlicerNext starten, mit Drucker verbinden
- `extract_credentials --verbose` ausführen und alle Kandidaten prüfen
- Richtigen Wert manuell in `.env` eintragen, Bridge neu starten
Docker: Permission denied:
```bash
sudo usermod -aG docker $USER
# Neu einloggen, dann erneut versuchen
```
---
## Konfigurationsreferenz (.env)
| Parameter | Beschreibung | Beispiel |
|-----------|-------------|---------|
| `PRINTER_IP` | IP-Adresse des Druckers | `192.168.1.100` |
| `MQTT_PORT` | MQTT-Port (nicht ändern) | `9883` |
| `MQTT_USERNAME` | Benutzername (beginnt mit „user") | `userXXXXXXXXXX` |
| `MQTT_PASSWORD` | Passwort (~15 Zeichen) | `***` |
| `DEVICE_ID` | Geräte-ID (32 Hex-Zeichen) | `xxxxxxxx...` |
| `MODE_ID` | Modell-ID (Kobra X Standard) | `20030` |
---
## Sicherheitshinweise
- Die Bridge bindet auf Port 7125 – nur im lokalen Netzwerk betreiben
- `.env` enthält Drucker-Zugangsdaten – nicht teilen oder ins Repo committen
- Alle Zugangsdaten werden ausschließlich lokal verarbeitet
---
## Hinweis zur Nutzung
Dieses Projekt dient der privaten Nutzung und der Herstellung von Interoperabilität zwischen dem Anycubic Kobra X und freier Software (OrcaSlicer).
`extract_credentials` liest ausschließlich den Arbeitsspeicher des auf deinem eigenen PC laufenden AnycubicSlicerNext-Prozesses. Es werden keine Daten übertragen oder gespeichert, außer in die lokale `.env`-Datei. Das Tool funktioniert nur für den Prozess des Druckers, dem du selbst gehörst.
Das Projekt steht in keiner Verbindung zu Anycubic und wird nicht kommerziell betrieben.

1
VERSION Normal file
View File

@@ -0,0 +1 @@
0.9.0-beta1

42
bridge.sh Executable file
View File

@@ -0,0 +1,42 @@
#!/bin/bash
# Bridge-Manager: start | stop | restart | status | log
SCRIPT="$(dirname "$0")/kobrax_moonraker_bridge.py"
LOGFILE="/tmp/bridge.log"
PRINTER_IP="192.168.178.94"
case "${1:-restart}" in
start)
if ss -tlnp | grep -q 7125; then
echo "Port 7125 belegt – beende alte Instanz..."
fuser -k 7125/tcp 2>/dev/null || pkill -f kobrax_moonraker_bridge 2>/dev/null
sleep 1
fi
nohup python3 "$SCRIPT" --printer-ip "$PRINTER_IP" > "$LOGFILE" 2>&1 &
echo "Bridge gestartet PID=$!"
sleep 2; tail -3 "$LOGFILE"
;;
stop)
pkill -f kobrax_moonraker_bridge && echo "Bridge beendet" || echo "Nicht aktiv"
fuser -k 7125/tcp 2>/dev/null
;;
restart)
"$0" stop
sleep 1
"$0" start
;;
status)
if ss -tlnp | grep -q 7125; then
echo "Bridge läuft (Port 7125 aktiv)"
ps aux | grep kobrax_moonraker | grep -v grep
else
echo "Bridge nicht aktiv"
fi
;;
log)
tail -${2:-20} "$LOGFILE"
;;
*)
echo "Usage: $0 {start|stop|restart|status|log [N]}"
exit 1
;;
esac

13
docker-compose.yml Normal file
View File

@@ -0,0 +1,13 @@
services:
kx-bridge:
image: kx-bridge:latest
build: .
env_file: .env
ports:
- "7125:7125"
restart: unless-stopped
logging:
driver: json-file
options:
max-size: "10m"
max-file: "3"

46
env_loader.py Normal file
View File

@@ -0,0 +1,46 @@
"""
env_loader.py – lädt Verbindungsparameter aus .env (Repo-Root oder Arbeitsverzeichnis).
Umgebungsvariablen haben Vorrang vor .env-Werten.
"""
import os
import pathlib
def _find_env_file() -> pathlib.Path | None:
# Suche .env im selben Verzeichnis, dann im Parent (Repo-Root)
for base in (pathlib.Path(__file__).parent, pathlib.Path(__file__).parent.parent):
p = base / ".env"
if p.is_file():
return p
return None
def _load_env_file(path: pathlib.Path):
with open(path, encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, _, val = line.partition("=")
key = key.strip()
val = val.strip()
if key and key not in os.environ:
os.environ[key] = val
_env_path = _find_env_file()
if _env_path:
_load_env_file(_env_path)
def get(key: str, default: str = "") -> str:
return os.environ.get(key, default)
# Häufig verwendete Shortcuts
PRINTER_IP = get("PRINTER_IP", "")
MQTT_PORT = int(get("MQTT_PORT", "9883"))
USERNAME = get("MQTT_USERNAME", "")
PASSWORD = get("MQTT_PASSWORD", "")
MODE_ID = get("MODE_ID", "")
DEVICE_ID = get("DEVICE_ID", "")

454
extract_credentials.py Normal file
View File

@@ -0,0 +1,454 @@
"""
extract_credentials.py – Extrahiert Anycubic LAN-MQTT-Credentials aus dem RAM
des laufenden AnycubicSlicerNext-Prozesses.
Voraussetzungen:
- AnycubicSlicerNext läuft und ist mit dem Drucker verbunden
- Gleiches Benutzerkonto wie der Slicer-Prozess (kein Admin nötig)
Verwendung:
python3 extract_credentials.py [--write-env] [--env-file ../.env]
Funktionsweise:
1. Prozess "AnycubicSlicer.exe" (Windows) bzw. "AnycubicSlicer" (Linux) finden
2. Speicherseiten des Prozesses durchsuchen (nur r/rw, keine Exec-Pages)
3. Nach MQTT-Credential-Patterns suchen:
Username: user[A-Za-z0-9]{8,12}
Password: [A-Za-z0-9+/]{13,18}
Drucker-IP: d{1,3}.d{1,3}.d{1,3}.d{1,3}
4. Kandidaten nach Plausibilität filtern und ausgeben
5. Optional: .env-Datei schreiben
"""
import argparse
import os
import re
import struct
import sys
import platform
# ---------------------------------------------------------------------------
# Plattform-Erkennung
# ---------------------------------------------------------------------------
IS_WINDOWS = platform.system() == "Windows"
IS_LINUX = platform.system() == "Linux"
# ---------------------------------------------------------------------------
# Pattern
# ---------------------------------------------------------------------------
# Username: "user" + 8–12 alphanumerische Zeichen (drucker-generiert)
RE_USERNAME = re.compile(rb'user[A-Za-z0-9]{8,12}(?=[^A-Za-z0-9]|$)')
# Password: 13–20 alphanumerische Zeichen (kein / da kein RTSP-Pfad)
# Anycubic-Passwörter: gemischte Groß/Klein/Ziffern, kein Slash
RE_PASSWORD = re.compile(rb'[A-Za-z0-9]{13,20}(?=[^A-Za-z0-9]|$)')
# Kontext-Pattern: sucht Passwort das direkt nach "password" im Speicher steht
RE_PASSWORD_CTX = re.compile(rb'(?:password|passwd|Password)\x00{0,4}([A-Za-z0-9]{10,25})(?=[^A-Za-z0-9]|$)', re.IGNORECASE)
# Proximity-Pattern: Username gefolgt von Passwort in naher Umgebung (<512 Bytes)
RE_USER_PASS_PROXIMITY = re.compile(
rb'(user[A-Za-z0-9]{8,12}).{1,512}?([A-Za-z0-9]{13,20})(?=[^A-Za-z0-9]|$)',
re.DOTALL
)
# IPv4-Adresse (kein localhost, kein Broadcast)
RE_IP = re.compile(rb'(?<![.\d])(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})(?![.\d])')
# mode_id: 5-stellige Zahl (z.B. 20030)
RE_MODE_ID = re.compile(rb'(?<!\d)(2\d{4})(?!\d)')
# device_id: 32 Hex-Zeichen (MD5-Format)
RE_DEVICE_ID = re.compile(rb'[0-9a-f]{32}(?=[^0-9a-f]|$)')
# ---------------------------------------------------------------------------
# Windows – Speicher lesen via ctypes / ReadProcessMemory
# ---------------------------------------------------------------------------
def _win_find_pid(name: str) -> "int | None":
"""Findet die PID eines Prozesses anhand des Namens (case-insensitive)."""
import ctypes
import ctypes.wintypes
TH32CS_SNAPPROCESS = 0x00000002
class PROCESSENTRY32(ctypes.Structure):
_fields_ = [
("dwSize", ctypes.wintypes.DWORD),
("cntUsage", ctypes.wintypes.DWORD),
("th32ProcessID", ctypes.wintypes.DWORD),
("th32DefaultHeapID", ctypes.POINTER(ctypes.c_ulong)),
("th32ModuleID", ctypes.wintypes.DWORD),
("cntThreads", ctypes.wintypes.DWORD),
("th32ParentProcessID", ctypes.wintypes.DWORD),
("pcPriClassBase", ctypes.c_long),
("dwFlags", ctypes.wintypes.DWORD),
("szExeFile", ctypes.c_char * 260),
]
snap = ctypes.windll.kernel32.CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0)
if snap == ctypes.wintypes.HANDLE(-1).value:
return None
entry = PROCESSENTRY32()
entry.dwSize = ctypes.sizeof(PROCESSENTRY32)
try:
if not ctypes.windll.kernel32.Process32First(snap, ctypes.byref(entry)):
return None
while True:
if entry.szExeFile.decode("utf-8", errors="replace").lower() == name.lower():
return entry.th32ProcessID
if not ctypes.windll.kernel32.Process32Next(snap, ctypes.byref(entry)):
break
finally:
ctypes.windll.kernel32.CloseHandle(snap)
return None
def _win_read_memory(pid: int, chunk_size: int = 0x10000) -> "list[bytes]":
"""Liest alle lesbaren Speicherseiten eines Windows-Prozesses."""
import ctypes
import ctypes.wintypes
PROCESS_VM_READ = 0x0010
PROCESS_QUERY_INFORMATION = 0x0400
MEM_COMMIT = 0x1000
PAGE_NOACCESS = 0x01
PAGE_GUARD = 0x100
class MEMORY_BASIC_INFORMATION(ctypes.Structure):
_fields_ = [
("BaseAddress", ctypes.c_void_p),
("AllocationBase", ctypes.c_void_p),
("AllocationProtect", ctypes.wintypes.DWORD),
("RegionSize", ctypes.c_size_t),
("State", ctypes.wintypes.DWORD),
("Protect", ctypes.wintypes.DWORD),
("Type", ctypes.wintypes.DWORD),
]
k32 = ctypes.windll.kernel32
handle = k32.OpenProcess(PROCESS_VM_READ | PROCESS_QUERY_INFORMATION, False, pid)
if not handle:
raise PermissionError(f"OpenProcess fehlgeschlagen (PID {pid}): {ctypes.GetLastError()}")
chunks = []
addr = 0
mbi = MEMORY_BASIC_INFORMATION()
try:
while k32.VirtualQueryEx(handle, ctypes.c_void_p(addr),
ctypes.byref(mbi), ctypes.sizeof(mbi)):
skip = (
mbi.State != MEM_COMMIT or
mbi.Protect & PAGE_NOACCESS or
mbi.Protect & PAGE_GUARD or
mbi.RegionSize > 256 * 1024 * 1024 # >256 MB überspringen
)
if not skip:
buf = ctypes.create_string_buffer(mbi.RegionSize)
read = ctypes.c_size_t(0)
if k32.ReadProcessMemory(handle, ctypes.c_void_p(addr),
buf, mbi.RegionSize, ctypes.byref(read)):
chunks.append(bytes(buf[:read.value]))
addr += mbi.RegionSize
if addr >= 0x7FFFFFFFFFFF: # Ende des User-Space (64-bit)
break
finally:
k32.CloseHandle(handle)
return chunks
# ---------------------------------------------------------------------------
# Linux – Speicher lesen via /proc/{pid}/mem
# ---------------------------------------------------------------------------
def _linux_find_pid(name: str) -> "int | None":
"""Findet PID anhand des Prozessnamens in /proc."""
for entry in os.listdir("/proc"):
if not entry.isdigit():
continue
try:
cmdline = open(f"/proc/{entry}/cmdline", "rb").read()
if name.lower().encode() in cmdline.lower():
return int(entry)
except (PermissionError, FileNotFoundError):
continue
return None
def _linux_read_memory(pid: int) -> "list[bytes]":
"""Liest lesbare Speichersegmente aus /proc/{pid}/mem."""
chunks = []
maps_path = f"/proc/{pid}/maps"
mem_path = f"/proc/{pid}/mem"
try:
maps = open(maps_path).readlines()
mem = open(mem_path, "rb")
except PermissionError:
raise PermissionError(
f"Kein Zugriff auf /proc/{pid}/mem — "
"Script als gleicher Benutzer wie der Slicer starten."
)
for line in maps:
parts = line.split()
if len(parts) < 2:
continue
perms = parts[1]
if "r" not in perms: # nur lesbare Seiten
continue
if "x" in perms: # Code-Seiten überspringen (keine Strings)
continue
try:
start, end = [int(x, 16) for x in parts[0].split("-")]
except ValueError:
continue
size = end - start
if size > 256 * 1024 * 1024: # >256 MB überspringen
continue
try:
mem.seek(start)
data = mem.read(size)
if data:
chunks.append(data)
except (OSError, ValueError):
continue
mem.close()
return chunks
# ---------------------------------------------------------------------------
# Pattern-Suche
# ---------------------------------------------------------------------------
def _is_valid_ip(ip_bytes: bytes) -> bool:
parts = ip_bytes.split(b".")
if len(parts) != 4:
return False
try:
nums = [int(p) for p in parts]
except ValueError:
return False
if nums[0] in (0, 127, 255):
return False
if nums == [0, 0, 0, 0]:
return False
return all(0 <= n <= 255 for n in nums)
def search_chunks(chunks: "list[bytes]") -> dict:
"""Durchsucht Speicher-Chunks nach Credential-Patterns."""
usernames = {} # value → count
passwords = {}
ips = {}
device_ids = {}
total = len(chunks)
for i, chunk in enumerate(chunks):
if i % 50 == 0 or i == total - 1:
pct = (i + 1) * 100 // total
mb_done = sum(len(c) for c in chunks[:i+1]) / 1024 / 1024
print(f"\r[*] Analysiere ... {pct:3d}% ({mb_done:.0f} MB)", end="", flush=True)
for m in RE_USERNAME.finditer(chunk):
v = m.group().decode("ascii", errors="replace")
usernames[v] = usernames.get(v, 0) + 1
# Proximity: Passwort das innerhalb von 512 Bytes nach einem Username steht
for m in RE_USER_PASS_PROXIMITY.finditer(chunk):
pw = m.group(2).decode("ascii", errors="replace")
has_upper = any(c.isupper() for c in pw)
has_lower = any(c.islower() for c in pw)
has_digit = any(c.isdigit() for c in pw)
if has_upper and has_lower and has_digit:
passwords[pw] = passwords.get(pw, 0) + 500
for m in RE_PASSWORD.finditer(chunk):
v = m.group().decode("ascii", errors="replace")
# Filter: mindestens 2 Großbuchstaben + 2 Kleinbuchstaben + 1 Ziffer
has_upper = sum(1 for c in v if c.isupper()) >= 2
has_lower = sum(1 for c in v if c.islower()) >= 2
has_digit = sum(1 for c in v if c.isdigit()) >= 1
if has_upper and has_lower and has_digit:
passwords[v] = passwords.get(v, 0) + 1
for m in RE_IP.finditer(chunk):
ip = m.group(1)
if _is_valid_ip(ip):
v = ip.decode("ascii")
ips[v] = ips.get(v, 0) + 1
for m in RE_DEVICE_ID.finditer(chunk):
v = m.group().decode("ascii")
device_ids[v] = device_ids.get(v, 0) + 1
print() # Zeilenumbruch nach Fortschrittszeile
# Nach Häufigkeit sortieren, häufigste zuerst
def top(d, n=10):
return sorted(d.items(), key=lambda x: -x[1])[:n]
return {
"usernames": top(usernames),
"passwords": top(passwords),
"ips": top(ips, 10),
"device_ids": top(device_ids),
}
# ---------------------------------------------------------------------------
# Hauptprogramm
# ---------------------------------------------------------------------------
SLICER_NAMES = [
"AnycubicSlicerNext.exe", # Windows
"AnycubicSlicer.exe",
"AnycubicSlicerNext", # Linux
"AnycubicSlicer",
]
def find_slicer_pid() -> "tuple[int, str] | tuple[None, None]":
for name in SLICER_NAMES:
if IS_WINDOWS:
pid = _win_find_pid(name)
else:
pid = _linux_find_pid(name)
if pid:
return pid, name
return None, None
def read_process(pid: int) -> "list[bytes]":
if IS_WINDOWS:
return _win_read_memory(pid)
else:
return _linux_read_memory(pid)
def write_env(results: dict, env_path: str,
username: str, password: str, ip: str,
mode_id: str = "20030", device_id: str = ""):
lines = []
if os.path.isfile(env_path):
with open(env_path) as f:
lines = f.readlines()
def set_val(key, val):
nonlocal lines
for i, line in enumerate(lines):
if line.startswith(f"{key}="):
lines[i] = f"{key}={val}\n"
return
lines.append(f"{key}={val}\n")
set_val("PRINTER_IP", ip)
set_val("MQTT_USERNAME", username)
set_val("MQTT_PASSWORD", password)
if device_id:
set_val("DEVICE_ID", device_id)
if mode_id:
set_val("MODE_ID", mode_id)
with open(env_path, "w") as f:
f.writelines(lines)
print(f"\n✓ Credentials in '{env_path}' gespeichert.")
def main():
parser = argparse.ArgumentParser(
description="Extrahiert MQTT-Credentials aus dem RAM des AnycubicSlicer-Prozesses"
)
parser.add_argument("--write-env", action="store_true",
help="Gefundene Credentials in .env schreiben")
parser.add_argument("--env-file", default=None,
help="Pfad zur .env-Datei (Standard: ../. env relativ zu diesem Script)")
parser.add_argument("--pid", type=int, default=None,
help="Prozess-PID direkt angeben (überspringt Auto-Erkennung)")
parser.add_argument("--verbose", action="store_true",
help="Alle Kandidaten ausgeben, nicht nur den besten")
args = parser.parse_args()
# .env-Pfad bestimmen
if args.env_file:
env_path = args.env_file
else:
env_path = os.path.join(os.path.dirname(__file__), "..", ".env")
env_path = os.path.normpath(env_path)
# Prozess finden
if args.pid:
pid, proc_name = args.pid, f"PID {args.pid}"
else:
print("[*] Suche AnycubicSlicer-Prozess ...")
pid, proc_name = find_slicer_pid()
if not pid:
print("✗ AnycubicSlicer nicht gefunden. Bitte den Slicer starten und "
"mit dem Drucker verbinden, dann erneut ausführen.")
sys.exit(1)
print(f"[*] Prozess gefunden: {proc_name} (PID {pid})")
print(f"[*] Lese Prozess-Speicher ...")
try:
chunks = read_process(pid)
except PermissionError as e:
print(f"✗ Zugriffsfehler: {e}")
sys.exit(1)
total_mb = sum(len(c) for c in chunks) / 1024 / 1024
print(f"[*] {len(chunks)} Speichersegmente gelesen ({total_mb:.1f} MB)")
print(f"[*] Durchsuche nach Credentials ...")
results = search_chunks(chunks)
# Ausgabe
print("\n" + "="*55)
print(" ERGEBNISSE")
print("="*55)
def show(label, items, verbose):
if not items:
print(f" {label:12s} — nicht gefunden")
return items[0][0] if items else ""
best = items[0][0]
print(f" {label:12s} {best} (Treffer: {items[0][1]})")
if verbose and len(items) > 1:
for val, cnt in items[1:]:
print(f" {'':12s} {val} (Treffer: {cnt})")
return best
best_user = show("Username", results["usernames"], args.verbose)
best_pass = show("Password", results["passwords"], args.verbose)
best_device = show("Device-ID", results["device_ids"], args.verbose)
# IP: 192.168.x.x bevorzugen
lan_ips = [(ip, cnt) for ip, cnt in results["ips"]
if ip.startswith("192.168.") or ip.startswith("10.") or ip.startswith("172.")]
if not lan_ips:
lan_ips = results["ips"]
best_ip = show("Drucker-IP", lan_ips, args.verbose)
print("="*55)
if not best_user or not best_pass:
print("\n⚠ Keine vollständigen Credentials gefunden.")
print(" Stelle sicher dass der Slicer MIT dem Drucker verbunden ist.")
sys.exit(1)
if args.write_env:
write_env(results, env_path, best_user, best_pass, best_ip,
device_id=best_device)
else:
print(f"\nHinweis: --write-env übergeben um Credentials in '{env_path}' zu speichern.")
if __name__ == "__main__":
main()

BIN
kobra_x_orcaslicer_preset.zip Normal file
View File

Binary file not shown.

554
kobrax_client.py Normal file
View File

@@ -0,0 +1,554 @@
"""
kobrax_client.py – Anycubic Kobra X LAN-MQTT-Client
Protokoll vollständig rekonstruiert via Sniffer 2026-04-17 (953 Nachrichten).
Voraussetzungen:
- /tmp/anycubic_slicer.crt und .key (aus cloud_mqtt.dll @ 0x2ed5b0 / 0x2edce0)
- Drucker im LAN-Modus erreichbar auf Port 9883
Verwendung:
client = KobraXClient(env_loader.PRINTER_IP, mode_id=env_loader.MODE_ID,
device_id=env_loader.DEVICE_ID)
client.connect()
info = client.query_info()
print(info["data"]["temp"])
client.disconnect()
"""
import json
import os
import socket
import ssl
import threading
import time
import uuid
from datetime import datetime
import env_loader
_SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
CERT_FILE = os.path.join(_SCRIPT_DIR, "anycubic_slicer.crt")
KEY_FILE = os.path.join(_SCRIPT_DIR, "anycubic_slicer.key")
# ---------------------------------------------------------------------------
# Low-level MQTT framing
# ---------------------------------------------------------------------------
def _enc_str(s: str) -> bytes:
b = s.encode("utf-8")
return len(b).to_bytes(2, "big") + b
def _enc_len(n: int) -> bytes:
out = bytearray()
while True:
d = n % 128
n //= 128
if n > 0:
d |= 0x80
out.append(d)
if n == 0:
break
return bytes(out)
def _build_connect(client_id: str, username: str, password: str) -> bytes:
proto = b"\x00\x04MQTT\x04"
ka = b"\x00\x3c" # keepalive = 60s
flags = 0xC2 # username + password, clean session
payload = _enc_str(client_id) + _enc_str(username) + _enc_str(password)
body = proto + bytes([flags]) + ka + payload
return bytes([0x10]) + _enc_len(len(body)) + body
def _build_subscribe(topic: str, pid: int) -> bytes:
p = pid.to_bytes(2, "big") + _enc_str(topic) + b"\x00"
return bytes([0x82]) + _enc_len(len(p)) + p
def _build_publish(topic: str, payload: str) -> bytes:
body = _enc_str(topic) + payload.encode("utf-8")
return bytes([0x30]) + _enc_len(len(body)) + body
def _build_pingreq() -> bytes:
return bytes([0xC0, 0x00])
def _parse_publish(pkt: bytes):
if len(pkt) < 2:
return None, None
tlen = (pkt[0] << 8) | pkt[1]
if 2 + tlen > len(pkt):
return None, None
topic = pkt[2:2 + tlen].decode("utf-8", errors="replace")
payload = pkt[2 + tlen:]
return topic, payload
# ---------------------------------------------------------------------------
# KobraXClient
# ---------------------------------------------------------------------------
class KobraXClient:
def __init__(self, host: str, username: str, password: str,
mode_id: str, device_id: str,
port: int = 9883, client_id: str = "kobrax_py"):
self.host = host
self.port = port
self.username = username
self.password = password
self.mode_id = mode_id
self.device_id = device_id
self.client_id = client_id
self._sock = None
self._buf = b""
self._pid = 1
self._lock = threading.Lock()
self._running = False
# Pending requests by msgid (for response ACK)
self._pending_msgid: dict[str, dict] = {}
# Pending requests by msg_type/report topic suffix
self._pending_report: dict[str, dict] = {}
# Optional callbacks: topic_suffix → callable(payload_dict)
self.callbacks: dict[str, callable] = {}
# -- Topics --------------------------------------------------------------
def _pub_topic(self, msg_type: str) -> str:
return (f"anycubic/anycubicCloud/v1/slicer/printer/"
f"{self.mode_id}/{self.device_id}/{msg_type}")
def _sub_topic(self) -> str:
return (f"anycubic/anycubicCloud/v1/printer/public/"
f"{self.mode_id}/{self.device_id}/#")
# -- Connection ----------------------------------------------------------
def _do_connect(self):
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
ctx.set_ciphers("DEFAULT:@SECLEVEL=0")
ctx.load_cert_chain(CERT_FILE, KEY_FILE)
raw = socket.create_connection((self.host, self.port), timeout=5)
self._sock = ctx.wrap_socket(raw)
print(f"[kobrax] TLS: {self._sock.cipher()[0]}")
self._sock.sendall(_build_connect(self.client_id, self.username, self.password))
self._sock.settimeout(3)
r = self._sock.recv(64)
if len(r) < 4 or r[0] != 0x20 or r[3] != 0:
raise RuntimeError(f"CONNACK failed: {r.hex()}")
print(f"[kobrax] CONNACK rc=0")
self._sock.settimeout(0.2)
self._buf = b""
self._subscribe(self._sub_topic())
def connect(self):
self._do_connect()
self._running = True
t = threading.Thread(target=self._read_loop, daemon=True)
t.start()
time.sleep(0.3)
def disconnect(self):
self._running = False
try:
self._sock.close()
except Exception:
pass
def _reconnect(self):
print("[kobrax] Verbindung verloren – reconnect…")
try:
self._sock.close()
except Exception:
pass
for delay in [2, 4, 8, 15, 30]:
try:
self._do_connect()
print("[kobrax] Reconnect erfolgreich")
return True
except Exception as e:
print(f"[kobrax] Reconnect fehlgeschlagen ({e}), warte {delay}s…")
time.sleep(delay)
return False
def _subscribe(self, topic: str):
with self._lock:
pid = self._pid
self._pid += 1
self._sock.sendall(_build_subscribe(topic, pid))
print(f"[kobrax] SUB {topic}")
# -- Read loop -----------------------------------------------------------
def _read_loop(self):
last_ping = time.time()
while self._running:
if time.time() - last_ping > 30:
with self._lock:
try:
self._sock.sendall(_build_pingreq())
except Exception:
if self._running and not self._reconnect():
break
last_ping = time.time()
continue
last_ping = time.time()
try:
data = self._sock.recv(65536)
if not data:
raise ConnectionResetError("EOF")
self._buf += data
self._drain()
except ssl.SSLWantReadError:
continue
except socket.timeout:
continue
except Exception as e:
if self._running:
print(f"[kobrax] reader error: {e}")
if not self._reconnect():
break
last_ping = time.time()
else:
break
def _drain(self):
buf = self._buf
idx = 0
while idx < len(buf):
ptype = buf[idx] & 0xF0
i = idx + 1
mul = 1
rem = 0
while i < len(buf):
b = buf[i]
rem += (b & 0x7F) * mul
mul *= 128
i += 1
if not (b & 0x80):
break
if i + rem > len(buf):
break
pkt = buf[i:i + rem]
idx = i + rem
if ptype == 0x30:
topic, raw_payload = _parse_publish(pkt)
if topic is None:
continue
try:
payload = json.loads(raw_payload)
except Exception:
payload = {"_raw": raw_payload.decode("utf-8", errors="replace")}
self._dispatch(topic, payload)
self._buf = buf[idx:]
def _dispatch(self, topic: str, payload: dict):
# Resolve by report topic suffix (e.g. "info/report")
suffix = "/".join(topic.split("/")[-2:])
if suffix in self._pending_report:
entry = self._pending_report[suffix]
entry["result"] = payload
entry["event"].set()
# Resolve by msgid (for generic response ACK)
msgid = payload.get("msgid")
if msgid and msgid in self._pending_msgid:
entry = self._pending_msgid[msgid]
entry["result"] = payload
entry["event"].set()
# User callbacks by topic suffix (last two path components)
suffix = "/".join(topic.split("/")[-2:])
if suffix in self.callbacks:
try:
self.callbacks[suffix](payload)
except Exception as e:
print(f"[kobrax] callback error for {suffix}: {e}")
# Generic wildcard callback
if "*" in self.callbacks:
try:
self.callbacks["*"](topic, payload)
except Exception as e:
print(f"[kobrax] wildcard callback error: {e}")
# -- Publish + request/response ------------------------------------------
def publish(self, msg_type: str, action: str, data=None, timeout: float = 5.0) -> dict | None:
msgid = str(uuid.uuid4())
payload = json.dumps({
"type": msg_type,
"action": action,
"msgid": msgid,
"timestamp": int(time.time() * 1000),
"data": data,
}, separators=(",", ":"))
# Wait by msgid only — avoids collisions when multiple threads
# call publish() for the same msg_type concurrently.
# Also register by report topic as fallback for responses without msgid.
report_key = f"{msg_type}/report"
event = threading.Event()
entry = {"event": event, "result": None}
self._pending_msgid[msgid] = entry
# Only register report-key waiter if nobody else is waiting on it
report_registered = False
if report_key not in self._pending_report:
self._pending_report[report_key] = entry
report_registered = True
topic = self._pub_topic(msg_type)
try:
with self._lock:
self._sock.sendall(_build_publish(topic, payload))
except Exception as e:
print(f"[kobrax] send error: {e}, reconnecting…")
self._pending_msgid.pop(msgid, None)
if report_registered:
self._pending_report.pop(report_key, None)
if not self._reconnect():
return None
# retry once after reconnect
try:
with self._lock:
self._sock.sendall(_build_publish(topic, payload))
self._pending_msgid[msgid] = entry
if report_registered:
self._pending_report[report_key] = entry
except Exception:
return None
if timeout <= 0:
self._pending_msgid.pop(msgid, None)
if report_registered:
self._pending_report.pop(report_key, None)
return None
received = event.wait(timeout)
self._pending_msgid.pop(msgid, None)
if report_registered:
self._pending_report.pop(report_key, None)
if not received:
return None
return entry["result"]
# -- High-level commands -------------------------------------------------
def query_info(self) -> dict | None:
return self.publish("info", "query")
def query_status(self) -> dict | None:
return self.publish("status", "query")
def query_multicolor_box(self) -> dict | None:
return self.publish("multiColorBox", "getInfo")
def set_temperature(self, nozzle: int, bed: int) -> dict | None:
return self.publish("tempature", "set",
{"target_nozzle_temp": nozzle, "target_hotbed_temp": bed})
def set_fan(self, pct: int) -> dict | None:
return self.publish("fan", "set", {"fan_speed_pct": pct})
def set_light(self, on: bool, brightness: int = 80) -> dict | None:
return self.publish("light", "control",
{"type": 2, "status": 1 if on else 0, "brightness": brightness})
def start_camera(self) -> dict | None:
return self.publish("video", "startCapture")
def stop_camera(self) -> dict | None:
return self.publish("video", "stopCapture")
def pause_print(self) -> dict | None:
return self.publish("print", "pause")
def resume_print(self) -> dict | None:
return self.publish("print", "resume")
def stop_print(self) -> dict | None:
return self.publish("print", "stop")
# -- G-Code Upload -------------------------------------------------------
def upload_gcode(self, filepath: str, remote_filename: str | None = None,
upload_url: str | None = None) -> dict:
"""Upload a G-Code or .3mf file via HTTP POST to port 18910.
Returns the parsed JSON response from the printer.
Raises RuntimeError on HTTP or connection errors.
Protocol captured via Wireshark 2026-04-18:
POST /gcode_upload?s={session_token}
Multipart fields: 'filename' (text) + 'gcode' (file bytes)
Required headers: X-File-Length, X-BBL-* (BambuLab heritage)
"""
if not upload_url:
info = self.query_info()
if not info:
raise RuntimeError("Could not get info/report for upload URL")
upload_url = info["data"]["urls"]["fileUploadurl"]
# parse token from URL query string
token = upload_url.split("?s=")[1] if "?s=" in upload_url else ""
with open(filepath, "rb") as f:
file_data = f.read()
if remote_filename is None:
remote_filename = os.path.basename(filepath)
boundary = "------------------------a3a050b927d92a4c"
sep = f"--{boundary}\r\n".encode()
end = f"--{boundary}--\r\n".encode()
part_filename = (
sep +
f'Content-Disposition: form-data; name="filename"\r\n\r\n'.encode() +
remote_filename.encode() + b"\r\n"
)
part_gcode = (
sep +
f'Content-Disposition: form-data; name="gcode"; filename="{remote_filename}"\r\n'
f'Content-Type: application/octet-stream\r\n\r\n'.encode() +
file_data + b"\r\n"
)
body = part_filename + part_gcode + end
headers = (
f"POST /gcode_upload?s={token} HTTP/1.1\r\n"
f"Host: {self.host}:18910\r\n"
f"User-Agent: AnycubicSlicerNext/1.3.9.4\r\n"
f"Accept: */*\r\n"
f"X-BBL-Client-Name: AnycubicSlicerNext\r\n"
f"X-BBL-Client-Type: slicer\r\n"
f"X-BBL-Client-Version: 01.03.09.04\r\n"
f"X-BBL-Device-ID: {str(uuid.uuid4())}\r\n"
f"X-BBL-Language: de-DE\r\n"
f"X-BBL-OS-Type: windows\r\n"
f"X-BBL-OS-Version: 10.0.26200\r\n"
f"X-File-Length: {len(file_data)}\r\n"
f"Content-Type: multipart/form-data; boundary={boundary}\r\n"
f"Content-Length: {len(body)}\r\n"
f"Connection: close\r\n\r\n"
).encode()
sock = socket.create_connection((self.host, 18910), timeout=30)
sock.sendall(headers + body)
sock.settimeout(10)
response = b""
try:
while True:
chunk = sock.recv(65536)
if not chunk:
break
response += chunk
except socket.timeout:
pass
sock.close()
# parse HTTP response body
if b"\r\n\r\n" in response:
body_start = response.index(b"\r\n\r\n") + 4
resp_body = response[body_start:]
else:
resp_body = response
try:
return json.loads(resp_body)
except Exception:
raise RuntimeError(f"Upload: unerwartete Antwort: {resp_body[:200]}")
def move_axis(self, axis: int, move_type: int = 2, distance: int = 0) -> dict | None:
return self.publish("axis", "move",
{"axis": axis, "move_type": move_type, "distance": distance})
def home_all(self) -> dict | None:
# axis=4 move_type=2 = Home all axes (~4-15s)
return self.publish("axis", "move", {"axis": 4, "move_type": 2, "distance": 0}, timeout=30.0)
def home_axis(self, axis: int) -> dict | None:
# axis: 1=Y, 2=X, 3=Z
return self.publish("axis", "move", {"axis": axis, "move_type": 2, "distance": 0}, timeout=30.0)
def jog(self, axis: int, direction: int, distance_mm: int = 1) -> dict | None:
# axis: 1=Y, 2=X, 3=Z direction: 0=neg, 1=pos
return self.move_axis(axis=axis, move_type=direction, distance=distance_mm)
# ---------------------------------------------------------------------------
# CLI Demo
# ---------------------------------------------------------------------------
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="Anycubic Kobra X LAN-Client")
parser.add_argument("--ip", default=env_loader.PRINTER_IP)
parser.add_argument("--port", type=int, default=env_loader.MQTT_PORT)
parser.add_argument("--username", default=env_loader.USERNAME)
parser.add_argument("--password", default=env_loader.PASSWORD)
parser.add_argument("--mode-id", default=env_loader.MODE_ID)
parser.add_argument("--device-id", default=env_loader.DEVICE_ID)
parser.add_argument("--monitor", action="store_true",
help="Dauerhaft mithören und alle Reports ausgeben")
args = parser.parse_args()
client = KobraXClient(
host=args.ip, port=args.port,
username=args.username, password=args.password,
mode_id=args.mode_id, device_id=args.device_id,
)
if args.monitor:
def on_msg(topic, payload):
suffix = "/".join(topic.split("/")[-2:])
ts = datetime.now().strftime("%H:%M:%S")
state = payload.get("state", "")
data = payload.get("data") or {}
if "progress" in data:
print(f"[{ts}] {suffix:25} state={state:12} progress={data['progress']}% layer={data.get('curr_layer','?')}/{data.get('total_layers','?')}")
elif "curr_nozzle_temp" in data:
print(f"[{ts}] {suffix:25} nozzle={data['curr_nozzle_temp']}°C/{data.get('target_nozzle_temp',0)}°C bed={data['curr_hotbed_temp']}°C/{data.get('target_hotbed_temp',0)}°C")
else:
print(f"[{ts}] {suffix:25} state={state}")
client.callbacks["*"] = on_msg
client.connect()
print("[kobrax] Monitor-Modus aktiv (Ctrl-C zum Beenden)")
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
pass
client.disconnect()
else:
client.connect()
print("\n--- query_info ---")
info = client.query_info()
if info:
d = info.get("data", {})
print(f" Drucker: {d.get('printerName')} FW {d.get('version')}")
print(f" Status: {d.get('state')}")
t = d.get("temp", {})
print(f" Nozzle: {t.get('curr_nozzle_temp')}°C → {t.get('target_nozzle_temp')}°C")
print(f" Bett: {t.get('curr_hotbed_temp')}°C → {t.get('target_hotbed_temp')}°C")
urls = d.get("urls", {})
print(f" Upload: {urls.get('fileUploadurl')}")
print(f" Kamera: {urls.get('rtspUrl')}")
else:
print(" Keine Antwort")
client.disconnect()

2067
kobrax_moonraker_bridge.py Normal file
View File

File diff suppressed because it is too large Load Diff

3
releases/0.9.0-beta1/SHA256SUMS.txt Normal file
View File

@@ -0,0 +1,3 @@
fb4bf06b0cfb5bcac81e2faf99d8ace1c15771ea009837802a08a4dd5ba77a8f /home/coding/Source/kobrax/releases/0.9.0-beta1/extract_credentials
68f9bf800d1df0e71423edd35e90a8f5f7fb6e9e5220a8c12ed98cc6c4fb4833 /home/coding/Source/kobrax/releases/0.9.0-beta1/extract_credentials.exe
7c1a99953e21fc3881f60df444940d66a4689b009e9a17ec936396857a6b9dc0 /home/coding/Source/kobrax/releases/0.9.0-beta1/kx-bridge

1
requirements.txt Normal file
View File

@@ -0,0 +1 @@
aiohttp>=3.9