mirror of
https://github.com/esphome/esphome.git
synced 2026-09-17 01:58:39 +00:00
Merge remote-tracking branch 'upstream/dev' into collapse_intermediary
This commit is contained in:
@@ -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/<env>/)
|
||||
|
||||
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)
|
||||
|
||||
@@ -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<char *>(msg), len);
|
||||
printk("%.*s", static_cast<int>(len), msg);
|
||||
#endif
|
||||
if (this->uart_dev_ == nullptr) {
|
||||
return;
|
||||
|
||||
@@ -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<float>(p_lsb));
|
||||
}
|
||||
return RetryResult::DONE;
|
||||
}
|
||||
|
||||
} // namespace lps22
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<char, MQTT_DISCOVERY_TOPIC_MAX_LEN> 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<char, MQTT_DEFAULT_TOPIC_MAX_LEN> 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<char, OBJECT_ID_MAX_LEN> buf) const {
|
||||
return this->get_entity()->get_object_id_to(buf);
|
||||
}
|
||||
|
||||
@@ -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<char, MQTT_DISCOVERY_TOPIC_MAX_LEN> 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;
|
||||
|
||||
@@ -258,8 +258,6 @@ StringRef AsyncWebServerRequest::url_to(std::span<char, URL_BUF_SIZE> 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());
|
||||
}
|
||||
|
||||
@@ -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; }
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
+1
-2
@@ -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
|
||||
|
||||
|
||||
@@ -20,6 +20,8 @@ lvgl:
|
||||
- id: lvgl_0
|
||||
default_font: space16
|
||||
displays: sdl0
|
||||
top_layer:
|
||||
|
||||
- id: lvgl_1
|
||||
displays: sdl1
|
||||
on_idle:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user