Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
120 changes: 110 additions & 10 deletions micro/config_loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,21 @@ def generate_hostname():
"button": ("pin", "debounce_ms"),
}
_KNOWN_SECTIONS = tuple(DEFAULT_CONFIG.keys())
_SUPPORTED_GPIO_PINS = tuple(range(14)) + (20, 21)
MAX_API_REQUEST_TIMEOUT_SECONDS = 10
_FIELD_RANGES = {
("wifi", "timeout_seconds"): (1, 120),
# app_main's watchdog expires after 60 seconds. Keep blocking API calls
# short enough that request setup and response reads retain headroom.
("api", "request_timeout_seconds"): (1, MAX_API_REQUEST_TIMEOUT_SECONDS),
("printer", "poll_interval_seconds"): (1, 3600),
("led", "flash_interval_ms"): (10, 60000),
("button", "debounce_ms"): (0, 5000),
}
_ALLOWED_VALUES = {
("button", "pull"): ("", "none", "up", "down"),
("button", "trigger"): ("rising", "falling", "both"),
}


def load_config(path="config.json"):
Expand Down Expand Up @@ -123,8 +138,7 @@ def is_runtime_config_ready(config):
def save_config(path, config):
"""Replace a config atomically, retaining the previous copy until verified."""
path = str(path)
if not isinstance(config, dict):
raise ValueError("Configuration must be an object")
validate_config(config)

temporary_path = path + ".tmp"
backup_path = path + ".bak"
Expand Down Expand Up @@ -186,27 +200,105 @@ def _load_config_file(path):


def _is_valid_config_shape(config):
if not isinstance(config, dict):
try:
validate_config(config)
except (TypeError, ValueError):
return False
return True


def validate_config(config):
"""Reject settings that would fail hardware or network initialization."""
if not isinstance(config, dict):
raise ValueError("Configuration must be an object")

for section in _KNOWN_SECTIONS:
if section in config and not isinstance(config[section], dict):
return False
raise ValueError("{} must be an object".format(section))

for section, fields in _STRING_FIELDS.items():
values = config.get(section, {})
for field in fields:
if field in values and not isinstance(values[field], str):
return False
if field not in values:
continue
value = values[field]
if not isinstance(value, str):
raise ValueError("{}.{} must be text".format(section, field))
allowed = _ALLOWED_VALUES.get((section, field))
if allowed and value.lower() not in allowed:
raise ValueError(
"{}.{} must be one of: {}".format(
section, field, ", ".join(allowed)
)
)

wifi = config.get("wifi", {})
for field, maximum in (("ssid", 32), ("hostname", 63)):
value = wifi.get(field)
if value is not None and len(value.encode("utf-8")) > maximum:
raise ValueError(
"wifi.{} must be at most {} UTF-8 bytes".format(field, maximum)
)
hostname = wifi.get("hostname")
if hostname is not None:
if not hostname or len(hostname) > 63:
raise ValueError("wifi.hostname must be between 1 and 63 characters")
if not _is_ascii_alphanumeric(hostname[0]) or not _is_ascii_alphanumeric(
hostname[-1]
):
raise ValueError("wifi.hostname must start and end with a letter or number")
if any(
not _is_ascii_alphanumeric(character) and character != "-"
for character in hostname
):
raise ValueError(
"wifi.hostname may contain only letters, numbers, and hyphens"
)
station_password = wifi.get("password")
if station_password:
length = len(station_password.encode("utf-8"))
is_hex_psk = length == 64 and all(
character in "0123456789abcdefABCDEF" for character in station_password
)
if (length < 8 or length > 63) and not is_hex_psk:
raise ValueError(
"wifi.password must be empty, 8 to 63 bytes, or a 64-digit hex key"
)

ap_password = wifi.get("ap_password")
if ap_password:
length = len(ap_password.encode("utf-8"))
if length < 8 or length > 63:
raise ValueError("wifi.ap_password must be empty or 8 to 63 bytes")
web_password = config.get("web", {}).get("password")
if web_password is not None and not web_password:
raise ValueError("web.password must not be empty")

for section, fields in _NUMBER_FIELDS.items():
values = config.get(section, {})
for field in fields:
value = values.get(field)
if field in values and (
not isinstance(value, (int, float)) or isinstance(value, bool)
not isinstance(value, int) or isinstance(value, bool)
):
return False
raise ValueError("{}.{} must be an integer".format(section, field))
bounds = _FIELD_RANGES.get((section, field))
if field in values and bounds:
minimum, maximum = bounds
if value < minimum or value > maximum:
raise ValueError(
"{}.{} must be between {} and {}".format(
section, field, minimum, maximum
)
)

for section in ("led", "button"):
pin = config.get(section, {}).get("pin")
if pin is not None and pin not in _SUPPORTED_GPIO_PINS:
raise ValueError(
"{}.pin must be a GPIO supported by the shipped ESP32-C3 firmware"
.format(section)
)

if "id" in config.get("printer", {}):
printer_id = config["printer"]["id"]
Expand All @@ -215,9 +307,13 @@ def _is_valid_config_shape(config):
or isinstance(printer_id, bool)
or printer_id < 0
):
return False
raise ValueError("printer.id must be a non-negative integer or null")

return True
effective_config = _copy_dict(DEFAULT_CONFIG)
_deep_update(effective_config, config)
if effective_config["led"]["pin"] == effective_config["button"]["pin"]:
raise ValueError("led.pin and button.pin must use different GPIO pins")
return config


def _remove_file(path):
Expand Down Expand Up @@ -245,3 +341,7 @@ def _deep_update(target, source):
_deep_update(target[key], value)
else:
target[key] = value


def _is_ascii_alphanumeric(value):
return ("a" <= value <= "z") or ("A" <= value <= "Z") or ("0" <= value <= "9")
4 changes: 2 additions & 2 deletions micro/ota_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,9 +63,9 @@
"https://github.com/EdwardChamberlain/Bambutton/releases/latest/"
"download/bambutton-ota.json"
)
MAX_FILE_BYTES = 40 * 1024
MAX_FILE_BYTES = 36 * 1024
MAX_BUNDLE_BYTES = 128 * 1024
# Keep the downloaded JSON below the web server's 128 KiB request-body ceiling.
# Keep OTA payloads compatible with the 96 KiB reader in installed v1 loaders.
MAX_BUNDLE_BODY_BYTES = 96 * 1024
OTA_STAGING_OVERHEAD_BYTES = 48 * 1024

Expand Down
1 change: 1 addition & 0 deletions micro/web_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -292,6 +292,7 @@ def build_config(form, current_config):
web.get("password", DEFAULT_WEB_PASSWORD) or DEFAULT_WEB_PASSWORD,
)

config_loader.validate_config(config)
return config


Expand Down
10 changes: 10 additions & 0 deletions src/bambutton/gui.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,9 @@
else:
RESOURCE_ROOT = SOURCE_ROOT if (SOURCE_ROOT / "micro").exists() else PACKAGE_ROOT

if str(RESOURCE_ROOT) not in sys.path:
sys.path.insert(0, str(RESOURCE_ROOT))

MICRO_DIR = RESOURCE_ROOT / "micro"
CONFIG_EXAMPLE_PATH = MICRO_DIR / "config_example.json"
DEFAULT_FIRMWARE_DIR = RESOURCE_ROOT / "firmware"
Expand Down Expand Up @@ -536,6 +539,13 @@ def validate_config_file(path):
if not isinstance(config, dict):
raise ValueError("Configuration file must contain a JSON object.")

from micro.config_loader import validate_config as validate_firmware_config

try:
validate_firmware_config(config)
except ValueError as exc:
raise ValueError("Invalid configuration: {}".format(exc))

return config_path


Expand Down
108 changes: 108 additions & 0 deletions tests/test_config_loader.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
import json
import re

import pytest

from micro import config_loader


Expand Down Expand Up @@ -88,3 +90,109 @@ def test_non_numeric_printer_id_is_rejected_and_uses_safe_defaults(tmp_path):

assert config["printer"]["id"] is None
assert not config_loader.is_runtime_config_ready(config)


@pytest.mark.parametrize("value", ["invalid", "upside-down", 1])
def test_invalid_button_pull_recovers_to_safe_defaults(tmp_path, value):
config_path = tmp_path / "config.json"
config_path.write_text(json.dumps({"button": {"pull": value}}))

config = config_loader.load_config(config_path)

assert config["button"]["pull"] == "down"
assert not config_loader.is_runtime_config_ready(config)


@pytest.mark.parametrize(
"section,field,value",
[
("button", "pin", 4.5),
("button", "pin", 22),
("button", "pin", 17),
("led", "pin", -1),
("api", "request_timeout_seconds", 0),
("api", "request_timeout_seconds", 11),
("printer", "poll_interval_seconds", 1.5),
("wifi", "timeout_seconds", 121),
("wifi", "hostname", "bad hostname"),
],
)
def test_invalid_hardware_and_timer_values_are_rejected(
tmp_path, section, field, value
):
config_path = tmp_path / "config.json"
config_path.write_text(json.dumps({section: {field: value}}))

config = config_loader.load_config(config_path)

assert config == config_loader._copy_dict(config_loader.DEFAULT_CONFIG)


def test_invalid_active_values_recover_from_a_valid_backup(tmp_path):
config_path = tmp_path / "config.json"
backup_path = tmp_path / "config.json.bak"
config_path.write_text(json.dumps({"button": {"trigger": "sideways"}}))
backup_path.write_text(json.dumps({"button": {"trigger": "both"}}))

config = config_loader.load_config(config_path)

assert config["button"]["trigger"] == "both"


def test_config_validation_reports_the_invalid_field():
with pytest.raises(ValueError, match="button.pull"):
config_loader.validate_config({"button": {"pull": "invalid"}})


@pytest.mark.parametrize("section", ("led", "button"))
@pytest.mark.parametrize("pin", (14, 15, 16, 17, 18, 19))
def test_gpio_pins_missing_from_shipped_firmware_are_rejected(section, pin):
with pytest.raises(ValueError, match="supported by the shipped ESP32-C3"):
config_loader.validate_config({section: {"pin": pin}})


@pytest.mark.parametrize("pin", (0, 3, 13, 20, 21))
def test_gpio_pins_supported_by_shipped_firmware_are_accepted(pin):
config_loader.validate_config({"led": {"pin": pin}})


def test_invalid_gpio_in_active_config_recovers_from_safe_backup(tmp_path):
config_path = tmp_path / "config.json"
backup_path = tmp_path / "config.json.bak"
config_path.write_text(json.dumps({"led": {"pin": 17}}))
backup_path.write_text(json.dumps({"led": {"pin": 3}, "button": {"pin": 4}}))

config = config_loader.load_config(config_path)

assert config["led"]["pin"] == 3
assert config["button"]["pin"] == 4


def test_api_timeout_is_bounded_below_the_watchdog_budget():
config_loader.validate_config(
{
"api": {
"request_timeout_seconds": config_loader.MAX_API_REQUEST_TIMEOUT_SECONDS
}
}
)
with pytest.raises(ValueError, match="request_timeout_seconds must be between"):
config_loader.validate_config(
{
"api": {
"request_timeout_seconds": (
config_loader.MAX_API_REQUEST_TIMEOUT_SECONDS + 1
)
}
}
)


def test_station_psk_64_digit_hex_is_allowed_but_ap_password_is_not():
config_loader.validate_config({"wifi": {"password": "a" * 64}})
with pytest.raises(ValueError, match="ap_password must be empty or 8 to 63 bytes"):
config_loader.validate_config({"wifi": {"ap_password": "a" * 64}})


def test_ap_password_maximum_supported_length_is_accepted():
config_loader.validate_config({"wifi": {"ap_password": "a" * 63}})
10 changes: 7 additions & 3 deletions tests/test_config_startup.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
import copy
import sys
from pathlib import Path
from types import SimpleNamespace
Expand All @@ -13,8 +12,13 @@ class StopMainLoop(BaseException):
pass


def test_incomplete_config_starts_setup_ap_without_printer_api_requests(monkeypatch):
config = copy.deepcopy(config_loader.DEFAULT_CONFIG)
def test_invalid_persisted_config_starts_setup_ap_without_printer_api_requests(
tmp_path, monkeypatch
):
config_path = tmp_path / "config.json"
config_path.write_text('{"button": {"pull": "invalid"}}')
config = config_loader.load_config(config_path)
assert config["button"]["pull"] == "down"
api_calls = []

class FakeNetwork:
Expand Down
22 changes: 22 additions & 0 deletions tests/test_gui.py
Original file line number Diff line number Diff line change
Expand Up @@ -297,6 +297,28 @@ def test_validate_config_file_rejects_invalid_files(tmp_path, contents, error):
gui.validate_config_file(config_path)


def test_validate_config_file_rejects_invalid_hardware_values(tmp_path):
config_path = tmp_path / "config.json"
config_path.write_text(json.dumps({"button": {"pull": "invalid"}}))

with pytest.raises(ValueError, match="button.pull"):
gui.validate_config_file(config_path)


def test_flash_board_rejects_invalid_config_before_erasing(tmp_path, monkeypatch):
firmware = tmp_path / "firmware.bin"
firmware.write_bytes(b"firmware")
config_path = tmp_path / "config.json"
config_path.write_text(json.dumps({"button": {"pull": "invalid"}}))
flashed = []
monkeypatch.setattr(gui, "flash_firmware", lambda path: flashed.append(path))

with pytest.raises(ValueError, match="button.pull"):
gui.flash_board(firmware, config_path)

assert flashed == []


def test_push_micro_files_installs_selected_file_as_config_json(tmp_path, monkeypatch):
micro_dir = tmp_path / "micro"
micro_dir.mkdir()
Expand Down
12 changes: 12 additions & 0 deletions tests/test_web_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -158,6 +158,18 @@ def test_save_config_rejects_non_object_config(tmp_path):
web_config.save_config(tmp_path / "config.json", [])


def test_save_config_rejects_invalid_hardware_values_before_writing(tmp_path):
path = tmp_path / "config.json"
original = base_config()
path.write_text(json.dumps(original))

with pytest.raises(ValueError, match="button.pull"):
web_config.save_config(path, {"button": {"pull": "invalid"}})

assert json.loads(path.read_text()) == original
assert not (tmp_path / "config.json.tmp").exists()


def test_config_page_contains_form_and_does_not_require_formatting_js():
page = web_config.render_config_page(base_config())

Expand Down
Loading