viewit/KX-Bridge-Release
12
18
Fork 11
Code Issues 5 Pull Requests Actions Packages Projects Releases 83 Wiki Activity

fix(config): don't crash the whole bridge on a typo'd numeric config value

Browse Source 
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).
This commit is contained in:
viewit
2026-08-04 15:31:46 +02:00
parent 5eeed97514
commit ce670f5f93
2 changed files with 109 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
config_loader.py
View File

@@ -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)

66
tests/test_config_loader_robustness.py Normal file
View File

@@ -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