From 60156fd5891518b79ea92c12a54cc33df83efbeb Mon Sep 17 00:00:00 2001 From: viewit Date: Tue, 4 Aug 2026 15:31:46 +0200 Subject: [PATCH] fix(config): don't crash the whole bridge on a typo'd numeric config value Found during a targeted code review, not from a user report. MQTT_PORT, POLL_INTERVAL, and the other numeric module-level shortcuts in config_loader.py ran int(get(...)) unguarded at import time. A hand-edited config.ini with a typo (e.g. "mqtt_port = 98833x") raised an uncaught ValueError before the bridge even started, with a raw traceback instead of a usable diagnostic - list_printers() already guarded this exact class of input the same way, but the module-level constants didn't. Added _safe_int() (same try/except-with-fallback pattern) and applied it everywhere int() was called unguarded on a config value. Also wrapped migrate_env_to_config()'s filesystem writes (which also run at import time during first-run .env migration) in try/except, so a permission or disk-full error logs a clear message before re-raising instead of surfacing as a bare traceback pointing into configparser. New tests in tests/test_config_loader_robustness.py cover both, including an end-to-end subprocess test that imports config_loader against a malformed config.ini (a plain re-import wouldn't re-exercise the import-time code path due to Python's module caching). --- config_loader.py | 58 ++++++++++++++++------ tests/test_config_loader_robustness.py | 66 ++++++++++++++++++++++++++ 2 files changed, 109 insertions(+), 15 deletions(-) create mode 100644 tests/test_config_loader_robustness.py diff --git a/config_loader.py b/config_loader.py index eaa674a..729b33d 100644 --- a/config_loader.py +++ b/config_loader.py @@ -7,8 +7,11 @@ import os import sys import pathlib import configparser +import logging from typing import Optional +log = logging.getLogger("kobrax.config") + _BASE = pathlib.Path(sys.executable).parent if getattr(sys, "frozen", False) else pathlib.Path(__file__).parent CONFIG_SECTION_CONNECTION = "connection" @@ -116,7 +119,6 @@ def migrate_env_to_config(env_path: pathlib.Path, config_path: pathlib.Path): k, _, v = line.partition("=") env_vals[k.strip()] = v.strip() - config_path.parent.mkdir(parents=True, exist_ok=True) cfg = configparser.ConfigParser(interpolation=None) cfg[CONFIG_SECTION_CONNECTION] = { "printer_ip": env_vals.get("PRINTER_IP", ""), @@ -136,10 +138,20 @@ def migrate_env_to_config(env_path: pathlib.Path, config_path: pathlib.Path): cfg[CONFIG_SECTION_BRIDGE] = { "poll_interval": "3", } - with open(config_path, "w", encoding="utf-8") as f: - f.write("# KX-Bridge Konfigurationsdatei\n") - f.write("# Automatically migrated from .env\n\n") - cfg.write(f) + # This runs at module import time (see the "Laden" section below) - an + # uncaught mkdir/write failure (e.g. a read-only filesystem) would crash + # the whole bridge at startup with a raw traceback. Log a clear diagnostic + # before re-raising, so the actual cause (permissions, disk full) is + # visible instead of a bare stack trace pointing into configparser. + try: + config_path.parent.mkdir(parents=True, exist_ok=True) + with open(config_path, "w", encoding="utf-8") as f: + f.write("# KX-Bridge Konfigurationsdatei\n") + f.write("# Automatically migrated from .env\n\n") + cfg.write(f) + except OSError as e: + log.error("Failed to write migrated config.ini to %s: %s", config_path, e) + raise def find_config_path() -> pathlib.Path: @@ -445,9 +457,25 @@ def get(key: str, default: str = "") -> str: return os.environ.get(key, default) +def _safe_int(value: str, default: int) -> int: + """Falls back to `default` instead of raising on a non-numeric value. + + All of these run at module import time - an uncaught ValueError here + (e.g. from a hand-edited config.ini with a typo like `mqtt_port = 98833x`) + would crash the entire bridge before it even starts, with a raw traceback + instead of a clear diagnostic. list_printers() already guards this same + class of input the same way; this applies it to the module-level + shortcuts too.""" + try: + return int(value) + except (TypeError, ValueError): + log.warning("config: expected a number, got %r - using default %r", value, default) + return default + + # Frequently used shortcuts PRINTER_IP = get("PRINTER_IP", "") -MQTT_PORT = int(get("MQTT_PORT", "9883")) +MQTT_PORT = _safe_int(get("MQTT_PORT", "9883"), 9883) USERNAME = get("MQTT_USERNAME", "") PASSWORD = get("MQTT_PASSWORD", "") MODE_ID = get("MODE_ID", "") @@ -456,14 +484,14 @@ POWER_ON_URL = get("POWER_ON_URL", "") POWER_OFF_URL = get("POWER_OFF_URL", "") POWER_STATUS_URL = get("POWER_STATUS_URL", "") DEFAULT_AMS_SLOT = get("DEFAULT_AMS_SLOT", "auto") -AUTO_LEVELING = int(get("AUTO_LEVELING", "1")) -VIBRATION_COMPENSATION = int(get("VIBRATION_COMPENSATION", "0")) -CAMERA_ON_PRINT = int(get("CAMERA_ON_PRINT", "0")) -WEB_UPLOAD_WARNING = int(get("WEB_UPLOAD_WARNING", "1")) -DELETE_PRINTER_FILE_AFTER_PRINT = int(get("DELETE_PRINTER_FILE_AFTER_PRINT", "0")) -PRINT_START_DIALOG = int(get("PRINT_START_DIALOG", get("FILE_READY_DIALOG", "1"))) +AUTO_LEVELING = _safe_int(get("AUTO_LEVELING", "1"), 1) +VIBRATION_COMPENSATION = _safe_int(get("VIBRATION_COMPENSATION", "0"), 0) +CAMERA_ON_PRINT = _safe_int(get("CAMERA_ON_PRINT", "0"), 0) +WEB_UPLOAD_WARNING = _safe_int(get("WEB_UPLOAD_WARNING", "1"), 1) +DELETE_PRINTER_FILE_AFTER_PRINT = _safe_int(get("DELETE_PRINTER_FILE_AFTER_PRINT", "0"), 0) +PRINT_START_DIALOG = _safe_int(get("PRINT_START_DIALOG", get("FILE_READY_DIALOG", "1")), 1) SPOOLMAN_SERVER = get("SPOOLMAN_SERVER", "") -SPOOLMAN_SYNC_RATE = int(get("SPOOLMAN_SYNC_RATE", "0")) +SPOOLMAN_SYNC_RATE = _safe_int(get("SPOOLMAN_SYNC_RATE", "0"), 0) BRIDGE_HOST_IP = get("BRIDGE_HOST_IP", "") -POLL_INTERVAL = int(get("POLL_INTERVAL", "3")) -VERBOSE_HTTP_LOG = int(get("VERBOSE_HTTP_LOG", "0")) +POLL_INTERVAL = _safe_int(get("POLL_INTERVAL", "3"), 3) +VERBOSE_HTTP_LOG = _safe_int(get("VERBOSE_HTTP_LOG", "0"), 0) diff --git a/tests/test_config_loader_robustness.py b/tests/test_config_loader_robustness.py new file mode 100644 index 0000000..106f0d7 --- /dev/null +++ b/tests/test_config_loader_robustness.py @@ -0,0 +1,66 @@ +"""config_loader.py robustness fixes found during code review: + +1. _safe_int() must fall back to a default instead of raising - the + module-level numeric shortcuts (MQTT_PORT, POLL_INTERVAL, etc.) run this + at import time, so an uncaught ValueError there previously crashed the + entire bridge on startup if config.ini had a typo'd numeric value + (e.g. "mqtt_port = 98833x"), with a raw traceback instead of a clear + diagnostic. list_printers() already guarded this same class of input the + same way; this applies it to the module-level shortcuts too. +2. migrate_env_to_config() must log a clear error (not a bare traceback) + if writing the migrated config.ini fails (e.g. permission error). +""" +import subprocess +import sys +import textwrap + +import config_loader + + +def test_safe_int_returns_value_for_valid_numeric_string(): + assert config_loader._safe_int("42", 0) == 42 + + +def test_safe_int_falls_back_to_default_on_garbage(): + assert config_loader._safe_int("98833x", 9883) == 9883 + + +def test_safe_int_falls_back_to_default_on_empty_string(): + assert config_loader._safe_int("", 3) == 3 + + +def test_safe_int_falls_back_to_default_on_none(): + assert config_loader._safe_int(None, 5) == 5 + + +def test_config_loader_import_survives_malformed_config_ini(tmp_path): + """End-to-end regression guard: importing config_loader with a + config.ini containing a non-numeric mqtt_port must not raise - it must + fall back to the default instead. Run in a subprocess since + config_loader executes its migration/loading logic at import time and + Python caches modules, so a plain re-import in this test process + wouldn't actually re-exercise the import-time code path.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + (config_dir / "config.ini").write_text( + "[connection]\n" + "printer_ip = 192.168.1.50\n" + "mqtt_port = 98833x\n" # malformed - not a number + ) + + script = textwrap.dedent(f""" + import sys + sys.path.insert(0, {str(tmp_path.parent.parent)!r}) + import config_loader + config_loader._BASE = __import__("pathlib").Path({str(tmp_path)!r}) + config_loader._find_config_file = lambda: {str(config_dir / "config.ini")!r} \ + and __import__("pathlib").Path({str(config_dir / "config.ini")!r}) + import importlib + importlib.reload(config_loader) + assert config_loader.MQTT_PORT == 9883, config_loader.MQTT_PORT + print("OK") + """) + + result = subprocess.run([sys.executable, "-c", script], capture_output=True, text=True, timeout=10) + assert result.returncode == 0, f"stdout={result.stdout!r} stderr={result.stderr!r}" + assert "OK" in result.stdout