[packet_interface] Fixes from code review

- Reject an addressing destination or peer that is this device's own
  address, whose packets would be dropped as echoes
- Show 0 fragments, not SIZE_MAX, for a link with no room
- Test the check that applications use a peer's id

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
clydebarrow
2026-10-20 08:41:59 +11:00
co-authored by Claude Opus 5.5
parent 19c415f253
commit 0d904fc47e
3 changed files with 54 additions and 1 deletions
@@ -44,7 +44,8 @@ void BasicFragmentation::setup() {
}
void BasicFragmentation::dump_config() {
size_t fragments = this->fragments_for_(this->max_packet_size_);
// A link with no room for fragments fails setup, which core reports below
size_t fragments = this->fragment_size_() == 0 ? 0 : this->fragments_for_(this->max_packet_size_);
ESP_LOGCONFIG(TAG,
"Basic Fragmentation on '%s':\n"
" Max Packet Size: %zu\n"
@@ -49,6 +49,22 @@ def _unique_destinations(peers: list[ConfigType]) -> list[ConfigType]:
return peers
def _not_own_address(config: ConfigType) -> ConfigType:
"""Packets from this device's own address are dropped as echoes, so it cannot be a destination."""
address = config[CONF_ADDRESS]
if config.get(CONF_DESTINATION) == address:
raise cv.Invalid(
"Must not be this device's own address", path=[CONF_DESTINATION]
)
for index, peer in enumerate(config.get(CONF_PEERS, [])):
if peer[CONF_DESTINATION] == address:
raise cv.Invalid(
"Must not be this device's own address",
path=[CONF_PEERS, index, CONF_DESTINATION],
)
return config
# A function, as each peer holds the other layers, whose schemas exist only once every layer type is registered
def _basic_addressing_schema() -> cv.Schema:
peer_schema = cv.All(
@@ -74,6 +90,7 @@ def _basic_addressing_schema() -> cv.Schema:
}
).extend(cv.COMPONENT_SCHEMA),
cv.has_at_most_one_key(CONF_DESTINATION, CONF_PEERS),
_not_own_address,
)
+35
View File
@@ -23,6 +23,8 @@ from esphome.components.packet_interface.layers.hdlc_reliability import (
from esphome.components.uart import packet_interface as uart_packet_interface
from esphome.components.udp import packet_interface as udp_packet_interface
from esphome.const import CONF_ENCRYPTION
from esphome.core import ID
import esphome.final_validate as fv
@pytest.mark.parametrize(
@@ -292,3 +294,36 @@ def test_auto_load_follows_layers(
def test_auto_load_without_config() -> None:
"""Tooling that lists dependencies gets everything a layer could load."""
assert packet_interface.AUTO_LOAD() == ["binary_sensor", "xxtea"]
@pytest.mark.parametrize(
"addressing",
[
{"address": 1, "destination": 1},
{"address": 1, "peers": [_peer("a", 2), _peer("b", 1)]},
],
)
def test_addressing_own_address_rejected(setup_core, addressing: dict) -> None:
"""A device cannot send to its own address, as its own packets are dropped as echoes."""
with pytest.raises(cv.Invalid, match="own address"):
uart_packet_interface.CONFIG_SCHEMA({CONF_ADDRESSING: addressing})
def test_application_needs_peer_id_with_several_peers(setup_core) -> None:
"""An application must use one peer's id, not the id of a link with several peers."""
validator = packet_interface.final_validate_interface("packet_interface")
token = fv.full_config.set(
{
"packet_interface": [
{"id": ID("bus"), CONF_ADDRESSING: {"peers": [_peer("a", 2)]}},
{"id": ID("single"), CONF_ADDRESSING: {"destination": 2}},
]
}
)
try:
with pytest.raises(cv.Invalid, match="several peers"):
validator({"packet_interface": ID("bus")})
assert validator({"packet_interface": ID("single")})
assert validator({"packet_interface": ID("a")})
finally:
fv.full_config.reset(token)