public_git/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-10-22 03:49:18 +02:00
Code Issues Packages Projects Releases Wiki Activity

Fix virtual printer model codes and serial prefixes

Browse Source 
- Correct SSDP model codes: C11=P1P, C12=P1S, N7=P2S, C13=X1E
  - Fix serial prefixes based on actual Bambu serial format
  - Add confirmation modal for pending upload discard
  - Sort model dropdown alphabetically, remove internal codes
  - Add "Setup Required" warning with link to wiki documentation
  - Update wiki with certificate installation and platform setup guides
This commit is contained in:
maziggy
2026-01-04 14:05:08 +01:00
parent 81c8dd7d0c
commit bb37a8c8a8
13 changed files with 418 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/services/virtual_printer/certificate.py
+100 -22
View File
@@ -3,6 +3,9 @@
Generates certificates that mimic real Bambu printer certificate format:
- CA certificate mimics "BBL CA" from "BBL Technologies Co., Ltd"
- Printer certificate has CN = serial number, signed by the CA
The CA certificate is persistent and only regenerated if missing or expired.
This allows users to add the CA to their slicer's trust store once.
"""
import logging
@@ -21,6 +24,9 @@ logger = logging.getLogger(__name__)
# Default serial number for virtual printer (matches SSDP/MQTT config)
DEFAULT_SERIAL = "00M09A391800001"
# Minimum days remaining before CA is considered expired and needs regeneration
CA_EXPIRY_THRESHOLD_DAYS = 30
def _get_local_ip() -> str:
"""Get the local IP address."""
@@ -67,8 +73,70 @@ class CertificateService:
return self.cert_path, self.key_path
return self.generate_certificates()
def _load_existing_ca(self) -> tuple[rsa.RSAPrivateKey, x509.Certificate] | None:
"""Try to load existing CA certificate and key.
Returns:
Tuple of (ca_private_key, ca_certificate) if valid CA exists, None otherwise
"""
if not self.ca_cert_path.exists() or not self.ca_key_path.exists():
logger.debug("CA certificate or key not found")
return None
try:
# Load CA certificate
ca_cert_pem = self.ca_cert_path.read_bytes()
ca_cert = x509.load_pem_x509_certificate(ca_cert_pem)
# Check if CA is expired or about to expire
now = datetime.now(UTC)
days_remaining = (ca_cert.not_valid_after_utc - now).days
if days_remaining < CA_EXPIRY_THRESHOLD_DAYS:
logger.warning(f"CA certificate expires in {days_remaining} days, will regenerate")
return None
# Load CA private key
ca_key_pem = self.ca_key_path.read_bytes()
ca_key = serialization.load_pem_private_key(ca_key_pem, password=None)
logger.info(f"Using existing CA certificate (expires in {days_remaining} days)")
return ca_key, ca_cert
except Exception as e:
logger.warning(f"Failed to load existing CA: {e}")
return None
def _get_or_create_ca(self) -> tuple[rsa.RSAPrivateKey, x509.Certificate]:
"""Get existing CA or create a new one.
Returns:
Tuple of (ca_private_key, ca_certificate)
"""
# Try to load existing CA first
existing = self._load_existing_ca()
if existing:
return existing
# Generate new CA
ca_key, ca_cert = self._generate_ca_certificate()
# Save CA certificate and key
self.cert_dir.mkdir(parents=True, exist_ok=True)
self.ca_key_path.write_bytes(
ca_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption(),
)
)
self.ca_key_path.chmod(0o600)
self.ca_cert_path.write_bytes(ca_cert.public_bytes(serialization.Encoding.PEM))
logger.info("Saved new CA certificate")
return ca_key, ca_cert
def _generate_ca_certificate(self) -> tuple[rsa.RSAPrivateKey, x509.Certificate]:
"""Generate a CA certificate for the virtual printer.
"""Generate a new CA certificate for the virtual printer.
We use a generic name instead of mimicking BBL CA, since the slicer
may specifically reject certificates claiming to be from BBL but
@@ -77,7 +145,7 @@ class CertificateService:
Returns:
Tuple of (ca_private_key, ca_certificate)
"""
logger.info("Generating Virtual Printer CA certificate...")
logger.info("Generating new Virtual Printer CA certificate...")
# Generate CA private key
ca_key = rsa.generate_private_key(
@@ -126,11 +194,11 @@ class CertificateService:
return ca_key, ca_cert
def generate_certificates(self) -> tuple[Path, Path]:
"""Generate CA and printer certificates.
"""Generate printer certificate (reusing existing CA if available).
Creates a certificate chain mimicking real Bambu printers:
- BBL CA (self-signed root)
- Printer certificate (CN=serial, signed by BBL CA)
- CA certificate (reused if exists and valid, otherwise generated)
- Printer certificate (CN=serial, signed by CA)
Returns:
Tuple of (cert_path, key_path)
@@ -140,19 +208,8 @@ class CertificateService:
# Ensure directory exists
self.cert_dir.mkdir(parents=True, exist_ok=True)
# Generate or load CA
ca_key, ca_cert = self._generate_ca_certificate()
# Save CA certificate and key
self.ca_key_path.write_bytes(
ca_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption(),
)
)
self.ca_key_path.chmod(0o600)
self.ca_cert_path.write_bytes(ca_cert.public_bytes(serialization.Encoding.PEM))
# Get or create CA (reuses existing if valid)
ca_key, ca_cert = self._get_or_create_ca()
# Generate printer private key
printer_key = rsa.generate_private_key(
@@ -246,9 +303,30 @@ class CertificateService:
logger.info(f" Printer: CN={self.serial}")
return self.cert_path, self.key_path
def delete_certificates(self) -> None:
"""Delete existing certificates."""
for path in [self.cert_path, self.key_path, self.ca_cert_path, self.ca_key_path]:
def delete_printer_certificate(self) -> None:
"""Delete only the printer certificate (preserves CA)."""
for path in [self.cert_path, self.key_path]:
if path.exists():
path.unlink()
logger.info("Deleted virtual printer certificates")
logger.info("Deleted printer certificate (CA preserved)")
def delete_certificates(self, include_ca: bool = False) -> None:
"""Delete existing certificates.
Args:
include_ca: If True, also delete CA certificate and key.
If False (default), only delete printer certificate.
"""
# Always delete printer certificate
for path in [self.cert_path, self.key_path]:
if path.exists():
path.unlink()
# Only delete CA if explicitly requested
if include_ca:
for path in [self.ca_cert_path, self.ca_key_path]:
if path.exists():
path.unlink()
logger.info("Deleted all certificates including CA")
else:
logger.info("Deleted printer certificate (CA preserved)")
backend/app/services/virtual_printer/ftp_server.py
+75 -1
View File
@@ -196,6 +196,40 @@ class FTPSession:
else:
await self.send(504, "Type not supported")
async def cmd_EPSV(self, arg: str) -> None:
"""Handle EPSV command - Extended Passive Mode (IPv6 compatible)."""
if not self.authenticated:
await self.send(530, "Not logged in")
return
# Close any existing data connection/server
await self._close_data_connection()
# Reset connection state
self._data_connected.clear()
self._data_reader = None
self._data_writer = None
# Find a free port for passive data connection
self.data_port = random.randint(50000, 60000)
try:
# Create data server with TLS - use same context for session reuse
self.data_server = await asyncio.start_server(
self._handle_data_connection,
"0.0.0.0",
self.data_port,
ssl=self.ssl_context,
)
# EPSV response format: 229 Entering Extended Passive Mode (|||port|)
await self.send(229, f"Entering Extended Passive Mode (|||{self.data_port}|)")
logger.info(f"FTP EPSV listening on port {self.data_port}")
except Exception as e:
logger.error(f"Failed to create EPSV data connection: {e}")
await self.send(425, "Cannot open data connection")
async def cmd_PASV(self, arg: str) -> None:
"""Handle PASV command - set up passive data connection."""
if not self.authenticated:
@@ -244,6 +278,16 @@ class FTPSession:
async def _handle_data_connection(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
"""Handle incoming data connection (used by PASV)."""
# Log TLS details for debugging
ssl_obj = writer.get_extra_info("ssl_object")
if ssl_obj:
logger.info(
f"FTP data TLS from {self.remote_ip}: cipher={ssl_obj.cipher()}, "
f"version={ssl_obj.version()}, session_reused={ssl_obj.session_reused}"
)
else:
logger.warning(f"FTP data connection from {self.remote_ip} has no SSL!")
logger.info(f"FTP data connection established from {self.remote_ip}")
self._data_reader = reader
self._data_writer = writer
@@ -252,6 +296,8 @@ class FTPSession:
async def _close_data_connection(self) -> None:
"""Close the data connection and server."""
had_connection = self._data_writer is not None or self.data_server is not None
if self._data_writer:
try:
self._data_writer.close()
@@ -269,6 +315,10 @@ class FTPSession:
pass
self.data_server = None
# Only delay if we actually closed something
if had_connection:
await asyncio.sleep(0.1)
async def cmd_STOR(self, arg: str) -> None:
"""Handle STOR command - receive file upload."""
if not self.authenticated:
@@ -436,6 +486,7 @@ class VirtualPrinterFTPServer:
self._server: asyncio.Server | None = None
self._running = False
self._ssl_context: ssl.SSLContext | None = None
self._active_sessions: list[asyncio.Task] = []
async def start(self) -> None:
"""Start the implicit FTPS server."""
@@ -454,6 +505,11 @@ class VirtualPrinterFTPServer:
self._ssl_context.load_cert_chain(str(self.cert_path), str(self.key_path))
self._ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2
# Use standard TLS settings for compatibility
self._ssl_context.set_ciphers("HIGH:!aNULL:!MD5:!RC4")
logger.info("FTP SSL context created with standard settings")
try:
# Create server with SSL - TLS handshake happens before any FTP data
self._server = await asyncio.start_server(
@@ -495,13 +551,31 @@ class VirtualPrinterFTPServer:
on_file_received=self.on_file_received,
)
await session.handle()
# Track the session task so we can cancel it on stop
task = asyncio.current_task()
if task:
self._active_sessions.append(task)
try:
await session.handle()
finally:
if task and task in self._active_sessions:
self._active_sessions.remove(task)
async def stop(self) -> None:
"""Stop the FTPS server."""
logger.info("Stopping FTP server")
self._running = False
# Cancel all active sessions first
for task in self._active_sessions[:]: # Copy list to avoid modification during iteration
task.cancel()
# Wait briefly for sessions to clean up
if self._active_sessions:
await asyncio.sleep(0.1)
self._active_sessions.clear()
if self._server:
try:
self._server.close()
backend/app/services/virtual_printer/manager.py
+91 -31
View File
@@ -17,15 +17,18 @@ logger = logging.getLogger(__name__)
# Mapping of SSDP model codes to display names
# These are the codes that slicers expect during discovery
# Sources:
# - https://gist.github.com/Alex-Schaefer/72a9e2491a42da2ef99fb87601955cc3
# - https://github.com/psychoticbeef/BambuLabOrcaSlicerDiscovery
VIRTUAL_PRINTER_MODELS = {
# X1 Series
"BL-P001": "X1C", # X1 Carbon
"BL-P002": "X1", # X1
"BL-P003": "X1E", # X1E
"3DPrinter-X1-Carbon": "X1C", # X1 Carbon
"3DPrinter-X1": "X1", # X1
"C13": "X1E", # X1E
# P Series
"C11": "P1S", # P1S
"C12": "P1P", # P1P
"C13": "P2S", # P2S
"C11": "P1P", # P1P
"C12": "P1S", # P1S
"N7": "P2S", # P2S
# A1 Series
"N2S": "A1", # A1
"N1": "A1 Mini", # A1 Mini
@@ -35,8 +38,35 @@ VIRTUAL_PRINTER_MODELS = {
"O1S": "H2S", # H2S
}
# Serial number prefixes for each model (based on Bambu Lab serial number format)
# Format: MMM??RYMDDUUUUU (15 chars total)
# MMM = Model prefix (3 chars)
# ?? = Unknown/revision code (2 chars)
# R = Revision letter (1 char)
# Y = Year digit (1 char)
# M = Month (1 char, hex: 1-9, A=Oct, B=Nov, C=Dec)
# DD = Day (2 chars)
# UUUUU = Unit number (5 chars)
MODEL_SERIAL_PREFIXES = {
# X1 Series
"3DPrinter-X1-Carbon": "00M00A", # X1C
"3DPrinter-X1": "00M00A", # X1
"C13": "03W00A", # X1E
# P Series
"C11": "01S00A", # P1P
"C12": "01P00A", # P1S
"N7": "22E00A", # P2S
# A1 Series
"N2S": "03900A", # A1
"N1": "03000A", # A1 Mini
# H2 Series
"O1D": "09400A", # H2D
"O1C": "09400A", # H2C
"O1S": "09400A", # H2S
}
# Default model
DEFAULT_VIRTUAL_PRINTER_MODEL = "BL-P001" # X1C
DEFAULT_VIRTUAL_PRINTER_MODEL = "3DPrinter-X1-Carbon" # X1C
class VirtualPrinterManager:
@@ -44,7 +74,7 @@ class VirtualPrinterManager:
# Fixed configuration
PRINTER_NAME = "Bambuddy"
PRINTER_SERIAL = "00M09A391800001" # X1C serial format
SERIAL_SUFFIX = "391800001" # Fixed suffix for virtual printer
def __init__(self):
"""Initialize the virtual printer manager."""
@@ -67,12 +97,29 @@ class VirtualPrinterManager:
self._upload_dir = self._base_dir / "uploads"
self._cert_dir = self._base_dir / "certs"
# Certificate service - pass serial to match CN in certificate
self._cert_service = CertificateService(self._cert_dir, serial=self.PRINTER_SERIAL)
# Certificate service
self._cert_service = CertificateService(self._cert_dir)
# Track pending uploads for MQTT correlation
self._pending_files: dict[str, Path] = {}
def _get_serial_for_model(self, model: str) -> str:
"""Get appropriate serial number for the given model.
Args:
model: SSDP model code (e.g., 'BL-P001', 'C11')
Returns:
Serial number with correct prefix for the model
"""
prefix = MODEL_SERIAL_PREFIXES.get(model, "00M09A")
return f"{prefix}{self.SERIAL_SUFFIX}"
@property
def printer_serial(self) -> str:
"""Get the current printer serial number based on model."""
return self._get_serial_for_model(self._model)
def set_session_factory(self, session_factory: Callable) -> None:
"""Set the database session factory.
@@ -112,6 +159,7 @@ class VirtualPrinterManager:
# Validate model if provided
new_model = model if model and model in VIRTUAL_PRINTER_MODELS else self._model
model_changed = new_model != self._model
old_model = self._model
self._access_code = access_code
self._mode = mode
@@ -123,8 +171,12 @@ class VirtualPrinterManager:
await self._stop()
elif enabled and self._enabled and model_changed:
# Model changed while running - restart services
logger.info(f"Model changed from {old_model} to {new_model}, restarting...")
await self._stop()
# Give time for ports to be released
await asyncio.sleep(0.5)
await self._start()
logger.info("Virtual printer restarted with new model")
self._enabled = enabled
@@ -132,8 +184,14 @@ class VirtualPrinterManager:
"""Start all virtual printer services."""
logger.info("Starting virtual printer services...")
# Ensure certificates exist
cert_path, key_path = self._cert_service.ensure_certificates()
# Update certificate service with current serial (based on model)
current_serial = self.printer_serial
self._cert_service.serial = current_serial
# Regenerate printer cert if serial changed (CA is preserved)
self._cert_service.delete_printer_certificate()
cert_path, key_path = self._cert_service.generate_certificates()
logger.info(f"Generated certificate for serial: {current_serial}")
# Create directories
self._upload_dir.mkdir(parents=True, exist_ok=True)
@@ -142,7 +200,7 @@ class VirtualPrinterManager:
# Initialize services
self._ssdp = VirtualPrinterSSDPServer(
name=self.PRINTER_NAME,
serial=self.PRINTER_SERIAL,
serial=self.printer_serial,
model=self._model,
)
@@ -155,7 +213,7 @@ class VirtualPrinterManager:
)
self._mqtt = SimpleMQTTServer(
serial=self.PRINTER_SERIAL,
serial=self.printer_serial,
access_code=self._access_code,
cert_path=cert_path,
key_path=key_path,
@@ -176,27 +234,13 @@ class VirtualPrinterManager:
asyncio.create_task(run_with_logging(self._mqtt.start(), "MQTT"), name="virtual_printer_mqtt"),
]
logger.info(f"Virtual printer '{self.PRINTER_NAME}' started (serial: {self.PRINTER_SERIAL})")
logger.info(f"Virtual printer '{self.PRINTER_NAME}' started (serial: {self.printer_serial})")
async def _stop(self) -> None:
"""Stop all virtual printer services."""
logger.info("Stopping virtual printer services...")
# Cancel all tasks
for task in self._tasks:
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
self._tasks = []
# Stop services
if self._ssdp:
await self._ssdp.stop()
self._ssdp = None
# Stop services first - this closes servers and cancels active sessions
if self._ftp:
await self._ftp.stop()
self._ftp = None
@@ -205,6 +249,22 @@ class VirtualPrinterManager:
await self._mqtt.stop()
self._mqtt = None
if self._ssdp:
await self._ssdp.stop()
self._ssdp = None
# Cancel remaining tasks with short timeout
for task in self._tasks:
task.cancel()
if self._tasks:
try:
await asyncio.wait_for(asyncio.gather(*self._tasks, return_exceptions=True), timeout=1.0)
except TimeoutError:
logger.debug("Some tasks didn't stop in time")
self._tasks = []
logger.info("Virtual printer stopped")
async def _on_file_received(self, file_path: Path, source_ip: str) -> None:
@@ -348,7 +408,7 @@ class VirtualPrinterManager:
"running": self.is_running,
"mode": self._mode,
"name": self.PRINTER_NAME,
"serial": self.PRINTER_SERIAL,
"serial": self.printer_serial,
"model": self._model,
"model_name": VIRTUAL_PRINTER_MODELS.get(self._model, self._model),
"pending_files": len(self._pending_files),
backend/app/services/virtual_printer/mqtt_server.py
+2 -2
View File
@@ -289,8 +289,8 @@ class SimpleMQTTServer:
pass
self._status_push_task = None
# Close all client connections
for _client_id, writer in self._clients.items():
# Close all client connections (iterate over copy to avoid modification during iteration)
for _client_id, writer in list(self._clients.items()):
try:
writer.close()
await writer.wait_closed()