From 5e6d74f170504543688432ee7422318b4f931f2b Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Sat, 5 Sep 2026 13:50:46 +0200 Subject: [PATCH] Key the web_server warning off the OTA platform and tell plaintext uploaders about the offer --- THREAT_MODEL.md | 9 ++-- esphome/components/esphome/ota/__init__.py | 25 +++++++---- esphome/espota2.py | 6 +++ tests/component_tests/ota/test_esphome_ota.py | 41 +++++++++++++++++++ tests/unit_tests/test_espota2_noise.py | 12 ++++++ 5 files changed, 82 insertions(+), 11 deletions(-) diff --git a/THREAT_MODEL.md b/THREAT_MODEL.md index 2ff6437ecf..25e6bc29da 100644 --- a/THREAT_MODEL.md +++ b/THREAT_MODEL.md @@ -151,10 +151,11 @@ The following are **not** vulnerabilities, by design: encryption, so turning on `ota: encryption:` is itself an encrypted upload. Older firmware needs one last plaintext upload of an offering build, with the pre-existing plaintext exposure. -- The web OTA `/update` endpoint alongside encryption. The `web_server` - component keeps it always reachable and validation warns about that - combination; `captive_portal:` auto-loads it only for the fallback AP - window, which is the intended recovery path, so that is not warned about. +- The web OTA `/update` endpoint alongside encryption. The `web_server` OTA + platform keeps it always reachable and validation warns about that + combination; `captive_portal:` auto-loads that platform only for the + fallback AP window, which is the intended recovery path, so that alone is + not warned about. - CLI retry behavior on transport or MAC failures; every attempt renegotiates a fresh handshake with fresh ephemerals, so retrying does not weaken authentication. diff --git a/esphome/components/esphome/ota/__init__.py b/esphome/components/esphome/ota/__init__.py index 1f04008b94..6639b5f9ab 100644 --- a/esphome/components/esphome/ota/__init__.py +++ b/esphome/components/esphome/ota/__init__.py @@ -27,6 +27,7 @@ import esphome.final_validate as fv from esphome.types import ConfigType CONF_ALLOW_PARTITION_ACCESS = "allow_partition_access" +CONF_CAPTIVE_PORTAL = "captive_portal" _LOGGER = logging.getLogger(__name__) @@ -133,9 +134,9 @@ def ota_esphome_final_validate(config: ConfigType) -> None: "'%s' %s wastes significant flash and RAM (about 3.5 KB and 60 " "bytes plus the password on the heap): the device already offers " "encryption with the '%s' %s %s, which authenticates any uploader " - "that takes it, so the password only serves older clients that do " - "not support encryption; remove '%s' and add '%s' under '%s' to " - "require encryption", + "that takes it, and a password only matters for uploaders without " + "encryption support; remove '%s' and add '%s' under '%s' so " + "uploads use the key and encryption is required", CONF_OTA, CONF_PASSWORD, CONF_API, @@ -147,9 +148,18 @@ def ota_esphome_final_validate(config: ConfigType) -> None: ) # The captive_portal auto-loads the web_server ota platform too, but that # endpoint only exists while the fallback AP is active and is the intended - # recovery path, so only an explicit web_server component warns - if CONF_WEB_SERVER in full_conf and any( - CONF_ENCRYPTION in conf for conf in merged_ota_esphome_configs_by_port.values() + # recovery path, so it does not warn unless the web_server component is + # configured as well + captive_portal_only = ( + CONF_CAPTIVE_PORTAL in full_conf and CONF_WEB_SERVER not in full_conf + ) + if ( + not captive_portal_only + and any(conf.get(CONF_PLATFORM) == CONF_WEB_SERVER for conf in full_ota_conf) + and any( + CONF_ENCRYPTION in conf + for conf in merged_ota_esphome_configs_by_port.values() + ) ): _warn_web_server_ota() @@ -306,13 +316,14 @@ async def to_code(config: ConfigType) -> None: if (encryption_conf := config.get(CONF_ENCRYPTION)) is not None: # A missing key was resolved from the api component in final validate. key = encryption_conf[CONF_KEY] - cg.add_define("USE_OTA_ENCRYPTION_REQUIRED") else: # An api key alone makes the device offer encryption while still # accepting plaintext, so the upload that adds the block is encrypted key = _api_static_key(CORE.config.get(CONF_API) or {}) if key is not None: cg.add_define("USE_OTA_ENCRYPTION") + if encryption_conf is not None: + cg.add_define("USE_OTA_ENCRYPTION_REQUIRED") cg.add(var.set_noise_psk(new_psk_progmem(config[CONF_ID], key))) # Build flag so lwip_fast_select.c (a .c file that can't include defines.h) sees it. diff --git a/esphome/espota2.py b/esphome/espota2.py index e18359c32a..f7c931a436 100644 --- a/esphome/espota2.py +++ b/esphome/espota2.py @@ -528,6 +528,12 @@ def perform_ota( else: features = 0 + if not noise_psk and extended_proto and features & SERVER_FEATURE_SUPPORTS_NOISE: + _LOGGER.warning( + "The device offers OTA encryption but this upload is plaintext; " + "add 'encryption:' under 'ota: platform: esphome' to use it" + ) + if noise_psk: # Fail closed: never fall back to a plaintext upload when an # encryption key is configured, an active attacker could otherwise diff --git a/tests/component_tests/ota/test_esphome_ota.py b/tests/component_tests/ota/test_esphome_ota.py index 26d22472f7..9baf125b56 100644 --- a/tests/component_tests/ota/test_esphome_ota.py +++ b/tests/component_tests/ota/test_esphome_ota.py @@ -387,6 +387,47 @@ def test_password_without_static_api_key_no_warning( fv.full_config.reset(token) +def test_web_server_component_without_ota_platform_does_not_warn( + caplog: pytest.LogCaptureFixture, +) -> None: + """The web_server component alone has no /update endpoint.""" + full_conf = { + "web_server": {}, + CONF_OTA: [ + _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}}) + ], + } + token = fv.full_config.set(full_conf) + try: + with caplog.at_level(logging.WARNING): + ota_esphome_final_validate({}) + assert not any( + "OTA encryption does not cover" in r.message for r in caplog.records + ) + finally: + fv.full_config.reset(token) + + +def test_web_server_ota_platform_alone_warns( + caplog: pytest.LogCaptureFixture, +) -> None: + """An explicit web_server ota platform exposes /update permanently, with + or without the web_server component.""" + full_conf = { + CONF_OTA: [ + _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}}), + {CONF_PLATFORM: "web_server", CONF_ID: ID("ota_ws", is_manual=False)}, + ], + } + token = fv.full_config.set(full_conf) + try: + with caplog.at_level(logging.WARNING): + ota_esphome_final_validate({}) + assert any("plaintext /update" in r.message for r in caplog.records) + finally: + fv.full_config.reset(token) + + def test_web_server_ota_without_encryption_unaffected() -> None: """web_server ota stays valid alongside an unencrypted esphome entry.""" full_conf = { diff --git a/tests/unit_tests/test_espota2_noise.py b/tests/unit_tests/test_espota2_noise.py index 61bc87fcfe..8ca3e4a553 100644 --- a/tests/unit_tests/test_espota2_noise.py +++ b/tests/unit_tests/test_espota2_noise.py @@ -13,6 +13,7 @@ import base64 from collections.abc import Callable import hashlib import io +import logging from pathlib import Path import socket import sys @@ -264,6 +265,17 @@ def test_client_fails_closed_when_device_lacks_encryption() -> None: device.join_and_check() +def test_plaintext_upload_to_offering_device_warns( + caplog: pytest.LogCaptureFixture, +) -> None: + """A keyless client is told the device could have encrypted the upload.""" + device = FakeEncryptedDevice(offer_noise=True, require_noise=False) + with patch("time.sleep"), caplog.at_level(logging.WARNING): + _upload(device, b"firmware", None) + device.join_and_check() + assert any("offers OTA encryption" in r.message for r in caplog.records) + + @pytest.mark.parametrize("noise_psk", [None, PSK], ids=["plaintext", "encrypted"]) def test_offering_device_accepts_either_transport(noise_psk: str | None) -> None: """A device that offers but does not require encryption takes a plaintext