import io import logging import os import zipfile from datetime import datetime from pathlib import Path from fastapi import APIRouter, Depends, File, HTTPException, UploadFile from fastapi.responses import FileResponse, JSONResponse from pydantic import BaseModel, Field from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from backend.app.core.auth import RequirePermissionIfAuthEnabled, caller_is_api_key, require_energy_cost_update from backend.app.core.config import settings as app_settings from backend.app.core.database import get_db from backend.app.core.permissions import Permission from backend.app.models.settings import Settings from backend.app.models.user import User from backend.app.schemas.settings import AppSettings, AppSettingsUpdate logger = logging.getLogger(__name__) router = APIRouter(prefix="/settings", tags=["settings"]) DEFAULT_SETTINGS = AppSettings() # Sensitive credential fields blanked for API-key callers _SENSITIVE_FIELDS_FOR_API_KEY = ( "mqtt_password", "ha_token", "prometheus_token", "virtual_printer_access_code", "ldap_bind_password", ) async def get_setting(db: AsyncSession, key: str) -> str | None: """Get a single setting value by key.""" result = await db.execute(select(Settings).where(Settings.key == key)) setting = result.scalar_one_or_none() return setting.value if setting else None # Accepted spellings for a boolean settings value. Settings live in a VARCHAR # column and every reader compares them as strings, so these are normalised to # "true"/"false" on the way in. The sets are deliberately generous: these # endpoints are part of the documented REST surface, reached by scripts and by # Home Assistant rest_command, where "True", "1" and "on" are all natural. _TRUTHY_SETTING_VALUES = frozenset({"true", "1", "yes", "on"}) _FALSY_SETTING_VALUES = frozenset({"false", "0", "no", "off"}) def setting_is_true(value: object) -> bool: """Return True if a *stored* settings value means "on". Deliberately narrower than the spellings ``normalize_bool_setting`` accepts: it matches only what every other reader in the codebase treats as on (``value.lower() == "true"``). Submitted values are canonicalised on write, so a stored value is always "true"/"false"/""; accepting "1" or "on" here would make this function disagree with the rest of the app about any legacy row containing them. A bool is tolerated for the case of a row written before values were normalised, where SQLite coerced a raw bool into the VARCHAR column. """ if isinstance(value, bool): return value if value is None: return False return str(value).strip().lower() == "true" def normalize_bool_setting(key: str, value: object) -> str: """Coerce a boolean-ish settings value to the canonical "true"/"false". Raises HTTPException(400) for values with no sensible interpretation, so an API client gets a message naming the field instead of a 500. A JSON boolean is the natural thing for an API client to send, and before this normalisation it caused two distinct failures on ``PUT /settings/spoolman``: ``bool.lower()`` raised AttributeError, and the raw bool was written into a VARCHAR column, which SQLite silently coerces to 1/0 while asyncpg rejects outright. Both surfaced as an opaque 500. """ if isinstance(value, bool): # must precede the int branch — bool is an int return "true" if value else "false" if isinstance(value, int): if value in (0, 1): return "true" if value else "false" raise HTTPException(400, f"{key} must be a boolean; got the number {value}") if isinstance(value, str): candidate = value.strip().lower() if not candidate: # Empty is stored verbatim rather than normalised to "false". # get_spoolman_settings reads these with ``or ""``, so an # empty stored value means "use the default" — and two of them # (spoolman_report_partial_usage, auto_add_unknown_rfid) default to # ON. Rewriting "" to "false" would silently switch them off for any # client that submits a blank value. return "" if candidate in _TRUTHY_SETTING_VALUES: return "true" if candidate in _FALSY_SETTING_VALUES: return "false" raise HTTPException(400, f"{key} must be a boolean; got {value!r}") raise HTTPException(400, f"{key} must be a boolean; got {type(value).__name__}") def normalize_str_setting(key: str, value: object) -> str: """Return a string settings value, rejecting types that would store garbage. ``str()`` on a dict or list would persist its repr, so those are refused rather than silently written. Numbers are accepted and stringified: a port or a bare host submitted unquoted is a plausible client mistake, not a reason to fail the request. """ if isinstance(value, str): return value if value is None: return "" if isinstance(value, bool | int | float): return str(value) raise HTTPException(400, f"{key} must be a string; got {type(value).__name__}") async def get_external_login_url(db: AsyncSession) -> str: """Get the external URL for the login page. Uses external_url from settings if available, otherwise falls back to APP_URL env var. Args: db: Database session Returns: Full URL to the login page """ import os external_url = await get_setting(db, "external_url") if external_url: external_url = external_url.rstrip("/") else: external_url = os.environ.get("APP_URL", "http://localhost:5173") return external_url + "/login" async def set_setting(db: AsyncSession, key: str, value: str) -> None: """Set a single setting value.""" from backend.app.core.db_dialect import upsert_setting await upsert_setting(db, Settings, key, value) async def _build_settings_response(db: AsyncSession, is_api_key: bool = False) -> AppSettings: """Build the full settings response, scrubbing secrets for API-key callers.""" settings_dict = DEFAULT_SETTINGS.model_dump() result = await db.execute(select(Settings)) for setting in result.scalars().all(): if setting.key not in settings_dict: continue if setting.key in [ "auto_archive", "save_thumbnails", "capture_finish_photo", "finish_photo_restore_plate", "spoolman_enabled", "spoolman_disable_weight_sync", "spoolman_report_partial_usage", "auto_add_unknown_rfid", "disable_filament_warnings", "prefer_lowest_filament", "check_updates", "check_printer_firmware", "include_beta_updates", "virtual_printer_enabled", "ftp_retry_enabled", "mqtt_enabled", "mqtt_use_tls", "ha_enabled", "per_printer_mapping_expanded", "prometheus_enabled", "user_notifications_enabled", "queue_drying_enabled", "queue_drying_block", "ambient_drying_enabled", "print_drying_enabled", "require_plate_clear", "queue_shortest_first", # default_bed_levelling / default_flow_cali / default_nozzle_offset_cali # are tri-state strings (off/on/auto) — parsed via the raw-string else # branch; the TriState validator coerces legacy "true"/"false" rows. "default_vibration_cali", "default_layer_inspect", "default_timelapse", "billing_enabled", "printer_kill_switch_enabled", "ldap_enabled", "ldap_auto_provision", "local_login_enabled", "preheat_enabled", "queue_keep_bed_warm", ]: settings_dict[setting.key] = setting.value.lower() == "true" elif setting.key in [ "default_filament_cost", "energy_cost_per_kwh", "ams_temp_good", "ams_temp_fair", "library_disk_warning_gb", "low_stock_threshold", ]: settings_dict[setting.key] = float(setting.value) elif setting.key in [ # Nullable floats. Settings storage stringifies None to the literal # "None", so these cannot go in the list above -- float("None") # raises and would take the whole settings response with it (#2905). "ams_temp_alarm", ]: try: settings_dict[setting.key] = float(setting.value) except (TypeError, ValueError): settings_dict[setting.key] = None elif setting.key in [ "ams_humidity_good", "ams_humidity_fair", "ams_history_retention_days", "printer_sensor_history_retention_days", "ftp_retry_count", "ftp_retry_delay", "ftp_timeout", "mqtt_port", "stagger_group_size", "stagger_interval_minutes", "forecast_global_lead_time_days", "location_sensor_poll_interval", "finance_budget_reset_day", "session_max_hours", "pipeline_max_copies", "preheat_max_wait_seconds", "preheat_soak_seconds", "queue_keep_warm_bed_temp", "queue_keep_warm_max_minutes", "queue_max_concurrent_uploads", ]: settings_dict[setting.key] = int(setting.value) elif setting.key == "default_printer_id": settings_dict[setting.key] = int(setting.value) if setting.value and setting.value != "None" else None elif setting.key == "open_in_slicer": # None means "inherit from preferred_slicer" (#1329). The PUT path # serializes None as the literal string "None"; strip it back so # the frontend sees a true null and falls back as intended. settings_dict[setting.key] = setting.value if setting.value and setting.value != "None" else None else: settings_dict[setting.key] = setting.value ha_settings = await get_homeassistant_settings(db) settings_dict.update(ha_settings) # ldap_bind_password is never returned to any caller settings_dict["ldap_bind_password"] = "" if is_api_key: for field in _SENSITIVE_FIELDS_FOR_API_KEY: if field in settings_dict: settings_dict[field] = "" return AppSettings(**settings_dict) @router.get("", response_model=AppSettings) @router.get("/", response_model=AppSettings) async def get_settings( db: AsyncSession = Depends(get_db), _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ), _is_api_key: bool = Depends(caller_is_api_key), ): """Get all application settings.""" return await _build_settings_response(db, is_api_key=_is_api_key) @router.put("/", response_model=AppSettings) async def update_settings( settings_update: AppSettingsUpdate, db: AsyncSession = Depends(get_db), current_user: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE), ): """Update application settings.""" update_data = settings_update.model_dump(exclude_unset=True) # Safety refusals on disabling local login (#1589). Two failure modes # would otherwise lock everyone out of the install: # 1. No enabled OIDC provider exists — nobody could authenticate. # 2. The caller has no UserOIDCLink — they would lock themselves out # even if other admins are linked. # Either case returns HTTP 400 instead of silently saving. The # ``BAMBUDDY_LOCAL_LOGIN=true`` env-var bypass on /auth/login is a # separate recovery path; the refusals here protect the *default* # configuration where the env var is absent. if update_data.get("local_login_enabled") is False: from backend.app.models.oidc_provider import OIDCProvider, UserOIDCLink enabled_count = await db.scalar(select(func.count(OIDCProvider.id)).where(OIDCProvider.is_enabled.is_(True))) if not enabled_count: raise HTTPException( status_code=400, detail="Cannot disable local login: no OIDC provider is enabled.", ) if current_user is not None: caller_links = await db.scalar( select(func.count(UserOIDCLink.id)).where(UserOIDCLink.user_id == current_user.id) ) if not caller_links: raise HTTPException( status_code=400, detail="Cannot disable local login: your account has no OIDC link, so you would lock yourself out.", ) # Check if any MQTT settings are being updated mqtt_keys = { "mqtt_enabled", "mqtt_broker", "mqtt_port", "mqtt_username", "mqtt_password", "mqtt_topic_prefix", "mqtt_use_tls", } mqtt_updated = bool(mqtt_keys & set(update_data.keys())) for key, value in update_data.items(): # Convert value to string for storage if isinstance(value, bool): str_value = "true" if value else "false" elif value is None: str_value = "None" else: str_value = str(value) await set_setting(db, key, str_value) await db.commit() # Expire all objects to ensure fresh reads after commit db.expire_all() # Reconfigure MQTT relay if any MQTT settings changed if mqtt_updated: try: from backend.app.services.mqtt_relay import mqtt_relay mqtt_settings = { "mqtt_enabled": (await get_setting(db, "mqtt_enabled") or "false") == "true", "mqtt_broker": await get_setting(db, "mqtt_broker") or "", "mqtt_port": int(await get_setting(db, "mqtt_port") or "1883"), "mqtt_username": await get_setting(db, "mqtt_username") or "", "mqtt_password": await get_setting(db, "mqtt_password") or "", "mqtt_topic_prefix": await get_setting(db, "mqtt_topic_prefix") or "bambuddy", "mqtt_use_tls": (await get_setting(db, "mqtt_use_tls") or "false") == "true", } await mqtt_relay.configure(mqtt_settings) except Exception: pass # Don't fail the settings update if MQTT reconfiguration fails # Return updated settings (never scrub secrets on PUT — caller has SETTINGS_UPDATE permission) return await _build_settings_response(db, is_api_key=False) @router.patch("/", response_model=AppSettings) @router.patch("", response_model=AppSettings) async def patch_settings( settings_update: AppSettingsUpdate, db: AsyncSession = Depends(get_db), _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE), ): """Partially update application settings (same as PUT, for REST compatibility).""" return await update_settings(settings_update, db, _) class ElectricityPriceUpdate(BaseModel): """Payload for ``POST /settings/electricity-price`` (#1356). Mirrors the field name documented in ``wiki/features/energy.md`` so the Home Assistant ``rest_command`` example needs only a URL change, not a payload change. Plain non-negative float; tariffs can go as low as 0.0 in some markets (e.g. free hours). """ energy_cost_per_kwh: float = Field(ge=0) @router.post("/electricity-price", response_model=AppSettings) async def update_electricity_price( payload: ElectricityPriceUpdate, db: AsyncSession = Depends(get_db), _: User | None = Depends(require_energy_cost_update()), _is_api_key: bool = Depends(caller_is_api_key), ): """Update the per-kWh electricity cost used by the energy-tracking pipeline. This is the only settings field writable via API key, gated by the ``can_update_energy_cost`` toggle on the key. JWT users still need the standard ``SETTINGS_UPDATE`` permission. See #1356 for the rationale — the general ``PATCH /settings`` route remains denied for API keys because it can rewrite SMTP/LDAP/MQTT credentials, which is a much wider surface than the documented dynamic-tariff use case requires. """ await set_setting(db, "energy_cost_per_kwh", str(payload.energy_cost_per_kwh)) await db.commit() db.expire_all() return await _build_settings_response(db, is_api_key=_is_api_key) @router.post("/reset", response_model=AppSettings) async def reset_settings( db: AsyncSession = Depends(get_db), _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE), ): """Reset all settings to defaults.""" # Delete all settings result = await db.execute(select(Settings)) for setting in result.scalars().all(): await db.delete(setting) await db.commit() return DEFAULT_SETTINGS @router.get("/default-sidebar-order") async def get_default_sidebar_order( db: AsyncSession = Depends(get_db), ): """Get the admin-set default sidebar order. Intentionally unauthenticated: non-admin users need to read this value to apply the default sidebar order, but may lack SETTINGS_READ permission. The value is non-sensitive (sidebar item IDs only). """ value = await get_setting(db, "default_sidebar_order") return {"default_sidebar_order": value or ""} # Fields exposed via /ui-preferences without SETTINGS_READ. Each entry MUST be # non-sensitive (no credentials, no PII, no secret tokens) — granting SETTINGS_READ # also grants visibility of SMTP/LDAP/MQTT passwords and similar, so the goal of # this endpoint is exactly to NOT require that permission for UI rendering hints. # When adding a field here, confirm it doesn't carry anything sensitive. _UI_PREFERENCE_FIELDS: tuple[str, ...] = ( "require_plate_clear", "check_printer_firmware", "camera_view_mode", "time_format", "date_format", "drying_presets", "ams_humidity_thresholds", "ams_humidity_good", "ams_humidity_fair", "ams_temp_good", "ams_temp_fair", # ams_temp_alarm is deliberately NOT here. This endpoint is unauthenticated # and exists so the UI can colour readings without SETTINGS_READ; the good / # fair bands are what the printer card colours by. The alarm threshold # changes no rendering anywhere -- only SettingsPage reads it, and that is # behind the settings permissions already (#2905). "bed_cooled_threshold", # Temperature / fan-speed presets for the printer-card popovers. Numbers # only; no PII / credentials. "nozzle_temp_presets", "bed_temp_presets", "chamber_temp_presets", "fan_speed_presets", ) @router.get("/ui-preferences") async def get_ui_preferences(db: AsyncSession = Depends(get_db)): """Get the curated subset of settings that any page needs to render correctly. Intentionally not gated on SETTINGS_READ — every authenticated user (and every page that loads for them) needs these fields, but granting SETTINGS_READ would also grant visibility of secrets (SMTP/LDAP/MQTT credentials, etc.). Same pattern as /default-sidebar-order (#1293). Reuses _build_settings_response so the typed values match what /settings returns for fields with the same name — bool/int/float/str types stay in sync without a separate type-coercion path. """ full = await _build_settings_response(db, is_api_key=False) dumped = full.model_dump() return {key: dumped[key] for key in _UI_PREFERENCE_FIELDS if key in dumped} @router.get("/check-ffmpeg") async def check_ffmpeg( _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ), ): """Check if ffmpeg is installed and available. Gated on ``SETTINGS_READ`` (audit finding I4 — the binary path was leaking the host filesystem layout to unauthenticated callers). ``require_permission_if_auth_enabled`` returns ``None`` only when auth is disabled (in which case there's no privacy boundary to enforce); otherwise it raises 401/403 before we get here. """ from backend.app.services.camera import get_ffmpeg_path ffmpeg_path = get_ffmpeg_path() return { "installed": ffmpeg_path is not None, "path": ffmpeg_path, } @router.get("/spoolman") async def get_spoolman_settings( db: AsyncSession = Depends(get_db), _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ), ): """Get Spoolman integration settings.""" spoolman_enabled = await get_setting(db, "spoolman_enabled") or "false" spoolman_url = await get_setting(db, "spoolman_url") or "" spoolman_sync_mode = await get_setting(db, "spoolman_sync_mode") or "auto" spoolman_disable_weight_sync = await get_setting(db, "spoolman_disable_weight_sync") or "false" spoolman_report_partial_usage = await get_setting(db, "spoolman_report_partial_usage") or "true" auto_add_unknown_rfid = await get_setting(db, "auto_add_unknown_rfid") or "true" return { "spoolman_enabled": spoolman_enabled, "spoolman_url": spoolman_url, "spoolman_sync_mode": spoolman_sync_mode, "spoolman_disable_weight_sync": spoolman_disable_weight_sync, "spoolman_report_partial_usage": spoolman_report_partial_usage, "auto_add_unknown_rfid": auto_add_unknown_rfid, } @router.put("/spoolman") async def update_spoolman_settings( settings: dict, db: AsyncSession = Depends(get_db), _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE), ): """Update Spoolman integration settings. The body is a free-form dict rather than a schema, so each value is normalised before it is persisted — see ``normalize_bool_setting`` for why a JSON boolean used to produce a 500 here. """ if "spoolman_enabled" in settings: was_enabled = setting_is_true(await get_setting(db, "spoolman_enabled")) new_val = normalize_bool_setting("spoolman_enabled", settings["spoolman_enabled"]) now_enabled = new_val == "true" await set_setting(db, "spoolman_enabled", new_val) # Nothing is deleted on a mode change (#2812). Each mode keeps its slot # assignments in its own table, so both can hold rows at once and the # toggle is reversible: switching to Spoolman to see what it does, then # switching back, returns you to the assignments you had. # # This used to empty the other mode's table on every toggle. The reason # was real -- checks that read both tables would let a row in the mode # you are not using answer for the mode you are -- but the cost was that # inspecting a mode destroyed your configuration, with no confirmation # and no way back, and the deletion was unfiltered across every printer. # The readers that could be confused now ask which mode is active # (``spoolman_owns_assignments``), which is where that decision belongs: # the mode is a property of the install, not of the rows. if was_enabled != now_enabled: logger.info( "Inventory mode switched to %s; slot assignments in both tables kept", "Spoolman" if now_enabled else "built-in", ) if "spoolman_url" in settings: await set_setting(db, "spoolman_url", normalize_str_setting("spoolman_url", settings["spoolman_url"])) if "spoolman_sync_mode" in settings: await set_setting( db, "spoolman_sync_mode", normalize_str_setting("spoolman_sync_mode", settings["spoolman_sync_mode"]) ) for bool_key in ("spoolman_disable_weight_sync", "spoolman_report_partial_usage", "auto_add_unknown_rfid"): if bool_key in settings: await set_setting(db, bool_key, normalize_bool_setting(bool_key, settings[bool_key])) spoolman_changed = "spoolman_enabled" in settings or "spoolman_url" in settings await db.commit() db.expire_all() if spoolman_changed: from backend.app.services.location_service import maybe_sync_spoolman_locations if await maybe_sync_spoolman_locations(db): await db.commit() # Return updated settings return await get_spoolman_settings(db) async def get_homeassistant_settings(db: AsyncSession) -> dict: """ Get Home Assistant integration settings. Environment variables (HA_URL, HA_TOKEN) take precedence over database settings. """ import os # Check environment variables first ha_url_env = os.environ.get("HA_URL") ha_token_env = os.environ.get("HA_TOKEN") # Fall back to database values ha_url = ha_url_env or await get_setting(db, "ha_url") or "" ha_token = ha_token_env or await get_setting(db, "ha_token") or "" ha_enabled_db = await get_setting(db, "ha_enabled") or "false" # Track which settings come from environment ha_url_from_env = bool(ha_url_env) ha_token_from_env = bool(ha_token_env) ha_env_managed = ha_url_from_env and ha_token_from_env # Auto-enable when both env vars are set, otherwise use database value if ha_url_env and ha_token_env: ha_enabled = True else: ha_enabled = ha_enabled_db.lower() == "true" return { "ha_enabled": ha_enabled, "ha_url": ha_url, "ha_token": ha_token, "ha_url_from_env": ha_url_from_env, "ha_token_from_env": ha_token_from_env, "ha_env_managed": ha_env_managed, } async def create_backup_zip(output_path: Path | None = None) -> tuple[Path, str]: """Create a complete backup ZIP (database + all data directories). If output_path is given, the ZIP is written there. Otherwise a temporary file is created (caller must clean up). Returns (zip_path, filename). """ import shutil import tempfile from backend.app.core.db_dialect import is_sqlite base_dir = app_settings.base_dir filename = f"bambuddy-backup-{datetime.now().strftime('%Y%m%d-%H%M%S')}.zip" with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) if is_sqlite(): from sqlalchemy import text from backend.app.core.database import engine db_path = Path(app_settings.database_url.replace("sqlite+aiosqlite:///", "")) # Checkpoint WAL to ensure all data is in main db file async with engine.begin() as conn: await conn.execute(text("PRAGMA wal_checkpoint(TRUNCATE)")) # Copy database file shutil.copy2(db_path, temp_path / "bambuddy.db") else: # PostgreSQL: export to a portable SQLite file via SQLAlchemy. # This makes backups restorable on both SQLite and Postgres installs. import json import sqlite3 from sqlalchemy import create_engine as create_sync_engine from backend.app.core.database import Base, engine backup_db_path = temp_path / "bambuddy.db" metadata = Base.metadata # Build the portable SQLite schema with SQLAlchemy's own DDL rather # than a hand-rolled CREATE TABLE. metadata.create_all() emits the # exact schema a native SQLite install gets — NOT NULL, DEFAULT # (server_default=func.now() → CURRENT_TIMESTAMP), foreign keys, # unique constraints and indexes. The previous name+type-only # rebuild dropped all of these, so a Postgres→SQLite restore left # server_default columns (e.g. spoolbuddy_devices.created_at) with # no DEFAULT — SQLAlchemy omits such columns on INSERT and the DB # then wrote NULL, which 500'd on the next read (#2526). Using the # real DDL also keeps the #1333 BLOB guard: LargeBinary still # renders as BLOB, so OIDC icon bytes survive the round trip. schema_engine = create_sync_engine(f"sqlite:///{backup_db_path}") try: metadata.create_all(schema_engine) finally: schema_engine.dispose() dst = sqlite3.connect(str(backup_db_path)) # Export data from Postgres to SQLite async with engine.connect() as conn: for table in metadata.sorted_tables: result = await conn.execute(table.select()) rows = result.fetchall() if not rows: continue columns = list(result.keys()) placeholders = ", ".join(["?"] * len(columns)) col_list = ", ".join(columns) insert_sql = f"INSERT INTO {table.name} ({col_list}) VALUES ({placeholders})" # noqa: S608 # nosec B608 — table/column names from ORM metadata, not user input def _serialize_row(row): return tuple(json.dumps(v) if isinstance(v, (list, dict)) else v for v in row) dst.executemany(insert_sql, [_serialize_row(row) for row in rows]) dst.commit() dst.close() logger.info("PostgreSQL backup exported to portable SQLite format") # Copy data directories (if they exist) dirs_to_backup = [ ("archive", base_dir / "archive"), ("virtual_printer", base_dir / "virtual_printer"), ("plate_calibration", app_settings.plate_calibration_dir), ("icons", base_dir / "icons"), ("projects", base_dir / "projects"), ] for name, src_dir in dirs_to_backup: if src_dir.exists() and any(src_dir.iterdir()): try: shutil.copytree( src_dir, temp_path / name ) # SEC-PATH-OK: name iterates the dirs_to_backup tuple of constant strings ("archive", "virtual_printer", ...) except shutil.Error as e: logger.warning("Some files in %s could not be copied: %s", name, e) except PermissionError as e: logger.warning("Permission denied copying %s: %s", name, e) # Include the MFA encryption key as a ZIP top-level entry alongside # bambuddy.db. Without it, encrypted client_secret / TOTP secret rows # would be unrecoverable after restore on a host without MFA_ENCRYPTION_KEY set. from backend.app.core.paths import resolve_data_dir mfa_key_src = resolve_data_dir() / ".mfa_encryption_key" if mfa_key_src.exists() and mfa_key_src.is_file(): try: shutil.copy2(mfa_key_src, temp_path / ".mfa_encryption_key") except OSError as exc: logger.error( "Could not include MFA encryption key in backup (%s). " "The backup ZIP will not contain the key — restore on a " "keyless host will fail for encrypted secrets.", exc, ) raise # Create ZIP if output_path is not None: zip_file = ( output_path / filename ) # SEC-PATH-OK: filename = f"bambuddy-backup-{datetime.now()...}.zip" generated in create_backup_zip itself else: fd, tmp = tempfile.mkstemp(suffix=".zip") os.close(fd) zip_file = Path(tmp) with zipfile.ZipFile(zip_file, "w", zipfile.ZIP_DEFLATED) as zf: for file_path in temp_path.rglob("*"): if file_path.is_file(): arcname = file_path.relative_to(temp_path) zf.write(file_path, arcname) return zip_file, filename @router.get("/backup") async def create_backup( db: AsyncSession = Depends(get_db), _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_BACKUP), ): """Create a complete backup (database + all files) as a ZIP download.""" from starlette.background import BackgroundTask try: zip_file, filename = await create_backup_zip() return FileResponse( path=zip_file, filename=filename, media_type="application/zip", background=BackgroundTask(lambda: zip_file.unlink(missing_ok=True)), ) except Exception as e: logger.error("Backup failed: %s", e, exc_info=True) return JSONResponse( status_code=500, content={"success": False, "message": "Backup failed. Check server logs for details."}, ) async def _import_sqlite_to_postgres(sqlite_path: Path, postgres_url: str): """Import data from a SQLite database file into the current PostgreSQL database. Used for cross-database restore (SQLite backup → PostgreSQL). Reads all tables from the SQLite file and bulk-inserts into Postgres. """ import sqlite3 from sqlalchemy import text from backend.app.core.database import Base, _create_engine # Create a temporary engine for the import (current engine was disposed) pg_engine = _create_engine() try: # Open SQLite file directly (sync — it's a local file read) src = sqlite3.connect(str(sqlite_path)) src.row_factory = sqlite3.Row # Get list of tables from SQLite (skip internal/FTS tables) cursor = src.execute( "SELECT name FROM sqlite_master WHERE type='table' " "AND name NOT LIKE 'sqlite_%' AND name NOT LIKE 'archive_fts%'" ) src_tables = {row["name"] for row in cursor.fetchall()} # Get Postgres tables from our ORM models metadata = Base.metadata pg_tables = set(metadata.tables.keys()) # Only import tables that exist in both source and destination tables_to_import = src_tables & pg_tables sorted_tables = [t.name for t in metadata.sorted_tables if t.name in tables_to_import] # Phase 1: Drop all tables and recreate WITHOUT foreign keys. # This avoids all FK ordering/orphan issues during import; the # constraints go back on at the end, once every row has landed. async with pg_engine.begin() as conn: # Cap how long DROP TABLE will wait for AccessExclusiveLock so # any residual concurrent writer (per-printer MQTT clients # writing reactively, an AMS history recorder firing on its # hourly cadence) surfaces a fast `lock_timeout` error instead # of blocking the restore for 30 s or producing a deadlock. # SET LOCAL scopes to this transaction only; outside this # restore path the global default (no timeout) applies. await conn.execute(text("SET LOCAL lock_timeout = '10s'")) # Drop every existing table in the public schema with CASCADE # rather than `metadata.drop_all`. Two reasons: # 1. The user's live DB may carry orphan tables from removed # features (e.g. the legacy `spoolman_slot_assignments`, # `spoolman_k_profile`) that hold FK constraints back to # ORM tables. `drop_all` doesn't know they exist and emits # `DROP TABLE printers` without CASCADE — Postgres refuses # and the whole restore aborts (#XXXX). # 2. Even within the metadata, `drop_all` is FK-ordered and # breaks if a future schema rename leaves old constraints # around. CASCADE is the right tool for a destructive # restore: the user is intentionally wiping state. await conn.execute( text( "DO $$ DECLARE r RECORD; BEGIN " "FOR r IN (SELECT tablename FROM pg_tables WHERE schemaname = 'public') LOOP " "EXECUTE 'DROP TABLE IF EXISTS public.' || quote_ident(r.tablename) || ' CASCADE'; " "END LOOP; END $$;" ) ) await conn.run_sync(metadata.create_all) # Now strip the foreign keys, at the database level. # # This used to be done by discarding each ForeignKeyConstraint # from `table.constraints` before `create_all`. That only # suppresses the inline REFERENCES clause inside CREATE TABLE: # `Table.foreign_key_constraints` is derived from the *columns'* # ForeignKey objects, which the discard never touched. When # `create_all` meets a dependency cycle it can't sort -- and # library_files / library_folders / print_archives are exactly # such a cycle -- it falls back to emitting those tables' keys # as separate ALTER TABLE ... ADD FOREIGN KEY statements read # straight from that property. Twelve constraints survived, # including library_files.folder_id, and because the same cycle # also drops the ordering edge from `sorted_tables` the child # table was imported before its parent and the restore died on # a ForeignKeyViolationError. # # Dropping them from pg_constraint instead is indifferent to how # create_all chose to emit them, so a future model cycle cannot # reintroduce this. It also keeps the app's global Base.metadata # untouched: the old code only put the constraints back *after* # the transaction, so a failure in here left the running process # with an FK-less metadata until restart. await conn.execute( text( "DO $$ DECLARE r RECORD; BEGIN " "FOR r IN (SELECT conrelid::regclass AS tbl, conname FROM pg_constraint " "WHERE contype = 'f' AND connamespace = 'public'::regnamespace) LOOP " "EXECUTE 'ALTER TABLE ' || r.tbl || ' DROP CONSTRAINT ' || quote_ident(r.conname); " "END LOOP; END $$;" ) ) # Phase 2: Import data (no FKs to worry about) async with pg_engine.begin() as conn: # Import each table in dependency order (parents before children) for table_name in sorted_tables: rows = src.execute(f"SELECT * FROM {table_name}").fetchall() # noqa: S608 # nosec B608 if not rows: continue # Filter to columns that exist in the Postgres table src_columns = rows[0].keys() pg_table = metadata.tables.get(table_name) pg_columns = {c.name for c in pg_table.columns} if pg_table is not None else set() columns = [c for c in src_columns if c in pg_columns] if not columns: continue col_list = ", ".join(columns) param_list = ", ".join(f":{c}" for c in columns) # ON CONFLICT DO NOTHING handles duplicate rows from SQLite (which doesn't enforce unique constraints) insert_sql = text(f"INSERT INTO {table_name} ({col_list}) VALUES ({param_list}) ON CONFLICT DO NOTHING") # noqa: S608 # nosec B608 # Identify columns that need type conversion (SQLite stores booleans # as int and datetimes as str — asyncpg requires native Python types) from datetime import datetime as dt bool_columns = set() datetime_columns = set() not_null_defaults = {} # col_name -> default value for NOT NULL columns if pg_table is not None: for col in pg_table.columns: if col.name not in columns: continue col_type = str(col.type) if col_type == "BOOLEAN": bool_columns.add(col.name) elif col_type in ("DATETIME", "TIMESTAMP WITHOUT TIME ZONE", "TIMESTAMP WITH TIME ZONE"): datetime_columns.add(col.name) # Track NOT NULL columns with defaults — older backups may have NULL # for columns added after the backup was created if not col.nullable: if col.default is not None: default = col.default.arg if callable(default): default = default(None) not_null_defaults[col.name] = default elif col.server_default is not None: # server_default=func.now() → use current timestamp if col.name in datetime_columns: not_null_defaults[col.name] = "__now__" else: # Try to extract literal server default sd = str(col.server_default.arg) if hasattr(col.server_default, "arg") else None if sd is not None: not_null_defaults[col.name] = sd now = dt.now() def _convert_row( row, cols=columns, bools=bool_columns, dts=datetime_columns, nn_defaults=not_null_defaults, _now=now ): result = {} for c in cols: val = row[c] if val is None and c in nn_defaults: val = _now if nn_defaults[c] == "__now__" else nn_defaults[c] if val is not None: if c in bools: val = bool(val) elif c in dts and isinstance(val, str): try: val = dt.fromisoformat(val) except ValueError: pass result[c] = val return result batch = [_convert_row(row) for row in rows] await conn.execute(insert_sql, batch) logger.info("Imported %d rows into %s", len(batch), table_name) # Reset sequences to max(id) + 1 for each table with an id column for table_name in sorted_tables: try: async with conn.begin_nested(): result = await conn.execute(text(f"SELECT MAX(id) FROM {table_name}")) # noqa: S608 # nosec B608 max_id = result.scalar() if max_id is not None: seq_name = f"{table_name}_id_seq" await conn.execute(text(f"SELECT setval('{seq_name}', {max_id})")) # noqa: S608 except Exception: pass # Table may not have an id column or sequence src.close() logger.info("Cross-database import complete: %d tables imported", len(tables_to_import)) # Recreate FK constraints from ORM metadata, which Phase 1 left intact. # Use individual transactions so orphaned SQLite data doesn't block valid FKs. from sqlalchemy.schema import AddConstraint failed_fks = [] for table in metadata.sorted_tables: for fk in table.foreign_key_constraints: try: async with pg_engine.begin() as fk_conn: await fk_conn.execute(AddConstraint(fk)) except Exception as e: # Name the constraint by what it links, not by `fk.name`: # these are unnamed in the ORM, so that field is None and # the warning used to read "print_archives.None" for every # one of the five keys on that table -- unusable for # working out which rows to go and look at. cols = ", ".join(c.name for c in fk.columns) target = fk.elements[0].target_fullname if fk.elements else "unknown" failed_fks.append(f"{table.name}({cols}) -> {target}") # Postgres puts the offending key in a DETAIL line; it # names the exact orphan value, which is the one thing # that turns this into an actionable report. detail = next( (ln.strip() for ln in str(e).splitlines() if ln.startswith("DETAIL:")), str(e).splitlines()[0] if str(e) else e.__class__.__name__, ) logger.info("FK %s(%s) -> %s not restored: %s", table.name, cols, target, detail) if failed_fks: logger.warning( "Could not restore %d FK constraints (orphaned data in the backup): %s. " "The data is restored and usable; those columns are simply no longer " "enforced. See the INFO lines above for the offending key in each case.", len(failed_fks), ", ".join(failed_fks), ) finally: await pg_engine.dispose() @router.post("/restore") async def restore_backup( file: UploadFile = File(...), db: AsyncSession = Depends(get_db), _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_RESTORE), ): """Restore from a complete backup ZIP. Replaces the database and all data directories from the backup ZIP. Requires a restart after restore. """ import shutil import tempfile from fastapi import HTTPException from backend.app.core.database import close_all_connections, init_db, reinitialize_database from backend.app.core.db_dialect import is_sqlite from backend.app.services.virtual_printer import virtual_printer_manager base_dir = app_settings.base_dir with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) # 1. Read and extract ZIP content = await file.read() # Check if it's a valid ZIP if not file.filename or not file.filename.endswith(".zip"): raise HTTPException(400, "Invalid backup file: must be a .zip file") try: with zipfile.ZipFile(io.BytesIO(content), "r") as zf: for name in zf.namelist(): # Reject path-traversal payloads: any entry whose resolved # path escapes temp_path would allow writing arbitrary files # on the host (ZipSlip / CVE-2006-5456). dest = ( temp_path / name ).resolve() # SEC-PATH-OK: is_relative_to containment check below before extractall # is_relative_to (Python 3.9+) covers both relative # path-traversal (../etc/passwd) and absolute-path overrides # (/etc/passwd) — str.startswith was vulnerable to # prefix-collision attacks (e.g. /tmp/abc_evil/file passing # a /tmp/abc prefix check). if not dest.is_relative_to(temp_path.resolve()): raise HTTPException(400, f"Invalid backup: unsafe path in ZIP: {name!r}") zf.extractall(temp_path) except zipfile.BadZipFile: raise HTTPException(400, "Invalid backup file: not a valid ZIP") # 2. Validate backup backup_db = temp_path / "bambuddy.db" if not backup_db.exists(): raise HTTPException(400, "Invalid backup: missing bambuddy.db") try: import asyncio # 3. Stop virtual printer if running (releases file locks) try: if virtual_printer_manager.is_enabled: logger.info("Stopping virtual printer for restore...") await virtual_printer_manager.configure(enabled=False) await asyncio.sleep(1) except Exception as e: logger.warning("Failed to stop virtual printer: %s", e) # 3b. Pause timer-based background services BEFORE the DB swap. # close_all_connections() below only disposes the engine's pool, # not the asyncio tasks that opened sessions from it. The print # scheduler (30 s cadence), smart-plug snapshot loop (30 s), and # notification digest loop all # wake up and call async_session(), which lazily re-creates a # pool connection holding RowExclusiveLock on print_queue / # smart_plug_energy_snapshots / etc. The DROP TABLE CASCADE # pass in the PostgreSQL restore path needs AccessExclusiveLock # on every public table, producing an AB/BA deadlock and a # full restore rollback. Successful restore already requires a # container restart, so we don't restart the services here. try: from backend.app.services.notification_service import notification_service from backend.app.services.print_scheduler import scheduler as print_scheduler from backend.app.services.smart_plug_manager import smart_plug_manager logger.info("Pausing background services for restore...") print_scheduler.stop() smart_plug_manager.stop_scheduler() notification_service.stop_digest_scheduler() # In-flight loop iterations need a moment to commit + release # their DB sessions before we dispose() the engine pool. await asyncio.sleep(1.0) except Exception as e: logger.warning("Could not cleanly pause background services: %s", e) # 4. Close current database connections logger.info("Closing database connections...") await close_all_connections() # B1: Restore the MFA encryption key file BEFORE the database swap. # If the key write fails (OSError, RO disk, full disk, EACCES) we # can still abort while the live DB is intact. Doing this AFTER the # DB swap would leave the database with rows encrypted under the # backup's key but the running install holding only the old key — # every encrypted secret becomes unrecoverable. from backend.app.core.paths import resolve_data_dir mfa_key_src = temp_path / ".mfa_encryption_key" if mfa_key_src.exists() and mfa_key_src.is_file(): dst_key = resolve_data_dir() / ".mfa_encryption_key" tmp_key = dst_key.parent / ".mfa_encryption_key.restore-tmp" try: dst_key.parent.mkdir(parents=True, exist_ok=True) # S1: atomic write with restrictive mode from creation. # O_TRUNC because a stale tmp may exist from a prior # failed restore attempt — we want to overwrite it. fd = os.open(str(tmp_key), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) try: os.write(fd, mfa_key_src.read_bytes()) finally: os.close(fd) # POSIX rename(2) — atomic when source/dest are on the # same filesystem (we're staying inside dst_key.parent). os.replace(str(tmp_key), str(dst_key)) # S9: warn if the FS doesn't enforce 0o600 actual_mode = dst_key.stat().st_mode & 0o777 if actual_mode != 0o600: logger.warning( "Restored MFA key file %s: filesystem did not enforce 0o600 " "(actual: 0o%o). Key may be world-readable on Windows / SMB / FUSE.", dst_key, actual_mode, ) logger.info("Restored .mfa_encryption_key from backup") except OSError as e: logger.error( "Could not write restored MFA key file to %s: %s — " "aborting BEFORE database swap (DB unchanged).", dst_key, e, exc_info=True, ) raise HTTPException( status_code=500, detail=("Restore aborted: MFA key write failed. Database is unchanged. Check server logs."), ) from e # 5. Replace database logger.info("Restoring database from backup...") if is_sqlite(): db_path = Path(app_settings.database_url.replace("sqlite+aiosqlite:///", "")) # Use SQLite's online backup API instead of shutil.copy2. # The pragma at database.py:19 runs the live DB in WAL mode, # which means a naive file copy is unsafe: anything written # to the live DB before this call that hasn't been # checkpointed yet (seed_default_groups + init_db on first # start, plus whatever background heartbeats wrote during # the request window) sits in bambuddy.db-wal with valid # checksums. The route handler's own `db: Depends(get_db)` # session also keeps a connection checked out across # engine.dispose(), holding fds to the WAL inode. With # `shutil.copy2` SQLite finds the stale WAL on the next # open and silently re-applies those page-level writes on # top of the restored DB, partially clobbering it with # fresh-install state — the user sees a "successful" # restore where most rows and settings have reverted to # defaults (#1211 / #668). The page-by-page backup API # opens both DBs as real SQLite connections, takes the # right locks, and routes new pages through the live DB's # own WAL — so concurrent open sessions see their own # snapshot until they close (transaction isolation) but # can't corrupt the restored state. import sqlite3 src_conn = sqlite3.connect(str(backup_db)) try: dst_conn = sqlite3.connect(str(db_path)) try: src_conn.backup(dst_conn) finally: dst_conn.close() finally: src_conn.close() else: # Import SQLite backup into PostgreSQL logger.info("Importing SQLite backup into PostgreSQL...") await _import_sqlite_to_postgres(backup_db, app_settings.database_url) # 6. Replace data directories # For Docker compatibility: clear contents then copy (don't delete mount points) dirs_to_restore = [ ("archive", base_dir / "archive"), ("virtual_printer", base_dir / "virtual_printer"), ("plate_calibration", app_settings.plate_calibration_dir), ("icons", base_dir / "icons"), ("projects", base_dir / "projects"), ] skipped_dirs = [] for name, dest_dir in dirs_to_restore: src_dir = ( temp_path / name ) # SEC-PATH-OK: name iterates the dirs_to_restore tuple of constant strings ("archive", "virtual_printer", ...) if src_dir.exists(): logger.info("Restoring %s directory...", name) try: # Clear destination contents (not the dir itself - may be Docker mount) if dest_dir.exists(): for item in dest_dir.iterdir(): try: if item.is_dir(): shutil.rmtree(item) else: item.unlink() except OSError as e: logger.warning("Could not delete %s: %s", item, e) else: dest_dir.mkdir(parents=True, exist_ok=True) # Copy contents from backup for item in src_dir.iterdir(): dest_item = dest_dir / item.name if item.is_dir(): shutil.copytree(item, dest_item) else: shutil.copy2(item, dest_item) except OSError as e: logger.warning("Could not restore %s directory: %s", name, e) skipped_dirs.append(name) # 7. Reset the encryption singleton so the migration that runs # inside init_db() picks up the restored key file (if a new one # was written above). Without this reset, _get_fernet would # return the cached Fernet instance built from the previous key. import backend.app.core.encryption as _enc_mod _enc_mod._fernet_instance = None _enc_mod._key_source = None _enc_mod._warn_shown = False # 8. Reinitialize the database engine and apply schema migrations so that # tables added after the backup was created (e.g. ams_labels) exist # immediately, without requiring a manual restart. await reinitialize_database() await init_db() logger.info("Restore complete - restart required") message = "Backup restored successfully. Please restart Bambuddy for changes to take effect." if skipped_dirs: message += f" Note: Some directories could not be restored ({', '.join(skipped_dirs)})." return { "success": True, "message": message, } except HTTPException: # Preserve specific HTTP error responses raised inside the restore # body (e.g. the key-write OSError → 500). The blanket # except Exception below would otherwise swallow them and replace # the operator-facing detail with a generic message. raise except Exception as e: logger.error("Restore failed: %s", e, exc_info=True) return JSONResponse( status_code=500, content={"success": False, "message": "Restore failed. Check server logs for details."}, ) @router.get("/network-interfaces") async def get_network_interfaces( _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ), ): """Get available network interfaces with all IPs (primary + aliases).""" from backend.app.services.network_utils import get_all_interface_ips interfaces = get_all_interface_ips() return {"interfaces": interfaces} @router.get("/virtual-printer/models") async def get_virtual_printer_models( _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ), ): """Get available virtual printer models.""" from backend.app.services.virtual_printer import ( DEFAULT_VIRTUAL_PRINTER_MODEL, VIRTUAL_PRINTER_MODELS, ) return { "models": VIRTUAL_PRINTER_MODELS, "default": DEFAULT_VIRTUAL_PRINTER_MODEL, } @router.get("/virtual-printer") async def get_virtual_printer_settings( db: AsyncSession = Depends(get_db), _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ), ): """Get virtual printer settings and status.""" from backend.app.services.virtual_printer import ( DEFAULT_VIRTUAL_PRINTER_MODEL, virtual_printer_manager, ) enabled = await get_setting(db, "virtual_printer_enabled") access_code = await get_setting(db, "virtual_printer_access_code") mode = await get_setting(db, "virtual_printer_mode") model = await get_setting(db, "virtual_printer_model") target_printer_id = await get_setting(db, "virtual_printer_target_printer_id") remote_interface_ip = await get_setting(db, "virtual_printer_remote_interface_ip") tailscale_disabled_raw = await get_setting(db, "virtual_printer_tailscale_disabled") archive_name_source = await get_setting(db, "virtual_printer_archive_name_source") from backend.app.models.virtual_printer import VP_MODE_ARCHIVE, normalize_vp_mode return { "enabled": enabled == "true" if enabled else False, "access_code_set": bool(access_code), # Normalize on read so older settings rows (with `immediate` / # `print_queue`) come out as `archive` / `queue` for the frontend. "mode": normalize_vp_mode(mode) or VP_MODE_ARCHIVE, "model": model or DEFAULT_VIRTUAL_PRINTER_MODEL, "target_printer_id": int(target_printer_id) if target_printer_id else None, "remote_interface_ip": remote_interface_ip or "", "tailscale_disabled": tailscale_disabled_raw == "true" if tailscale_disabled_raw else True, "archive_name_source": archive_name_source if archive_name_source in ("metadata", "filename") else "metadata", "status": virtual_printer_manager.get_status(), } @router.put("/virtual-printer") async def update_virtual_printer_settings( enabled: bool = None, access_code: str = None, mode: str = None, model: str = None, target_printer_id: int = None, remote_interface_ip: str = None, tailscale_disabled: bool = None, archive_name_source: str = None, db: AsyncSession = Depends(get_db), _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE), ): """Update virtual printer settings and restart services if needed. For proxy mode with SSDP proxy (dual-homed setup): - remote_interface_ip: IP of interface on slicer's network (LAN B) - Local interface is auto-detected based on target printer IP """ from sqlalchemy import select from backend.app.models.printer import Printer from backend.app.services.virtual_printer import ( DEFAULT_VIRTUAL_PRINTER_MODEL, VIRTUAL_PRINTER_MODELS, virtual_printer_manager, ) # Get current values current_enabled = await get_setting(db, "virtual_printer_enabled") == "true" current_access_code = await get_setting(db, "virtual_printer_access_code") or "" # Default to `archive` (the canonical name) but tolerate legacy `immediate` # in the stored value — normalized later before validation. current_mode = await get_setting(db, "virtual_printer_mode") or "archive" current_model = await get_setting(db, "virtual_printer_model") or DEFAULT_VIRTUAL_PRINTER_MODEL current_target_id_str = await get_setting(db, "virtual_printer_target_printer_id") current_target_id = int(current_target_id_str) if current_target_id_str else None current_remote_iface = await get_setting(db, "virtual_printer_remote_interface_ip") or "" current_ts_disabled_raw = await get_setting(db, "virtual_printer_tailscale_disabled") # Default True (opt-in) when the setting has never been saved — matches the model default. current_ts_disabled = current_ts_disabled_raw == "true" if current_ts_disabled_raw else True # Apply updates new_enabled = enabled if enabled is not None else current_enabled new_access_code = access_code if access_code is not None else current_access_code new_mode = mode if mode is not None else current_mode new_model = model if model is not None else current_model new_target_id = target_printer_id if target_printer_id is not None else current_target_id new_remote_iface = remote_interface_ip if remote_interface_ip is not None else current_remote_iface new_ts_disabled = tailscale_disabled if tailscale_disabled is not None else current_ts_disabled # Validate mode. Canonical wire values are `archive` / `review` / `queue` # / `proxy`; legacy `immediate` and `print_queue` are accepted as aliases # and translated before storage so support bundles stop showing the old # confusing pair (#1429 mode-label discrepancy). from backend.app.models.virtual_printer import VP_MODE_VALUES, normalize_vp_mode canonical_mode = normalize_vp_mode(new_mode) if canonical_mode not in VP_MODE_VALUES: return JSONResponse( status_code=400, content={ "detail": f"Mode must be one of: {', '.join(VP_MODE_VALUES)}", }, ) new_mode = canonical_mode # Validate archive_name_source if archive_name_source is not None and archive_name_source not in ("metadata", "filename"): return JSONResponse( status_code=400, content={"detail": "archive_name_source must be 'metadata' or 'filename'"}, ) # Validate model if model is not None and model not in VIRTUAL_PRINTER_MODELS: return JSONResponse( status_code=400, content={"detail": f"Invalid model. Must be one of: {', '.join(VIRTUAL_PRINTER_MODELS.keys())}"}, ) # Mode-specific validation and printer lookup target_printer_ip = "" target_printer_serial = "" if new_mode == "proxy": # Proxy mode requires target printer when enabling if new_enabled and not new_target_id: # If just switching to proxy mode (not explicitly enabling), auto-disable if enabled is None: new_enabled = False else: return JSONResponse( status_code=400, content={"detail": "Target printer is required for proxy mode"}, ) # Look up printer IP and serial if we have a target if new_target_id: result = await db.execute(select(Printer).where(Printer.id == new_target_id)) printer = result.scalar_one_or_none() if not printer: return JSONResponse( status_code=400, content={"detail": f"Printer with ID {new_target_id} not found"}, ) target_printer_ip = printer.ip_address target_printer_serial = printer.serial_number # Access code not required for proxy mode else: # Non-proxy modes require access code when enabling if new_enabled and not new_access_code: # If just switching modes (not explicitly enabling), auto-disable if enabled is None: new_enabled = False else: return JSONResponse( status_code=400, content={"detail": "Access code is required when enabling virtual printer"}, ) # Validate access code length (Bambu Studio requires exactly 8 characters) if access_code is not None and access_code and len(access_code) != 8: return JSONResponse( status_code=400, content={"detail": "Access code must be exactly 8 characters"}, ) # Save settings await set_setting(db, "virtual_printer_enabled", "true" if new_enabled else "false") if access_code is not None: await set_setting(db, "virtual_printer_access_code", access_code) await set_setting(db, "virtual_printer_mode", new_mode) if model is not None: await set_setting(db, "virtual_printer_model", model) if target_printer_id is not None: await set_setting(db, "virtual_printer_target_printer_id", str(target_printer_id)) if remote_interface_ip is not None: await set_setting(db, "virtual_printer_remote_interface_ip", remote_interface_ip) if tailscale_disabled is not None: await set_setting(db, "virtual_printer_tailscale_disabled", "true" if tailscale_disabled else "false") if archive_name_source is not None: await set_setting(db, "virtual_printer_archive_name_source", archive_name_source) # Propagate tailscale_disabled to the first VirtualPrinter row so sync_from_db() picks it up if tailscale_disabled is not None: from backend.app.models.virtual_printer import VirtualPrinter as VPModel vp_result = await db.execute(select(VPModel).order_by(VPModel.position).limit(1)) first_vp = vp_result.scalar_one_or_none() if first_vp is not None: first_vp.tailscale_disabled = new_ts_disabled await db.commit() db.expire_all() # Reconfigure virtual printer try: await virtual_printer_manager.configure( enabled=new_enabled, access_code=new_access_code, mode=new_mode, model=new_model, target_printer_ip=target_printer_ip, target_printer_serial=target_printer_serial, remote_interface_ip=new_remote_iface, ) except ValueError as e: logger.warning("Virtual printer configuration validation error: %s", e) return JSONResponse( status_code=400, content={"detail": "Invalid virtual printer configuration. Check the provided values."}, ) except Exception as e: logger.error("Failed to configure virtual printer: %s", e, exc_info=True) return JSONResponse( status_code=500, content={"detail": "Failed to configure virtual printer. Check server logs for details."}, ) return await get_virtual_printer_settings(db) # ============================================================================= # MQTT Relay Settings # ============================================================================= @router.get("/mqtt/status") async def get_mqtt_status( _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ), ): """Get MQTT relay connection status.""" from backend.app.services.mqtt_relay import mqtt_relay return mqtt_relay.get_status()