diff --git a/esphome/analyze_memory/__init__.py b/esphome/analyze_memory/__init__.py index d8c941e76fe..d8abc8bafbc 100644 --- a/esphome/analyze_memory/__init__.py +++ b/esphome/analyze_memory/__init__.py @@ -397,47 +397,38 @@ class MemoryAnalyzer: return pioenvs_dir return None - def _scan_cswtch_in_objects( - self, obj_dir: Path - ) -> dict[str, list[tuple[str, int]]]: - """Scan object files for CSWTCH symbols using a single nm invocation. + @staticmethod + def _parse_nm_cswtch_output( + output: str, + base_dir: Path | None, + cswtch_map: dict[str, list[tuple[str, int]]], + ) -> None: + """Parse nm output for CSWTCH symbols and add to cswtch_map. - Uses ``nm --print-file-name -S`` on all ``.o`` files at once. - Output format: ``/path/to/file.o:address size type name`` + Handles both ``.o`` files and ``.a`` archives. + + nm output formats:: + + .o files: /path/file.o:hex_addr hex_size type name + .a files: /path/lib.a:member.o:hex_addr hex_size type name + + For ``.o`` files, paths are made relative to *base_dir* when possible. + For ``.a`` archives (detected by ``:`` in the file portion), paths are + formatted as ``archive_stem/member.o`` (e.g. ``liblwip2-536-feat/lwip-esp.o``). Args: - obj_dir: Directory containing object files (.pioenvs//) - - Returns: - Dict mapping "CSWTCH$NNN:size" to list of (source_file, size) tuples. + output: Raw stdout from ``nm --print-file-name -S``. + base_dir: Base directory for computing relative paths of ``.o`` files. + Pass ``None`` when scanning archives outside the build tree. + cswtch_map: Dict to populate, mapping ``"CSWTCH$N:size"`` to source list. """ - cswtch_map: dict[str, list[tuple[str, int]]] = defaultdict(list) - - if not self.nm_path: - return cswtch_map - - # Find all .o files recursively, sorted for deterministic output - obj_files = sorted(obj_dir.rglob("*.o")) - if not obj_files: - return cswtch_map - - _LOGGER.debug("Scanning %d object files for CSWTCH symbols", len(obj_files)) - - # Single nm call with --print-file-name for all object files - result = run_tool( - [self.nm_path, "--print-file-name", "-S"] + [str(f) for f in obj_files], - timeout=30, - ) - if result is None or result.returncode != 0: - return cswtch_map - - for line in result.stdout.splitlines(): + for line in output.splitlines(): if "CSWTCH$" not in line: continue # Split on last ":" that precedes a hex address. - # nm --print-file-name format: filepath:hex_addr hex_size type name - # We split from the right: find the last colon followed by hex digits. + # For .o: "filepath.o" : "hex_addr hex_size type name" + # For .a: "filepath.a:member.o" : "hex_addr hex_size type name" parts_after_colon = line.rsplit(":", 1) if len(parts_after_colon) != 2: continue @@ -457,16 +448,89 @@ class MemoryAnalyzer: except ValueError: continue - # Get relative path from obj_dir for readability - try: - rel_path = str(Path(file_path).relative_to(obj_dir)) - except ValueError: + # Determine readable source path + # Use ".a:" to detect archive format (not bare ":" which matches + # Windows drive letters like "C:\...\file.o"). + if ".a:" in file_path: + # Archive format: "archive.a:member.o" → "archive_stem/member.o" + archive_part, member = file_path.rsplit(":", 1) + archive_name = Path(archive_part).stem + rel_path = f"{archive_name}/{member}" + elif base_dir is not None: + try: + rel_path = str(Path(file_path).relative_to(base_dir)) + except ValueError: + rel_path = file_path + else: rel_path = file_path key = f"{sym_name}:{size}" cswtch_map[key].append((rel_path, size)) - return cswtch_map + def _run_nm_cswtch_scan( + self, + files: list[Path], + base_dir: Path | None, + cswtch_map: dict[str, list[tuple[str, int]]], + ) -> None: + """Run nm on *files* and add any CSWTCH symbols to *cswtch_map*. + + Args: + files: Object (``.o``) or archive (``.a``) files to scan. + base_dir: Base directory for relative path computation (see + :meth:`_parse_nm_cswtch_output`). + cswtch_map: Dict to populate with results. + """ + if not self.nm_path or not files: + return + + _LOGGER.debug("Scanning %d files for CSWTCH symbols", len(files)) + + result = run_tool( + [self.nm_path, "--print-file-name", "-S"] + [str(f) for f in files], + timeout=30, + ) + if result is None or result.returncode != 0: + _LOGGER.debug( + "nm failed or timed out scanning %d files for CSWTCH symbols", + len(files), + ) + return + + self._parse_nm_cswtch_output(result.stdout, base_dir, cswtch_map) + + def _scan_cswtch_in_sdk_archives( + self, cswtch_map: dict[str, list[tuple[str, int]]] + ) -> None: + """Scan SDK library archives (.a) for CSWTCH symbols. + + Prebuilt SDK libraries (e.g. lwip, bearssl) are not compiled from source, + so their CSWTCH symbols only exist inside ``.a`` archives. Results are + merged into *cswtch_map* for keys not already found in ``.o`` files. + + The same source file (e.g. ``lwip-esp.o``) often appears in multiple + library variants (``liblwip2-536.a``, ``liblwip2-1460-feat.a``, etc.), + so results are deduplicated by member name. + """ + sdk_dirs = self._find_sdk_library_dirs() + if not sdk_dirs: + return + + sdk_archives = sorted(a for sdk_dir in sdk_dirs for a in sdk_dir.glob("*.a")) + + sdk_map: dict[str, list[tuple[str, int]]] = defaultdict(list) + self._run_nm_cswtch_scan(sdk_archives, None, sdk_map) + + # Merge SDK results, deduplicating by member name. + for key, sources in sdk_map.items(): + if key in cswtch_map: + continue + seen: dict[str, tuple[str, int]] = {} + for path, sz in sources: + member = Path(path).name + if member not in seen: + seen[member] = (path, sz) + cswtch_map[key] = list(seen.values()) def _source_file_to_component(self, source_file: str) -> str: """Map a source object file path to its component name. @@ -505,17 +569,25 @@ class MemoryAnalyzer: CSWTCH symbols are compiler-generated lookup tables for switch statements. They are local symbols, so the same name can appear in different object files. - This method scans .o files to attribute them to their source components. + This method scans .o files and SDK archives to attribute them to their + source components. """ obj_dir = self._find_object_files_dir() if obj_dir is None: _LOGGER.debug("No object files directory found, skipping CSWTCH analysis") return - # Scan object files for CSWTCH symbols - cswtch_map = self._scan_cswtch_in_objects(obj_dir) + # Scan build-dir object files for CSWTCH symbols + cswtch_map: dict[str, list[tuple[str, int]]] = defaultdict(list) + self._run_nm_cswtch_scan(sorted(obj_dir.rglob("*.o")), obj_dir, cswtch_map) + + # Also scan SDK library archives (.a) for CSWTCH symbols. + # Prebuilt SDK libraries (e.g. lwip, bearssl) are not compiled from source + # so their symbols only exist inside .a archives, not as loose .o files. + self._scan_cswtch_in_sdk_archives(cswtch_map) + if not cswtch_map: - _LOGGER.debug("No CSWTCH symbols found in object files") + _LOGGER.debug("No CSWTCH symbols found in object files or SDK archives") return # Collect CSWTCH symbols from the ELF (already parsed in sections) diff --git a/esphome/components/logger/logger_zephyr.cpp b/esphome/components/logger/logger_zephyr.cpp index ef1702c5c1c..1fc0acd5733 100644 --- a/esphome/components/logger/logger_zephyr.cpp +++ b/esphome/components/logger/logger_zephyr.cpp @@ -68,7 +68,7 @@ void HOT Logger::write_msg_(const char *msg, uint16_t len) { #ifdef CONFIG_PRINTK // Requires the debug component and an active SWD connection. // It is used for pyocd rtt -t nrf52840 - k_str_out(const_cast(msg), len); + printk("%.*s", static_cast(len), msg); #endif if (this->uart_dev_ == nullptr) { return; diff --git a/esphome/components/lps22/lps22.cpp b/esphome/components/lps22/lps22.cpp index 526286ba726..7fc5774b08c 100644 --- a/esphome/components/lps22/lps22.cpp +++ b/esphome/components/lps22/lps22.cpp @@ -38,22 +38,29 @@ void LPS22Component::dump_config() { LOG_UPDATE_INTERVAL(this); } +static constexpr uint32_t INTERVAL_READ = 0; + void LPS22Component::update() { uint8_t value = 0x00; this->read_register(CTRL_REG2, &value, 1); value |= CTRL_REG2_ONE_SHOT_MASK; this->write_register(CTRL_REG2, &value, 1); - this->set_retry(READ_INTERVAL, READ_ATTEMPTS, [this](uint8_t _) { return this->try_read_(); }); + this->read_attempts_remaining_ = READ_ATTEMPTS; + this->set_interval(INTERVAL_READ, READ_INTERVAL, [this]() { this->try_read_(); }); } -RetryResult LPS22Component::try_read_() { +void LPS22Component::try_read_() { uint8_t value = 0x00; this->read_register(STATUS, &value, 1); const uint8_t expected_status_mask = STATUS_T_DA_MASK | STATUS_P_DA_MASK; if ((value & expected_status_mask) != expected_status_mask) { ESP_LOGD(TAG, "STATUS not ready: %x", value); - return RetryResult::RETRY; + if (--this->read_attempts_remaining_ == 0) { + this->cancel_interval(INTERVAL_READ); + } + return; } + this->cancel_interval(INTERVAL_READ); if (this->temperature_sensor_ != nullptr) { uint8_t t_buf[2]{0}; @@ -68,7 +75,6 @@ RetryResult LPS22Component::try_read_() { uint32_t p_lsb = encode_uint24(p_buf[2], p_buf[1], p_buf[0]); this->pressure_sensor_->publish_state(PRESSURE_SCALE * static_cast(p_lsb)); } - return RetryResult::DONE; } } // namespace lps22 diff --git a/esphome/components/lps22/lps22.h b/esphome/components/lps22/lps22.h index 549ea524ea9..95ee4ad4427 100644 --- a/esphome/components/lps22/lps22.h +++ b/esphome/components/lps22/lps22.h @@ -17,10 +17,11 @@ class LPS22Component : public sensor::Sensor, public PollingComponent, public i2 void dump_config() override; protected: + void try_read_(); + sensor::Sensor *temperature_sensor_{nullptr}; sensor::Sensor *pressure_sensor_{nullptr}; - - RetryResult try_read_(); + uint8_t read_attempts_remaining_{0}; }; } // namespace lps22 diff --git a/esphome/components/lvgl/schemas.py b/esphome/components/lvgl/schemas.py index 45d933c00e6..2aeeedbd100 100644 --- a/esphome/components/lvgl/schemas.py +++ b/esphome/components/lvgl/schemas.py @@ -436,6 +436,7 @@ def container_schema(widget_type: WidgetType, extras=None): schema = schema.extend(widget_type.schema) def validator(value): + value = value or {} return append_layout_schema(schema, value)(value) return validator diff --git a/esphome/components/mqtt/mqtt_component.cpp b/esphome/components/mqtt/mqtt_component.cpp index e4b80de50a1..09570106df9 100644 --- a/esphome/components/mqtt/mqtt_component.cpp +++ b/esphome/components/mqtt/mqtt_component.cpp @@ -34,10 +34,7 @@ inline char *append_char(char *p, char c) { // MQTT_COMPONENT_TYPE_MAX_LEN, MQTT_SUFFIX_MAX_LEN, and MQTT_DEFAULT_TOPIC_MAX_LEN are in mqtt_component.h. // ESPHOME_DEVICE_NAME_MAX_LEN and OBJECT_ID_MAX_LEN are defined in entity_base.h. // This ensures the stack buffers below are always large enough. -static constexpr size_t DISCOVERY_PREFIX_MAX_LEN = 64; // Validated in Python: cv.Length(max=64) -// Format: prefix + "/" + type + "/" + name + "/" + object_id + "/config" + null -static constexpr size_t DISCOVERY_TOPIC_MAX_LEN = DISCOVERY_PREFIX_MAX_LEN + 1 + MQTT_COMPONENT_TYPE_MAX_LEN + 1 + - ESPHOME_DEVICE_NAME_MAX_LEN + 1 + OBJECT_ID_MAX_LEN + 7 + 1; +// MQTT_DISCOVERY_PREFIX_MAX_LEN and MQTT_DISCOVERY_TOPIC_MAX_LEN are defined in mqtt_component.h // Function implementation of LOG_MQTT_COMPONENT macro to reduce code size void log_mqtt_component(const char *tag, MQTTComponent *obj, bool state_topic, bool command_topic) { @@ -54,15 +51,15 @@ void MQTTComponent::set_subscribe_qos(uint8_t qos) { this->subscribe_qos_ = qos; void MQTTComponent::set_retain(bool retain) { this->retain_ = retain; } -std::string MQTTComponent::get_discovery_topic_(const MQTTDiscoveryInfo &discovery_info) const { +StringRef MQTTComponent::get_discovery_topic_to_(std::span buf, + const MQTTDiscoveryInfo &discovery_info) const { char sanitized_name[ESPHOME_DEVICE_NAME_MAX_LEN + 1]; str_sanitize_to(sanitized_name, App.get_name().c_str()); const char *comp_type = this->component_type(); char object_id_buf[OBJECT_ID_MAX_LEN]; StringRef object_id = this->get_default_object_id_to_(object_id_buf); - char buf[DISCOVERY_TOPIC_MAX_LEN]; - char *p = buf; + char *p = buf.data(); p = append_str(p, discovery_info.prefix.data(), discovery_info.prefix.size()); p = append_char(p, '/'); @@ -72,8 +69,9 @@ std::string MQTTComponent::get_discovery_topic_(const MQTTDiscoveryInfo &discove p = append_char(p, '/'); p = append_str(p, object_id.c_str(), object_id.size()); p = append_str(p, "/config", 7); + *p = '\0'; - return std::string(buf, p - buf); + return StringRef(buf.data(), p - buf.data()); } StringRef MQTTComponent::get_default_topic_for_to_(std::span buf, const char *suffix, @@ -182,16 +180,19 @@ bool MQTTComponent::publish_json(const char *topic, const json::json_build_t &f) bool MQTTComponent::send_discovery_() { const MQTTDiscoveryInfo &discovery_info = global_mqtt_client->get_discovery_info(); + char discovery_topic_buf[MQTT_DISCOVERY_TOPIC_MAX_LEN]; + StringRef discovery_topic = this->get_discovery_topic_to_(discovery_topic_buf, discovery_info); + if (discovery_info.clean) { ESP_LOGV(TAG, "'%s': Cleaning discovery", this->friendly_name_().c_str()); - return global_mqtt_client->publish(this->get_discovery_topic_(discovery_info), "", 0, this->qos_, true); + return global_mqtt_client->publish(discovery_topic.c_str(), "", 0, this->qos_, true); } ESP_LOGV(TAG, "'%s': Sending discovery", this->friendly_name_().c_str()); // NOLINTBEGIN(clang-analyzer-cplusplus.NewDeleteLeaks) false positive with ArduinoJson return global_mqtt_client->publish_json( - this->get_discovery_topic_(discovery_info), + discovery_topic.c_str(), [this](JsonObject root) { SendDiscoveryConfig config; config.state_topic = true; @@ -204,7 +205,7 @@ bool MQTTComponent::send_discovery_() { } // Fields from EntityBase - root[MQTT_NAME] = this->get_entity()->has_own_name() ? this->friendly_name_() : ""; + root[MQTT_NAME] = this->get_entity()->has_own_name() ? this->friendly_name_() : StringRef(); if (this->is_disabled_by_default_()) root[MQTT_ENABLED_BY_DEFAULT] = false; @@ -248,7 +249,7 @@ bool MQTTComponent::send_discovery_() { if (discovery_info.unique_id_generator == MQTT_MAC_ADDRESS_UNIQUE_ID_GENERATOR) { char friendly_name_hash[9]; buf_append_printf(friendly_name_hash, sizeof(friendly_name_hash), 0, "%08" PRIx32, - fnv1_hash(this->friendly_name_())); + fnv1_hash(this->friendly_name_().c_str())); // Format: mac-component_type-hash (e.g. "aabbccddeeff-sensor-12345678") // MAC (12) + "-" (1) + domain (max 20) + "-" (1) + hash (8) + null (1) = 43 char unique_id[MAC_ADDRESS_BUFFER_SIZE + ESPHOME_DOMAIN_MAX_LEN + 11]; @@ -414,7 +415,7 @@ void MQTTComponent::schedule_resend_state() { this->resend_state_ = true; } bool MQTTComponent::is_connected_() const { return global_mqtt_client->is_connected(); } // Pull these properties from EntityBase if not overridden -std::string MQTTComponent::friendly_name_() const { return this->get_entity()->get_name(); } +const StringRef &MQTTComponent::friendly_name_() const { return this->get_entity()->get_name(); } StringRef MQTTComponent::get_default_object_id_to_(std::span buf) const { return this->get_entity()->get_object_id_to(buf); } diff --git a/esphome/components/mqtt/mqtt_component.h b/esphome/components/mqtt/mqtt_component.h index 2cec6fda7ed..eb98158647e 100644 --- a/esphome/components/mqtt/mqtt_component.h +++ b/esphome/components/mqtt/mqtt_component.h @@ -32,6 +32,10 @@ static constexpr size_t MQTT_TOPIC_PREFIX_MAX_LEN = 64; // Validated in Python: // Format: prefix + "/" + type + "/" + object_id + "/" + suffix + null static constexpr size_t MQTT_DEFAULT_TOPIC_MAX_LEN = MQTT_TOPIC_PREFIX_MAX_LEN + 1 + MQTT_COMPONENT_TYPE_MAX_LEN + 1 + OBJECT_ID_MAX_LEN + 1 + MQTT_SUFFIX_MAX_LEN + 1; +static constexpr size_t MQTT_DISCOVERY_PREFIX_MAX_LEN = 64; // Validated in Python: cv.Length(max=64) +// Format: prefix + "/" + type + "/" + name + "/" + object_id + "/config" + null +static constexpr size_t MQTT_DISCOVERY_TOPIC_MAX_LEN = MQTT_DISCOVERY_PREFIX_MAX_LEN + 1 + MQTT_COMPONENT_TYPE_MAX_LEN + + 1 + ESPHOME_DEVICE_NAME_MAX_LEN + 1 + OBJECT_ID_MAX_LEN + 7 + 1; class MQTTComponent; // Forward declaration void log_mqtt_component(const char *tag, MQTTComponent *obj, bool state_topic, bool command_topic); @@ -263,8 +267,9 @@ class MQTTComponent : public Component { void subscribe_json(const std::string &topic, const mqtt_json_callback_t &callback, uint8_t qos = 0); protected: - /// Helper method to get the discovery topic for this component. - std::string get_discovery_topic_(const MQTTDiscoveryInfo &discovery_info) const; + /// Helper method to get the discovery topic for this component into a buffer. + StringRef get_discovery_topic_to_(std::span buf, + const MQTTDiscoveryInfo &discovery_info) const; /** Get this components state/command/... topic into a buffer. * @@ -288,7 +293,7 @@ class MQTTComponent : public Component { virtual const EntityBase *get_entity() const = 0; /// Get the friendly name of this MQTT component. - std::string friendly_name_() const; + const StringRef &friendly_name_() const; /// Get the icon field of this component as StringRef StringRef get_icon_ref_() const; diff --git a/esphome/components/web_server_idf/web_server_idf.cpp b/esphome/components/web_server_idf/web_server_idf.cpp index 98608104527..39e6b7a790c 100644 --- a/esphome/components/web_server_idf/web_server_idf.cpp +++ b/esphome/components/web_server_idf/web_server_idf.cpp @@ -258,8 +258,6 @@ StringRef AsyncWebServerRequest::url_to(std::span buffer) co return StringRef(buffer.data(), decoded_len); } -std::string AsyncWebServerRequest::host() const { return this->get_header("Host").value(); } - void AsyncWebServerRequest::send(AsyncWebServerResponse *response) { httpd_resp_send(*this, response->get_content_data(), response->get_content_size()); } diff --git a/esphome/components/web_server_idf/web_server_idf.h b/esphome/components/web_server_idf/web_server_idf.h index 817f47da796..17605449632 100644 --- a/esphome/components/web_server_idf/web_server_idf.h +++ b/esphome/components/web_server_idf/web_server_idf.h @@ -121,7 +121,6 @@ class AsyncWebServerRequest { char buffer[URL_BUF_SIZE]; return std::string(this->url_to(buffer)); } - std::string host() const; // NOLINTNEXTLINE(readability-identifier-naming) size_t contentLength() const { return this->req_->content_len; } diff --git a/esphome/dashboard/web_server.py b/esphome/dashboard/web_server.py index da502798649..00974bf4608 100644 --- a/esphome/dashboard/web_server.py +++ b/esphome/dashboard/web_server.py @@ -1054,17 +1054,26 @@ class DownloadBinaryRequestHandler(BaseHandler): # fallback to type=, but prioritize file= file_name = self.get_argument("type", None) file_name = self.get_argument("file", file_name) - if file_name is None: + if file_name is None or not file_name.strip(): self.send_error(400) return - file_name = file_name.replace("..", "").lstrip("/") # get requested download name, or build it based on filename download_name = self.get_argument( "download", f"{storage_json.name}-{file_name}", ) - path = storage_json.firmware_bin_path.parent.joinpath(file_name) + if storage_json.firmware_bin_path is None: + self.send_error(404) + return + + base_dir = storage_json.firmware_bin_path.parent.resolve() + path = base_dir.joinpath(file_name).resolve() + try: + path.relative_to(base_dir) + except ValueError: + self.send_error(403) + return if not path.is_file(): args = ["esphome", "idedata", settings.rel_path(configuration)] @@ -1078,7 +1087,7 @@ class DownloadBinaryRequestHandler(BaseHandler): found = False for image in idedata.extra_flash_images: - if image.path.endswith(file_name): + if image.path.as_posix().endswith(file_name): path = image.path download_name = file_name found = True diff --git a/esphome/wizard.py b/esphome/wizard.py index 4b748479960..f83342cc6a5 100644 --- a/esphome/wizard.py +++ b/esphome/wizard.py @@ -1,6 +1,5 @@ import base64 from pathlib import Path -import random import secrets import string from typing import Literal, NotRequired, TypedDict, Unpack @@ -130,7 +129,7 @@ def wizard_file(**kwargs: Unpack[WizardFileKwargs]) -> str: if len(ap_name) > 32: ap_name = ap_name_base kwargs["fallback_name"] = ap_name - kwargs["fallback_psk"] = "".join(random.choice(letters) for _ in range(12)) + kwargs["fallback_psk"] = "".join(secrets.choice(letters) for _ in range(12)) base = BASE_CONFIG_FRIENDLY if kwargs.get("friendly_name") else BASE_CONFIG diff --git a/tests/components/lvgl/test.host.yaml b/tests/components/lvgl/test.host.yaml index 00a8cd8c012..f84156c9d82 100644 --- a/tests/components/lvgl/test.host.yaml +++ b/tests/components/lvgl/test.host.yaml @@ -20,6 +20,8 @@ lvgl: - id: lvgl_0 default_font: space16 displays: sdl0 + top_layer: + - id: lvgl_1 displays: sdl1 on_idle: diff --git a/tests/dashboard/test_web_server.py b/tests/dashboard/test_web_server.py index 7642876ee57..9ea7a5164b7 100644 --- a/tests/dashboard/test_web_server.py +++ b/tests/dashboard/test_web_server.py @@ -8,6 +8,7 @@ import gzip import json import os from pathlib import Path +import sys from unittest.mock import AsyncMock, MagicMock, Mock, patch import pytest @@ -421,7 +422,7 @@ async def test_download_binary_handler_idedata_fallback( # Mock idedata response mock_image = Mock() - mock_image.path = str(bootloader_file) + mock_image.path = bootloader_file mock_idedata_instance = Mock() mock_idedata_instance.extra_flash_images = [mock_image] mock_idedata.return_value = mock_idedata_instance @@ -528,14 +529,22 @@ async def test_download_binary_handler_subdirectory_file_url_encoded( @pytest.mark.asyncio @pytest.mark.usefixtures("mock_ext_storage_path") @pytest.mark.parametrize( - "attack_path", + ("attack_path", "expected_code"), [ - pytest.param("../../../secrets.yaml", id="basic_traversal"), - pytest.param("..%2F..%2F..%2Fsecrets.yaml", id="url_encoded"), - pytest.param("zephyr/../../../secrets.yaml", id="traversal_with_prefix"), - pytest.param("/etc/passwd", id="absolute_path"), - pytest.param("//etc/passwd", id="double_slash_absolute"), - pytest.param("....//secrets.yaml", id="multiple_dots"), + pytest.param("../../../secrets.yaml", 403, id="basic_traversal"), + pytest.param("..%2F..%2F..%2Fsecrets.yaml", 403, id="url_encoded"), + pytest.param("zephyr/../../../secrets.yaml", 403, id="traversal_with_prefix"), + pytest.param("/etc/passwd", 403, id="absolute_path"), + pytest.param("//etc/passwd", 403, id="double_slash_absolute"), + pytest.param( + "....//secrets.yaml", + # On Windows, Path.resolve() treats "..." and "...." as parent + # traversal (like ".."), so the path escapes base_dir -> 403. + # On Unix, "...." is a literal directory name that stays inside + # base_dir but doesn't exist -> 404. + 403 if sys.platform == "win32" else 404, + id="multiple_dots", + ), ], ) async def test_download_binary_handler_path_traversal_protection( @@ -543,11 +552,14 @@ async def test_download_binary_handler_path_traversal_protection( tmp_path: Path, mock_storage_json: MagicMock, attack_path: str, + expected_code: int, ) -> None: """Test that DownloadBinaryRequestHandler prevents path traversal attacks. - Verifies that attempts to use '..' in file paths are sanitized to prevent - accessing files outside the build directory. Tests multiple attack vectors. + Verifies that attempts to escape the build directory via '..' are rejected + using resolve()/relative_to() validation. Tests multiple attack vectors. + Real traversals that escape the base directory get 403. Paths like '....' + that resolve inside the base directory but don't exist get 404. """ # Create build structure build_dir = get_build_path(tmp_path, "test") @@ -565,14 +577,67 @@ async def test_download_binary_handler_path_traversal_protection( mock_storage.firmware_bin_path = firmware_file mock_storage_json.load.return_value = mock_storage - # Attempt path traversal attack - should be blocked - with pytest.raises(HTTPClientError) as exc_info: + # Mock async_run_system_command so paths that pass validation but don't exist + # return 404 deterministically without spawning a real subprocess. + with ( + patch( + "esphome.dashboard.web_server.async_run_system_command", + new_callable=AsyncMock, + return_value=(2, "", ""), + ), + pytest.raises(HTTPClientError) as exc_info, + ): await dashboard.fetch( f"/download.bin?configuration=test.yaml&file={attack_path}", method="GET", ) - # Should get 404 (file not found after sanitization) or 500 (idedata fails) - assert exc_info.value.code in (404, 500) + assert exc_info.value.code == expected_code + + +@pytest.mark.asyncio +@pytest.mark.usefixtures("mock_ext_storage_path") +async def test_download_binary_handler_no_firmware_bin_path( + dashboard: DashboardTestHelper, + mock_storage_json: MagicMock, +) -> None: + """Test that download returns 404 when firmware_bin_path is None. + + This covers configs created by StorageJSON.from_wizard() where no + firmware has been compiled yet. + """ + mock_storage = Mock() + mock_storage.name = "test_device" + mock_storage.firmware_bin_path = None + mock_storage_json.load.return_value = mock_storage + + with pytest.raises(HTTPClientError) as exc_info: + await dashboard.fetch( + "/download.bin?configuration=test.yaml&file=firmware.bin", + method="GET", + ) + assert exc_info.value.code == 404 + + +@pytest.mark.asyncio +@pytest.mark.usefixtures("mock_ext_storage_path") +@pytest.mark.parametrize("file_value", ["", "%20%20", "%20"]) +async def test_download_binary_handler_empty_file_name( + dashboard: DashboardTestHelper, + mock_storage_json: MagicMock, + file_value: str, +) -> None: + """Test that download returns 400 for empty or whitespace-only file names.""" + mock_storage = Mock() + mock_storage.name = "test_device" + mock_storage.firmware_bin_path = Path("/fake/firmware.bin") + mock_storage_json.load.return_value = mock_storage + + with pytest.raises(HTTPClientError) as exc_info: + await dashboard.fetch( + f"/download.bin?configuration=test.yaml&file={file_value}", + method="GET", + ) + assert exc_info.value.code == 400 @pytest.mark.asyncio diff --git a/tests/unit_tests/test_wizard.py b/tests/unit_tests/test_wizard.py index eb44c1c20f4..0ce89230d80 100644 --- a/tests/unit_tests/test_wizard.py +++ b/tests/unit_tests/test_wizard.py @@ -2,7 +2,7 @@ from pathlib import Path from typing import Any -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch import pytest from pytest import MonkeyPatch @@ -632,3 +632,14 @@ def test_wizard_accepts_rpipico_board(tmp_path: Path, monkeypatch: MonkeyPatch): # rpipico doesn't support WiFi, so no api_encryption_key or ota_password assert "api_encryption_key" not in call_kwargs assert "ota_password" not in call_kwargs + + +def test_fallback_psk_uses_secrets_choice( + default_config: dict[str, Any], +) -> None: + """Test that fallback PSK is generated using secrets.choice.""" + with patch("esphome.wizard.secrets.choice", return_value="X") as mock_choice: + config = wz.wizard_file(**default_config) + + assert 'password: "XXXXXXXXXXXX"' in config + assert mock_choice.call_count == 12