preen

esphome/components/esphome/ota/__init__.py

@@ -24,7 +24,7 @@ _LOGGER = logging.getLogger(__name__)
 
 
 CODEOWNERS = ["@esphome/core"]
-AUTO_LOAD = ["md5", "socket"]
+AUTO_LOAD = ["md5", "sha256", "socket"]
 DEPENDENCIES = ["network"]
 
 esphome = cg.esphome_ns.namespace("esphome")

esphome/components/esphome/ota/ota_esphome.cpp

@@ -1,6 +1,9 @@
 #include "ota_esphome.h"
 #ifdef USE_OTA
 #include "esphome/components/md5/md5.h"
+#ifdef USE_SHA256
+#include "esphome/components/sha256/sha256.h"
+#endif
 #include "esphome/components/network/util.h"
 #include "esphome/components/ota/ota_backend.h"
 #include "esphome/components/ota/ota_backend_arduino_esp32.h"
@@ -95,6 +98,111 @@ void ESPHomeOTAComponent::loop() {
 }
 
 static const uint8_t FEATURE_SUPPORTS_COMPRESSION = 0x01;
+#ifdef USE_SHA256
+static const uint8_t FEATURE_SUPPORTS_SHA256_AUTH = 0x02;
+#endif
+
+// Template traits for hash algorithms
+template<typename HashClass> struct HashTraits;
+
+template<> struct HashTraits<md5::MD5Digest> {
+  static constexpr int nonce_size = 8;
+  static constexpr int hex_size = 32;
+  static constexpr const char *name = "MD5";
+  static constexpr ota::OTAResponseTypes auth_request = ota::OTA_RESPONSE_REQUEST_AUTH;
+};
+
+#ifdef USE_SHA256
+template<> struct HashTraits<sha256::SHA256> {
+  static constexpr int nonce_size = 16;
+  static constexpr int hex_size = 64;
+  static constexpr const char *name = "SHA256";
+  static constexpr ota::OTAResponseTypes auth_request = ota::OTA_RESPONSE_REQUEST_SHA256_AUTH;
+};
+#endif
+
+// Template helper for hash-based authentication
+template<typename HashClass> bool perform_hash_auth(ESPHomeOTAComponent *ota, const std::string &password) {
+  using Traits = HashTraits<HashClass>;
+
+  // Minimize stack usage by reusing buffers
+  // We only need 2 buffers at most at the same time
+  constexpr size_t hex_buffer_size = Traits::hex_size + 1;
+
+  // These two buffers are reused throughout the function
+  char hex_buffer1[hex_buffer_size];  // Used for: nonce -> expected result
+  char hex_buffer2[hex_buffer_size];  // Used for: cnonce -> response
+
+  // Small stack buffer for auth request and nonce seed
+  uint8_t buf[1];
+  char nonce_seed[17];  // Max: "%08x%08x" = 16 chars + null
+
+  // Send auth request type
+  buf[0] = Traits::auth_request;
+  ota->writeall_(buf, 1);
+
+  HashClass hasher;
+  hasher.init();
+
+  // Generate nonce seed
+  if (Traits::nonce_size == 8) {
+    sprintf(nonce_seed, "%08" PRIx32, random_uint32());
+  } else {
+    sprintf(nonce_seed, "%08" PRIx32 "%08" PRIx32, random_uint32(), random_uint32());
+  }
+  hasher.add(nonce_seed, Traits::nonce_size);
+  hasher.calculate();
+
+  // Use hex_buffer1 for nonce
+  hasher.get_hex(hex_buffer1);
+  hex_buffer1[Traits::hex_size] = '\0';
+  ESP_LOGV("esphome.ota", "Auth: %s Nonce is %s", Traits::name, hex_buffer1);
+
+  // Send nonce
+  if (!ota->writeall_(reinterpret_cast<uint8_t *>(hex_buffer1), Traits::hex_size)) {
+    ESP_LOGW("esphome.ota", "Auth: Writing %s nonce failed", Traits::name);
+    return false;
+  }
+
+  // Prepare challenge
+  hasher.init();
+  hasher.add(password.c_str(), password.length());
+  hasher.add(hex_buffer1, Traits::hex_size);  // Add nonce
+
+  // Receive cnonce into hex_buffer2
+  if (!ota->readall_(reinterpret_cast<uint8_t *>(hex_buffer2), Traits::hex_size)) {
+    ESP_LOGW("esphome.ota", "Auth: Reading %s cnonce failed", Traits::name);
+    return false;
+  }
+  hex_buffer2[Traits::hex_size] = '\0';
+  ESP_LOGV("esphome.ota", "Auth: %s CNonce is %s", Traits::name, hex_buffer2);
+
+  // Add cnonce to hash
+  hasher.add(hex_buffer2, Traits::hex_size);
+
+  // Calculate result - reuse hex_buffer1 for expected
+  hasher.calculate();
+  hasher.get_hex(hex_buffer1);
+  hex_buffer1[Traits::hex_size] = '\0';
+  ESP_LOGV("esphome.ota", "Auth: %s Result is %s", Traits::name, hex_buffer1);
+
+  // Receive response - reuse hex_buffer2
+  if (!ota->readall_(reinterpret_cast<uint8_t *>(hex_buffer2), Traits::hex_size)) {
+    ESP_LOGW("esphome.ota", "Auth: Reading %s response failed", Traits::name);
+    return false;
+  }
+  hex_buffer2[Traits::hex_size] = '\0';
+  ESP_LOGV("esphome.ota", "Auth: %s Response is %s", Traits::name, hex_buffer2);
+
+  // Compare
+  bool matches = memcmp(hex_buffer1, hex_buffer2, Traits::hex_size) == 0;
+
+  if (!matches) {
+    ESP_LOGW("esphome.ota", "Auth failed! %s passwords do not match", Traits::name);
+  }
+
+  return matches;
+}
 
 void ESPHomeOTAComponent::handle_handshake_() {
   /// Handle the initial OTA handshake.
@@ -225,57 +333,23 @@ void ESPHomeOTAComponent::handle_data_() {
 
 #ifdef USE_OTA_PASSWORD
   if (!this->password_.empty()) {
-    buf[0] = ota::OTA_RESPONSE_REQUEST_AUTH;
+    bool auth_success = false;
-    this->writeall_(buf, 1);
-    md5::MD5Digest md5{};
-    md5.init();
-    sprintf(sbuf, "%08" PRIx32, random_uint32());
-    md5.add(sbuf, 8);
-    md5.calculate();
-    md5.get_hex(sbuf);
-    ESP_LOGV(TAG, "Auth: Nonce is %s", sbuf);
 
-    // Send nonce, 32 bytes hex MD5
+#ifdef USE_SHA256
-    if (!this->writeall_(reinterpret_cast<uint8_t *>(sbuf), 32)) {
+    // Check if client supports SHA256 auth
-      ESP_LOGW(TAG, "Auth: Writing nonce failed");
+    bool use_sha256 = (ota_features & FEATURE_SUPPORTS_SHA256_AUTH) != 0;
-      goto error;  // NOLINT(cppcoreguidelines-avoid-goto)
+
+    if (use_sha256) {
+      // Use SHA256 for authentication
+      auth_success = perform_hash_auth<sha256::SHA256>(this, this->password_);
+    } else
+#endif  // USE_SHA256
+    {
+      // Fall back to MD5 for backward compatibility (or when SHA256 is not available)
+      auth_success = perform_hash_auth<md5::MD5Digest>(this, this->password_);
     }
 
-    // prepare challenge
+    if (!auth_success) {
-    md5.init();
-    md5.add(this->password_.c_str(), this->password_.length());
-    // add nonce
-    md5.add(sbuf, 32);
-
-    // Receive cnonce, 32 bytes hex MD5
-    if (!this->readall_(buf, 32)) {
-      ESP_LOGW(TAG, "Auth: Reading cnonce failed");
-      goto error;  // NOLINT(cppcoreguidelines-avoid-goto)
-    }
-    sbuf[32] = '\0';
-    ESP_LOGV(TAG, "Auth: CNonce is %s", sbuf);
-    // add cnonce
-    md5.add(sbuf, 32);
-
-    // calculate result
-    md5.calculate();
-    md5.get_hex(sbuf);
-    ESP_LOGV(TAG, "Auth: Result is %s", sbuf);
-
-    // Receive result, 32 bytes hex MD5
-    if (!this->readall_(buf + 64, 32)) {
-      ESP_LOGW(TAG, "Auth: Reading response failed");
-      goto error;  // NOLINT(cppcoreguidelines-avoid-goto)
-    }
-    sbuf[64 + 32] = '\0';
-    ESP_LOGV(TAG, "Auth: Response is %s", sbuf + 64);
-
-    bool matches = true;
-    for (uint8_t i = 0; i < 32; i++)
-      matches = matches && buf[i] == buf[64 + i];
-
-    if (!matches) {
-      ESP_LOGW(TAG, "Auth failed! Passwords do not match");
       error_code = ota::OTA_RESPONSE_ERROR_AUTH_INVALID;
       goto error;  // NOLINT(cppcoreguidelines-avoid-goto)
     }

esphome/components/ota/ota_backend.h

@@ -14,6 +14,7 @@ namespace ota {
 enum OTAResponseTypes {
   OTA_RESPONSE_OK = 0x00,
   OTA_RESPONSE_REQUEST_AUTH = 0x01,
+  OTA_RESPONSE_REQUEST_SHA256_AUTH = 0x02,
 
   OTA_RESPONSE_HEADER_OK = 0x40,
   OTA_RESPONSE_AUTH_OK = 0x41,

esphome/components/sha256/__init__.py

@@ -0,0 +1,14 @@
+import esphome.codegen as cg
+import esphome.config_validation as cv
+from esphome.core import coroutine_with_priority
+
+CODEOWNERS = ["@esphome/core"]
+
+sha256_ns = cg.esphome_ns.namespace("sha256")
+
+CONFIG_SCHEMA = cv.All(cv.Schema({}))
+
+
+@coroutine_with_priority(1.0)
+async def to_code(config):
+    cg.add_define("USE_SHA256")

esphome/components/sha256/sha256.cpp

@@ -0,0 +1,131 @@
+#include "sha256.h"
+#include "esphome/core/helpers.h"
+#include <cstring>
+
+#ifdef USE_ESP32
+#include "mbedtls/sha256.h"
+#elif defined(USE_ESP8266) || defined(USE_RP2040)
+#include <SHA256.h>
+#endif
+
+namespace esphome {
+namespace sha256 {
+
+#ifdef USE_ESP32
+struct SHA256::SHA256Context {
+  mbedtls_sha256_context ctx;
+  uint8_t hash[32];
+};
+
+SHA256::~SHA256() {
+  if (this->ctx_) {
+    mbedtls_sha256_free(&this->ctx_->ctx);
+  }
+}
+
+void SHA256::init() {
+  if (!this->ctx_) {
+    this->ctx_ = std::make_unique<SHA256Context>();
+  }
+  mbedtls_sha256_init(&this->ctx_->ctx);
+  mbedtls_sha256_starts(&this->ctx_->ctx, 0);  // 0 = SHA256, not SHA224
+}
+
+void SHA256::add(const uint8_t *data, size_t len) {
+  if (!this->ctx_) {
+    this->init();
+  }
+  mbedtls_sha256_update(&this->ctx_->ctx, data, len);
+}
+
+void SHA256::calculate() {
+  if (!this->ctx_) {
+    this->init();
+  }
+  mbedtls_sha256_finish(&this->ctx_->ctx, this->ctx_->hash);
+}
+
+#elif defined(USE_ESP8266) || defined(USE_RP2040)
+
+struct SHA256::SHA256Context {
+  ::SHA256 sha;
+  uint8_t hash[32];
+  bool calculated{false};
+};
+
+SHA256::~SHA256() = default;
+
+void SHA256::init() {
+  if (!this->ctx_) {
+    this->ctx_ = std::make_unique<SHA256Context>();
+  }
+  this->ctx_->sha.reset();
+  this->ctx_->calculated = false;
+}
+
+void SHA256::add(const uint8_t *data, size_t len) {
+  if (!this->ctx_) {
+    this->init();
+  }
+  this->ctx_->sha.update(data, len);
+}
+
+void SHA256::calculate() {
+  if (!this->ctx_) {
+    this->init();
+  }
+  if (!this->ctx_->calculated) {
+    this->ctx_->sha.finalize(this->ctx_->hash, 32);
+    this->ctx_->calculated = true;
+  }
+}
+
+#else
+#error "SHA256 not supported on this platform"
+#endif
+
+void SHA256::get_bytes(uint8_t *output) {
+  if (!this->ctx_) {
+    memset(output, 0, 32);
+    return;
+  }
+  memcpy(output, this->ctx_->hash, 32);
+}
+
+void SHA256::get_hex(char *output) {
+  if (!this->ctx_) {
+    memset(output, '0', 64);
+    output[64] = '\0';
+    return;
+  }
+  for (size_t i = 0; i < 32; i++) {
+    sprintf(output + i * 2, "%02x", this->ctx_->hash[i]);
+  }
+}
+
+std::string SHA256::get_hex_string() {
+  char buf[65];
+  this->get_hex(buf);
+  return std::string(buf);
+}
+
+bool SHA256::equals_bytes(const uint8_t *expected) {
+  if (!this->ctx_) {
+    return false;
+  }
+  return memcmp(this->ctx_->hash, expected, 32) == 0;
+}
+
+bool SHA256::equals_hex(const char *expected) {
+  if (!this->ctx_) {
+    return false;
+  }
+  uint8_t parsed[32];
+  if (!parse_hex(expected, parsed, 32)) {
+    return false;
+  }
+  return this->equals_bytes(parsed);
+}
+
+}  // namespace sha256
+}  // namespace esphome

esphome/components/sha256/sha256.h

@@ -0,0 +1,36 @@
+#pragma once
+
+#include "esphome/core/defines.h"
+#include <cstdint>
+#include <string>
+#include <memory>
+
+namespace esphome {
+namespace sha256 {
+
+class SHA256 {
+ public:
+  SHA256() = default;
+  ~SHA256();
+
+  void init();
+  void add(const uint8_t *data, size_t len);
+  void add(const char *data, size_t len) { this->add((const uint8_t *) data, len); }
+  void add(const std::string &data) { this->add(data.c_str(), data.length()); }
+
+  void calculate();
+
+  void get_bytes(uint8_t *output);
+  void get_hex(char *output);
+  std::string get_hex_string();
+
+  bool equals_bytes(const uint8_t *expected);
+  bool equals_hex(const char *expected);
+
+ protected:
+  struct SHA256Context;
+  std::unique_ptr<SHA256Context> ctx_;
+};
+
+}  // namespace sha256
+}  // namespace esphome

esphome/core/defines.h

@@ -115,6 +115,7 @@
 #define USE_API_PLAINTEXT
 #define USE_API_SERVICES
 #define USE_MD5
+#define USE_SHA256
 #define USE_MQTT
 #define USE_NETWORK
 #define USE_ONLINE_IMAGE_BMP_SUPPORT

esphome/espota2.py

@@ -14,6 +14,7 @@ from esphome.helpers import resolve_ip_address
 
 RESPONSE_OK = 0x00
 RESPONSE_REQUEST_AUTH = 0x01
+RESPONSE_REQUEST_SHA256_AUTH = 0x02
 
 RESPONSE_HEADER_OK = 0x40
 RESPONSE_AUTH_OK = 0x41
@@ -44,6 +45,7 @@ OTA_VERSION_2_0 = 2
 MAGIC_BYTES = [0x6C, 0x26, 0xF7, 0x5C, 0x45]
 
 FEATURE_SUPPORTS_COMPRESSION = 0x01
+FEATURE_SUPPORTS_SHA256_AUTH = 0x02
 
 
 UPLOAD_BLOCK_SIZE = 8192
@@ -209,10 +211,14 @@ def perform_ota(
             f"Device uses unsupported OTA version {version}, this ESPHome supports {supported_versions}"
         )
 
-    # Features
+    # Features - send both compression and SHA256 auth support
-    send_check(sock, FEATURE_SUPPORTS_COMPRESSION, "features")
+    features_to_send = FEATURE_SUPPORTS_COMPRESSION | FEATURE_SUPPORTS_SHA256_AUTH
+    send_check(sock, features_to_send, "features")
     features = receive_exactly(
-        sock, 1, "features", [RESPONSE_HEADER_OK, RESPONSE_SUPPORTS_COMPRESSION]
+        sock,
+        1,
+        "features",
+        None,  # Accept any response
     )[0]
 
     if features == RESPONSE_SUPPORTS_COMPRESSION:
@@ -221,31 +227,46 @@ def perform_ota(
     else:
         upload_contents = file_contents
 
-    (auth,) = receive_exactly(
+    def perform_auth(sock, password, hash_func, nonce_size, hash_name):
-        sock, 1, "auth", [RESPONSE_REQUEST_AUTH, RESPONSE_AUTH_OK]
+        """Perform challenge-response authentication using specified hash algorithm."""
-    )
-    if auth == RESPONSE_REQUEST_AUTH:
         if not password:
             raise OTAError("ESP requests password, but no password given!")
+
         nonce = receive_exactly(
-            sock, 32, "authentication nonce", [], decode=False
+            sock, nonce_size, f"{hash_name} authentication nonce", [], decode=False
         ).decode()
-        _LOGGER.debug("Auth: Nonce is %s", nonce)
+        _LOGGER.debug("Auth: %s Nonce is %s", hash_name, nonce)
-        cnonce = hashlib.md5(str(random.random()).encode()).hexdigest()
+
-        _LOGGER.debug("Auth: CNonce is %s", cnonce)
+        # Generate cnonce
+        cnonce = hash_func(str(random.random()).encode()).hexdigest()
+        _LOGGER.debug("Auth: %s CNonce is %s", hash_name, cnonce)
 
         send_check(sock, cnonce, "auth cnonce")
 
-        result_md5 = hashlib.md5()
+        # Calculate challenge response
-        result_md5.update(password.encode("utf-8"))
+        hasher = hash_func()
-        result_md5.update(nonce.encode())
+        hasher.update(password.encode("utf-8"))
-        result_md5.update(cnonce.encode())
+        hasher.update(nonce.encode())
-        result = result_md5.hexdigest()
+        hasher.update(cnonce.encode())
-        _LOGGER.debug("Auth: Result is %s", result)
+        result = hasher.hexdigest()
+        _LOGGER.debug("Auth: %s Result is %s", hash_name, result)
 
         send_check(sock, result, "auth result")
         receive_exactly(sock, 1, "auth result", RESPONSE_AUTH_OK)
 
+    (auth,) = receive_exactly(
+        sock,
+        1,
+        "auth",
+        [RESPONSE_REQUEST_AUTH, RESPONSE_REQUEST_SHA256_AUTH, RESPONSE_AUTH_OK],
+    )
+    if auth == RESPONSE_REQUEST_SHA256_AUTH:
+        # SHA256 authentication
+        perform_auth(sock, password, hashlib.sha256, 64, "SHA256")
+    elif auth == RESPONSE_REQUEST_AUTH:
+        # MD5 authentication (backward compatibility)
+        perform_auth(sock, password, hashlib.md5, 32, "MD5")
+
     # Set higher timeout during upload
     sock.settimeout(30.0)