[ota.esphome] Handle blank password the same as no password defined (#11271)

esphome/__main__.py

@@ -606,7 +606,7 @@ def upload_program(
     from esphome import espota2
 
     remote_port = int(ota_conf[CONF_PORT])
-    password = ota_conf.get(CONF_PASSWORD, "")
+    password = ota_conf.get(CONF_PASSWORD)
     if getattr(args, "file", None) is not None:
         binary = Path(args.file)
     else:

esphome/components/esphome/ota/__init__.py

@@ -19,6 +19,7 @@ from esphome.const import (
 from esphome.core import CORE, coroutine_with_priority
 from esphome.coroutine import CoroPriority
 import esphome.final_validate as fv
+from esphome.types import ConfigType
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -136,11 +137,12 @@ FINAL_VALIDATE_SCHEMA = ota_esphome_final_validate
 
 
 @coroutine_with_priority(CoroPriority.OTA_UPDATES)
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = cg.new_Pvariable(config[CONF_ID])
     cg.add(var.set_port(config[CONF_PORT]))
 
-    if CONF_PASSWORD in config:
+    # Password could be set to an empty string and we can assume that means no password
+    if config.get(CONF_PASSWORD):
         cg.add(var.set_auth_password(config[CONF_PASSWORD]))
         cg.add_define("USE_OTA_PASSWORD")
         # Only include hash algorithms when password is configured

esphome/espota2.py

@@ -242,7 +242,7 @@ def send_check(
 
 
 def perform_ota(
-    sock: socket.socket, password: str, file_handle: io.IOBase, filename: Path
+    sock: socket.socket, password: str | None, file_handle: io.IOBase, filename: Path
 ) -> None:
     file_contents = file_handle.read()
     file_size = len(file_contents)
@@ -278,13 +278,13 @@ def perform_ota(
 
     def perform_auth(
         sock: socket.socket,
-        password: str,
+        password: str | None,
         hash_func: Callable[..., Any],
         nonce_size: int,
         hash_name: str,
     ) -> None:
         """Perform challenge-response authentication using specified hash algorithm."""
-        if not password:
+        if password is None:
             raise OTAError("ESP requests password, but no password given!")
 
         nonce_bytes = receive_exactly(
@@ -385,7 +385,7 @@ def perform_ota(
 
 
 def run_ota_impl_(
-    remote_host: str | list[str], remote_port: int, password: str, filename: Path
+    remote_host: str | list[str], remote_port: int, password: str | None, filename: Path
 ) -> tuple[int, str | None]:
     from esphome.core import CORE
 
@@ -436,7 +436,7 @@ def run_ota_impl_(
 
 
 def run_ota(
-    remote_host: str | list[str], remote_port: int, password: str, filename: Path
+    remote_host: str | list[str], remote_port: int, password: str | None, filename: Path
 ) -> tuple[int, str | None]:
     try:
         return run_ota_impl_(remote_host, remote_port, password, filename)

tests/unit_tests/test_espota2.py

@@ -287,7 +287,7 @@ def test_perform_ota_no_auth(mock_socket: Mock, mock_file: io.BytesIO) -> None:
 
     mock_socket.recv.side_effect = recv_responses
 
-    espota2.perform_ota(mock_socket, "", mock_file, "test.bin")
+    espota2.perform_ota(mock_socket, None, mock_file, "test.bin")
 
     # Should not send any auth-related data
     auth_calls = [
@@ -317,7 +317,7 @@ def test_perform_ota_with_compression(mock_socket: Mock) -> None:
 
     mock_socket.recv.side_effect = recv_responses
 
-    espota2.perform_ota(mock_socket, "", mock_file, "test.bin")
+    espota2.perform_ota(mock_socket, None, mock_file, "test.bin")
 
     # Verify compressed content was sent
     # Get the binary size that was sent (4 bytes after features)
@@ -347,7 +347,7 @@ def test_perform_ota_auth_without_password(mock_socket: Mock) -> None:
     with pytest.raises(
         espota2.OTAError, match="ESP requests password, but no password given"
     ):
-        espota2.perform_ota(mock_socket, "", mock_file, "test.bin")
+        espota2.perform_ota(mock_socket, None, mock_file, "test.bin")
 
 
 @pytest.mark.usefixtures("mock_time")
@@ -413,7 +413,7 @@ def test_perform_ota_sha256_auth_without_password(mock_socket: Mock) -> None:
     with pytest.raises(
         espota2.OTAError, match="ESP requests password, but no password given"
     ):
-        espota2.perform_ota(mock_socket, "", mock_file, "test.bin")
+        espota2.perform_ota(mock_socket, None, mock_file, "test.bin")
 
 
 def test_perform_ota_unexpected_auth_response(mock_socket: Mock) -> None:
@@ -450,7 +450,7 @@ def test_perform_ota_unsupported_version(mock_socket: Mock) -> None:
     mock_socket.recv.side_effect = responses
 
     with pytest.raises(espota2.OTAError, match="Device uses unsupported OTA version"):
-        espota2.perform_ota(mock_socket, "", mock_file, "test.bin")
+        espota2.perform_ota(mock_socket, None, mock_file, "test.bin")
 
 
 @pytest.mark.usefixtures("mock_time")
@@ -471,7 +471,7 @@ def test_perform_ota_upload_error(mock_socket: Mock, mock_file: io.BytesIO) -> N
     mock_socket.recv.side_effect = recv_responses
 
     with pytest.raises(espota2.OTAError, match="Error receiving acknowledge chunk OK"):
-        espota2.perform_ota(mock_socket, "", mock_file, "test.bin")
+        espota2.perform_ota(mock_socket, None, mock_file, "test.bin")
 
 
 @pytest.mark.usefixtures("mock_socket_constructor", "mock_resolve_ip")
@@ -706,7 +706,7 @@ def test_perform_ota_version_differences(
     ]
 
     mock_socket.recv.side_effect = recv_responses
-    espota2.perform_ota(mock_socket, "", mock_file, "test.bin")
+    espota2.perform_ota(mock_socket, None, mock_file, "test.bin")
 
     # For v1.0, verify that we only get the expected number of recv calls
     # v1.0 doesn't have chunk acknowledgments, so fewer recv calls
@@ -732,7 +732,7 @@ def test_perform_ota_version_differences(
     ]
 
     mock_socket.recv.side_effect = recv_responses_v2
-    espota2.perform_ota(mock_socket, "", mock_file, "test.bin")
+    espota2.perform_ota(mock_socket, None, mock_file, "test.bin")
 
     # For v2.0, verify more recv calls due to chunk acknowledgments
     assert mock_socket.recv.call_count == 9  # v2.0 has 9 recv calls (includes chunk OK)

tests/unit_tests/test_main.py

@@ -1062,7 +1062,7 @@ def test_upload_program_ota_with_file_arg(
     assert exit_code == 0
     assert host == "192.168.1.100"
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.100"], 3232, "", Path("custom.bin")
+        ["192.168.1.100"], 3232, None, Path("custom.bin")
     )
 
 
@@ -1119,7 +1119,9 @@ def test_upload_program_ota_with_mqtt_resolution(
     expected_firmware = (
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
     )
-    mock_run_ota.assert_called_once_with(["192.168.1.100"], 3232, "", expected_firmware)
+    mock_run_ota.assert_called_once_with(
+        ["192.168.1.100"], 3232, None, expected_firmware
+    )
 
 
 @patch("esphome.__main__.importlib.import_module")