[cli] Add --prompt-ota-key to present a different key for one upload (#19397)

esphome/__main__.py

@@ -48,8 +48,11 @@ from esphome.const import (
     CONF_WEB_SERVER,
     CONF_WIFI,
     ENV_NOGITIGNORE,
+    KEY_CORE,
     KEY_ESP32,
+    KEY_TARGET_PLATFORM,
     KEY_VARIANT,
+    PLATFORM_HOST,
     SECRETS_FILES,
     ErrorFormat,
     Toolchain,
@@ -70,6 +73,7 @@ from esphome.util import (
     get_serial_ports,
     is_picotool_usb_permission_error,
     list_yaml_files,
+    read_secret_line,
     run_external_command,
     run_external_process,
     safe_print,
@@ -1232,14 +1236,28 @@ def upload_program(
     config: ConfigType, args: ArgsProtocol, devices: list[str]
 ) -> tuple[int, str | None]:
     host = devices[0]
+    port_type = get_port_type(host)
+    ota_key = getattr(args, "ota_key", None)
+    if ota_key and port_type in (PortType.SERIAL, PortType.BOOTSEL):
+        _LOGGER.warning(
+            "The presented OTA key only applies to network uploads; ignored for %s",
+            host,
+        )
+
     platform_upload = platform_hooks.get_platform_hook(
         CORE.target_platform, "upload_program"
     )
     if platform_upload is not None and platform_upload(config, args, host):
+        if ota_key and port_type not in (PortType.SERIAL, PortType.BOOTSEL):
+            # A platform's own network path (nrf52 mcumgr) has no key handshake
+            _LOGGER.warning(
+                "The presented OTA key only applies to the %s OTA platform; "
+                "ignored for %s",
+                CONF_ESPHOME,
+                host,
+            )
         return 0, host
 
-    port_type = get_port_type(host)
-
     # MQTT and MQTTIP are also OTA paths; MQTTIP gets resolved to a real IP later by
     # _resolve_network_devices(). Only SERIAL and BOOTSEL are non-OTA upload paths.
     is_partition_table = getattr(args, "partition_table", False)
@@ -1282,6 +1300,8 @@ def upload_program(
     network_devices = _resolve_network_devices(devices, config, args)
 
     if chosen_platform == CONF_WEB_SERVER:
+        if ota_key:
+            raise EsphomeError(OTA_KEY_PLATFORM_ERROR)
         if is_partition_table or is_bootloader:
             raise EsphomeError(
                 f"{option_string} is only supported with the esphome OTA platform; "
@@ -1295,6 +1315,17 @@ def upload_program(
     return _upload_via_native_api(config, network_devices, args)
 
 
+def _ota_upload_platforms(config: ConfigType) -> dict[str, None]:
+    """The network OTA platforms the config provides, one entry each, in
+    config order; the web_server final-validate hook merges duplicates."""
+    return {
+        ota_platform: None
+        for ota_item in config.get(CONF_OTA, [])
+        if (ota_platform := ota_item.get(CONF_PLATFORM))
+        in (CONF_ESPHOME, CONF_WEB_SERVER)
+    }
+
+
 def _choose_ota_platform(config: ConfigType, requested: str | None) -> str:
     """Pick the OTA platform to use, optionally honoring ``--ota-platform``.
 
@@ -1307,16 +1338,7 @@ def _choose_ota_platform(config: ConfigType, requested: str | None) -> str:
     stream on ESP32/LibreTiny, which inflate it as it arrives.) Falls back to
     ``web_server`` only when that is the only available platform.
     """
-    # Use a dict (insertion-ordered) instead of a list so error messages and
-    # membership checks see one entry per platform even if the user has
-    # multiple ``ota:`` items of the same platform; the web_server OTA
-    # platform's final-validate hook merges duplicates anyway.
-    available: dict[str, None] = {}
-    for ota_item in config.get(CONF_OTA, []):
-        ota_platform = ota_item.get(CONF_PLATFORM)
-        if ota_platform in (CONF_ESPHOME, CONF_WEB_SERVER):
-            available[ota_platform] = None
-
+    available = _ota_upload_platforms(config)
     if not available:
         raise EsphomeError(
             f"Cannot upload Over the Air as the {CONF_OTA} configuration is not "
@@ -1356,7 +1378,12 @@ def _upload_via_native_api(
     noise_psk = None
     plaintext_fallback = False
     allow_plaintext_upload = False
-    if (encryption_conf := ota_conf.get(CONF_ENCRYPTION)) is not None:
+    if ota_key := getattr(args, "ota_key", None):
+        # The user presents a key the device still runs instead of the
+        # configured one (a key change in progress); an explicit key never
+        # downgrades to plaintext
+        noise_psk = ota_key
+    elif (encryption_conf := ota_conf.get(CONF_ENCRYPTION)) is not None:
         noise_psk = encryption_conf.get(CONF_KEY)
         allow_plaintext_upload = bool(
             encryption_conf.get(espota2.CONF_ALLOW_PLAINTEXT_UPLOAD)
@@ -1738,7 +1765,60 @@ def command_compile(args: ArgsProtocol, config: ConfigType) -> int | None:
     return 0
 
 
+ENV_OTA_KEY = "ESPHOME_OTA_KEY"
+OTA_KEY_PLATFORM_ERROR = (
+    f"--prompt-ota-key and {ENV_OTA_KEY} only apply to the {CONF_ESPHOME} OTA "
+    "platform; to flash by serial instead, name the port with --device"
+)
+
+
+def _read_ota_key(args: ArgsProtocol, config: ConfigType) -> None:
+    """Take the key to present to the device for this upload, from the
+    prompt or from ESPHOME_OTA_KEY, before any compile so the prompt is
+    not buried in build output."""
+    prompt = getattr(args, "prompt_ota_key", False)
+    env_key = getattr(args, "env_ota_key", None)
+    if not prompt and env_key is None:
+        return
+    if CORE.data.get(KEY_CORE, {}).get(KEY_TARGET_PLATFORM) == PLATFORM_HOST:
+        _LOGGER.warning("A host program runs locally; the presented OTA key is ignored")
+        return
+    # The HTTP path has no key handshake; decide before prompting or
+    # compiling. A serial target is a serial flash: the key is not even
+    # asked for, and --ota-platform means nothing to it. Every entry must
+    # be serial: a missing port is skipped and the next entry may be a host
+    devices = getattr(args, "device", None) or []
+    serial_target = bool(devices) and all(
+        device == "SERIAL"
+        or get_port_type(device) in (PortType.SERIAL, PortType.BOOTSEL)
+        for device in devices
+    )
+    chosen = (
+        _choose_ota_platform(config, getattr(args, "ota_platform", None))
+        if not serial_target and _ota_upload_platforms(config)
+        else None
+    )
+    if chosen == CONF_WEB_SERVER:
+        raise EsphomeError(OTA_KEY_PLATFORM_ERROR)
+    from esphome.components.noise import validate_encryption_key
+    from esphome.config_validation import Invalid
+
+    if serial_target:
+        _LOGGER.warning(
+            "The presented OTA key only applies to network uploads; ignored for %s",
+            ", ".join(devices),
+        )
+        return
+    key = read_secret_line("OTA encryption key: ") if prompt else env_key
+    try:
+        validate_encryption_key(key)
+    except Invalid as err:
+        raise EsphomeError(f"Invalid OTA encryption key: {err}") from err
+    args.ota_key = key
+
+
 def command_upload(args: ArgsProtocol, config: ConfigType) -> int | None:
+    _read_ota_key(args, config)
     # Get devices, resolving special identifiers like OTA
     devices = choose_upload_log_host(
         default=args.device,
@@ -1786,6 +1866,7 @@ def command_run(args: ArgsProtocol, config: ConfigType) -> int | None:
         # Elsewhere the flag is ignored, so serial stays fine there.
         _LOGGER.error("--skip-bootloader builds cannot be flashed over serial")
         return 1
+    _read_ota_key(args, config)
     exit_code = write_cpp(config)
     if exit_code != 0:
         return exit_code
@@ -2216,6 +2297,15 @@ def _error_format(value: str) -> ErrorFormat:
         ) from None
 
 
+ARGUMENT_HELP_PROMPT_OTA_KEY = (
+    "Ask for the OTA encryption key to present to the device instead of using "
+    "the one in the configuration, for a device that still runs a previous key. "
+    "The key is read without echo, or as one line from stdin when that is not "
+    "a terminal; it is never taken from the command line. A calling program "
+    f"can hand the key over in the {ENV_OTA_KEY} environment variable instead."
+)
+
+
 def parse_args(argv):
     options_parser = argparse.ArgumentParser(add_help=False)
     options_parser.add_argument(
@@ -2384,6 +2474,11 @@ def parse_args(argv):
         help="Upload as partition table (OTA).",
         action="store_true",
     )
+    parser_upload.add_argument(
+        "--prompt-ota-key",
+        action="store_true",
+        help=ARGUMENT_HELP_PROMPT_OTA_KEY,
+    )
     parser_upload.add_argument(
         "--bootloader",
         help="Upload as bootloader (OTA).",
@@ -2464,6 +2559,11 @@ def parse_args(argv):
             "(HTTP Basic auth) when that is the only configured platform."
         ),
     )
+    parser_run.add_argument(
+        "--prompt-ota-key",
+        action="store_true",
+        help=ARGUMENT_HELP_PROMPT_OTA_KEY,
+    )
 
     parser_clean = subparsers.add_parser(
         "clean-mqtt",
@@ -2689,6 +2789,23 @@ def run_esphome(argv):
         args.log_level = "CRITICAL"
 
     setup_log(log_level=args.log_level)
+    # Taken out of the environment before any command runs, so no build
+    # tool inherits a key a calling program handed over
+    args.env_ota_key = os.environ.pop(ENV_OTA_KEY, None)
+    if args.command in ("upload", "run"):
+        # One key is one device's; several configs run in child processes
+        # that would never see it
+        if (args.env_ota_key is not None or args.prompt_ota_key) and len(
+            args.configuration
+        ) > 1:
+            raise EsphomeError(
+                f"--prompt-ota-key and {ENV_OTA_KEY} take one configuration; "
+                f"{len(args.configuration)} were given"
+            )
+    elif args.env_ota_key is not None:
+        _LOGGER.warning(
+            "%s is set but the %s command does not use it", ENV_OTA_KEY, args.command
+        )
     _warn_if_source_tree_mismatch()
     _warn_if_intel_macos()
 

esphome/util.py

@@ -130,6 +130,17 @@ def safe_input(prompt=""):
     return input()
 
 
+def read_secret_line(prompt: str) -> str:
+    """Read a secret without echo on a terminal, or one line from a pipe."""
+    if sys.stdin.isatty():
+        import getpass
+
+        return getpass.getpass(prompt).strip()
+    # Still say what is awaited, so a pipe held open is not a silent hang
+    print(prompt, end="", file=sys.stderr, flush=True)
+    return sys.stdin.readline().strip()
+
+
 def shlex_quote(s: str | Path) -> str:
     # Convert Path objects to strings
     if isinstance(s, Path):

tests/unit_tests/test_main.py

@@ -25,6 +25,7 @@ from esphome.__main__ import (
     Purpose,
     _get_configured_xtal_freq,
     _make_crystal_freq_callback,
+    _read_ota_key,
     _redact_with_legacy_fallback,
     _resolve_network_devices,
     _should_subscribe_states,
@@ -44,6 +45,7 @@ from esphome.__main__ import (
     command_idedata,
     command_run,
     command_update_all,
+    command_upload,
     command_wizard,
     compile_program,
     detect_external_components,
@@ -126,6 +128,7 @@ from esphome.espota2 import (
 )
 from esphome.platformio import toolchain
 from esphome.types import ConfigType
+from esphome.upload_targets import PortType
 from esphome.util import BootselResult, FlashImage
 from esphome.zeroconf import _await_discovery, discover_mdns_devices
 
@@ -1550,6 +1553,9 @@ class MockArgs:
     bootloader: bool = False
     states: bool | None = None
     device: list[str] | None = None
+    prompt_ota_key: bool = False
+    ota_key: str | None = None
+    env_ota_key: str | None = None
 
 
 def test_upload_program_serial_esp32(
@@ -2223,6 +2229,384 @@ def test_upload_program_ota_allow_plaintext_upload(
     }
 
 
+@pytest.mark.parametrize(
+    "ota_conf",
+    [
+        {
+            CONF_ENCRYPTION: {
+                CONF_KEY: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=",
+                "allow_plaintext_upload": True,
+            }
+        },
+        {},
+    ],
+    ids=["with_encryption_block", "without_encryption_block"],
+)
+def test_upload_program_ota_prompted_key_is_presented(
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+    ota_conf: dict[str, Any],
+) -> None:
+    """A prompted key replaces whatever the config holds for this upload and
+    never downgrades, whatever the config allows."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+    mock_run_ota.return_value = (0, "192.168.1.100")
+
+    old_key = "AQIDBAUGBwgJCgsMDQ4PEBESExQVFhcYGRobHB0eHyA="
+    config = {
+        CONF_API: {
+            CONF_ENCRYPTION: {CONF_KEY: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="}
+        },
+        CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 3232, **ota_conf}],
+    }
+    exit_code, _ = upload_program(config, MockArgs(ota_key=old_key), ["192.168.1.100"])
+
+    assert exit_code == 0
+    assert mock_run_ota.call_args.args[5] == old_key
+    assert mock_run_ota.call_args.kwargs == {
+        "plaintext_fallback": False,
+        "allow_plaintext_upload": False,
+    }
+
+
+def test_upload_program_serial_ignores_prompted_key(
+    mock_upload_using_esptool: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A serial upload has no handshake to present the key to."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = PortType.SERIAL
+    mock_upload_using_esptool.return_value = 0
+
+    config = {CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 3232}]}
+    with (
+        patch("esphome.__main__.check_permissions"),
+        caplog.at_level(logging.WARNING),
+    ):
+        exit_code, _ = upload_program(config, MockArgs(ota_key="key"), ["/dev/ttyUSB0"])
+
+    assert exit_code == 0
+    assert any("ignored for /dev/ttyUSB0" in r.message for r in caplog.records)
+
+
+def test_command_run_prompts_for_ota_key_before_compile() -> None:
+    """The prompt comes first so it is not buried in minutes of build output."""
+    setup_core(
+        config={CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME}]}, platform=PLATFORM_ESP32
+    )
+    key = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+    args = MockArgs(prompt_ota_key=True)
+    calls: list[str] = []
+
+    def read(prompt: str) -> str:
+        calls.append("prompt")
+        return key
+
+    def compile_program(*_: Any) -> int:
+        calls.append("compile")
+        assert args.ota_key == key
+        return 1  # stop before the upload
+
+    with (
+        patch("esphome.__main__.read_secret_line", side_effect=read),
+        patch("esphome.__main__.write_cpp", return_value=0),
+        patch("esphome.__main__.compile_program", side_effect=compile_program),
+    ):
+        assert command_run(args, CORE.config) == 1
+    assert calls == ["prompt", "compile"]
+
+
+def test_upload_program_platform_hook_ignores_prompted_key(
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A platform's own network upload path (nrf52 mcumgr) has no key
+    handshake; the key is dropped with a warning, never silently."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+    config = {CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 3232}]}
+    with (
+        patch(
+            "esphome.__main__.platform_hooks.get_platform_hook",
+            return_value=lambda *_: True,
+        ),
+        caplog.at_level(logging.WARNING),
+    ):
+        exit_code, host = upload_program(config, MockArgs(ota_key="key"), ["dev.local"])
+
+    assert (exit_code, host) == (0, "dev.local")
+    assert any("ignored for dev.local" in r.message for r in caplog.records)
+
+
+def test_command_upload_reads_the_ota_key_first() -> None:
+    """The upload command takes the key before choosing a device, like run."""
+    setup_core(
+        config={CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME}]}, platform=PLATFORM_ESP32
+    )
+    key = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+    args = MockArgs(prompt_ota_key=True)
+    args.device = None
+    with (
+        patch("esphome.__main__.read_secret_line", return_value=key),
+        patch("esphome.__main__.choose_upload_log_host", return_value=["dev.local"]),
+        patch(
+            "esphome.__main__.upload_program", return_value=(0, "dev.local")
+        ) as upload,
+    ):
+        assert command_upload(args, CORE.config) == 0
+    assert args.ota_key == key
+    upload.assert_called_once()
+
+
+def test_read_ota_key_stores_prompted_key() -> None:
+    """The key lands on args and never in argv."""
+    key = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+    args = MockArgs(prompt_ota_key=True)
+    config = {CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME}]}
+    with patch("esphome.__main__.read_secret_line", return_value=key) as read:
+        _read_ota_key(args, config)
+    read.assert_called_once_with("OTA encryption key: ")
+    assert args.ota_key == key
+
+
+def test_read_ota_key_not_requested() -> None:
+    args = MockArgs()
+    with patch("esphome.__main__.read_secret_line") as read:
+        _read_ota_key(args, {CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME}]})
+    read.assert_not_called()
+    assert args.ota_key is None
+
+
+def test_read_ota_key_from_environment() -> None:
+    """A calling program hands the key over in the environment."""
+    key = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+    args = MockArgs(env_ota_key=key)
+    with patch("esphome.__main__.read_secret_line") as read:
+        _read_ota_key(args, {CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME}]})
+    read.assert_not_called()
+    assert args.ota_key == key
+
+
+def test_read_ota_key_empty_environment_value_is_an_error() -> None:
+    """An exported empty value is a failed lookup upstream, not "unset"."""
+    args = MockArgs(env_ota_key="")
+    with pytest.raises(EsphomeError, match="Invalid OTA encryption key"):
+        _read_ota_key(args, {CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME}]})
+
+
+def test_read_ota_key_prompt_wins_over_environment() -> None:
+    key = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+    args = MockArgs(
+        prompt_ota_key=True, env_ota_key="AQIDBAUGBwgJCgsMDQ4PEBESExQVFhcYGRobHB0eHyA="
+    )
+    with patch("esphome.__main__.read_secret_line", return_value=key):
+        _read_ota_key(args, {CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME}]})
+    assert args.ota_key == key
+
+
+def test_read_ota_key_without_ota_block_is_left_to_the_upload() -> None:
+    """A serial flash config has no OTA; the key is ignored with a warning
+    at upload time instead of refusing here."""
+    key = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+    args = MockArgs(prompt_ota_key=True)
+    with patch("esphome.__main__.read_secret_line", return_value=key):
+        _read_ota_key(args, {})
+    assert args.ota_key == key
+
+
+@pytest.mark.parametrize(
+    "argv",
+    [
+        ["esphome", "run", "--prompt-ota-key", "a.yaml", "b.yaml"],
+        ["esphome", "upload", "a.yaml", "b.yaml"],
+    ],
+    ids=["prompt", "environment"],
+)
+def test_run_esphome_refuses_one_key_for_several_configs(
+    monkeypatch: pytest.MonkeyPatch, argv: list[str]
+) -> None:
+    """Several configs run in child processes that never see the key."""
+    from esphome.__main__ import run_esphome
+
+    if "--prompt-ota-key" not in argv:
+        monkeypatch.setenv("ESPHOME_OTA_KEY", "x")
+    with pytest.raises(EsphomeError, match="take one configuration; 2 were given"):
+        run_esphome(argv)
+
+
+def test_run_esphome_only_warns_for_a_command_with_one_string_config(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """The wizard takes one config as a string; only upload and run count."""
+    from esphome.__main__ import run_esphome
+
+    monkeypatch.setenv("ESPHOME_OTA_KEY", "x")
+    wizard = Mock(return_value=0)
+    with (
+        patch.dict("esphome.__main__.PRE_CONFIG_ACTIONS", {"wizard": wizard}),
+        caplog.at_level(logging.WARNING),
+    ):
+        assert run_esphome(["esphome", "wizard", "living-room.yaml"]) == 0
+    assert any("does not use it" in r.message for r in caplog.records)
+
+
+def test_run_esphome_scrubs_the_key_from_the_environment(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Every command takes the key out before any tool it runs could see it."""
+    from esphome.__main__ import run_esphome
+
+    monkeypatch.setenv("ESPHOME_OTA_KEY", "x")
+    version = Mock(return_value=0)
+    with (
+        patch.dict("esphome.__main__.PRE_CONFIG_ACTIONS", {"version": version}),
+        caplog.at_level(logging.WARNING),
+    ):
+        run_esphome(["esphome", "version"])
+    assert "ESPHOME_OTA_KEY" not in os.environ
+    assert version.call_args.args[0].env_ota_key == "x"
+    # A command that never uploads says so rather than dropping it silently
+    assert any("does not use it" in r.message for r in caplog.records)
+
+
+@pytest.mark.parametrize("line", ["", "not-base64"], ids=["empty", "invalid"])
+def test_read_ota_key_rejects_bad_input(line: str) -> None:
+    args = MockArgs(prompt_ota_key=True)
+    config = {CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME}]}
+    with (
+        patch("esphome.__main__.read_secret_line", return_value=line),
+        pytest.raises(EsphomeError, match="Invalid OTA encryption key"),
+    ):
+        _read_ota_key(args, config)
+    assert args.ota_key is None
+
+
+@pytest.mark.parametrize(
+    ("device", "port_type", "ota_platform"),
+    [
+        ("/dev/ttyUSB0", PortType.SERIAL, None),
+        ("/dev/ttyUSB0", PortType.SERIAL, CONF_WEB_SERVER),
+        ("SERIAL", PortType.NETWORK, None),
+    ],
+    ids=["port", "port_with_ota_platform", "selector"],
+)
+def test_read_ota_key_lets_a_serial_target_through(
+    mock_get_port_type: Mock,
+    caplog: pytest.LogCaptureFixture,
+    device: str,
+    port_type: PortType,
+    ota_platform: str | None,
+) -> None:
+    """A serial flash of a web_server only config is not refused, whatever
+    --ota-platform says; the key is not asked for and the user is told it is
+    ignored. The SERIAL selector only resolves to a port later, so it counts
+    by name."""
+    mock_get_port_type.return_value = port_type
+    args = MockArgs(prompt_ota_key=True, ota_platform=ota_platform)
+    args.device = [device]
+    with (
+        patch("esphome.__main__.read_secret_line") as read,
+        caplog.at_level(logging.WARNING),
+    ):
+        _read_ota_key(args, {CONF_OTA: [{CONF_PLATFORM: CONF_WEB_SERVER}]})
+    read.assert_not_called()
+    assert args.ota_key is None
+    assert any(f"ignored for {device}" in r.message for r in caplog.records)
+
+
+def test_read_ota_key_serial_then_network_still_asks(
+    mock_get_port_type: Mock,
+) -> None:
+    """`--device SERIAL --device OTA` with no port attached ends up on the
+    network, so the key is read and a web_server only config is refused."""
+    mock_get_port_type.side_effect = lambda d: (
+        PortType.SERIAL if d.startswith("/dev") else PortType.NETWORK
+    )
+    key = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+    args = MockArgs(prompt_ota_key=True)
+    args.device = ["/dev/ttyUSB0", "OTA"]
+    with patch("esphome.__main__.read_secret_line", return_value=key) as read:
+        _read_ota_key(args, {CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME}]})
+    read.assert_called_once()
+    assert args.ota_key == key
+    args = MockArgs(prompt_ota_key=True)
+    args.device = ["SERIAL", "OTA"]
+    with (
+        patch("esphome.__main__.read_secret_line") as read,
+        pytest.raises(EsphomeError, match="only apply to the esphome OTA platform"),
+    ):
+        _read_ota_key(args, {CONF_OTA: [{CONF_PLATFORM: CONF_WEB_SERVER}]})
+    read.assert_not_called()
+
+
+def test_read_ota_key_ignored_for_a_host_build(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A host run never uploads, so nothing is asked for."""
+    setup_core(platform=PLATFORM_HOST)
+    args = MockArgs(prompt_ota_key=True)
+    with (
+        patch("esphome.__main__.read_secret_line") as read,
+        caplog.at_level(logging.WARNING),
+    ):
+        _read_ota_key(args, {CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME}]})
+    read.assert_not_called()
+    assert args.ota_key is None
+    assert any("host program" in r.message for r in caplog.records)
+
+
+def test_read_ota_key_reports_a_bad_platform_before_prompting() -> None:
+    args = MockArgs(prompt_ota_key=True, ota_platform=CONF_WEB_SERVER)
+    with (
+        patch("esphome.__main__.read_secret_line") as read,
+        pytest.raises(EsphomeError, match="only provides"),
+    ):
+        _read_ota_key(args, {CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME}]})
+    read.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    ("ota_platform", "config"),
+    [
+        (
+            CONF_WEB_SERVER,
+            {
+                CONF_OTA: [
+                    {CONF_PLATFORM: CONF_ESPHOME},
+                    {CONF_PLATFORM: CONF_WEB_SERVER},
+                ]
+            },
+        ),
+        (None, {CONF_OTA: [{CONF_PLATFORM: CONF_WEB_SERVER}]}),
+    ],
+    ids=["requested", "only_platform"],
+)
+@pytest.mark.parametrize("devices", [["192.168.1.100"], None], ids=["host", "chooser"])
+def test_read_ota_key_refuses_web_server_platform(
+    mock_get_port_type: Mock,
+    ota_platform: str | None,
+    config: dict[str, Any],
+    devices: list[str] | None,
+) -> None:
+    """The HTTP path has no key handshake, so the flag is refused before the
+    prompt and before any compile, whether web_server was asked for or is the
+    only platform in the config, for a network target or none named yet."""
+    mock_get_port_type.return_value = PortType.NETWORK
+    args = MockArgs(prompt_ota_key=True, ota_platform=ota_platform)
+    args.device = devices
+    with (
+        patch("esphome.__main__.read_secret_line") as read,
+        pytest.raises(EsphomeError, match="only apply to the esphome OTA"),
+    ):
+        _read_ota_key(args, config)
+    read.assert_not_called()
+
+
 def test_upload_program_ota_api_key_opportunistic(
     mock_run_ota: Mock,
     mock_get_port_type: Mock,
@@ -2889,6 +3273,22 @@ def test_upload_program_web_server_only_auto_dispatches(
     mock_run_ota.assert_not_called()
 
 
+def test_upload_program_web_server_refuses_a_presented_key(
+    mock_run_web_server_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """A network target picked in the chooser reaches the HTTP path with the
+    key; it is refused rather than silently dropped."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+    config = {CONF_OTA: [{CONF_PLATFORM: CONF_WEB_SERVER}], CONF_WEB_SERVER: {}}
+    args = MockArgs(ota_key="AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=")
+    with pytest.raises(EsphomeError, match="only apply to the esphome OTA"):
+        upload_program(config, args, ["192.168.1.100"])
+    mock_run_web_server_ota.assert_not_called()
+
+
 def test_upload_program_web_server_no_auth(
     mock_run_web_server_ota: Mock,
     mock_get_port_type: Mock,
@@ -6445,6 +6845,17 @@ def test_parse_args_run_states() -> None:
     assert args.states is True
 
 
+@pytest.mark.parametrize("command", ["upload", "run"])
+def test_parse_args_prompt_ota_key(command: str) -> None:
+    """The flag takes no value: the key is never accepted on the command line."""
+    args = parse_args(["esphome", command, "--prompt-ota-key", "device.yaml"])
+    assert args.prompt_ota_key is True
+    args = parse_args(["esphome", command, "device.yaml"])
+    assert args.prompt_ota_key is False
+    with pytest.raises(SystemExit):
+        parse_args(["esphome", command, "--prompt-ota-key=abc", "device.yaml"])
+
+
 def test_parse_args_run_states_default() -> None:
     """Test that states defaults to None (unset) for the run command."""
     args = parse_args(["esphome", "run", "device.yaml"])

tests/unit_tests/test_util.py

@@ -1177,3 +1177,26 @@ class TestSafePrint:
         assert (
             capsys.readouterr().out == "Cannot print line because of invalid locale!\n"
         )
+
+
+@pytest.mark.parametrize("is_tty", [True, False], ids=["tty", "pipe"])
+def test_read_secret_line(is_tty: bool) -> None:
+    """A terminal reads without echo, a pipe reads one line; both strip."""
+    with (
+        patch("esphome.util.sys.stdin") as stdin,
+        patch("getpass.getpass", return_value="  secret \n") as getpass,
+    ):
+        stdin.isatty.return_value = is_tty
+        stdin.readline.return_value = "  secret \r\n"
+        assert util.read_secret_line("Key: ") == "secret"
+    assert getpass.called is is_tty
+    assert stdin.readline.called is not is_tty
+
+
+def test_read_secret_line_empty_pipe(capsys: pytest.CaptureFixture[str]) -> None:
+    """The prompt still goes to stderr on a pipe so a blocked read is diagnosable."""
+    with patch("esphome.util.sys.stdin") as stdin:
+        stdin.isatty.return_value = False
+        stdin.readline.return_value = ""
+        assert util.read_secret_line("Key: ") == ""
+    assert capsys.readouterr().err == "Key: "