[packet_interface] Stretch the XXTEA key text with PBKDF2

A single SHA-256 of the key text let a captured packet check about 2^30
guesses for every one that passes, so a weak key text fell quickly to
an offline search. Derive the key with PBKDF2-HMAC-SHA256 at 600,000
iterations instead; this runs at build time, so the device pays nothing.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

CODEOWNERS

@@ -689,6 +689,7 @@ esphome/components/uart/packet_transport/* @clydebarrow
 esphome/components/uart_mux/* @kbx81
 esphome/components/uart_tcp/* @Bascht74
 esphome/components/udp/* @clydebarrow
+esphome/components/udp/packet_interface/* @clydebarrow
 esphome/components/ufire_ec/* @pvizeli
 esphome/components/ufire_ise/* @pvizeli
 esphome/components/ufm01/* @ljungqvist

esphome/components/packet_interface/__init__.py

@@ -1,3 +1,5 @@
+import importlib
+import pkgutil
 from typing import Any
 
 import esphome.codegen as cg
@@ -7,14 +9,27 @@ from esphome.const import CONF_ID
 from esphome.cpp_generator import MockObj, MockObjClass
 from esphome.types import ConfigType
 
+from . import layers
+from .layers import (
+    PacketInterface,
+    build_stack,
+    check_session,
+    packet_interface_ns,
+    stack_auto_load,
+    stack_schema,
+)
+
 CODEOWNERS = ["@clydebarrow"]
 DOMAIN = "packet_interface"
 IS_PLATFORM_COMPONENT = True
 
 CONF_PACKET_INTERFACE = "packet_interface"
 
-packet_interface_ns = cg.esphome_ns.namespace("packet_interface")
-PacketInterface = packet_interface_ns.class_("PacketInterface", cg.Component)
+# Each module in layers registers a layer type; import them all before building the schemas from the registry
+for module_info in pkgutil.iter_modules(layers.__path__):
+    importlib.import_module(f".layers.{module_info.name}", package=__package__)
+
+LINK_LAYER_SCHEMAS = stack_schema()
 
 __all__ = [
     "CONF_MAX_PACKET_SIZE",
@@ -26,24 +41,47 @@ __all__ = [
 ]
 
 
+def AUTO_LOAD(config: list[ConfigType] | None = None) -> list[str]:
+    """Load only what the configured layers use.
+
+    Without a config, as tooling that lists dependencies calls it, everything a layer could load.
+    """
+    if config is None:
+        return ["binary_sensor", "xxtea"]
+    loads: set[str] = set()
+    for interface in config:
+        loads |= stack_auto_load(interface)
+    return sorted(loads)
+
+
+def _validate_layers(config: ConfigType) -> ConfigType:
+    check_session(config)
+    return config
+
+
 def packet_interface_schema(
     class_: MockObjClass,
     default_max_packet_size: int,
     max_packet_size_limit: int = 4096,
 ) -> cv.Schema:
-    """Schema for a link platform."""
-    return cv.Schema(
-        {
-            cv.GenerateID(): cv.declare_id(class_),
-            cv.Optional(
-                CONF_MAX_PACKET_SIZE, default=default_max_packet_size
-            ): cv.int_range(16, max_packet_size_limit),
-        }
-    ).extend(cv.COMPONENT_SCHEMA)
+    """Schema for a link platform, with the layers that can be stacked on it."""
+    return (
+        cv.Schema(
+            {
+                cv.GenerateID(): cv.declare_id(class_),
+                cv.Optional(
+                    CONF_MAX_PACKET_SIZE, default=default_max_packet_size
+                ): cv.int_range(16, max_packet_size_limit),
+                **LINK_LAYER_SCHEMAS,
+            }
+        )
+        .extend(cv.COMPONENT_SCHEMA)
+        .add_extra(_validate_layers)
+    )
 
 
 async def new_packet_interface(config: ConfigType, *args: Any) -> MockObj:
-    """Create the link layer.
+    """Create the link layer and the layers configured on it, from the bottom up.
 
     The link's constructor takes its maximum packet size, usually as a std::span over a buffer that code generation
     sized from the configuration, so the two cannot differ.
@@ -53,4 +91,5 @@ async def new_packet_interface(config: ConfigType, *args: Any) -> MockObj:
     var = cg.new_Pvariable(config[CONF_ID], *args)
     cg.add(var.set_name(config[CONF_ID].id))
     await cg.register_component(var, config)
+    await build_stack(config, var)
     return var

esphome/components/packet_interface/basic_session.cpp

@@ -0,0 +1,135 @@
+#include "basic_session.h"
+#ifdef USE_PACKET_INTERFACE_BASIC_SESSION
+#include "esphome/core/application.h"
+#include "esphome/core/helpers.h"
+#include "esphome/core/log.h"
+
+namespace esphome::packet_interface {
+
+ESPHOME_LOG_TAG(TAG, "packet_interface.session");
+
+static constexpr uint8_t PACKET_TYPE_DATA = 0x00;
+static constexpr uint8_t PACKET_TYPE_KEEP_ALIVE = 0x01;
+// Type byte and the 32-bit boot id, little endian
+static constexpr size_t KEEP_ALIVE_SIZE = 5;
+
+BasicSession::BasicSession(PacketInterface *lower)
+    : PacketInterface(remaining_packet_size(lower->get_max_packet_size(), HEADER_SIZE)), lower_(lower) {
+  this->stack_on_(lower, CAPABILITY_SESSION);
+  // Data lost by a reliability layer below must be sent again, as after a restart of the peer
+  if (lower->has_capability(CAPABILITY_RELIABILITY)) {
+    lower->add_on_session_state_callback([this](bool connected) {
+      if (!connected && this->connected_)
+        this->set_connected_(false);
+    });
+  }
+}
+
+void BasicSession::setup() {
+  if (this->lower_->get_max_packet_size() < KEEP_ALIVE_SIZE) {
+    this->mark_failed(LOG_STR("max_packet_size too small for the configured layers"));
+    return;
+  }
+  this->boot_id_ = random_uint32();
+#ifdef USE_BINARY_SENSOR
+  if (this->connected_sensor_ != nullptr)
+    this->connected_sensor_->publish_initial_state(false);
+#endif
+  this->lower_->add_packet_interface_listener(
+      [this](const PacketBuffer &data, const PacketMetaData &meta_data) { this->receive_(data, meta_data); });
+  // A reliability layer below queues this until its link is up, so the peer hears of a restart straight away
+  this->send_keep_alive_();
+  this->set_interval(this->keep_alive_ms_, [this] { this->on_interval_(); });
+}
+
+void BasicSession::dump_config() {
+  ESP_LOGCONFIG(TAG,
+                "Basic Session on '%s':\n"
+                "  Max Packet Size: %zu\n"
+                "  Keep Alive: %" PRIu32 "ms\n"
+                "  Timeout: %" PRIu32 "ms",
+                this->name_, this->max_packet_size_, this->keep_alive_ms_, this->timeout_ms_);
+#ifdef USE_BINARY_SENSOR
+  LOG_BINARY_SENSOR("  ", "Connected", this->connected_sensor_);
+#endif
+}
+
+bool BasicSession::send_to_interface(const PacketBuffer &data, const PacketMetaData &meta_data) {
+  if (this->is_failed())
+    return false;
+  uint8_t header = PACKET_TYPE_DATA;
+  PacketBuffer packet(&header, HEADER_SIZE, &data, nullptr, 0);
+  return this->lower_->send_to_interface(packet, meta_data);
+}
+
+void BasicSession::send_keep_alive_() {
+  uint8_t packet[KEEP_ALIVE_SIZE] = {
+      PACKET_TYPE_KEEP_ALIVE, static_cast<uint8_t>(this->boot_id_), static_cast<uint8_t>(this->boot_id_ >> 8),
+      static_cast<uint8_t>(this->boot_id_ >> 16), static_cast<uint8_t>(this->boot_id_ >> 24)};
+  if (!this->lower_->send_to_interface(PacketBuffer(packet, sizeof(packet)), {}))
+    ESP_LOGV(TAG, "%s: Lower layer did not accept keep-alive", this->name_);
+}
+
+void BasicSession::on_interval_() {
+  this->send_keep_alive_();
+  if (this->connected_ && App.get_loop_component_start_time() - this->last_receive_ms_ > this->timeout_ms_) {
+    ESP_LOGW(TAG, "%s: Peer not heard for %" PRIu32 "ms", this->name_, this->timeout_ms_);
+    this->set_connected_(false);
+  }
+}
+
+void BasicSession::set_connected_(bool connected) {
+  this->connected_ = connected;
+  ESP_LOGD(TAG, "%s: Session %s", this->name_, connected ? LOG_STR_LITERAL("connected") : LOG_STR_LITERAL("lost"));
+#ifdef USE_BINARY_SENSOR
+  if (this->connected_sensor_ != nullptr)
+    this->connected_sensor_->publish_state(connected);
+#endif
+  this->session_callback_.call(connected);
+}
+
+void BasicSession::receive_(const PacketBuffer &data, const PacketMetaData &meta_data) {
+  // Lower layers deliver packets in one piece; the payload is passed up without copying
+  if (!data.is_contiguous()) {
+    ESP_LOGW(TAG, "%s: Packet split across buffers is not supported", this->name_);
+    return;
+  }
+  auto packet = data.as_span();
+  if (packet.empty()) {
+    this->note_received_(false);
+    return;
+  }
+  uint8_t type = packet[0];
+  if (type == PACKET_TYPE_KEEP_ALIVE) {
+    if (packet.size() != KEEP_ALIVE_SIZE) {
+      ESP_LOGV(TAG, "%s: Keep-alive of %zu bytes dropped", this->name_, packet.size());
+      this->note_received_(false);
+      return;
+    }
+    uint32_t boot_id = encode_uint32(packet[4], packet[3], packet[2], packet[1]);
+    if (!this->peer_known_ || boot_id != this->peer_boot_id_) {
+      // Data may arrive before the first keep-alive, so only a changed id means a restart
+      if (this->connected_ && this->peer_known_) {
+        ESP_LOGD(TAG, "%s: Peer has restarted", this->name_);
+        this->set_connected_(false);
+      }
+      this->peer_boot_id_ = boot_id;
+      this->peer_known_ = true;
+      // Answer a new peer at once; it answers only if our id is new to it, so this cannot repeat
+      this->send_keep_alive_();
+    }
+  } else if (type != PACKET_TYPE_DATA) {
+    ESP_LOGV(TAG, "%s: Unknown packet type 0x%02X dropped", this->name_, type);
+    this->note_received_(false);
+    return;
+  }
+  this->note_received_(true);
+  this->last_receive_ms_ = App.get_loop_component_start_time();
+  if (!this->connected_)
+    this->set_connected_(true);
+  if (type == PACKET_TYPE_DATA)
+    this->on_receive_from_interface_(PacketBuffer(packet.data() + HEADER_SIZE, packet.size() - HEADER_SIZE), meta_data);
+}
+
+}  // namespace esphome::packet_interface
+#endif

esphome/components/packet_interface/basic_session.h

@@ -0,0 +1,61 @@
+#pragma once
+
+#include "esphome/core/defines.h"
+#ifdef USE_PACKET_INTERFACE_BASIC_SESSION
+#include "packet_interface.h"
+#ifdef USE_BINARY_SENSOR
+#include "esphome/components/binary_sensor/binary_sensor.h"
+#endif
+
+namespace esphome::packet_interface {
+
+/**
+ * A session layer that reports whether the peer can be reached.
+ *
+ * Every packet starts with a type byte. A data packet carries the payload of the layer above. A keep-alive packet
+ * carries a random number chosen at boot, so that a restart of the peer is noticed even if it is quicker than the
+ * timeout. A keep-alive is sent every keep_alive interval, and at once when a new peer is heard. The peer is
+ * considered gone when nothing has been received from it for the timeout; this is checked every keep_alive interval.
+ * A restart of the peer is reported as a disconnect followed by a connect. So is a reset of a reliability layer below,
+ * which discards data: the session is reported lost, and connected again when the peer is next heard.
+ */
+class BasicSession : public PacketInterface {
+ public:
+  /// The packet type byte.
+  static constexpr size_t HEADER_SIZE = 1;
+
+  explicit BasicSession(PacketInterface *lower);
+
+  void setup() override;
+  void dump_config() override;
+  bool send_to_interface(const PacketBuffer &data, const PacketMetaData &meta_data) override;
+
+  void set_keep_alive(uint32_t keep_alive_ms) { this->keep_alive_ms_ = keep_alive_ms; }
+  void set_session_timeout(uint32_t timeout_ms) { this->timeout_ms_ = timeout_ms; }
+#ifdef USE_BINARY_SENSOR
+  void set_connected_sensor(binary_sensor::BinarySensor *sensor) { this->connected_sensor_ = sensor; }
+#endif
+
+  bool is_connected() const { return this->connected_; }
+
+ protected:
+  void receive_(const PacketBuffer &data, const PacketMetaData &meta_data);
+  void on_interval_();
+  void send_keep_alive_();
+  void set_connected_(bool connected);
+
+  PacketInterface *lower_;
+#ifdef USE_BINARY_SENSOR
+  binary_sensor::BinarySensor *connected_sensor_{nullptr};
+#endif
+  uint32_t keep_alive_ms_{10000};
+  uint32_t timeout_ms_{30000};
+  uint32_t last_receive_ms_{0};
+  uint32_t boot_id_{0};
+  uint32_t peer_boot_id_{0};
+  bool peer_known_{false};
+  bool connected_{false};
+};
+
+}  // namespace esphome::packet_interface
+#endif

esphome/components/packet_interface/hdlc_reliability.cpp

@@ -0,0 +1,365 @@
+#include "hdlc_reliability.h"
+#ifdef USE_PACKET_INTERFACE_HDLC
+#include "esphome/core/helpers.h"
+#include "esphome/core/log.h"
+
+namespace esphome::packet_interface {
+
+ESPHOME_LOG_TAG(TAG, "packet_interface.hdlc");
+
+static constexpr uint32_t RETRANSMIT_TIMER_ID = 0;
+static constexpr uint32_t ACK_TIMER_ID = 1;
+static constexpr uint32_t SEND_RETRY_TIMER_ID = 2;
+// A frame or acknowledgement the lower layer refused, for example while a radio was receiving, is sent again after this
+static constexpr uint32_t SEND_RETRY_MS = 50;
+// SABM is repeated at this interval, or the timeout if longer, until the peer answers
+static constexpr uint32_t RESET_INTERVAL_MS = 1000;
+
+// Control byte formats, with the poll/final bit (0x10) ignored
+static constexpr uint8_t POLL_FINAL_BIT = 0x10;
+static constexpr uint8_t SUPERVISORY_FRAME = 0x01;
+static constexpr uint8_t SUPERVISORY_RR = 0x00;
+static constexpr uint8_t SUPERVISORY_REJ = 0x02;
+static constexpr uint8_t UNNUMBERED_SABM = 0x2F;
+static constexpr uint8_t UNNUMBERED_UA = 0x63;
+
+static uint16_t frame_crc(uint8_t control, const uint8_t *payload, size_t length) {
+  uint16_t crc = crc16(&control, 1);
+  return crc16(payload, length, crc);
+}
+
+HdlcReliability::HdlcReliability(PacketInterface *lower)
+    : PacketInterface(remaining_packet_size(lower->get_max_packet_size(), FRAME_OVERHEAD)), lower_(lower) {
+  this->stack_on_(lower, CAPABILITY_RELIABILITY);
+}
+
+void HdlcReliability::setup() {
+  if (this->max_packet_size_ == 0) {
+    this->mark_failed(LOG_STR("max_packet_size too small for the configured layers"));
+    return;
+  }
+  RAMAllocator<uint8_t> allocator;
+  this->tx_slots_ = allocator.allocate(size_t{this->max_queue_length_} * this->max_packet_size_);
+  if (this->tx_slots_ == nullptr) {
+    this->mark_failed(LOG_STR("Not enough memory for the send queue"));
+    return;
+  }
+  this->lower_->add_packet_interface_listener(
+      [this](const PacketBuffer &data, const PacketMetaData &meta_data) { this->receive_(data, meta_data); });
+}
+
+void HdlcReliability::dump_config() {
+  ESP_LOGCONFIG(TAG,
+                "HDLC Reliability on '%s':\n"
+                "  Max Packet Size: %zu\n"
+                "  Timeout: %" PRIu32 "ms\n"
+                "  Ack Delay: %" PRIu32 "ms\n"
+                "  Max Queue Length: %u\n"
+                "  Max Retries: %u",
+                this->name_, this->max_packet_size_, this->timeout_ms_, this->ack_delay_ms_, this->max_queue_length_,
+                this->max_retries_);
+}
+
+void HdlcReliability::loop() {
+  // Disabled first, as sending may deliver a reply to this layer and request the loop again
+  this->disable_loop();
+  if (this->state_ == LinkState::LINK_STATE_RESET_PENDING) {
+    this->state_ = LinkState::LINK_STATE_RESETTING;
+    this->start_timer_();
+    this->send_frame_(UNNUMBERED_SABM, nullptr, 0);
+  }
+  if (this->state_ != LinkState::LINK_STATE_CONNECTED)
+    return;
+  // Queued frames go first, as each carries the acknowledgement
+  if (this->send_pending_() && this->ack_due_ && !this->send_supervisory_(SUPERVISORY_RR))
+    this->retry_send_();
+}
+
+void HdlcReliability::retry_send_() {
+  this->set_timeout(SEND_RETRY_TIMER_ID, SEND_RETRY_MS, [this]() { this->request_loop_(); });
+}
+
+void HdlcReliability::owe_ack_(bool now) {
+  if (now || this->ack_delay_ms_ == 0) {
+    this->ack_due_ = true;
+    this->request_loop_();
+  } else if (!this->ack_pending_) {
+    // Waiting lets one acknowledgement cover several frames, or a reply carry it
+    this->set_timeout(ACK_TIMER_ID, this->ack_delay_ms_, [this]() {
+      this->ack_due_ = true;
+      this->request_loop_();
+    });
+  }
+  this->ack_pending_ = true;
+}
+
+void HdlcReliability::ack_sent_() {
+  this->ack_pending_ = false;
+  this->ack_due_ = false;
+  this->cancel_timeout(ACK_TIMER_ID);
+}
+
+void HdlcReliability::request_loop_() { this->enable_loop(); }
+
+bool HdlcReliability::send_to_interface(const PacketBuffer &data, const PacketMetaData &meta_data) {
+  if (this->tx_slots_ == nullptr)
+    return false;
+  size_t size = data.size();
+  if (size > this->max_packet_size_) {
+    ESP_LOGW(TAG, "%s: Packet of %zu bytes is larger than %zu", this->name_, size, this->max_packet_size_);
+    return false;
+  }
+  uint8_t outstanding = this->outstanding_();
+  if (outstanding >= this->max_queue_length_) {
+    if (this->state_ == LinkState::LINK_STATE_CONNECTED) {
+      ESP_LOGV(TAG, "%s: Send queue full", this->name_);
+    } else if (!this->reset_wait_logged_) {
+      // Once for each reset; the layers above keep trying while the peer is away
+      this->reset_wait_logged_ = true;
+      ESP_LOGD(TAG, "%s: Send queue full, waiting for the peer to answer the link reset", this->name_);
+    }
+    return false;
+  }
+  uint8_t seq = this->vs_;
+  data.copy_to(this->slot_(seq), size);
+  this->tx_lengths_[this->slot_index_(seq)] = size;
+  this->vs_ = (seq + 1) & 7;
+  // While the link is being reset, the frame waits in the queue and is sent when UA arrives
+  if (this->state_ == LinkState::LINK_STATE_CONNECTED) {
+    this->send_pending_();
+    if (outstanding == 0)
+      this->start_timer_();
+  }
+  return true;
+}
+
+bool HdlcReliability::send_frame_(uint8_t control, const uint8_t *payload, size_t length) {
+  uint16_t crc = frame_crc(control, payload, length);
+  uint8_t trailer[2] = {static_cast<uint8_t>(crc), static_cast<uint8_t>(crc >> 8)};
+  PacketBuffer body(payload, length);
+  PacketBuffer frame(&control, 1, length != 0 ? &body : nullptr, trailer, sizeof(trailer));
+  if (this->lower_->send_to_interface(frame, {}))
+    return true;
+  ESP_LOGV(TAG, "%s: Lower layer did not accept frame 0x%02X", this->name_, control);
+  return false;
+}
+
+bool HdlcReliability::send_supervisory_(uint8_t type) {
+  if (!this->send_frame_(SUPERVISORY_FRAME | (type << 2) | (this->vr_ << 5), nullptr, 0))
+    return false;
+  this->ack_sent_();
+  return true;
+}
+
+// Sends the next queued frame not yet sent, and returns true if none are left. One frame per loop pass lets other
+// components run between frames: on a radio, sending blocks, and the peer's frames are received in between. If the
+// lower layer refuses the frame, it is sent again shortly.
+bool HdlcReliability::send_pending_() {
+  if (this->vn_ == this->vs_)
+    return true;
+  uint8_t control = (this->vn_ << 1) | (this->vr_ << 5);
+  if (!this->send_frame_(control, this->slot_(this->vn_), this->tx_lengths_[this->slot_index_(this->vn_)])) {
+    this->retry_send_();
+    return false;
+  }
+  // Every I-frame carries an acknowledgement, so a pending RR is no longer needed
+  this->ack_sent_();
+  this->vn_ = (this->vn_ + 1) & 7;
+  if (this->vn_ == this->vs_)
+    return true;
+  this->request_loop_();
+  return false;
+}
+
+void HdlcReliability::resend_outstanding_() {
+  this->vn_ = this->va_;
+  this->send_pending_();
+}
+
+void HdlcReliability::start_timer_() {
+  uint32_t interval = this->timeout_ms_;
+  if (this->state_ != LinkState::LINK_STATE_CONNECTED && interval < RESET_INTERVAL_MS)
+    interval = RESET_INTERVAL_MS;
+  // Up to a quarter longer at random, so devices whose frames collided on a shared channel do not resend together
+  interval += random_uint32() % (interval / 4 + 1);
+  this->set_timeout(RETRANSMIT_TIMER_ID, interval, [this]() { this->on_timeout_(); });
+}
+
+void HdlcReliability::on_timeout_() {
+  if (this->state_ != LinkState::LINK_STATE_CONNECTED) {
+    // Counted up to max_retries_ only, so the warning is logged once until the link comes up
+    if (this->retries_ < this->max_retries_) {
+      this->retries_++;
+      if (this->retries_ == this->max_retries_) {
+        ESP_LOGW(TAG, "%s: No answer to the link reset after %u tries; check that the peer has a reliability layer",
+                 this->name_, this->max_retries_);
+      }
+    }
+    this->send_frame_(UNNUMBERED_SABM, nullptr, 0);
+    this->start_timer_();
+    return;
+  }
+  if (this->outstanding_() == 0)
+    return;
+  if (this->retries_ == this->max_retries_) {
+    ESP_LOGW(TAG, "%s: No acknowledgement after %u retries, resetting link", this->name_, this->max_retries_);
+    this->start_reset_();
+    return;
+  }
+  this->retries_++;
+  ESP_LOGV(TAG, "%s: Timeout, resending %u frames", this->name_, this->outstanding_());
+  this->resend_outstanding_();
+  this->start_timer_();
+}
+
+// Unacknowledged frames are discarded; frames queued from now on are numbered from 0 and sent once the link is up.
+// Resetting a connected link can lose data both ways, which is reported to the layers above.
+void HdlcReliability::start_reset_() {
+  bool was_connected = this->state_ == LinkState::LINK_STATE_CONNECTED;
+  this->state_ = LinkState::LINK_STATE_RESET_PENDING;
+  this->va_ = this->vs_ = this->vn_ = 0;
+  this->head_ = 0;
+  this->vr_ = 0;
+  this->retries_ = 0;
+  this->rej_sent_ = false;
+  this->ack_sent_();
+  this->cancel_timeout(RETRANSMIT_TIMER_ID);
+  this->request_loop_();
+  if (was_connected)
+    this->session_callback_.call(false);
+}
+
+void HdlcReliability::link_up_() {
+  ESP_LOGD(TAG, "%s: Link connected", this->name_);
+  this->state_ = LinkState::LINK_STATE_CONNECTED;
+  this->reset_wait_logged_ = false;
+  this->peer_frame_seen_ = false;
+  this->retries_ = 0;
+  this->cancel_timeout(RETRANSMIT_TIMER_ID);
+  if (this->outstanding_() != 0) {
+    this->resend_outstanding_();
+    this->start_timer_();
+  }
+}
+
+void HdlcReliability::process_ack_(uint8_t nr) {
+  uint8_t acked = (nr - this->va_) & 7;
+  if (acked > this->outstanding_()) {
+    ESP_LOGV(TAG, "%s: Ignoring invalid N(R) %u", this->name_, nr);
+    return;
+  }
+  if (acked == 0)
+    return;
+  // The peer may have had frames that were refused when last resent
+  if (acked > ((this->vn_ - this->va_) & 7))
+    this->vn_ = nr;
+  this->head_ = (this->head_ + acked) % this->max_queue_length_;
+  this->va_ = nr;
+  this->retries_ = 0;
+  if (this->outstanding_() == 0) {
+    this->cancel_timeout(RETRANSMIT_TIMER_ID);
+  } else {
+    this->start_timer_();
+  }
+}
+
+void HdlcReliability::receive_(const PacketBuffer &data, const PacketMetaData &meta_data) {
+  // Lower layers deliver packets in one piece; the payload is passed up without copying
+  if (!data.is_contiguous()) {
+    ESP_LOGW(TAG, "%s: Packet split across buffers is not supported", this->name_);
+    return;
+  }
+  auto frame = data.as_span();
+  if (frame.size() < FRAME_OVERHEAD) {
+    this->note_received_(false);
+    return;
+  }
+  size_t length = frame.size() - FRAME_OVERHEAD;
+  uint8_t control = frame[0];
+  const uint8_t *payload = frame.data() + 1;
+  uint16_t crc = frame[length + 1] | (frame[length + 2] << 8);
+  if (frame_crc(control, payload, length) != crc) {
+    ESP_LOGV(TAG, "%s: CRC error, frame dropped", this->name_);
+    this->note_received_(false);
+    return;
+  }
+  // The peer's packets may only be larger if it has a larger max_packet_size, which the layers above cannot take
+  if (length > this->max_packet_size_) {
+    if (!this->oversize_warned_) {
+      this->oversize_warned_ = true;
+      ESP_LOGW(TAG,
+               "%s: Frame of %zu bytes is larger than %zu, dropped; check that both devices use the same "
+               "max_packet_size",
+               this->name_, length, this->max_packet_size_);
+    } else {
+      ESP_LOGV(TAG, "%s: Frame of %zu bytes is larger than %zu, dropped", this->name_, length, this->max_packet_size_);
+    }
+    this->note_received_(false);
+    return;
+  }
+  this->note_received_(true);
+
+  uint8_t command = control & ~POLL_FINAL_BIT;
+  if (command == UNNUMBERED_SABM) {
+    // A peer that has not heard our UA repeats its SABM, and cannot have sent a numbered frame meanwhile; answering
+    // again keeps this side connected, and the queued frames, instead of treating the repeat as a restart.
+    if (this->state_ == LinkState::LINK_STATE_CONNECTED && this->peer_frame_seen_) {
+      // The peer has restarted, so frames it has not acknowledged are lost
+      ESP_LOGD(TAG, "%s: Link reset by peer", this->name_);
+      this->start_reset_();
+    }
+    this->vr_ = 0;
+    this->rej_sent_ = false;
+    this->send_frame_(UNNUMBERED_UA, nullptr, 0);
+    // Both sides now start from 0, so the link is up even if our own SABM is unanswered
+    this->link_up_();
+    return;
+  }
+  if (command == UNNUMBERED_UA) {
+    if (this->state_ == LinkState::LINK_STATE_RESETTING)
+      this->link_up_();
+    return;
+  }
+  // Numbered frames are ignored until the link is reset; the peer answers our SABM first
+  if (this->state_ != LinkState::LINK_STATE_CONNECTED)
+    return;
+  this->peer_frame_seen_ = true;
+
+  uint8_t nr = control >> 5;
+  if ((control & 1) == 0) {
+    this->process_ack_(nr);
+    this->receive_info_(control, payload, length, meta_data);
+    return;
+  }
+  if ((control & 3) != SUPERVISORY_FRAME)
+    return;
+  this->process_ack_(nr);
+  if (((control >> 2) & 3) == SUPERVISORY_REJ && this->outstanding_() != 0) {
+    ESP_LOGV(TAG, "%s: Peer rejected from %u", this->name_, nr);
+    this->resend_outstanding_();
+    this->start_timer_();
+  }
+}
+
+void HdlcReliability::receive_info_(uint8_t control, const uint8_t *payload, size_t length,
+                                    const PacketMetaData &meta_data) {
+  uint8_t ns = (control >> 1) & 7;
+  if (ns != this->vr_) {
+    // A gap or a duplicate; ask once for a resend from the expected frame, then just acknowledge at once
+    if (!this->rej_sent_) {
+      this->rej_sent_ = this->send_supervisory_(SUPERVISORY_REJ);
+      if (this->rej_sent_)
+        return;
+    }
+    this->owe_ack_(true);
+    return;
+  }
+  this->vr_ = (this->vr_ + 1) & 7;
+  this->rej_sent_ = false;
+  // The acknowledgement is sent from loop(), so a reply sent from the callback can carry it instead
+  this->owe_ack_(false);
+  this->on_receive_from_interface_(PacketBuffer(payload, length), meta_data);
+}
+
+}  // namespace esphome::packet_interface
+#endif

esphome/components/packet_interface/hdlc_reliability.h

@@ -0,0 +1,99 @@
+#pragma once
+
+#include "esphome/core/defines.h"
+#ifdef USE_PACKET_INTERFACE_HDLC
+#include <array>
+
+#include "packet_interface.h"
+
+namespace esphome::packet_interface {
+
+/**
+ * A reliability layer that delivers packets in order, without loss or duplicates, between two peers.
+ *
+ * It follows HDLC asynchronous balanced mode, with sequence numbers modulo 8 and go-back-N retransmission:
+ * - I-frames carry a payload, the sequence number N(S) and an acknowledgement N(R) of the peer's frames.
+ * - RR frames acknowledge without a payload; REJ frames ask the peer to resend from N(R).
+ * - SABM resets both peers' sequence numbers, and UA answers it.
+ *
+ * Each frame is the HDLC control byte, the payload, then a CRC-16 of both in little endian order.
+ * Unacknowledged frames are resent after the timeout; after max_retries the link is reset.
+ */
+class HdlcReliability : public PacketInterface {
+ public:
+  /// Most frames that can be unacknowledged at once, set by the 3 bit sequence numbers.
+  static constexpr uint8_t MAX_WINDOW = 7;
+  /// Control byte plus CRC-16.
+  static constexpr size_t FRAME_OVERHEAD = 3;
+
+  explicit HdlcReliability(PacketInterface *lower);
+
+  void setup() override;
+  void loop() override;
+  void dump_config() override;
+  bool send_to_interface(const PacketBuffer &data, const PacketMetaData &meta_data) override;
+  /// No room while the layer has no send queue: before setup(), or when setup() failed.
+  size_t get_send_capacity() const override {
+    return this->tx_slots_ == nullptr ? 0 : this->max_queue_length_ - this->outstanding_();
+  }
+
+  void set_ack_timeout(uint32_t timeout_ms) { this->timeout_ms_ = timeout_ms; }
+  void set_ack_delay(uint32_t ack_delay_ms) { this->ack_delay_ms_ = ack_delay_ms; }
+  void set_max_queue_length(uint8_t length) { this->max_queue_length_ = length; }
+  void set_max_retries(uint8_t retries) { this->max_retries_ = retries; }
+
+  bool is_connected() const { return this->state_ == LinkState::LINK_STATE_CONNECTED; }
+
+ protected:
+  enum class LinkState : uint8_t {
+    LINK_STATE_RESET_PENDING,  // SABM to be sent from loop()
+    LINK_STATE_RESETTING,      // SABM sent, waiting for UA
+    LINK_STATE_CONNECTED,
+  };
+
+  void receive_(const PacketBuffer &data, const PacketMetaData &meta_data);
+  void receive_info_(uint8_t control, const uint8_t *payload, size_t length, const PacketMetaData &meta_data);
+  bool send_frame_(uint8_t control, const uint8_t *payload, size_t length);
+  bool send_supervisory_(uint8_t type);
+  bool send_pending_();
+  void resend_outstanding_();
+  void retry_send_();
+  /// An RR is owed: at once, or after the ack delay unless an I-frame carries it first.
+  void owe_ack_(bool now);
+  void ack_sent_();
+  void process_ack_(uint8_t nr);
+  void on_timeout_();
+  void start_reset_();
+  void link_up_();
+  void start_timer_();
+  void request_loop_();
+  uint8_t outstanding_() const { return (this->vs_ - this->va_) & 7; }
+  uint8_t slot_index_(uint8_t seq) const { return (this->head_ + ((seq - this->va_) & 7)) % this->max_queue_length_; }
+  uint8_t *slot_(uint8_t seq) { return this->tx_slots_ + this->slot_index_(seq) * this->max_packet_size_; }
+
+  PacketInterface *lower_;
+  // max_queue_length_ slots of max_packet_size_ bytes, a ring starting at head_ which holds frame va_; allocated in
+  // setup() and never freed
+  uint8_t *tx_slots_{nullptr};
+  std::array<uint16_t, MAX_WINDOW> tx_lengths_{};  // indexed like the slots
+  uint32_t timeout_ms_{200};
+  uint32_t ack_delay_ms_{0};
+  uint8_t max_queue_length_{4};
+  uint8_t max_retries_{5};
+  uint8_t retries_{0};
+  uint8_t head_{0};
+  uint8_t vs_{0};  // sequence number of the next new frame
+  uint8_t vn_{0};  // next frame to send; frames from va_ up to it have been sent
+  uint8_t va_{0};  // oldest unacknowledged frame
+  uint8_t vr_{0};  // next frame expected from the peer
+  LinkState state_{LinkState::LINK_STATE_RESET_PENDING};
+  bool ack_pending_{false};  // an RR is owed
+  bool ack_due_{false};      // and should be sent now
+  bool rej_sent_{false};
+  bool peer_frame_seen_{false};  // a numbered frame has arrived since the link came up
+  bool oversize_warned_{false};
+  bool reset_wait_logged_{false};  // the queue filled while waiting for this reset to be answered
+};
+
+}  // namespace esphome::packet_interface
+#endif

esphome/components/packet_interface/layers/__init__.py

@@ -0,0 +1,155 @@
+"""The kinds of layer that can be stacked on a packet interface, and the registry of their implementations.
+
+Each module in this package implements one layer type and registers it with its kind, for example:
+
+    @RELIABILITY.register("hdlc", schema=HDLC_SCHEMA)
+    async def build_hdlc(config: ConfigType, lower: MockObj) -> MockObj:
+        ...
+
+The packet_interface component imports every module here before it builds its schemas, so a new layer type needs
+only a new module. The kinds, and the order in which they are stacked, are fixed here.
+"""
+
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from typing import Any
+
+import esphome.codegen as cg
+import esphome.config_validation as cv
+from esphome.const import CONF_ENCRYPTION, CONF_KEY, CONF_TYPE
+from esphome.cpp_generator import MockObj
+from esphome.types import ConfigType
+
+packet_interface_ns = cg.esphome_ns.namespace("packet_interface")
+PacketInterface = packet_interface_ns.class_("PacketInterface", cg.Component)
+
+CONF_RELIABILITY = "reliability"
+CONF_SESSION = "session"
+
+# Builds a layer above lower and returns it
+LayerBuilder = Callable[[ConfigType, MockObj], Awaitable[Any]]
+
+
+def _never(config: ConfigType) -> bool:
+    return False
+
+
+def _nothing(config: ConfigType) -> list[str]:
+    return []
+
+
+@dataclass(frozen=True)
+class LayerType:
+    """One implementation of a kind of layer."""
+
+    schema: Any
+    build: LayerBuilder
+    # True if this configuration works only when the peer sends packets back, which a session layer makes sure of
+    needs_session: Callable[[ConfigType], bool]
+    # The components this configuration needs loaded
+    auto_load: Callable[[ConfigType], list[str]]
+
+
+class LayerKind:
+    """A kind of layer, such as encryption, with the types that implement it."""
+
+    def __init__(
+        self, key: str, default_type: str, shorthand: str | None = None
+    ) -> None:
+        self.key = key
+        self.default_type = default_type
+        # The option that a plain string value sets, as in `encryption: my key`
+        self.shorthand = shorthand
+        self.types: dict[str, LayerType] = {}
+        # Set once the schema is built, as types registered after that would not be in it
+        self.frozen = False
+
+    def register(
+        self,
+        name: str,
+        *,
+        schema: Any,
+        needs_session: Callable[[ConfigType], bool] = _never,
+        auto_load: Callable[[ConfigType], list[str]] = _nothing,
+    ) -> Callable[[LayerBuilder], LayerBuilder]:
+        """Register a type with its schema."""
+
+        def decorator(build: LayerBuilder) -> LayerBuilder:
+            if self.frozen:
+                raise ValueError(
+                    f"{self.key} layer type '{name}' is registered after the schemas were built; register it "
+                    "before the packet_interface component is imported"
+                )
+            if name in self.types:
+                raise ValueError(
+                    f"{self.key} layer type '{name}' is already registered"
+                )
+            self.types[name] = LayerType(schema, build, needs_session, auto_load)
+            return build
+
+        return decorator
+
+    def _expand(self, value: Any) -> Any:
+        """Allow a layer to be enabled with no options, as in `reliability:`, or with its shorthand."""
+        if value is None:
+            return {}
+        if self.shorthand is not None and isinstance(value, str):
+            return {self.shorthand: value}
+        return value
+
+    def schema(self) -> Callable[[Any], ConfigType]:
+        """The schema for this kind; no type can be registered after this."""
+        self.frozen = True
+        return cv.All(
+            self._expand,
+            cv.typed_schema(
+                {name: layer.schema for name, layer in self.types.items()},
+                default_type=self.default_type,
+            ),
+        )
+
+    def get_type(self, config: ConfigType) -> LayerType:
+        return self.types[config[CONF_TYPE]]
+
+
+ENCRYPTION = LayerKind(CONF_ENCRYPTION, "xxtea", shorthand=CONF_KEY)
+RELIABILITY = LayerKind(CONF_RELIABILITY, "hdlc")
+SESSION = LayerKind(CONF_SESSION, "basic")
+# The kinds stacked above the link, from the bottom up
+STACK_ORDER = (ENCRYPTION, RELIABILITY, SESSION)
+
+
+def stack_schema() -> dict:
+    """Schema entries for the layers of one stack; call only once all layer types are registered."""
+    return {cv.Optional(kind.key): kind.schema() for kind in STACK_ORDER}
+
+
+def check_session(config: ConfigType) -> None:
+    """Layers that need packets back from the peer need a session layer, whose keep-alives send them."""
+    if CONF_SESSION in config:
+        return
+    for kind in STACK_ORDER:
+        if (layer := config.get(kind.key)) is not None and kind.get_type(
+            layer
+        ).needs_session(layer):
+            raise cv.Invalid(
+                f"This '{kind.key}' needs packets from the peer, so it needs a '{CONF_SESSION}' layer, whose "
+                "keep-alives make sure they are sent",
+                path=[kind.key],
+            )
+
+
+def stack_auto_load(config: ConfigType) -> set[str]:
+    """The components that the layers configured in config need loaded."""
+    loads: set[str] = set()
+    for kind in STACK_ORDER:
+        if (layer := config.get(kind.key)) is not None:
+            loads.update(kind.get_type(layer).auto_load(layer))
+    return loads
+
+
+async def build_stack(config: ConfigType, lower: MockObj) -> None:
+    """Create the layers configured in config, from the bottom up, above lower."""
+    for kind in STACK_ORDER:
+        if (layer := config.get(kind.key)) is not None:
+            lower = await kind.get_type(layer).build(layer, lower)

esphome/components/packet_interface/layers/basic_session.py

@@ -0,0 +1,70 @@
+"""Session layer that sends keep-alives and reports whether the peer can be reached."""
+
+import esphome.codegen as cg
+from esphome.components import binary_sensor
+import esphome.config_validation as cv
+from esphome.const import (
+    CONF_BINARY_SENSOR,
+    CONF_ID,
+    CONF_TIMEOUT,
+    DEVICE_CLASS_CONNECTIVITY,
+    ENTITY_CATEGORY_DIAGNOSTIC,
+)
+from esphome.cpp_generator import MockObj
+from esphome.types import ConfigType
+
+from . import SESSION, PacketInterface, packet_interface_ns
+
+CONF_KEEP_ALIVE = "keep_alive"
+
+BasicSession = packet_interface_ns.class_("BasicSession", PacketInterface)
+
+
+def _session_timeout(config: ConfigType) -> ConfigType:
+    """Default the timeout to three keep-alive intervals, and require it to be longer than one."""
+    keep_alive = config[CONF_KEEP_ALIVE].total_milliseconds
+    if (timeout := config.get(CONF_TIMEOUT)) is None:
+        config[CONF_TIMEOUT] = cv.positive_time_period_milliseconds(
+            f"{keep_alive * 3}ms"
+        )
+    elif timeout.total_milliseconds <= keep_alive:
+        raise cv.Invalid("Must be longer than keep_alive", path=[CONF_TIMEOUT])
+    return config
+
+
+BASIC_SESSION_SCHEMA = cv.All(
+    cv.Schema(
+        {
+            cv.GenerateID(): cv.declare_id(BasicSession),
+            cv.Optional(CONF_KEEP_ALIVE, default="10s"): cv.All(
+                cv.positive_time_period_milliseconds,
+                cv.Range(min=cv.TimePeriod(seconds=1), max=cv.TimePeriod(hours=1)),
+            ),
+            cv.Optional(CONF_TIMEOUT): cv.All(
+                cv.positive_time_period_milliseconds,
+                cv.Range(max=cv.TimePeriod(hours=3)),
+            ),
+            cv.Optional(CONF_BINARY_SENSOR): binary_sensor.binary_sensor_schema(
+                device_class=DEVICE_CLASS_CONNECTIVITY,
+                entity_category=ENTITY_CATEGORY_DIAGNOSTIC,
+            ),
+        }
+    ).extend(cv.COMPONENT_SCHEMA),
+    _session_timeout,
+)
+
+
+@SESSION.register(
+    "basic",
+    schema=BASIC_SESSION_SCHEMA,
+    auto_load=lambda config: ["binary_sensor"] if CONF_BINARY_SENSOR in config else [],
+)
+async def build_basic_session(config: ConfigType, lower: MockObj) -> MockObj:
+    cg.add_define("USE_PACKET_INTERFACE_BASIC_SESSION")
+    var = cg.new_Pvariable(config[CONF_ID], lower)
+    cg.add(var.set_keep_alive(config[CONF_KEEP_ALIVE]))
+    cg.add(var.set_session_timeout(config[CONF_TIMEOUT]))
+    await cg.register_component(var, config)
+    binary_sensors = binary_sensor.sub_binary_sensors(config)
+    await binary_sensors(CONF_BINARY_SENSOR, var.set_connected_sensor)
+    return var

esphome/components/packet_interface/layers/hdlc_reliability.py

@@ -0,0 +1,59 @@
+"""Reliability layer using HDLC style acknowledgement and go-back-N retransmission."""
+
+import esphome.codegen as cg
+import esphome.config_validation as cv
+from esphome.const import CONF_ID, CONF_TIMEOUT
+from esphome.cpp_generator import MockObj
+from esphome.types import ConfigType
+
+from . import RELIABILITY, PacketInterface, packet_interface_ns
+
+CONF_ACK_DELAY = "ack_delay"
+CONF_MAX_QUEUE_LENGTH = "max_queue_length"
+CONF_MAX_RETRIES = "max_retries"
+
+HdlcReliability = packet_interface_ns.class_("HdlcReliability", PacketInterface)
+
+
+def _ack_delay_within_timeout(config: ConfigType) -> ConfigType:
+    """The peer, configured the same, must hear the acknowledgement before its timeout."""
+    if (
+        config[CONF_ACK_DELAY].total_milliseconds
+        >= config[CONF_TIMEOUT].total_milliseconds
+    ):
+        raise cv.Invalid(
+            f"Must be shorter than '{CONF_TIMEOUT}'", path=[CONF_ACK_DELAY]
+        )
+    return config
+
+
+HDLC_SCHEMA = cv.All(
+    cv.Schema(
+        {
+            cv.GenerateID(): cv.declare_id(HdlcReliability),
+            # Retransmitting more often than the main loop runs would exhaust the retries at once
+            cv.Optional(CONF_TIMEOUT, default="200ms"): cv.All(
+                cv.positive_time_period_milliseconds,
+                cv.Range(min=cv.TimePeriod(milliseconds=50)),
+            ),
+            cv.Optional(
+                CONF_ACK_DELAY, default="0ms"
+            ): cv.positive_time_period_milliseconds,
+            cv.Optional(CONF_MAX_QUEUE_LENGTH, default=4): cv.int_range(1, 7),
+            cv.Optional(CONF_MAX_RETRIES, default=5): cv.int_range(1, 255),
+        }
+    ).extend(cv.COMPONENT_SCHEMA),
+    _ack_delay_within_timeout,
+)
+
+
+@RELIABILITY.register("hdlc", schema=HDLC_SCHEMA)
+async def build_hdlc(config: ConfigType, lower: MockObj) -> MockObj:
+    cg.add_define("USE_PACKET_INTERFACE_HDLC")
+    var = cg.new_Pvariable(config[CONF_ID], lower)
+    cg.add(var.set_ack_timeout(config[CONF_TIMEOUT]))
+    cg.add(var.set_ack_delay(config[CONF_ACK_DELAY]))
+    cg.add(var.set_max_queue_length(config[CONF_MAX_QUEUE_LENGTH]))
+    cg.add(var.set_max_retries(config[CONF_MAX_RETRIES]))
+    await cg.register_component(var, config)
+    return var

esphome/components/packet_interface/layers/xxtea_encryption.py

@@ -0,0 +1,50 @@
+"""Encryption layer using XXTEA, with an optional rolling code."""
+
+import hashlib
+
+import esphome.codegen as cg
+import esphome.config_validation as cv
+from esphome.const import CONF_ID, CONF_KEY
+from esphome.cpp_generator import MockObj
+from esphome.types import ConfigType
+
+from . import ENCRYPTION, PacketInterface, packet_interface_ns
+
+CONF_ROLLING_CODE_ENABLE = "rolling_code_enable"
+
+# A captured packet lets a guessed key be checked offline, so the key text is stretched to make each guess costly.
+# Both ends derive the same key from the same text, so the salt is fixed; it keeps tables built for other uses of
+# PBKDF2 from applying here.
+KEY_SALT = b"esphome packet_interface xxtea"
+KEY_ITERATIONS = 600_000
+
+XxteaEncryption = packet_interface_ns.class_("XxteaEncryption", PacketInterface)
+
+XXTEA_SCHEMA = cv.Schema(
+    {
+        cv.GenerateID(): cv.declare_id(XxteaEncryption),
+        cv.Required(CONF_KEY): cv.All(cv.sensitive(cv.string), cv.Length(min=1)),
+        cv.Optional(CONF_ROLLING_CODE_ENABLE, default=False): cv.boolean,
+    }
+).extend(cv.COMPONENT_SCHEMA)
+
+
+def derive_key(text: str) -> list[int]:
+    """The 256-bit key for the key text, as 32-bit words in little endian order."""
+    digest = hashlib.pbkdf2_hmac("sha256", text.encode(), KEY_SALT, KEY_ITERATIONS)
+    return [int.from_bytes(digest[i : i + 4], "little") for i in range(0, 32, 4)]
+
+
+@ENCRYPTION.register(
+    "xxtea",
+    schema=XXTEA_SCHEMA,
+    # The rolling code accepts a packet only once the peer has sent back this device's boot number
+    needs_session=lambda config: config[CONF_ROLLING_CODE_ENABLE],
+    auto_load=lambda config: ["xxtea"],
+)
+async def build_xxtea(config: ConfigType, lower: MockObj) -> MockObj:
+    cg.add_define("USE_PACKET_INTERFACE_XXTEA")
+    var = cg.new_Pvariable(config[CONF_ID], lower, config[CONF_ROLLING_CODE_ENABLE])
+    cg.add(var.set_key(derive_key(config[CONF_KEY])))
+    await cg.register_component(var, config)
+    return var

esphome/components/packet_interface/xxtea_encryption.cpp

@@ -0,0 +1,182 @@
+#include "xxtea_encryption.h"
+#ifdef USE_PACKET_INTERFACE_XXTEA
+#include <algorithm>
+#include <cstring>
+#include "esphome/components/xxtea/xxtea.h"
+#include "esphome/core/helpers.h"
+#include "esphome/core/log.h"
+
+namespace esphome::packet_interface {
+
+ESPHOME_LOG_TAG(TAG, "packet_interface.xxtea");
+
+static constexpr uint32_t PREF_HASH = fnv1a_hash("packet_interface.xxtea");
+// Different for each mode, so devices set up differently drop each other's packets
+static constexpr uint8_t CHECK_BYTES[3] = {0x58, 0x54, 0x45};
+static constexpr uint8_t ROLLING_CHECK_BYTES[3] = {0x58, 0x54, 0x52};
+static constexpr size_t CHECK_SIZE = 4;
+// Boot number, counter, and the echo of the receiver's boot number
+static constexpr size_t ROLLING_SIZE = 12;
+// XXTEA needs at least two words
+static constexpr size_t MIN_LENGTH = 8;
+static constexpr size_t MAX_PADDING = 4;
+
+static void put_uint32(uint8_t *buf, uint32_t value) {
+  buf[0] = value;
+  buf[1] = value >> 8;
+  buf[2] = value >> 16;
+  buf[3] = value >> 24;
+}
+
+static uint32_t get_uint32(const uint8_t *buf) { return encode_uint32(buf[3], buf[2], buf[1], buf[0]); }
+
+static constexpr size_t trailer_size(bool rolling_code) {
+  return rolling_code ? ROLLING_SIZE + CHECK_SIZE : CHECK_SIZE;
+}
+
+XxteaEncryption::XxteaEncryption(PacketInterface *lower, bool rolling_code)
+    // XXTEA works on whole 32-bit words
+    : PacketInterface(remaining_packet_size(lower->get_max_packet_size() & ~size_t{3}, trailer_size(rolling_code))),
+      lower_(lower),
+      trailer_size_(trailer_size(rolling_code)),
+      rolling_code_(rolling_code) {
+  this->stack_on_(lower, CAPABILITY_ENCRYPTION);
+}
+
+void XxteaEncryption::set_key(std::initializer_list<uint32_t> key) {
+  std::copy_n(key.begin(), std::min(key.size(), this->key_.size()), this->key_.begin());
+}
+
+void XxteaEncryption::setup() {
+  if (this->max_packet_size_ == 0) {
+    this->mark_failed(LOG_STR("max_packet_size too small for the configured layers"));
+    return;
+  }
+  size_t words = this->lower_->get_max_packet_size() / 4;
+  RAMAllocator<uint32_t> allocator;
+  this->tx_buffer_ = allocator.allocate(words * 2);
+  if (this->tx_buffer_ == nullptr) {
+    this->mark_failed(LOG_STR("Not enough memory for the packet buffers"));
+    return;
+  }
+  this->rx_buffer_ = this->tx_buffer_ + words;
+  if (this->rolling_code_) {
+    this->pref_ = global_preferences->make_preference<uint32_t>(fnv1a_hash_extend(PREF_HASH, this->name_), true);
+    this->pref_.load(&this->boot_);
+    this->next_boot_();
+  }
+  this->lower_->add_packet_interface_listener(
+      [this](const PacketBuffer &data, const PacketMetaData &meta_data) { this->receive_(data, meta_data); });
+}
+
+void XxteaEncryption::dump_config() {
+  ESP_LOGCONFIG(TAG,
+                "XXTEA Encryption on '%s':\n"
+                "  Max Packet Size: %zu\n"
+                "  Rolling Code: %s",
+                this->name_, this->max_packet_size_, YESNO(this->rolling_code_));
+  if (this->rolling_code_) {
+    ESP_LOGCONFIG(TAG, "  Boot Number: %" PRIu32, this->boot_);
+  }
+}
+
+// A new boot number must be saved before it is used, or a crash could make the peer see it again
+void XxteaEncryption::next_boot_() {
+  this->boot_++;
+  this->counter_ = 0;
+  if (!this->pref_.save(&this->boot_) || !global_preferences->sync()) {
+    ESP_LOGW(TAG, "%s: Could not save the boot number, so a restart could reuse it", this->name_);
+  }
+}
+
+bool XxteaEncryption::send_to_interface(const PacketBuffer &data, const PacketMetaData &meta_data) {
+  if (this->tx_buffer_ == nullptr)
+    return false;
+  size_t size = data.size();
+  if (size > this->max_packet_size_) {
+    ESP_LOGW(TAG, "%s: Packet of %zu bytes is larger than %zu", this->name_, size, this->max_packet_size_);
+    return false;
+  }
+  size_t padding = (4 - (size + this->trailer_size_) % 4) % 4;
+  if (size + padding + this->trailer_size_ < MIN_LENGTH)
+    padding += 4;
+  size_t length = size + padding + this->trailer_size_;
+  auto *buf = reinterpret_cast<uint8_t *>(this->tx_buffer_);
+  data.copy_to(buf, size);
+  memset(buf + size, 0, padding);
+  uint8_t *check = buf + length - CHECK_SIZE;
+  if (this->rolling_code_) {
+    put_uint32(check - ROLLING_SIZE, this->boot_);
+    put_uint32(check - ROLLING_SIZE + 4, this->counter_);
+    put_uint32(check - ROLLING_SIZE + 8, this->echo_boot_);
+    memcpy(check, ROLLING_CHECK_BYTES, sizeof(ROLLING_CHECK_BYTES));
+    if (++this->counter_ == 0)
+      this->next_boot_();
+  } else {
+    memcpy(check, CHECK_BYTES, sizeof(CHECK_BYTES));
+  }
+  check[CHECK_SIZE - 1] = padding;
+  xxtea::encrypt(this->tx_buffer_, length / 4, this->key_.data());
+  return this->lower_->send_to_interface(PacketBuffer(buf, length), meta_data);
+}
+
+bool XxteaEncryption::accept_counter_(uint32_t boot, uint32_t counter) {
+  if (this->peer_seen_ && (boot < this->peer_boot_ || (boot == this->peer_boot_ && counter <= this->peer_counter_)))
+    return false;
+  this->peer_seen_ = true;
+  this->peer_boot_ = boot;
+  this->peer_counter_ = counter;
+  return true;
+}
+
+void XxteaEncryption::receive_(const PacketBuffer &data, const PacketMetaData &meta_data) {
+  size_t length = data.size();
+  if (length % 4 != 0 || length < std::max(MIN_LENGTH, this->trailer_size_) ||
+      length > this->lower_->get_max_packet_size()) {
+    ESP_LOGV(TAG, "%s: Packet of %zu bytes dropped", this->name_, length);
+    this->note_received_(false);
+    return;
+  }
+  auto *buf = reinterpret_cast<uint8_t *>(this->rx_buffer_);
+  data.copy_to(buf, length);
+  xxtea::decrypt(this->rx_buffer_, length / 4, this->key_.data());
+  const uint8_t *check = buf + length - CHECK_SIZE;
+  const uint8_t *expected = this->rolling_code_ ? ROLLING_CHECK_BYTES : CHECK_BYTES;
+  size_t padding = check[CHECK_SIZE - 1];
+  if (memcmp(check, expected, sizeof(CHECK_BYTES)) != 0 || padding > MAX_PADDING ||
+      padding > length - this->trailer_size_) {
+    ESP_LOGV(TAG, "%s: Packet failed authentication", this->name_);
+    this->note_received_(false);
+    return;
+  }
+  this->note_received_(true);
+  if (this->rolling_code_) {
+    uint32_t boot = get_uint32(check - ROLLING_SIZE);
+    uint32_t counter = get_uint32(check - ROLLING_SIZE + 4);
+    uint32_t echo = get_uint32(check - ROLLING_SIZE + 8);
+    // Learn the peer's boot number even from a packet that is then dropped, so both sides can catch up
+    if (boot > this->echo_boot_)
+      this->echo_boot_ = boot;
+    // A packet recorded before this boot cannot carry this boot's number
+    if (echo != this->boot_) {
+      ESP_LOGV(TAG, "%s: Packet for boot %" PRIu32 " dropped", this->name_, echo);
+      return;
+    }
+    if (!this->accept_counter_(boot, counter)) {
+      // A boot number lower than one already heard means the peer lost its saved boot number, or a replay
+      if (boot < this->peer_boot_ && !this->old_boot_warned_) {
+        this->old_boot_warned_ = true;
+        ESP_LOGW(TAG,
+                 "%s: Packets from an older boot %" PRIu32 " of the peer are dropped; if its flash was erased, "
+                 "restart this device",
+                 this->name_, boot);
+      }
+      ESP_LOGV(TAG, "%s: Old packet %" PRIu32 ":%" PRIu32 " dropped", this->name_, boot, counter);
+      return;
+    }
+  }
+  this->on_receive_from_interface_(PacketBuffer(buf, length - this->trailer_size_ - padding), meta_data);
+}
+
+}  // namespace esphome::packet_interface
+#endif

esphome/components/packet_interface/xxtea_encryption.h

@@ -0,0 +1,61 @@
+#pragma once
+
+#include "esphome/core/defines.h"
+#ifdef USE_PACKET_INTERFACE_XXTEA
+#include <array>
+#include <initializer_list>
+
+#include "esphome/core/preferences.h"
+#include "packet_interface.h"
+
+namespace esphome::packet_interface {
+
+/**
+ * An encryption layer that encrypts and authenticates each packet with XXTEA and a 256-bit shared key.
+ *
+ * The packet is padded with zeros and followed by a trailer, then the whole is encrypted as one block. With the
+ * rolling code enabled, the trailer starts with:
+ * - a 32-bit boot number, saved in flash and increased at every boot;
+ * - a 32-bit packet counter, starting at 0 at every boot;
+ * - the highest boot number heard from the peer, echoed back to it.
+ * The trailer always ends with 3 fixed check bytes and the number of padding bytes. XXTEA mixes the whole block, so
+ * a packet that was changed or encrypted with another key fails the check.
+ * With the rolling code, a packet is dropped as a replay unless it echoes this device's current boot number, and its
+ * boot number and counter are higher than those of the last packet accepted from the peer. A boot number is never
+ * reused, so the echo acts as a fresh challenge after every restart.
+ */
+class XxteaEncryption : public PacketInterface {
+ public:
+  XxteaEncryption(PacketInterface *lower, bool rolling_code);
+
+  void setup() override;
+  void dump_config() override;
+  bool send_to_interface(const PacketBuffer &data, const PacketMetaData &meta_data) override;
+
+  void set_key(std::initializer_list<uint32_t> key);
+
+ protected:
+  void receive_(const PacketBuffer &data, const PacketMetaData &meta_data);
+  bool accept_counter_(uint32_t boot, uint32_t counter);
+  void next_boot_();
+
+  PacketInterface *lower_;
+  const size_t trailer_size_;
+  std::array<uint32_t, 8> key_{};
+  // Separate buffers, as a layer above may send a reply while a received packet is being passed up; both are in one
+  // block allocated in setup(), and tx_buffer_ stays null if that fails
+  uint32_t *tx_buffer_{nullptr};
+  uint32_t *rx_buffer_{nullptr};
+  ESPPreferenceObject pref_;
+  uint32_t boot_{0};
+  uint32_t counter_{0};
+  uint32_t peer_boot_{0};
+  uint32_t peer_counter_{0};
+  uint32_t echo_boot_{0};  // highest boot number heard from the peer
+  bool peer_seen_{false};
+  bool old_boot_warned_{false};
+  const bool rolling_code_;
+};
+
+}  // namespace esphome::packet_interface
+#endif

esphome/components/udp/packet_interface/__init__.py

@@ -0,0 +1,40 @@
+import esphome.codegen as cg
+from esphome.components import packet_interface
+from esphome.components.packet_interface import (
+    CONF_MAX_PACKET_SIZE,
+    new_packet_interface,
+    packet_interface_schema,
+)
+from esphome.const import CONF_ID
+from esphome.types import ConfigType
+
+from .. import CONF_UDP_ID, UDP_SCHEMA, udp_ns
+
+CODEOWNERS = ["@clydebarrow"]
+
+DEPENDENCIES = ["udp"]
+
+# Size of the receive buffer of the udp component
+MAX_PACKET_SIZE = 508
+
+UdpPacketInterface = udp_ns.class_(
+    "UdpPacketInterface",
+    packet_interface.PacketInterface,
+)
+
+CONFIG_SCHEMA = packet_interface_schema(
+    UdpPacketInterface, MAX_PACKET_SIZE, MAX_PACKET_SIZE
+).extend(UDP_SCHEMA)
+
+
+async def to_code(config: ConfigType) -> None:
+    udp_var = await cg.get_variable(config[CONF_UDP_ID])
+    cg.add(udp_var.set_should_listen())
+    # The udp component sends to its addresses through its broadcast socket
+    cg.add(udp_var.set_should_broadcast())
+    # The copy buffer's size is known here, so it is a static array rather than allocated at run time
+    tx_buffer = f"{config[CONF_ID].id}__tx_buffer"
+    cg.add_global(
+        cg.RawStatement(f"static uint8_t {tx_buffer}[{config[CONF_MAX_PACKET_SIZE]}];")
+    )
+    await new_packet_interface(config, udp_var, cg.RawExpression(tx_buffer))

esphome/components/udp/packet_interface/udp_packet_interface.cpp

@@ -0,0 +1,43 @@
+#include "udp_packet_interface.h"
+#ifdef USE_NETWORK
+#include "esphome/components/network/util.h"
+#include "esphome/core/log.h"
+
+namespace esphome::udp {
+
+using packet_interface::PacketBuffer;
+using packet_interface::PacketMetaData;
+
+ESPHOME_LOG_TAG(TAG, "udp.packet_interface");
+
+void UdpPacketInterface::setup() {
+  this->parent_->add_listener([this](std::span<const uint8_t> data) { this->receive_(data); });
+}
+
+void UdpPacketInterface::receive_(std::span<const uint8_t> data) {
+  if (data.size() > this->max_packet_size_) {
+    ESP_LOGV(TAG, "%s: Packet of %zu bytes is larger than %zu, dropped", this->name_, data.size(),
+             this->max_packet_size_);
+    return;
+  }
+  this->on_receive_from_interface_(PacketBuffer(data.data(), data.size()), {});
+}
+
+bool UdpPacketInterface::send_to_interface(const PacketBuffer &data, const PacketMetaData &meta_data) {
+  size_t size = data.size();
+  if (size > this->max_packet_size_) {
+    ESP_LOGW(TAG, "%s: Packet of %zu bytes is larger than %zu", this->name_, size, this->max_packet_size_);
+    return false;
+  }
+  if (!network::is_connected())
+    return false;
+  if (data.is_contiguous()) {
+    auto span = data.as_span();
+    return this->parent_->send_packet(span.data(), span.size());
+  }
+  data.copy_to(this->tx_buffer_.data(), size);
+  return this->parent_->send_packet(this->tx_buffer_.data(), size);
+}
+
+}  // namespace esphome::udp
+#endif

esphome/components/udp/packet_interface/udp_packet_interface.h

@@ -0,0 +1,37 @@
+#pragma once
+
+#include "esphome/core/defines.h"
+
+// udp_component.h declares nothing without USE_NETWORK
+#ifdef USE_NETWORK
+#include <span>
+
+#include "esphome/components/packet_interface/packet_interface.h"
+#include "../udp_component.h"
+
+namespace esphome::udp {
+
+/**
+ * A packet interface link over UDP: each packet is one datagram, sent to every address of the UDP component.
+ *
+ * Every datagram received on the listen port is passed up, from any sender, so on a shared network an addressing
+ * or encryption layer is needed to tell the peer's packets from others.
+ */
+class UdpPacketInterface : public packet_interface::PacketInterface {
+ public:
+  /// @param tx_buffer Joins packets split across buffers; its size is the largest packet, and it must outlive the link
+  UdpPacketInterface(UDPComponent *udp_component, std::span<uint8_t> tx_buffer)
+      : packet_interface::PacketInterface(tx_buffer.size()), parent_(udp_component), tx_buffer_(tx_buffer) {}
+
+  void setup() override;
+  bool send_to_interface(const packet_interface::PacketBuffer &data,
+                         const packet_interface::PacketMetaData &meta_data) override;
+
+ protected:
+  void receive_(std::span<const uint8_t> data);
+
+  UDPComponent *parent_;
+  const std::span<uint8_t> tx_buffer_;
+};
+}  // namespace esphome::udp
+#endif

esphome/components/udp/packet_transport/udp_transport.cpp

@@ -1,12 +1,9 @@
-#include "esphome/core/log.h"
 #include "esphome/core/application.h"
 #include "esphome/components/network/util.h"
 #include "udp_transport.h"
 
 namespace esphome::udp {
 
-ESPHOME_LOG_TAG(TAG, "udp_transport");
-
 bool UDPTransport::should_send() { return network::is_connected(); }
 void UDPTransport::setup() {
   PacketTransport::setup();

esphome/components/udp/udp_component.cpp

@@ -205,18 +205,25 @@ void UDPComponent::set_poll_warning_(PollStep step, const LogString *what) {
 }
 
 void UDPComponent::loop() {
-  if (this->should_listen_) {
-    std::array<uint8_t, MAX_PACKET_SIZE> buf;
-    for (;;) {
-      if (!this->listen_socket_->ready())
-        break;
-      auto len = this->listen_socket_->read(buf.data(), buf.size());
-      if (len <= 0)
-        break;
-      size_t packet_len = static_cast<size_t>(len);
-      ESP_LOGV(TAG, "Received packet of length %zu", packet_len);
-      this->packet_listeners_.call(std::span<const uint8_t>(buf.data(), packet_len));
+  if (!this->should_listen_) {
+    this->disable_loop();
+    return;
+  }
+  // One byte more than the largest packet, as a read cuts a longer datagram to fit and so cannot report it
+  std::array<uint8_t, MAX_PACKET_SIZE + 1> buf;
+  for (;;) {
+    if (!this->listen_socket_->ready())
+      break;
+    auto len = this->listen_socket_->read(buf.data(), buf.size());
+    if (len <= 0)
+      break;
+    size_t packet_len = static_cast<size_t>(len);
+    if (packet_len > MAX_PACKET_SIZE) {
+      ESP_LOGV(TAG, "Packet longer than %zu bytes dropped", MAX_PACKET_SIZE);
+      continue;
     }
+    ESP_LOGV(TAG, "Received packet of length %zu", packet_len);
+    this->packet_listeners_.call(std::span<const uint8_t>(buf.data(), packet_len));
   }
 }
 
@@ -239,9 +246,10 @@ void UDPComponent::dump_config() {
                 YESNO(this->should_broadcast_), YESNO(this->should_listen_));
 }
 
-void UDPComponent::send_packet(const uint8_t *data, size_t size) {
+bool UDPComponent::send_packet(const uint8_t *data, size_t size) {
   if (this->is_failed())
-    return;  // a send socket may be missing
+    return false;  // a send socket may be missing
+  bool sent = true;
   for (const auto &entry : this->sockaddrs_) {
 #ifdef USE_UDP_IPV6_SEND
     const bool v6 = entry.sa.sa_family == AF_INET6;
@@ -256,11 +264,23 @@ void UDPComponent::send_packet(const uint8_t *data, size_t size) {
     socket::UDPSendSocket *sock = this->broadcast_socket_.get();
     socklen_t sockaddr_size = sizeof(sockaddr_in);
 #endif
+    if (sock == nullptr) {
+      ESP_LOGW(TAG, "Broadcast socket not initialized");
+      return false;
+    }
     auto result = sock->sendto(data, size, 0, &entry.sa, sockaddr_size);
     if (result < 0) {
-      ESP_LOGW(TAG, "sendto() error %d", errno);
+      // lwIP refuses a packet with ENOMEM while the destination's ARP lookup is pending and its queue is full;
+      // the packet can be sent again later
+      if (errno == ENOMEM) {
+        ESP_LOGV(TAG, "sendto() out of memory");
+      } else {
+        ESP_LOGW(TAG, "sendto() error %d", errno);
+      }
+      sent = false;
     }
   }
+  return sent;
 }
 }  // namespace esphome::udp
 

esphome/components/udp/udp_component.h

@@ -35,8 +35,9 @@ class UDPComponent final : public Component {
   void setup() override;
   void loop() override;
   void dump_config() override;
-  void send_packet(const uint8_t *data, size_t size);
-  void send_packet(const std::vector<uint8_t> &buf) { this->send_packet(buf.data(), buf.size()); }
+  /// Returns false if the packet could not be sent to every destination.
+  bool send_packet(const uint8_t *data, size_t size);
+  bool send_packet(const std::vector<uint8_t> &buf) { return this->send_packet(buf.data(), buf.size()); }
   float get_setup_priority() const override { return setup_priority::AFTER_WIFI; };
 
  protected:
@@ -53,7 +54,7 @@ class UDPComponent final : public Component {
   bool network_poll_();
   void set_poll_warning_(PollStep step, const LogString *what);
 
-  CallbackManager<void(std::span<const uint8_t>)> packet_listeners_{};
+  LazyCallbackManager<void(std::span<const uint8_t>)> packet_listeners_{};
   std::unique_ptr<socket::UDPSendSocket> broadcast_socket_ = nullptr;
   std::unique_ptr<socket::UDPSocket> listen_socket_ = nullptr;
 #ifdef USE_UDP_IPV6_SEND

esphome/core/defines.h

@@ -266,6 +266,9 @@
 #define USE_OTA_STATE_LISTENER
 #define USE_OUTPUT
 #define USE_OUTPUT_FLOAT_POWER_SCALING
+#define USE_PACKET_INTERFACE_BASIC_SESSION
+#define USE_PACKET_INTERFACE_HDLC
+#define USE_PACKET_INTERFACE_XXTEA
 #define USE_PCA9554_PCAL
 #define USE_POWER_SUPPLY
 // Only defined by key-lookup preference backends; the slot-based platforms

tests/components/packet_interface/__init__.py

@@ -0,0 +1,11 @@
+import esphome.codegen as cg
+from tests.testing_helpers import ComponentManifestOverride
+
+
+def override_manifest(manifest: ComponentManifestOverride) -> None:
+    async def to_code_testing(config: object) -> None:
+        cg.add_define("USE_PACKET_INTERFACE_BASIC_SESSION")
+        cg.add_define("USE_PACKET_INTERFACE_HDLC")
+        cg.add_define("USE_PACKET_INTERFACE_XXTEA")
+
+    manifest.to_code = to_code_testing

tests/components/packet_interface/basic_session.cpp

@@ -0,0 +1,252 @@
+#include <gtest/gtest.h>
+#include <cstdint>
+#include <memory>
+#include <vector>
+#include "esphome/components/packet_interface/basic_session.h"
+#include "esphome/core/application.h"
+
+namespace esphome::packet_interface::session_testing {
+
+// Copies a packet into a vector, for comparing in tests
+static std::vector<uint8_t> to_vector(const esphome::packet_interface::PacketBuffer &data) {
+  std::vector<uint8_t> result(data.size());
+  data.copy_to(result.data(), result.size());
+  return result;
+}
+
+using Packet = std::vector<uint8_t>;
+
+// A link layer that records what is sent, so a test can deliver it to the peer
+class MockLink : public PacketInterface {
+ public:
+  explicit MockLink(size_t max_packet_size = 64) : PacketInterface(max_packet_size) {}
+  bool send_to_interface(const PacketBuffer &data, const PacketMetaData &) override {
+    if (!this->accept)
+      return false;
+    this->sent.push_back(to_vector(data));
+    return true;
+  }
+  void inject(const Packet &packet) { this->on_receive_from_interface_(PacketBuffer(packet), {}); }
+  // Act as a reliability layer that resets and discards data
+  void set_reliable() { this->capabilities_ = CAPABILITY_RELIABILITY; }
+  void report_data_lost() { this->session_callback_.call(false); }
+
+  std::vector<Packet> sent;
+  bool accept{true};
+};
+
+class TestSession : public BasicSession {
+ public:
+  using BasicSession::BasicSession;
+  using BasicSession::on_interval_;
+  unsigned get_not_understood_count() const { return this->not_understood_count_; }
+};
+
+struct Peer {
+  Peer() {
+    this->session = std::make_unique<TestSession>(&this->link);
+    this->session->set_keep_alive(1000);
+    this->session->set_session_timeout(3000);
+    this->session->add_packet_interface_listener(
+        [this](const PacketBuffer &data, const PacketMetaData &) { this->received.push_back(to_vector(data)); });
+    this->session->add_on_session_state_callback([this](bool connected) { this->states.push_back(connected); });
+    this->session->setup();
+  }
+  bool send(const Packet &data) { return this->session->send_to_interface(PacketBuffer(data), {}); }
+
+  MockLink link;
+  std::unique_ptr<TestSession> session;
+  std::vector<Packet> received;
+  std::vector<bool> states;
+};
+
+static bool is_keep_alive(const Packet &packet) { return !packet.empty() && packet[0] == 0x01; }
+
+// Move packets between the peers until nothing more is sent
+static void pump(Peer &a, Peer &b) {
+  for (int i = 0; i != 20; i++) {
+    if (a.link.sent.empty() && b.link.sent.empty())
+      return;
+    auto from_a = std::move(a.link.sent);
+    auto from_b = std::move(b.link.sent);
+    a.link.sent.clear();
+    b.link.sent.clear();
+    for (auto &packet : from_a)
+      b.link.inject(packet);
+    for (auto &packet : from_b)
+      a.link.inject(packet);
+  }
+  FAIL() << "Peers did not go quiet";
+}
+
+// Each test replaces App with a fresh one; the keep-alive interval needs its scheduler
+class BasicSessionTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    App.~Application();
+    new (&App) Application();
+    static char name[] = "session-tests";
+    static char friendly_name[] = "";
+    App.pre_setup(name, sizeof(name) - 1, friendly_name, 0);
+  }
+  void TearDown() override {
+    App.~Application();
+    new (&App) Application();  // later tests and the test listener use App
+  }
+
+  static void set_time(uint32_t now) { LoopBlockingGuard guard{nullptr, nullptr, now}; }
+};
+
+TEST_F(BasicSessionTest, MaxPacketSizeAllowsForHeader) {
+  Peer a;
+  EXPECT_EQ(a.session->get_max_packet_size(), 64 - BasicSession::HEADER_SIZE);
+  EXPECT_EQ(a.link.top(), a.session.get());
+  EXPECT_EQ(a.session->get_capabilities(), CAPABILITY_SESSION);
+}
+
+TEST_F(BasicSessionTest, KeepAliveSentAtSetup) {
+  Peer a;
+  ASSERT_EQ(a.link.sent.size(), 1);
+  EXPECT_TRUE(is_keep_alive(a.link.sent[0]));
+  EXPECT_EQ(a.link.sent[0].size(), 5);
+  EXPECT_FALSE(a.session->is_connected());
+}
+
+TEST_F(BasicSessionTest, BothConnectWithoutRepeatedAnswers) {
+  Peer a, b;
+  pump(a, b);
+  EXPECT_TRUE(a.session->is_connected());
+  EXPECT_TRUE(b.session->is_connected());
+  EXPECT_EQ(a.states, std::vector<bool>{true});
+  EXPECT_EQ(b.states, std::vector<bool>{true});
+}
+
+TEST_F(BasicSessionTest, OneSideStartsLater) {
+  Peer a;
+  a.link.sent.clear();  // a's first keep-alive is lost
+  Peer b;
+  pump(a, b);
+  EXPECT_TRUE(a.session->is_connected());
+  EXPECT_TRUE(b.session->is_connected());
+}
+
+TEST_F(BasicSessionTest, DataPassesThroughWithHeader) {
+  Peer a, b;
+  pump(a, b);
+  ASSERT_TRUE(a.send({0x10, 0x11}));
+  ASSERT_EQ(a.link.sent.size(), 1);
+  EXPECT_EQ(a.link.sent[0], (Packet{0x00, 0x10, 0x11}));
+  pump(a, b);
+  EXPECT_EQ(b.received, (std::vector<Packet>{{0x10, 0x11}}));
+  EXPECT_TRUE(a.received.empty());
+}
+
+TEST_F(BasicSessionTest, NotUnderstoodPacketsCountedUntilOneIsUnderstood) {
+  Peer a;
+  for (int i = 0; i != 20; i++)
+    a.link.inject({0x7F, 0x00});
+  // The count stops where the mismatch is logged, so it is logged once
+  EXPECT_EQ(a.session->get_not_understood_count(), 10);
+  a.link.inject({0x00, 0x55});
+  EXPECT_EQ(a.session->get_not_understood_count(), 0);
+}
+
+TEST_F(BasicSessionTest, DataMarksPeerConnected) {
+  Peer a;
+  a.link.inject({0x00, 0x42});
+  EXPECT_TRUE(a.session->is_connected());
+  EXPECT_EQ(a.received, (std::vector<Packet>{{0x42}}));
+}
+
+TEST_F(BasicSessionTest, DataBeforeFirstKeepAliveIsNotARestart) {
+  Peer a, b;
+  a.link.sent.clear();
+  b.link.sent.clear();  // b's first keep-alive is lost
+  b.send({0x42});
+  pump(a, b);
+  b.session->on_interval_();
+  pump(a, b);
+  EXPECT_EQ(a.states, std::vector<bool>{true});
+}
+
+TEST_F(BasicSessionTest, SendFailureIsReturned) {
+  Peer a;
+  a.link.accept = false;
+  EXPECT_FALSE(a.send({0x01}));
+}
+
+TEST_F(BasicSessionTest, IntervalSendsKeepAlive) {
+  Peer a;
+  a.link.sent.clear();
+  a.session->on_interval_();
+  ASSERT_EQ(a.link.sent.size(), 1);
+  EXPECT_TRUE(is_keep_alive(a.link.sent[0]));
+}
+
+TEST_F(BasicSessionTest, PeerLostAfterTimeout) {
+  Peer a, b;
+  set_time(1000);
+  pump(a, b);
+  set_time(4000);
+  a.session->on_interval_();
+  EXPECT_TRUE(a.session->is_connected());
+  set_time(4001);
+  a.session->on_interval_();
+  EXPECT_FALSE(a.session->is_connected());
+  EXPECT_EQ(a.states, (std::vector<bool>{true, false}));
+  // Heard again
+  b.session->on_interval_();
+  pump(a, b);
+  EXPECT_TRUE(a.session->is_connected());
+  EXPECT_EQ(a.states, (std::vector<bool>{true, false, true}));
+}
+
+TEST_F(BasicSessionTest, PeerRestartIsReported) {
+  Peer a;
+  auto b = std::make_unique<Peer>();
+  pump(a, *b);
+  b = std::make_unique<Peer>();
+  pump(a, *b);
+  EXPECT_TRUE(a.session->is_connected());
+  EXPECT_EQ(a.states, (std::vector<bool>{true, false, true}));
+  EXPECT_EQ(b->states, std::vector<bool>{true});
+}
+
+TEST_F(BasicSessionTest, MalformedPacketsDropped) {
+  Peer a;
+  a.link.inject({});
+  a.link.inject({0x02, 0x00});
+  a.link.inject({0x01, 0x00, 0x00});
+  EXPECT_FALSE(a.session->is_connected());
+  EXPECT_TRUE(a.received.empty());
+}
+
+TEST_F(BasicSessionTest, DataLostBelowReportedAsSessionLost) {
+  MockLink link;
+  link.set_reliable();
+  TestSession session(&link);
+  std::vector<bool> states;
+  session.add_on_session_state_callback([&states](bool connected) { states.push_back(connected); });
+  session.setup();
+  // Not yet connected, so nothing to report
+  link.report_data_lost();
+  EXPECT_TRUE(states.empty());
+  link.inject({0x00, 0x55});
+  link.report_data_lost();
+  EXPECT_FALSE(session.is_connected());
+  // Connected again when the peer is next heard
+  link.inject({0x00, 0x56});
+  EXPECT_EQ(states, (std::vector<bool>{true, false, true}));
+}
+
+TEST_F(BasicSessionTest, LinkTooSmallForKeepAliveFailsSetup) {
+  MockLink link(4);
+  TestSession session(&link);
+  session.setup();
+  EXPECT_TRUE(session.is_failed());
+  EXPECT_TRUE(link.sent.empty());
+  Packet packet{1};
+  EXPECT_FALSE(session.send_to_interface(PacketBuffer(packet), {}));
+}
+
+}  // namespace esphome::packet_interface::session_testing

tests/components/packet_interface/common.yaml

@@ -1,7 +1,33 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+udp:
+  - id: packet_interface_udp
+
 packet_interface:
   - platform: uart
     id: packet_interface_uart
     max_packet_size: 256
+    reliability:
+      timeout: 300ms
+      ack_delay: 50ms
+      max_queue_length: 3
+      max_retries: 4
+    encryption:
+      key: Johnny Cash
+      rolling_code_enable: true
+    session:
+      keep_alive: 5s
+      timeout: 12s
+      binary_sensor:
+        name: UART session
+  - platform: udp
+    id: packet_interface_network
+    udp_id: packet_interface_udp
+    encryption: Johnny Cash
+    reliability:
+    session:
 
 sensor:
   - platform: template
@@ -13,3 +39,7 @@ packet_transport:
     packet_interface: packet_interface_uart
     sensors:
       - packet_interface_sensor
+  - platform: packet_interface
+    packet_interface: packet_interface_network
+    sensors:
+      - packet_interface_sensor

tests/components/packet_interface/hdlc_reliability.cpp

@@ -0,0 +1,464 @@
+#include <gtest/gtest.h>
+#include <cstdint>
+#include <functional>
+#include <memory>
+#include <vector>
+#include "esphome/components/packet_interface/hdlc_reliability.h"
+#include "esphome/core/application.h"
+#include "esphome/core/hal.h"
+#include "esphome/core/helpers.h"
+
+namespace esphome::packet_interface::testing {
+
+// Copies a packet into a vector, for comparing in tests
+static std::vector<uint8_t> to_vector(const esphome::packet_interface::PacketBuffer &data) {
+  std::vector<uint8_t> result(data.size());
+  data.copy_to(result.data(), result.size());
+  return result;
+}
+
+using Frame = std::vector<uint8_t>;
+
+// A link layer that records what is sent, so a test can deliver it to the peer
+class MockLink : public PacketInterface {
+ public:
+  explicit MockLink(size_t max_packet_size = 64) : PacketInterface(max_packet_size) {}
+  bool send_to_interface(const PacketBuffer &data, const PacketMetaData &) override {
+    if (this->budget == 0) {
+      this->refused++;
+      return false;
+    }
+    this->budget--;
+    this->sent.push_back(to_vector(data));
+    return true;
+  }
+  void inject(const Frame &frame) { this->on_receive_from_interface_(PacketBuffer(frame), {}); }
+
+  std::vector<Frame> sent;
+  size_t budget{SIZE_MAX};  // frames accepted before the link refuses, like a full network stack
+  size_t refused{0};
+};
+
+class TestHdlc : public HdlcReliability {
+ public:
+  using HdlcReliability::HdlcReliability;
+  using HdlcReliability::on_timeout_;
+  using HdlcReliability::outstanding_;
+};
+
+struct Peer {
+  explicit Peer(uint8_t queue_length = 4) {
+    this->hdlc = std::make_unique<TestHdlc>(&this->link);
+    this->hdlc->set_max_queue_length(queue_length);
+    this->hdlc->set_max_retries(3);
+    this->hdlc->setup();
+    this->hdlc->add_packet_interface_listener(
+        [this](const PacketBuffer &data, const PacketMetaData &) { this->received.push_back(to_vector(data)); });
+    this->hdlc->add_on_session_state_callback([this](bool connected) { this->states.push_back(connected); });
+  }
+  bool send(const Frame &data) { return this->hdlc->send_to_interface(PacketBuffer(data), {}); }
+
+  MockLink link;
+  std::unique_ptr<TestHdlc> hdlc;
+  std::vector<Frame> received;
+  std::vector<bool> states;
+};
+
+// Returns true to drop a frame on its way from one peer to the other
+using DropFilter = std::function<bool(const Frame &)>;
+
+static bool is_info(const Frame &frame) { return (frame[0] & 1) == 0; }
+static bool is_ack(const Frame &frame) { return (frame[0] & 0x0F) == 0x01; }
+
+// Runs a layer's loop only while it has asked for it, as the main loop does
+static void run_loop(TestHdlc &hdlc) {
+  if (!hdlc.is_idle())
+    hdlc.loop();
+}
+
+// Move frames between the peers and run their loops until nothing more is sent
+static void pump(Peer &a, Peer &b, const DropFilter &drop_a_to_b = nullptr, const DropFilter &drop_b_to_a = nullptr) {
+  for (int i = 0; i != 50; i++) {
+    run_loop(*a.hdlc);
+    run_loop(*b.hdlc);
+    if (a.link.sent.empty() && b.link.sent.empty())
+      return;
+    auto from_a = std::move(a.link.sent);
+    auto from_b = std::move(b.link.sent);
+    a.link.sent.clear();
+    b.link.sent.clear();
+    for (auto &frame : from_a) {
+      if (!drop_a_to_b || !drop_a_to_b(frame))
+        b.link.inject(frame);
+    }
+    for (auto &frame : from_b) {
+      if (!drop_b_to_a || !drop_b_to_a(frame))
+        a.link.inject(frame);
+    }
+  }
+  FAIL() << "Peers did not go quiet";
+}
+
+static Frame make_frame(uint8_t control, const Frame &payload) {
+  Frame frame{control};
+  frame.insert(frame.end(), payload.begin(), payload.end());
+  uint16_t crc = crc16(frame.data(), frame.size());
+  frame.push_back(crc & 0xFF);
+  frame.push_back(crc >> 8);
+  return frame;
+}
+
+// Each test replaces App with a fresh one; the retransmit timer needs its scheduler
+class HdlcReliabilityTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    App.~Application();
+    new (&App) Application();
+    static char name[] = "hdlc-tests";
+    static char friendly_name[] = "";
+    App.pre_setup(name, sizeof(name) - 1, friendly_name, 0);
+  }
+  void TearDown() override {
+    App.~Application();
+    new (&App) Application();  // later tests and the test listener use App
+  }
+};
+
+TEST_F(HdlcReliabilityTest, MaxPacketSizeAllowsForOverhead) {
+  Peer a;
+  EXPECT_EQ(a.hdlc->get_send_capacity(), 4);
+  EXPECT_EQ(a.hdlc->get_max_packet_size(), 64 - HdlcReliability::FRAME_OVERHEAD);
+  EXPECT_EQ(a.link.top(), a.hdlc.get());
+  EXPECT_EQ(a.hdlc->get_capabilities(), CAPABILITY_RELIABILITY);
+}
+
+TEST_F(HdlcReliabilityTest, ResetHandshake) {
+  Peer a, b;
+  a.hdlc->loop();
+  ASSERT_EQ(a.link.sent.size(), 1);
+  EXPECT_EQ(a.link.sent[0], make_frame(0x2F, {}));  // SABM
+  pump(a, b);
+  EXPECT_TRUE(a.hdlc->is_connected());
+  EXPECT_TRUE(b.hdlc->is_connected());
+}
+
+TEST_F(HdlcReliabilityTest, RepeatedSabmBeforeNumberedFramesKeepsQueue) {
+  Peer a, b;
+  pump(a, b);
+  // A frame is queued and sent, but has not reached A, so A has sent no numbered frame
+  ASSERT_TRUE(b.send({0x42}));
+  auto held = std::move(b.link.sent);
+  b.link.sent.clear();
+  EXPECT_EQ(b.hdlc->get_send_capacity(), 3);
+
+  // A repeats its SABM because it did not hear the UA; that is not a restart
+  b.link.inject(make_frame(0x2F, {}));  // SABM
+  EXPECT_TRUE(b.hdlc->is_connected());
+  EXPECT_TRUE(b.states.empty());              // no session lost reported
+  EXPECT_EQ(b.hdlc->get_send_capacity(), 3);  // the queued frame was kept
+  ASSERT_FALSE(b.link.sent.empty());
+  EXPECT_EQ(b.link.sent[0], make_frame(0x63, {}));  // UA, then the frame is sent again
+
+  // The frame is still delivered when it reaches A
+  for (auto &frame : held)
+    a.link.inject(frame);
+  EXPECT_EQ(a.received, std::vector<Frame>{{0x42}});
+}
+
+TEST_F(HdlcReliabilityTest, SabmAfterNumberedFramesIsARestart) {
+  Peer a, b;
+  pump(a, b);
+  // B has heard a numbered frame, so a SABM now means A has started again
+  ASSERT_TRUE(a.send({0x10}));
+  pump(a, b);
+  b.link.inject(make_frame(0x2F, {}));  // SABM
+  EXPECT_TRUE(b.hdlc->is_connected());
+  EXPECT_EQ(b.states, std::vector<bool>{false});
+}
+
+TEST_F(HdlcReliabilityTest, QueuedBeforeConnectIsSentAfter) {
+  Peer a, b;
+  ASSERT_TRUE(a.send({0x01}));
+  ASSERT_TRUE(a.send({0x02}));
+  pump(a, b);
+  EXPECT_EQ(b.received, (std::vector<Frame>{{0x01}, {0x02}}));
+}
+
+TEST_F(HdlcReliabilityTest, DeliversInOrderBothWays) {
+  Peer a, b;
+  pump(a, b);
+  a.send({0x10});
+  b.send({0x20});
+  a.send({0x11});
+  pump(a, b);
+  EXPECT_EQ(b.received, (std::vector<Frame>{{0x10}, {0x11}}));
+  EXPECT_EQ(a.received, (std::vector<Frame>{{0x20}}));
+}
+
+TEST_F(HdlcReliabilityTest, SequenceNumbersWrap) {
+  Peer a(3), b(3);
+  pump(a, b);
+  std::vector<Frame> expected;
+  for (uint8_t i = 0; i != 20; i++) {
+    expected.push_back({i, static_cast<uint8_t>(i * 3)});
+    ASSERT_TRUE(a.send(expected.back()));
+    if (i % 2 == 1)
+      pump(a, b);
+  }
+  pump(a, b);
+  EXPECT_EQ(b.received, expected);
+}
+
+TEST_F(HdlcReliabilityTest, GapRecoveredByReject) {
+  Peer a, b;
+  pump(a, b);
+  bool dropped = false;
+  a.send({0x01});
+  a.send({0x02});
+  a.send({0x03});
+  pump(a, b, [&dropped](const Frame &frame) {
+    if (dropped || !is_info(frame))
+      return false;
+    dropped = true;
+    return true;
+  });
+  EXPECT_EQ(b.received, (std::vector<Frame>{{0x01}, {0x02}, {0x03}}));
+}
+
+TEST_F(HdlcReliabilityTest, LossRecoveredByTimeout) {
+  Peer a, b;
+  pump(a, b);
+  a.send({0x42});
+  pump(a, b, [](const Frame &frame) { return is_info(frame); });
+  EXPECT_TRUE(b.received.empty());
+  a.hdlc->on_timeout_();
+  pump(a, b);
+  EXPECT_EQ(b.received, (std::vector<Frame>{{0x42}}));
+}
+
+TEST_F(HdlcReliabilityTest, LostAckDoesNotDuplicate) {
+  Peer a, b;
+  pump(a, b);
+  a.send({0x42});
+  pump(a, b, nullptr, [](const Frame &frame) { return is_ack(frame); });
+  ASSERT_EQ(b.received.size(), 1);
+  a.hdlc->on_timeout_();
+  pump(a, b);
+  EXPECT_EQ(b.received.size(), 1);
+  // The resend was acknowledged, so the full queue is available again
+  for (int i = 0; i != 4; i++)
+    EXPECT_TRUE(a.send({0x01}));
+}
+
+TEST_F(HdlcReliabilityTest, QueueFull) {
+  Peer a(2), b(2);
+  pump(a, b);
+  EXPECT_TRUE(a.send({0x01}));
+  EXPECT_TRUE(a.send({0x02}));
+  EXPECT_FALSE(a.send({0x03}));
+}
+
+TEST_F(HdlcReliabilityTest, OversizedPacketRejected) {
+  Peer a;
+  EXPECT_FALSE(a.send(Frame(a.hdlc->get_max_packet_size() + 1)));
+}
+
+TEST_F(HdlcReliabilityTest, CrcErrorDropped) {
+  Peer a, b;
+  pump(a, b);
+  Frame frame = make_frame(0x00, {0x55});  // I-frame N(S)=0 N(R)=0
+  frame[1] ^= 0xFF;
+  b.link.inject(frame);
+  EXPECT_TRUE(b.received.empty());
+  b.link.inject(make_frame(0x00, {0x55}));
+  EXPECT_EQ(b.received, (std::vector<Frame>{{0x55}}));
+}
+
+TEST_F(HdlcReliabilityTest, RetriesExhaustedResetsLink) {
+  Peer a, b;
+  pump(a, b);
+  // One frame is delivered first, so b can tell a reset from a repeated SABM
+  ASSERT_TRUE(a.send({0x00}));
+  pump(a, b);
+  ASSERT_EQ(b.received, std::vector<Frame>{{0x00}});
+  a.send({0x01});
+  pump(a, b, [](const Frame &frame) { return is_info(frame); });
+  for (int i = 0; i != 4; i++) {
+    a.hdlc->on_timeout_();
+    pump(a, b, [](const Frame &frame) { return is_info(frame); });
+  }
+  // The unacknowledged frame was discarded and the link was reset, which both report as data lost
+  EXPECT_EQ(b.received, std::vector<Frame>{{0x00}});
+  EXPECT_TRUE(a.hdlc->is_connected());
+  EXPECT_EQ(a.states, std::vector<bool>{false});
+  EXPECT_EQ(b.states, std::vector<bool>{false});
+  a.send({0x02});
+  pump(a, b);
+  EXPECT_EQ(b.received, (std::vector<Frame>{{0x00}, {0x02}}));
+}
+
+TEST_F(HdlcReliabilityTest, PeerRestart) {
+  Peer a;
+  auto b = std::make_unique<Peer>();
+  pump(a, *b);
+  a.send({0x01});
+  pump(a, *b);
+  // The peer restarts with fresh sequence numbers
+  b = std::make_unique<Peer>();
+  pump(a, *b);
+  a.send({0x02});
+  b->send({0x03});
+  pump(a, *b);
+  EXPECT_EQ(b->received, (std::vector<Frame>{{0x02}}));
+  EXPECT_EQ(a.received, (std::vector<Frame>{{0x03}}));
+  // The restarted peer's reset discarded what a had not yet had acknowledged
+  EXPECT_EQ(a.states, std::vector<bool>{false});
+  EXPECT_TRUE(b->states.empty());
+}
+
+// Waits past the retry after a refused send, servicing timers as the main loop does
+static void run_send_retry() {
+  const uint32_t start = millis();
+  while (millis() - start < 60) {
+    App.scheduler.call(millis());
+    delay(1);
+  }
+  App.scheduler.call(millis());
+}
+
+TEST_F(HdlcReliabilityTest, RefusedFramesSentAgainWithoutTimeout) {
+  Peer a, b;
+  // Only the retry of a refused send can fire while the test runs, not a retransmit
+  a.hdlc->set_ack_timeout(10000);
+  b.hdlc->set_ack_timeout(10000);
+  for (uint8_t i = 0; i != 3; i++)
+    ASSERT_TRUE(a.send({i}));
+  a.link.budget = 3;  // SABM, UA to the peer's SABM, and the first queued frame
+  pump(a, b);
+  EXPECT_GE(a.link.refused, 1);
+  EXPECT_EQ(b.received, (std::vector<Frame>{{0}}));
+  a.link.budget = SIZE_MAX;  // the lower layer accepts again, as when a radio stops receiving
+  // The layer waits for its retry rather than running every loop
+  pump(a, b);
+  EXPECT_EQ(b.received, (std::vector<Frame>{{0}}));
+  run_send_retry();
+  pump(a, b);
+  EXPECT_EQ(b.received, (std::vector<Frame>{{0}, {1}, {2}}));
+  EXPECT_EQ(a.hdlc->outstanding_(), 0);
+}
+
+TEST_F(HdlcReliabilityTest, RefusedAckSentAgain) {
+  Peer a, b;
+  a.hdlc->set_ack_timeout(10000);
+  b.hdlc->set_ack_timeout(10000);
+  pump(a, b);
+  a.send({1});
+  for (auto &frame : a.link.sent)
+    b.link.inject(frame);
+  a.link.sent.clear();
+  b.link.budget = 0;
+  b.hdlc->loop();
+  EXPECT_TRUE(b.link.sent.empty());
+  b.link.budget = SIZE_MAX;
+  pump(a, b);
+  EXPECT_EQ(a.hdlc->outstanding_(), 1);  // the RR waits for the retry
+  run_send_retry();
+  pump(a, b);
+  EXPECT_EQ(a.hdlc->outstanding_(), 0);
+}
+
+TEST_F(HdlcReliabilityTest, AckDelayLetsReplyCarryAck) {
+  Peer a, b;
+  b.hdlc->set_ack_delay(100);
+  pump(a, b);
+  a.send({1});
+  for (auto &frame : a.link.sent)
+    b.link.inject(frame);
+  a.link.sent.clear();
+  b.hdlc->loop();
+  EXPECT_TRUE(b.link.sent.empty());  // no RR yet
+  b.send({2});
+  ASSERT_EQ(b.link.sent.size(), 1);
+  EXPECT_TRUE(is_info(b.link.sent[0]));
+  EXPECT_EQ(b.link.sent[0][0] >> 5, 1);  // N(R) acknowledges the frame from a
+  b.link.sent.clear();
+  b.hdlc->loop();
+  EXPECT_TRUE(b.link.sent.empty());  // the RR is no longer owed
+}
+
+TEST_F(HdlcReliabilityTest, MaxRetriesOf255ResetsLink) {
+  Peer a, b;
+  a.hdlc->set_max_retries(255);
+  pump(a, b);
+  a.send({0x01});
+  a.link.sent.clear();
+  for (int i = 0; i != 255; i++) {
+    a.hdlc->on_timeout_();
+    a.link.sent.clear();
+  }
+  EXPECT_TRUE(a.states.empty());
+  a.hdlc->on_timeout_();
+  EXPECT_EQ(a.states, std::vector<bool>{false});
+}
+
+TEST_F(HdlcReliabilityTest, OversizedFrameDropped) {
+  Peer a, b;
+  pump(a, b);
+  // An I-frame, N(S) 0 and N(R) 0, one byte larger than this layer's packets
+  b.link.inject(make_frame(0x00, Frame(b.hdlc->get_max_packet_size() + 1, 0x55)));
+  EXPECT_TRUE(b.received.empty());
+  b.link.inject(make_frame(0x00, Frame(b.hdlc->get_max_packet_size(), 0x55)));
+  EXPECT_EQ(b.received.size(), 1);
+}
+
+TEST_F(HdlcReliabilityTest, LinkTooSmallFailsSetup) {
+  MockLink link(HdlcReliability::FRAME_OVERHEAD);
+  TestHdlc hdlc(&link);
+  EXPECT_EQ(hdlc.get_max_packet_size(), 0);
+  hdlc.setup();
+  EXPECT_TRUE(hdlc.is_failed());
+  Frame packet{1};
+  EXPECT_FALSE(hdlc.send_to_interface(PacketBuffer(packet), {}));
+  EXPECT_EQ(hdlc.get_send_capacity(), 0);
+}
+
+TEST_F(HdlcReliabilityTest, AckDelayTimerSendsAck) {
+  Peer a, b;
+  b.hdlc->set_ack_delay(5);
+  pump(a, b);
+  a.send({1});
+  for (auto &frame : a.link.sent)
+    b.link.inject(frame);
+  a.link.sent.clear();
+  b.hdlc->loop();
+  EXPECT_TRUE(b.link.sent.empty());
+  delay(10);
+  App.scheduler.call(millis());
+  b.hdlc->loop();
+  ASSERT_EQ(b.link.sent.size(), 1);
+  EXPECT_TRUE(is_ack(b.link.sent[0]));
+  EXPECT_EQ(b.link.sent[0][0] >> 5, 1);  // N(R) acknowledges the frame from a
+}
+
+TEST_F(HdlcReliabilityTest, RefusedRejectFallsBackToAck) {
+  Peer a, b;
+  pump(a, b);
+  a.send({1});
+  a.send({2});
+  ASSERT_EQ(a.link.sent.size(), 2);
+  // The first frame is lost, and the REJ for the gap is refused
+  b.link.budget = 0;
+  b.link.inject(a.link.sent[1]);
+  a.link.sent.clear();
+  EXPECT_EQ(b.link.refused, 1);
+  b.link.budget = SIZE_MAX;
+  b.hdlc->loop();
+  ASSERT_EQ(b.link.sent.size(), 1);
+  EXPECT_TRUE(is_ack(b.link.sent[0]));
+  EXPECT_EQ(b.link.sent[0][0] >> 5, 0);
+  a.hdlc->on_timeout_();
+  pump(a, b);
+  EXPECT_EQ(b.received, (std::vector<Frame>{{1}, {2}}));
+}
+
+}  // namespace esphome::packet_interface::testing

tests/components/packet_interface/xxtea_encryption.cpp

@@ -0,0 +1,282 @@
+#include <gtest/gtest.h>
+#include <algorithm>
+#include <cstdint>
+#include <memory>
+#include <vector>
+#include "esphome/components/packet_interface/xxtea_encryption.h"
+#include "esphome/core/application.h"
+
+namespace esphome::packet_interface::encryption_testing {
+
+// Copies a packet into a vector, for comparing in tests
+static std::vector<uint8_t> to_vector(const esphome::packet_interface::PacketBuffer &data) {
+  std::vector<uint8_t> result(data.size());
+  data.copy_to(result.data(), result.size());
+  return result;
+}
+
+using Packet = std::vector<uint8_t>;
+
+// A link layer that records what is sent, so a test can deliver it to the peer
+class MockLink : public PacketInterface {
+ public:
+  explicit MockLink(size_t max_packet_size = 64) : PacketInterface(max_packet_size) {}
+  bool send_to_interface(const PacketBuffer &data, const PacketMetaData &) override {
+    this->sent.push_back(to_vector(data));
+    return true;
+  }
+  void inject(const Packet &packet) { this->on_receive_from_interface_(PacketBuffer(packet), {}); }
+
+  std::vector<Packet> sent;
+};
+
+static constexpr size_t ROLLING_TRAILER_SIZE = 16;
+static constexpr size_t TRAILER_SIZE = 4;
+
+class TestEncryption : public XxteaEncryption {
+ public:
+  using XxteaEncryption::XxteaEncryption;
+  uint32_t get_boot() const { return this->boot_; }
+  void set_counter(uint32_t counter) { this->counter_ = counter; }
+};
+
+struct Peer {
+  explicit Peer(uint32_t key_word = 0x12345678, size_t max_packet_size = 64, bool rolling_code = true)
+      : link(max_packet_size) {
+    this->link.set_name("test_link");
+    this->encryption = std::make_unique<TestEncryption>(&this->link, rolling_code);
+    this->encryption->set_key({key_word, 2, 3, 4, 5, 6, 7, 8});
+    this->encryption->add_packet_interface_listener(
+        [this](const PacketBuffer &data, const PacketMetaData &) { this->received.push_back(to_vector(data)); });
+    this->encryption->setup();
+  }
+  bool send(const Packet &data) { return this->encryption->send_to_interface(PacketBuffer(data), {}); }
+  // Deliver everything sent so far to the peer
+  void deliver_to(Peer &peer) {
+    for (auto &packet : this->link.sent)
+      peer.link.inject(packet);
+    this->link.sent.clear();
+  }
+  // Exchange a packet each way so that each peer echoes the other's boot number
+  void introduce(Peer &peer) {
+    this->send({});
+    this->deliver_to(peer);
+    peer.send({});
+    peer.deliver_to(*this);
+    this->send({});
+    this->deliver_to(peer);
+    this->received.clear();
+    peer.received.clear();
+  }
+
+  MockLink link;
+  std::unique_ptr<TestEncryption> encryption;
+  std::vector<Packet> received;
+};
+
+// Each test replaces App with a fresh one
+class XxteaEncryptionTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    App.~Application();
+    new (&App) Application();
+    static char name[] = "xxtea-tests";
+    static char friendly_name[] = "";
+    App.pre_setup(name, sizeof(name) - 1, friendly_name, 0);
+  }
+  void TearDown() override {
+    App.~Application();
+    new (&App) Application();  // later tests and the test listener use App
+  }
+};
+
+TEST_F(XxteaEncryptionTest, MaxPacketSizeAllowsForTrailerAndWords) {
+  Peer a;
+  EXPECT_EQ(a.encryption->get_max_packet_size(), 64 - ROLLING_TRAILER_SIZE);
+  EXPECT_EQ(a.link.top(), a.encryption.get());
+  EXPECT_TRUE(a.encryption->has_capability(CAPABILITY_ENCRYPTION));
+  EXPECT_FALSE(a.encryption->has_capability(CAPABILITY_SESSION));
+  Peer b(0x12345678, 63);
+  EXPECT_EQ(b.encryption->get_max_packet_size(), 60 - ROLLING_TRAILER_SIZE);
+  Peer c(0x12345678, 64, false);
+  EXPECT_EQ(c.encryption->get_max_packet_size(), 64 - TRAILER_SIZE);
+}
+
+static void round_trip_all_sizes(bool rolling_code, size_t trailer_size) {
+  Peer a(0x12345678, 64, rolling_code), b(0x12345678, 64, rolling_code);
+  a.introduce(b);
+  std::vector<Packet> expected;
+  for (size_t size = 0; size <= a.encryption->get_max_packet_size(); size++) {
+    Packet packet(size);
+    for (size_t i = 0; i != size; i++)
+      packet[i] = static_cast<uint8_t>(size + i);
+    expected.push_back(packet);
+    ASSERT_TRUE(a.send(packet));
+    ASSERT_EQ(a.link.sent.back().size() % 4, 0);
+    ASSERT_GE(a.link.sent.back().size(), std::max<size_t>(8, size + trailer_size));
+  }
+  a.deliver_to(b);
+  EXPECT_EQ(b.received, expected);
+}
+
+TEST_F(XxteaEncryptionTest, RoundTripAllSizes) { round_trip_all_sizes(true, ROLLING_TRAILER_SIZE); }
+
+TEST_F(XxteaEncryptionTest, RoundTripAllSizesWithoutRollingCode) { round_trip_all_sizes(false, TRAILER_SIZE); }
+
+TEST_F(XxteaEncryptionTest, WithoutRollingCodeReplayAccepted) {
+  Peer a(0x12345678, 64, false), b(0x12345678, 64, false);
+  a.send({1});
+  b.link.inject(a.link.sent[0]);
+  b.link.inject(a.link.sent[0]);
+  EXPECT_EQ(b.received, (std::vector<Packet>{{1}, {1}}));
+}
+
+TEST_F(XxteaEncryptionTest, RollingCodeSettingMustMatch) {
+  Peer a(0x12345678, 64, true), b(0x12345678, 64, false);
+  a.send({1, 2, 3, 4, 5, 6, 7, 8});
+  b.send({1, 2, 3, 4, 5, 6, 7, 8});
+  a.deliver_to(b);
+  b.deliver_to(a);
+  EXPECT_TRUE(a.received.empty());
+  EXPECT_TRUE(b.received.empty());
+}
+
+TEST_F(XxteaEncryptionTest, SamePacketEncryptsDifferently) {
+  Peer a;
+  a.send({1, 2, 3, 4});
+  a.send({1, 2, 3, 4});
+  ASSERT_EQ(a.link.sent.size(), 2);
+  EXPECT_NE(a.link.sent[0], a.link.sent[1]);
+}
+
+TEST_F(XxteaEncryptionTest, OversizedPacketRefused) {
+  Peer a;
+  EXPECT_FALSE(a.send(Packet(a.encryption->get_max_packet_size() + 1)));
+  EXPECT_TRUE(a.link.sent.empty());
+}
+
+TEST_F(XxteaEncryptionTest, WrongKeyDropped) {
+  Peer a, b(0x87654321);
+  a.send({1, 2, 3});
+  a.deliver_to(b);
+  EXPECT_TRUE(b.received.empty());
+}
+
+TEST_F(XxteaEncryptionTest, ChangedPacketDropped) {
+  Peer a, b;
+  a.send({1, 2, 3});
+  a.link.sent[0][2] ^= 0x01;
+  a.deliver_to(b);
+  EXPECT_TRUE(b.received.empty());
+}
+
+TEST_F(XxteaEncryptionTest, BadLengthDropped) {
+  Peer a, b;
+  a.send({1, 2, 3});
+  Packet packet = a.link.sent[0];
+  b.link.inject(Packet(packet.begin(), packet.end() - 1));
+  b.link.inject(Packet(8));
+  EXPECT_TRUE(b.received.empty());
+}
+
+TEST_F(XxteaEncryptionTest, ReplayDropped) {
+  Peer a, b;
+  a.introduce(b);
+  a.send({1});
+  Packet first = a.link.sent[0];
+  a.send({2});
+  a.deliver_to(b);
+  b.link.inject(first);
+  EXPECT_EQ(b.received, (std::vector<Packet>{{1}, {2}}));
+}
+
+TEST_F(XxteaEncryptionTest, OlderPacketDropped) {
+  Peer a, b;
+  a.introduce(b);
+  a.send({1});
+  a.send({2});
+  b.link.inject(a.link.sent[1]);
+  b.link.inject(a.link.sent[0]);
+  EXPECT_EQ(b.received, (std::vector<Packet>{{2}}));
+}
+
+TEST_F(XxteaEncryptionTest, BootNumberIncreasesAndPeerRestartAccepted) {
+  Peer b;
+  auto a = std::make_unique<Peer>();
+  uint32_t boot = a->encryption->get_boot();
+  a->introduce(b);
+  a->send({1});
+  a->send({2});
+  a->deliver_to(b);
+  EXPECT_EQ(b.received, (std::vector<Packet>{{1}, {2}}));
+  a = std::make_unique<Peer>();  // the peer restarts, and its counter starts again from 0
+  EXPECT_EQ(a->encryption->get_boot(), boot + 1);
+  a->introduce(b);
+  a->send({3});
+  a->deliver_to(b);
+  EXPECT_EQ(b.received, std::vector<Packet>{{3}});
+}
+
+TEST_F(XxteaEncryptionTest, FirstPacketsDroppedUntilEchoed) {
+  Peer a, b;
+  a.send({1});
+  a.deliver_to(b);
+  EXPECT_TRUE(b.received.empty());  // a does not know b's boot number yet
+  b.send({2});
+  b.deliver_to(a);
+  EXPECT_EQ(a.received, std::vector<Packet>{{2}});  // b learnt a's from the dropped packet
+  a.send({3});
+  a.deliver_to(b);
+  EXPECT_EQ(b.received, std::vector<Packet>{{3}});
+}
+
+TEST_F(XxteaEncryptionTest, PacketsFromBeforeRestartDropped) {
+  Peer a;
+  auto b = std::make_unique<Peer>();
+  a.introduce(*b);
+  a.send({1});
+  a.send({2});
+  std::vector<Packet> recorded = a.link.sent;
+  a.link.sent.clear();
+  b = std::make_unique<Peer>();  // b restarts and forgets a's counter
+  for (auto &packet : recorded)
+    b->link.inject(packet);
+  EXPECT_TRUE(b->received.empty());
+  a.introduce(*b);
+  a.send({3});
+  a.deliver_to(*b);
+  EXPECT_EQ(b->received, std::vector<Packet>{{3}});
+}
+
+TEST_F(XxteaEncryptionTest, LinkTooSmallFailsSetup) {
+  MockLink link(ROLLING_TRAILER_SIZE + 3);
+  TestEncryption encryption(&link, true);
+  EXPECT_EQ(encryption.get_max_packet_size(), 0);
+  encryption.setup();
+  EXPECT_TRUE(encryption.is_failed());
+  Packet packet{1};
+  EXPECT_FALSE(encryption.send_to_interface(PacketBuffer(packet), {}));
+}
+
+TEST_F(XxteaEncryptionTest, SendWithoutBuffersRefused) {
+  MockLink link(64);
+  TestEncryption encryption(&link, false);
+  Packet packet{1};
+  EXPECT_FALSE(encryption.send_to_interface(PacketBuffer(packet), {}));
+  EXPECT_TRUE(link.sent.empty());
+}
+
+TEST_F(XxteaEncryptionTest, CounterWrapStartsNewBoot) {
+  Peer a, b(0x12345678);
+  a.introduce(b);
+  uint32_t boot = a.encryption->get_boot();
+  a.encryption->set_counter(UINT32_MAX);
+  ASSERT_TRUE(a.send({1}));
+  EXPECT_EQ(a.encryption->get_boot(), boot + 1);
+  ASSERT_TRUE(a.send({2}));
+  a.deliver_to(b);
+  // Both were accepted: the last packet of the old boot, then the first of the new one
+  EXPECT_EQ(b.received, (std::vector<Packet>{{1}, {2}}));
+}
+
+}  // namespace esphome::packet_interface::encryption_testing

tests/components/packet_transport/common.yaml

@@ -3,12 +3,14 @@ wifi:
   password: password1
 
 udp:
-  listen_address: 239.0.60.53
-  addresses: ["239.0.60.53"]
+  - id: transport_udp_component
+    listen_address: 239.0.60.53
+    addresses: ["239.0.60.53"]
 
 packet_transport:
   - platform: udp
     id: transport_udp
+    udp_id: transport_udp_component
     update_interval: 5s
     encryption: "our key goes here"
     rolling_code_enable: true

tests/components/udp/__init__.py

@@ -0,0 +1,6 @@
+from tests.testing_helpers import ComponentManifestOverride
+
+
+def override_manifest(manifest: ComponentManifestOverride) -> None:
+    # benchmark.yaml configures a udp packet_interface, whose to_code waits on the udp component variable.
+    manifest.enable_codegen()

tests/components/udp/benchmark.yaml

@@ -0,0 +1,5 @@
+udp:
+  addresses: ["192.168.1.255"]
+
+packet_interface:
+  - platform: udp

tests/components/udp/udp_packet_interface.cpp

@@ -0,0 +1,78 @@
+#include <gtest/gtest.h>
+
+#include <cstdint>
+#include <vector>
+
+#include "esphome/components/udp/packet_interface/udp_packet_interface.h"
+
+#ifdef USE_NETWORK
+
+namespace esphome::udp::testing {
+
+// Copies a packet into a vector, for comparing in tests
+static std::vector<uint8_t> to_vector(const esphome::packet_interface::PacketBuffer &data) {
+  std::vector<uint8_t> result(data.size());
+  data.copy_to(result.data(), result.size());
+  return result;
+}
+
+using packet_interface::PacketBuffer;
+using packet_interface::PacketMetaData;
+using Packet = std::vector<uint8_t>;
+
+// The copy buffer is a base, so it is set up before the link's constructor sees it
+struct UdpBufferStorage {
+  uint8_t tx_buffer_storage[8]{};
+};
+
+class TestUdp : public UdpBufferStorage, public UdpPacketInterface {
+ public:
+  explicit TestUdp(UDPComponent *udp) : UdpPacketInterface(udp, this->tx_buffer_storage) {
+    this->add_packet_interface_listener(
+        [this](const PacketBuffer &data, const PacketMetaData &) { this->received.push_back(to_vector(data)); });
+  }
+  using UdpPacketInterface::receive_;
+
+  std::vector<Packet> received;
+};
+
+// The udp component is not set up, so it has no destinations and a send does nothing
+class UdpPacketInterfaceTest : public ::testing::Test {
+ protected:
+  UDPComponent udp_;
+  TestUdp link_{&this->udp_};
+};
+
+TEST_F(UdpPacketInterfaceTest, ReceivedPacketsUpToMaxPassedUp) {
+  Packet packet{1, 2, 3, 4, 5, 6, 7, 8};
+  this->link_.receive_(packet);
+  Packet oversized{1, 2, 3, 4, 5, 6, 7, 8, 9};
+  this->link_.receive_(oversized);
+  EXPECT_EQ(this->link_.received, std::vector<Packet>{packet});
+}
+
+TEST_F(UdpPacketInterfaceTest, SendsContiguousAndSplitPackets) {
+  this->link_.setup();
+  Packet packet{1, 2, 3, 4};
+  EXPECT_TRUE(this->link_.send_to_interface(PacketBuffer(packet), {}));
+  uint8_t header = 0;
+  PacketBuffer body(packet);
+  EXPECT_TRUE(this->link_.send_to_interface(PacketBuffer(&header, 1, &body, nullptr, 0), {}));
+  // A packet split across buffers is joined in the copy buffer before it is sent
+  const auto *storage = this->link_.tx_buffer_storage;
+  EXPECT_EQ(Packet(storage, storage + 5), (Packet{0, 1, 2, 3, 4}));
+}
+
+TEST_F(UdpPacketInterfaceTest, OversizedPacketNotSent) {
+  this->link_.setup();
+  Packet packet{1, 2, 3, 4, 5, 6, 7, 8, 9};
+  EXPECT_FALSE(this->link_.send_to_interface(PacketBuffer(packet), {}));
+  uint8_t header = 0;
+  Packet body_data{1, 2, 3, 4, 5, 6, 7, 8};
+  PacketBuffer body(body_data);
+  EXPECT_FALSE(this->link_.send_to_interface(PacketBuffer(&header, 1, &body, nullptr, 0), {}));
+}
+
+}  // namespace esphome::udp::testing
+
+#endif

tests/unit_tests/test_packet_interface.py

@@ -1,17 +1,35 @@
 """Unit tests for the packet_interface component."""
 
+import hashlib
+from typing import Any
+
 import pytest
 
 from esphome import config_validation as cv
 from esphome.components import packet_interface
 from esphome.components.packet_interface import CONF_MAX_PACKET_SIZE
+from esphome.components.packet_interface.layers import (
+    CONF_RELIABILITY,
+    CONF_SESSION,
+    RELIABILITY,
+    STACK_ORDER,
+    LayerKind,
+    xxtea_encryption,
+)
+from esphome.components.packet_interface.layers.basic_session import CONF_KEEP_ALIVE
+from esphome.components.packet_interface.layers.hdlc_reliability import (
+    CONF_MAX_QUEUE_LENGTH,
+)
 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
 
 
 @pytest.mark.parametrize(
     ("schema", "default", "limit"),
     [
         (uart_packet_interface.CONFIG_SCHEMA, 508, 4096),
+        (udp_packet_interface.CONFIG_SCHEMA, 508, 508),
     ],
 )
 def test_platform_max_packet_size(
@@ -32,3 +50,167 @@ def test_schema_default(setup_core) -> None:
         packet_interface.PacketInterface, 100
     )
     assert schema({})[CONF_MAX_PACKET_SIZE] == 100
+
+
+def test_reliability_defaults(setup_core) -> None:
+    """An empty reliability key selects the HDLC layer with its defaults."""
+    config = uart_packet_interface.CONFIG_SCHEMA({CONF_RELIABILITY: None})
+    reliability = config[CONF_RELIABILITY]
+    assert reliability["type"] == "hdlc"
+    assert reliability["timeout"].total_milliseconds == 200
+    assert reliability[CONF_MAX_QUEUE_LENGTH] == 4
+    assert reliability["max_retries"] == 5
+
+
+def test_reliability_queue_length_limit(setup_core) -> None:
+    """The queue cannot hold more frames than 3 bit sequence numbers allow."""
+    with pytest.raises(cv.Invalid):
+        uart_packet_interface.CONFIG_SCHEMA(
+            {CONF_RELIABILITY: {CONF_MAX_QUEUE_LENGTH: 8}}
+        )
+
+
+def test_reliability_timeout_minimum(setup_core) -> None:
+    """A timeout shorter than a few loop passes cannot work."""
+    uart_packet_interface.CONFIG_SCHEMA({CONF_RELIABILITY: {"timeout": "50ms"}})
+    with pytest.raises(cv.Invalid):
+        uart_packet_interface.CONFIG_SCHEMA({CONF_RELIABILITY: {"timeout": "40ms"}})
+
+
+def test_ack_delay_shorter_than_timeout(setup_core) -> None:
+    """The acknowledgement must not wait longer than the peer's timeout."""
+    reliability = uart_packet_interface.CONFIG_SCHEMA(
+        {CONF_RELIABILITY: {"timeout": "500ms", "ack_delay": "100ms"}}
+    )[CONF_RELIABILITY]
+    assert reliability["ack_delay"].total_milliseconds == 100
+    with pytest.raises(cv.Invalid, match="shorter"):
+        uart_packet_interface.CONFIG_SCHEMA(
+            {CONF_RELIABILITY: {"timeout": "500ms", "ack_delay": "500ms"}}
+        )
+
+
+def test_layer_types_registered() -> None:
+    """Importing packet_interface registers the built-in layer types."""
+    assert {kind.key: set(kind.types) for kind in STACK_ORDER} == {
+        CONF_ENCRYPTION: {"xxtea"},
+        CONF_RELIABILITY: {"hdlc"},
+        CONF_SESSION: {"basic"},
+    }
+
+
+async def _build(config: dict, lower: Any) -> Any:
+    return lower
+
+
+def test_registered_type_selectable(setup_core) -> None:
+    """A newly registered layer type can be selected with type."""
+    kind = LayerKind("test_kind", "test")
+    kind.register("test", schema=cv.Schema({}))(_build)
+    config = kind.schema()({"type": "test"})
+    assert config["type"] == "test"
+    assert kind.get_type(config).build is _build
+
+
+def test_duplicate_layer_type_rejected(setup_core) -> None:
+    """Registering a type under a name already in use is an error, not a silent replacement."""
+    kind = LayerKind("test_kind", "test")
+    kind.register("test", schema=cv.Schema({}))(_build)
+    with pytest.raises(ValueError, match="already registered"):
+        kind.register("test", schema=cv.Schema({}))(_build)
+
+
+def test_late_registration_rejected(setup_core) -> None:
+    """A type registered after the schemas were built would be missing from them, so it is an error."""
+    with pytest.raises(ValueError, match="after the schemas were built"):
+        RELIABILITY.register("late", schema=cv.Schema({}))(_build)
+
+
+def test_session_defaults(setup_core) -> None:
+    """An empty session key selects the basic session, timing out after three keep-alives."""
+    session = uart_packet_interface.CONFIG_SCHEMA({CONF_SESSION: None})[CONF_SESSION]
+    assert session["type"] == "basic"
+    assert session[CONF_KEEP_ALIVE].total_milliseconds == 10000
+    assert session["timeout"].total_milliseconds == 30000
+
+
+def test_session_timeout(setup_core) -> None:
+    """The timeout follows keep_alive unless given, and must be longer than it."""
+    session = uart_packet_interface.CONFIG_SCHEMA(
+        {CONF_SESSION: {CONF_KEEP_ALIVE: "2s"}}
+    )[CONF_SESSION]
+    assert session["timeout"].total_milliseconds == 6000
+    with pytest.raises(cv.Invalid):
+        uart_packet_interface.CONFIG_SCHEMA(
+            {CONF_SESSION: {CONF_KEEP_ALIVE: "5s", "timeout": "5s"}}
+        )
+    with pytest.raises(cv.Invalid):
+        uart_packet_interface.CONFIG_SCHEMA({CONF_SESSION: {CONF_KEEP_ALIVE: "500ms"}})
+    with pytest.raises(cv.Invalid):
+        uart_packet_interface.CONFIG_SCHEMA({CONF_SESSION: {CONF_KEEP_ALIVE: "2h"}})
+    with pytest.raises(cv.Invalid):
+        uart_packet_interface.CONFIG_SCHEMA({CONF_SESSION: {"timeout": "4h"}})
+
+
+def test_encryption_key_only(setup_core) -> None:
+    """The key can be given on its own, and selects XXTEA."""
+    encryption = uart_packet_interface.CONFIG_SCHEMA({CONF_ENCRYPTION: "secret"})[
+        CONF_ENCRYPTION
+    ]
+    assert encryption["type"] == "xxtea"
+    assert encryption["key"] == "secret"
+    assert encryption["rolling_code_enable"] is False
+
+
+def test_encryption_requires_key(setup_core) -> None:
+    """An encryption layer without a key is rejected."""
+    with pytest.raises(cv.Invalid):
+        uart_packet_interface.CONFIG_SCHEMA({CONF_ENCRYPTION: {}})
+    with pytest.raises(cv.Invalid):
+        uart_packet_interface.CONFIG_SCHEMA({CONF_ENCRYPTION: ""})
+
+
+def test_rolling_code_requires_session(setup_core) -> None:
+    """The rolling code needs the session's keep-alives, so the peer sends this device's boot number back."""
+    rolling = {CONF_ENCRYPTION: {"key": "secret", "rolling_code_enable": True}}
+    with pytest.raises(cv.Invalid, match="session"):
+        uart_packet_interface.CONFIG_SCHEMA(rolling)
+    uart_packet_interface.CONFIG_SCHEMA({**rolling, CONF_SESSION: None})
+    uart_packet_interface.CONFIG_SCHEMA({CONF_ENCRYPTION: "secret"})
+
+
+@pytest.mark.parametrize(
+    ("config", "loads"),
+    [
+        ({}, []),
+        ({CONF_RELIABILITY: None, CONF_SESSION: None}, []),
+        ({CONF_SESSION: {"binary_sensor": {"name": "Link"}}}, ["binary_sensor"]),
+        ({CONF_ENCRYPTION: "secret"}, ["xxtea"]),
+    ],
+)
+def test_auto_load_follows_layers(
+    setup_core, config: dict[str, Any], loads: list[str]
+) -> None:
+    """Only the components that the configured layers use are loaded."""
+    validated = uart_packet_interface.CONFIG_SCHEMA(config)
+    assert packet_interface.AUTO_LOAD([validated]) == loads
+
+
+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"]
+
+
+def test_xxtea_key_stretched() -> None:
+    """The key comes from PBKDF2 of the key text, not a single hash a captured packet could check guesses against."""
+    key = xxtea_encryption.derive_key("Johnny Cash")
+    digest = hashlib.pbkdf2_hmac(
+        "sha256",
+        b"Johnny Cash",
+        b"esphome packet_interface xxtea",
+        600_000,
+    )
+    assert key == [int.from_bytes(digest[i : i + 4], "little") for i in range(0, 32, 4)]
+    assert len(key) == 8 and all(0 <= word < 2**32 for word in key)
+    assert key != xxtea_encryption.derive_key("Johnny Cash ")
+    sha = hashlib.sha256(b"Johnny Cash").digest()
+    assert key[0] != int.from_bytes(sha[:4], "little")