diff --git a/tests/component_tests/ota/test_esphome_ota.py b/tests/component_tests/ota/test_esphome_ota.py
index cdac430ff7..730a67ad35 100644
--- a/tests/component_tests/ota/test_esphome_ota.py
+++ b/tests/component_tests/ota/test_esphome_ota.py
@@ -8,17 +8,25 @@ from typing import Any
 import pytest
 
 from esphome import config_validation as cv
-from esphome.components.esphome.ota import ota_esphome_final_validate
+from esphome.components.esphome.ota import (
+    AUTO_LOAD,
+    FILTER_SOURCE_FILES,
+    _validate_no_password_with_encryption,
+    ota_esphome_final_validate,
+)
 from esphome.const import (
+    CONF_API,
+    CONF_ENCRYPTION,
     CONF_ESPHOME,
     CONF_ID,
+    CONF_KEY,
     CONF_OTA,
     CONF_PASSWORD,
     CONF_PLATFORM,
     CONF_PORT,
     CONF_VERSION,
 )
-from esphome.core import ID
+from esphome.core import CORE, ID
 import esphome.final_validate as fv
 
 
@@ -103,3 +111,299 @@ def test_non_esphome_ota_unaffected() -> None:
         assert len(updated[CONF_OTA]) == 3
     finally:
         fv.full_config.reset(token)
+
+
+API_KEY = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+OTHER_KEY = "AQIDBAUGBwgJCgsMDQ4PEBESExQVFhcYGRobHB0eHyA="
+ZEROS_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
+
+
+def test_encryption_key_inherited_from_api() -> None:
+    """A bare encryption block resolves to the api encryption key."""
+    full_conf = {
+        CONF_API: {CONF_ENCRYPTION: {CONF_KEY: API_KEY}},
+        CONF_OTA: [_make_ota_config(port=3232, **{CONF_ENCRYPTION: {}})],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        ota_esphome_final_validate({})
+        updated = fv.full_config.get()
+        assert updated[CONF_OTA][0][CONF_ENCRYPTION][CONF_KEY] == API_KEY
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_explicit_key_matching_api_accepted() -> None:
+    """An explicit ota key equal to the api key validates."""
+    full_conf = {
+        CONF_API: {CONF_ENCRYPTION: {CONF_KEY: API_KEY}},
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: API_KEY}})
+        ],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        ota_esphome_final_validate({})
+        updated = fv.full_config.get()
+        assert updated[CONF_OTA][0][CONF_ENCRYPTION][CONF_KEY] == API_KEY
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_key_differing_from_api_rejected() -> None:
+    """There is one key per device; an ota key differing from the api key raises."""
+    full_conf = {
+        CONF_API: {CONF_ENCRYPTION: {CONF_KEY: API_KEY}},
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}})
+        ],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(cv.Invalid, match="must match the 'api' encryption key"):
+            ota_esphome_final_validate({})
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_explicit_key_without_api_encryption_accepted() -> None:
+    """An explicit ota key with a plaintext api has nothing to match; it stands."""
+    full_conf = {
+        CONF_API: {},
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}})
+        ],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        ota_esphome_final_validate({})
+        updated = fv.full_config.get()
+        assert updated[CONF_OTA][0][CONF_ENCRYPTION][CONF_KEY] == OTHER_KEY
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_without_any_key_rejected() -> None:
+    """A bare encryption block with no api key to inherit raises."""
+    full_conf = {
+        CONF_API: {},
+        CONF_OTA: [_make_ota_config(port=3232, **{CONF_ENCRYPTION: {}})],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(cv.Invalid, match="no 'api' encryption key to inherit"):
+            ota_esphome_final_validate({})
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_explicit_all_zeros_key_rejected() -> None:
+    """The all-zeros key is the provisioning sentinel; the device would treat
+    it as no PSK and accept plaintext, so it must fail validation."""
+    full_conf = {
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: ZEROS_KEY}})
+        ],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(cv.Invalid, match="all-zeros key is reserved"):
+            ota_esphome_final_validate({})
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_inherited_all_zeros_key_rejected() -> None:
+    """An all-zeros api key must not silently disable ota encryption either."""
+    full_conf = {
+        CONF_API: {CONF_ENCRYPTION: {CONF_KEY: ZEROS_KEY}},
+        CONF_OTA: [_make_ota_config(port=3232, **{CONF_ENCRYPTION: {}})],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(cv.Invalid, match="all-zeros key is reserved"):
+            ota_esphome_final_validate({})
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_key_mismatch_between_merged_configs_rejected() -> None:
+    """Same-port configs with different encryption keys raise."""
+    full_conf = {
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: API_KEY}}),
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}}),
+        ]
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(cv.Invalid, match="encryption is inconsistent"):
+            ota_esphome_final_validate({})
+    finally:
+        fv.full_config.reset(token)
+
+
+@pytest.mark.parametrize("keyed_first", [True, False])
+def test_encryption_bare_and_keyed_blocks_merge(keyed_first: bool) -> None:
+    """A bare encryption block (package/device split) is compatible with a
+    keyed one on the same port; the merge resolves to the keyed result."""
+    keyed = _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}})
+    bare = _make_ota_config(port=3232, **{CONF_ENCRYPTION: {}})
+    full_conf = {
+        CONF_OTA: [keyed, bare] if keyed_first else [bare, keyed],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        ota_esphome_final_validate({})
+        updated = fv.full_config.get()
+        assert len(updated[CONF_OTA]) == 1
+        assert updated[CONF_OTA][0][CONF_ENCRYPTION][CONF_KEY] == OTHER_KEY
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_runtime_provisioned_api_key_not_inheritable() -> None:
+    """A keyless api encryption block provisions its key at runtime; a bare
+    ota encryption block cannot inherit it and the message says so."""
+    full_conf = {
+        CONF_API: {CONF_ENCRYPTION: {}},
+        CONF_OTA: [_make_ota_config(port=3232, **{CONF_ENCRYPTION: {}})],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(cv.Invalid, match="provisioned at runtime"):
+            ota_esphome_final_validate({})
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_explicit_key_with_runtime_provisioned_api_accepted() -> None:
+    """The documented remedy for a runtime-provisioned api key: set an
+    explicit ota key."""
+    full_conf = {
+        CONF_API: {CONF_ENCRYPTION: {}},
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: OTHER_KEY}})
+        ],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        ota_esphome_final_validate({})
+        updated = fv.full_config.get()
+        assert updated[CONF_OTA][0][CONF_ENCRYPTION][CONF_KEY] == OTHER_KEY
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_with_web_server_ota_rejected() -> None:
+    """With the web_server component the plaintext /update endpoint is always
+    on, a full bypass of the encryption; the combination fails closed."""
+    full_conf = {
+        "web_server": {},
+        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 pytest.raises(cv.Invalid, match="plaintext HTTP"):
+            ota_esphome_final_validate({})
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_encryption_with_captive_portal_web_server_ota_warns(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """captive_portal auto-loads the web_server ota platform without the
+    web_server component; encryption stays usable and only warns, so the
+    fallback AP recovery path is not lost."""
+    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("captive_portal" in record.message for record in caplog.records)
+        esphome_conf = next(
+            conf
+            for conf in fv.full_config.get()[CONF_OTA]
+            if conf.get(CONF_PLATFORM) == CONF_ESPHOME
+        )
+        assert esphome_conf[CONF_ENCRYPTION][CONF_KEY] == OTHER_KEY
+    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 = {
+        CONF_OTA: [
+            _make_ota_config(port=3232),
+            {CONF_PLATFORM: "web_server", CONF_ID: ID("ota_ws", is_manual=False)},
+        ],
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        ota_esphome_final_validate({})
+        assert len(fv.full_config.get()[CONF_OTA]) == 2
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_auto_load_pulls_noise_only_for_encryption() -> None:
+    """A plain ota entry must never pull noise-c into the build."""
+    assert AUTO_LOAD({CONF_PORT: 3232}) == ["sha256", "socket"]
+    assert "noise" in AUTO_LOAD({CONF_ENCRYPTION: {}})
+    # The dependency-resolution tooling calls with no config
+    assert "noise" in AUTO_LOAD(None)
+
+
+def test_filter_source_files_excludes_noise_without_encryption() -> None:
+    """The noise transport source compiles only for encrypted builds."""
+    old_config = CORE.config
+    try:
+        CORE.config = {CONF_OTA: [_make_ota_config(port=3232)]}
+        assert FILTER_SOURCE_FILES() == ["ota_esphome_noise.cpp"]
+        CORE.config = {
+            CONF_OTA: [
+                _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: API_KEY}})
+            ]
+        }
+        assert FILTER_SOURCE_FILES() == []
+    finally:
+        CORE.config = old_config
+
+
+def test_password_with_encryption_rejected() -> None:
+    """The password and encryption options are mutually exclusive."""
+    config = {CONF_PASSWORD: "pw", CONF_ENCRYPTION: {CONF_KEY: API_KEY}}
+    with pytest.raises(cv.Invalid, match="cannot be combined"):
+        _validate_no_password_with_encryption(config)
+
+
+def test_password_alone_accepted() -> None:
+    """A password without encryption still validates."""
+    config = {CONF_PASSWORD: "pw"}
+    assert _validate_no_password_with_encryption(config) is config
+
+
+def test_merged_password_and_encryption_rejected() -> None:
+    """A password block and an encryption block merged on one port raise."""
+    full_conf = {
+        CONF_OTA: [
+            _make_ota_config(port=3232, **{CONF_PASSWORD: "pw"}),
+            _make_ota_config(port=3232, **{CONF_ENCRYPTION: {CONF_KEY: API_KEY}}),
+        ]
+    }
+    token = fv.full_config.set(full_conf)
+    try:
+        with pytest.raises(cv.Invalid, match="cannot be combined"):
+            ota_esphome_final_validate({})
+    finally:
+        fv.full_config.reset(token)
diff --git a/tests/integration/fixtures/host_ota_encrypted.yaml b/tests/integration/fixtures/host_ota_encrypted.yaml
new file mode 100644
index 0000000000..0d11c99d3d
--- /dev/null
+++ b/tests/integration/fixtures/host_ota_encrypted.yaml
@@ -0,0 +1,11 @@
+esphome:
+  name: host-ota-test
+host:
+api:
+ota:
+  - platform: esphome
+    port: __OTA_PORT__
+    encryption:
+      key: "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+logger:
+  level: DEBUG
diff --git a/tests/integration/test_host_ota.py b/tests/integration/test_host_ota.py
index e1036fdf1c..4e74814534 100644
--- a/tests/integration/test_host_ota.py
+++ b/tests/integration/test_host_ota.py
@@ -10,6 +10,7 @@ from __future__ import annotations
 import asyncio
 from collections.abc import Generator
 from contextlib import contextmanager
+import functools
 import socket
 
 import pytest
@@ -111,6 +112,62 @@ async def test_host_ota_self_update(
             assert proc.pid == pid_before
 
 
+@pytest.mark.asyncio
+async def test_host_ota_encrypted(
+    yaml_config: str,
+    write_yaml_config: ConfigWriter,
+    compile_esphome: CompileFunction,
+    reserved_tcp_port: tuple[int, socket.socket],
+) -> None:
+    """Encrypted self-OTA succeeds; a plaintext upload to the same device fails."""
+    pytest.importorskip("aioesphomeapi.noise")
+    noise_psk = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+    api_port, api_socket = reserved_tcp_port
+    with _reserve_port() as (ota_port, ota_socket):
+        yaml_config = yaml_config.replace("__OTA_PORT__", str(ota_port))
+        config_path = await write_yaml_config(yaml_config)
+        binary_path = await compile_esphome(config_path)
+        api_socket.close()
+        ota_socket.close()
+
+        loop = asyncio.get_running_loop()
+        rebooted = loop.create_future()
+
+        def on_log(line: str) -> None:
+            if not rebooted.done() and "Rebooting safely" in line:
+                rebooted.set_result(True)
+
+        async with run_binary(binary_path, line_callback=on_log) as (proc, _lines):
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            pid_before = proc.pid
+
+            # A plaintext upload must be refused with the device unharmed
+            rc, _ = await loop.run_in_executor(
+                None, espota2.run_ota, LOCALHOST, ota_port, None, binary_path
+            )
+            assert rc == 1, "plaintext upload to an encrypted device must fail"
+            await asyncio.sleep(0.5)
+            assert proc.returncode is None, "process died on rejected plaintext OTA"
+
+            # The encrypted upload goes through and the device re-execs
+            rc, _ = await loop.run_in_executor(
+                None,
+                functools.partial(
+                    espota2.run_ota,
+                    LOCALHOST,
+                    ota_port,
+                    None,
+                    binary_path,
+                    noise_psk=noise_psk,
+                ),
+            )
+            assert rc == 0, "encrypted OTA reported failure"
+            await asyncio.wait_for(rebooted, timeout=10.0)
+            await _wait_for_port(LOCALHOST, api_port, PORT_WAIT_TIMEOUT)
+            assert proc.returncode is None, "process exited instead of execing"
+            assert proc.pid == pid_before
+
+
 @pytest.mark.asyncio
 async def test_host_ota_rejects_garbage(
     yaml_config: str,
diff --git a/tests/unit_tests/test_espota2_noise.py b/tests/unit_tests/test_espota2_noise.py
new file mode 100644
index 0000000000..3095ced1cd
--- /dev/null
+++ b/tests/unit_tests/test_espota2_noise.py
@@ -0,0 +1,407 @@
+"""Unit tests for encrypted OTA uploads in esphome.espota2.
+
+A fake device implementing the responder side of the wire protocol (via
+noiseprotocol, which esphome already has through aioesphomeapi) serves a real
+TCP loopback connection, so these exercise the actual handshake, framing, and
+cipher interop of the client code. Tests that need the client-side crypto skip
+when the installed aioesphomeapi predates the noise module.
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import io
+from pathlib import Path
+import socket
+import sys
+import threading
+from unittest.mock import Mock, patch
+
+import pytest
+
+from esphome import espota2
+
+PSK = base64.b64encode(bytes(range(32))).decode()
+OTHER_PSK = base64.b64encode(bytes(range(1, 33))).decode()
+
+MAGIC = bytes(espota2.MAGIC_BYTES)
+
+
+def _recv_exact(sock: socket.socket, amount: int) -> bytes:
+    data = b""
+    while len(data) < amount:
+        chunk = sock.recv(amount - len(data))
+        if not chunk:
+            raise ConnectionError("client closed")
+        data += chunk
+    return data
+
+
+def _frame(payload: bytes) -> bytes:
+    return (
+        bytes([espota2.NOISE_FRAME_INDICATOR, len(payload) >> 8, len(payload) & 0xFF])
+        + payload
+    )
+
+
+def _send_frame(sock: socket.socket, payload: bytes) -> None:
+    sock.sendall(_frame(payload))
+
+
+def _recv_frame(sock: socket.socket) -> bytes:
+    header = _recv_exact(sock, 3)
+    assert header[0] == 0x01
+    return _recv_exact(sock, (header[1] << 8) | header[2])
+
+
+class FakeEncryptedDevice(threading.Thread):
+    """Responder side of the encrypted OTA wire protocol."""
+
+    def __init__(
+        self,
+        psk: str = PSK,
+        version: int = 2,
+        offer_noise: bool = True,
+        require_noise: bool = True,
+        prologue_features_override: int | None = None,
+    ) -> None:
+        super().__init__(daemon=True)
+        self.psk = psk
+        self.version = version
+        self.offer_noise = offer_noise
+        self.require_noise = require_noise
+        self.prologue_features_override = prologue_features_override
+        self.received: bytes | None = None
+        self.error: Exception | None = None
+        self.listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.listener.bind(("127.0.0.1", 0))
+        self.listener.listen(1)
+        self.port = self.listener.getsockname()[1]
+
+    def run(self) -> None:
+        try:
+            sock, _ = self.listener.accept()
+            sock.settimeout(10)
+            with sock:
+                self._serve(sock)
+        except Exception as err:  # noqa: BLE001 - surfaced via join_and_check
+            self.error = err
+        finally:
+            self.listener.close()
+
+    def join_and_check(self) -> None:
+        self.join(timeout=10)
+        assert not self.is_alive(), "fake device did not finish"
+        if self.error is not None:
+            raise self.error
+
+    def _serve(self, sock: socket.socket) -> None:
+        assert _recv_exact(sock, 5) == MAGIC
+        sock.sendall(bytes([espota2.RESPONSE_OK, self.version]))
+        features = _recv_exact(sock, 1)[0]
+        noise_negotiated = bool(
+            features & espota2.CLIENT_FEATURE_SUPPORTS_NOISE
+            and features & espota2.CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL
+        )
+        if self.require_noise and not noise_negotiated:
+            sock.sendall(bytes([espota2.RESPONSE_ERROR_ENCRYPTION_REQUIRED]))
+            return
+        server_flags = espota2.SERVER_FEATURE_SUPPORTS_NOISE if self.offer_noise else 0
+        sock.sendall(bytes([espota2.RESPONSE_FEATURE_FLAGS, server_flags]))
+        if not (self.offer_noise and noise_negotiated):
+            return  # the client fails closed; nothing further arrives
+
+        from cryptography.exceptions import InvalidTag
+        from noise.connection import NoiseConnection
+
+        prologue_features = (
+            features
+            if self.prologue_features_override is None
+            else self.prologue_features_override
+        )
+        prologue = (
+            espota2.NOISE_PROLOGUE_INIT
+            + MAGIC
+            + bytes([espota2.RESPONSE_OK, self.version, prologue_features])
+            + bytes([espota2.RESPONSE_FEATURE_FLAGS, server_flags])
+        )
+        proto = NoiseConnection.from_name(b"Noise_NNpsk0_25519_ChaChaPoly_SHA256")
+        proto.set_as_responder()
+        proto.set_psks(base64.b64decode(self.psk))
+        proto.set_prologue(prologue)
+        proto.start_handshake()
+
+        msg1 = _recv_frame(sock)
+        assert msg1[0] == 0x00
+        try:
+            proto.read_message(msg1[1:])
+        except InvalidTag:
+            _send_frame(sock, b"\x01Handshake MAC failure")
+            return
+        _send_frame(sock, b"\x00" + bytes(proto.write_message()))
+
+        def send_byte(byte: int) -> None:
+            _send_frame(sock, proto.encrypt(bytes([byte])))
+
+        def recv_unit(length: int) -> bytes:
+            plaintext = proto.decrypt(_recv_frame(sock))
+            assert len(plaintext) == length, "control units must be one per frame"
+            return plaintext
+
+        send_byte(espota2.RESPONSE_AUTH_OK)
+        recv_unit(1)  # ota type
+        size = int.from_bytes(recv_unit(4), "big")
+        send_byte(espota2.RESPONSE_UPDATE_PREPARE_OK)
+        md5_hex = recv_unit(32)
+        send_byte(espota2.RESPONSE_BIN_MD5_OK)
+
+        received = b""
+        acked = 0
+        while len(received) < size:
+            plaintext = proto.decrypt(_recv_frame(sock))
+            assert 0 < len(plaintext) <= espota2.NOISE_MAX_PLAINTEXT
+            received += plaintext
+            if self.version >= espota2.OTA_VERSION_2_0:
+                while acked + espota2.UPLOAD_BLOCK_SIZE <= len(received) or (
+                    len(received) == size and acked < size
+                ):
+                    send_byte(espota2.RESPONSE_CHUNK_OK)
+                    acked += espota2.UPLOAD_BLOCK_SIZE
+        assert hashlib.md5(received).hexdigest().encode() == md5_hex
+        send_byte(espota2.RESPONSE_RECEIVE_OK)
+        send_byte(espota2.RESPONSE_UPDATE_END_OK)
+        assert recv_unit(1) == bytes([espota2.RESPONSE_OK])
+        self.received = received
+
+
+def _upload(
+    device: FakeEncryptedDevice, firmware: bytes, noise_psk: str | None
+) -> None:
+    device.start()
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock.settimeout(10)
+    sock.connect(("127.0.0.1", device.port))
+    try:
+        espota2.perform_ota(
+            sock, None, io.BytesIO(firmware), Path("firmware.bin"), noise_psk=noise_psk
+        )
+    finally:
+        sock.close()
+
+
+def test_encrypted_upload_success() -> None:
+    """A full encrypted v2 upload spanning several 8192-byte blocks."""
+    pytest.importorskip("aioesphomeapi.noise")
+    firmware = bytes(range(256)) * 80  # 20480 bytes, crosses chunk-ack boundaries
+    device = FakeEncryptedDevice()
+    with patch("time.sleep"):
+        _upload(device, firmware, PSK)
+    device.join_and_check()
+    assert device.received == firmware
+
+
+def test_encrypted_upload_version_1() -> None:
+    """Version 1 protocol (no chunk acks) works through the noise transport."""
+    pytest.importorskip("aioesphomeapi.noise")
+    firmware = b"v1 firmware image" * 100
+    device = FakeEncryptedDevice(version=1)
+    with patch("time.sleep"):
+        _upload(device, firmware, PSK)
+    device.join_and_check()
+    assert device.received == firmware
+
+
+def test_wrong_key_fails_with_clear_error() -> None:
+    """A key mismatch surfaces the device's handshake reject readably."""
+    pytest.importorskip("aioesphomeapi.noise")
+    device = FakeEncryptedDevice(psk=OTHER_PSK)
+    with pytest.raises(espota2.OTAError, match="encryption key correct"):
+        _upload(device, b"firmware", PSK)
+    device.join_and_check()
+
+
+def test_tampered_negotiation_breaks_handshake() -> None:
+    """A negotiation byte differing between the sides breaks the prologue MAC."""
+    pytest.importorskip("aioesphomeapi.noise")
+    device = FakeEncryptedDevice(
+        prologue_features_override=espota2.CLIENT_FEATURE_SUPPORTS_EXTENDED_PROTOCOL
+    )
+    with pytest.raises(espota2.OTAError, match="encryption key correct"):
+        _upload(device, b"firmware", PSK)
+    device.join_and_check()
+
+
+def test_client_fails_closed_when_device_lacks_encryption() -> None:
+    """With a key configured, a device not offering noise aborts the upload."""
+    device = FakeEncryptedDevice(offer_noise=False, require_noise=False)
+    with pytest.raises(espota2.OTAError, match="refusing to send the image"):
+        _upload(device, b"firmware", PSK)
+    device.join_and_check()
+
+
+def test_plaintext_client_gets_encryption_required_error() -> None:
+    """A client without a key gets the device's 0x94 error message."""
+    device = FakeEncryptedDevice()
+    with pytest.raises(espota2.OTAError, match="requires an encrypted OTA"):
+        _upload(device, b"firmware", None)
+    device.join_and_check()
+
+
+def test_missing_aioesphomeapi_noise_module_message() -> None:
+    """An aioesphomeapi without the noise module produces a clear error."""
+    with (
+        patch.dict(sys.modules, {"aioesphomeapi.noise": None}),
+        pytest.raises(espota2.OTAError, match="requires a newer aioesphomeapi"),
+    ):
+        espota2.NoiseSocketWrapper(Mock(), PSK, b"prologue")
+
+
+class ScriptedSocket:
+    """Serves scripted recv chunks; b"" means the peer closed."""
+
+    def __init__(self, *chunks: bytes | Exception) -> None:
+        self.chunks = list(chunks)
+        self.sent: list[bytes] = []
+
+    def sendall(self, data: bytes) -> None:
+        self.sent.append(data)
+
+    def settimeout(self, timeout: float) -> None:
+        pass
+
+    def recv(self, amount: int) -> bytes:
+        if not self.chunks:
+            return b""
+        chunk = self.chunks[0]
+        if isinstance(chunk, Exception):
+            self.chunks.pop(0)
+            raise chunk
+        take, rest = chunk[:amount], chunk[amount:]
+        if rest:
+            self.chunks[0] = rest
+        else:
+            self.chunks.pop(0)
+        return take
+
+
+def _wrapper(*chunks: bytes | Exception) -> espota2.NoiseSocketWrapper:
+    pytest.importorskip("aioesphomeapi.noise")
+    return espota2.NoiseSocketWrapper(ScriptedSocket(*chunks), PSK, b"prologue")
+
+
+def test_wrapper_rejects_malformed_psk() -> None:
+    pytest.importorskip("aioesphomeapi.noise")
+    with pytest.raises(espota2.OTAError, match="Invalid OTA encryption key"):
+        espota2.NoiseSocketWrapper(ScriptedSocket(), "not-base64!!!", b"prologue")
+
+
+def test_handshake_socket_error_is_network_error() -> None:
+    wrapper = _wrapper(OSError("boom"))
+    with pytest.raises(espota2.OTANetworkError, match="noise handshake"):
+        wrapper.do_handshake()
+
+
+def test_handshake_closed_at_frame_boundary() -> None:
+    wrapper = _wrapper()
+    with pytest.raises(espota2.OTANetworkError, match="closed connection during"):
+        wrapper.do_handshake()
+
+
+def test_handshake_reject_with_other_reason() -> None:
+    wrapper = _wrapper(_frame(b"\x01Handshake error"))
+    with pytest.raises(
+        espota2.OTAError, match="rejected the noise handshake: Handshake error"
+    ):
+        wrapper.do_handshake()
+
+
+def test_handshake_garbage_second_message() -> None:
+    """A valid-looking point with a garbage MAC fails cleanly."""
+    wrapper = _wrapper(_frame(b"\x00" + bytes(range(48))))
+    with pytest.raises(
+        espota2.OTAError, match="handshake failed; is the OTA encryption key"
+    ):
+        wrapper.do_handshake()
+
+
+def test_handshake_invalid_curve_point() -> None:
+    """An all-zero x25519 point is rejected as a clean error, not a crash."""
+    wrapper = _wrapper(_frame(b"\x00" + bytes(48)))
+    with pytest.raises(
+        espota2.OTAError, match="handshake failed; is the OTA encryption key"
+    ):
+        wrapper.do_handshake()
+
+
+def test_recv_closed_at_frame_boundary_returns_empty() -> None:
+    wrapper = _wrapper()
+    assert wrapper.recv(1) == b""
+
+
+def test_recv_corrupt_frame_is_retryable_network_error() -> None:
+    from cryptography.exceptions import InvalidTag
+
+    wrapper = _wrapper(_frame(b"ciphertext"))
+    wrapper._decrypt = Mock(decrypt=Mock(side_effect=InvalidTag()))
+    with pytest.raises(espota2.OTANetworkError, match="decryption failed"):
+        wrapper.recv(1)
+
+
+def test_wrapper_blocks_unencrypted_socket_methods() -> None:
+    """Byte-moving socket methods must not bypass the encrypted transport."""
+    wrapper = _wrapper()
+    # The harmless socket controls pass through to the wrapped socket
+    wrapper._sock = Mock()
+    wrapper.settimeout(1)
+    wrapper._sock.settimeout.assert_called_once_with(1)
+    wrapper.setsockopt(6, 1, 1)
+    wrapper._sock.setsockopt.assert_called_once_with(6, 1, 1)
+    wrapper.close()
+    wrapper._sock.close.assert_called_once_with()
+    with pytest.raises(AttributeError):
+        _ = wrapper.send
+    with pytest.raises(AttributeError):
+        _ = wrapper.recv_into
+
+
+def test_recv_empty_plaintext_frame_is_protocol_error() -> None:
+    """A MAC-only frame decrypts to nothing; b'' from recv must mean close."""
+    wrapper = _wrapper(_frame(bytes(16)))
+    wrapper._decrypt = Mock(decrypt=Mock(return_value=b""))
+    with pytest.raises(espota2.OTANetworkError, match="empty noise frame"):
+        wrapper.recv(1)
+
+
+def test_recv_frame_bad_indicator_is_retryable() -> None:
+    wrapper = _wrapper(b"\x02\x00\x01x")
+    with pytest.raises(espota2.OTANetworkError, match="Bad noise frame indicator"):
+        wrapper._recv_frame()
+
+
+def test_recv_frame_zero_length_is_retryable() -> None:
+    wrapper = _wrapper(bytes([espota2.NOISE_FRAME_INDICATOR, 0, 0]))
+    with pytest.raises(espota2.OTANetworkError, match="empty noise frame"):
+        wrapper._recv_frame()
+
+
+def test_perform_ota_blank_key_refuses_plaintext() -> None:
+    with pytest.raises(espota2.OTAError, match="empty OTA encryption key"):
+        espota2.perform_ota(
+            ScriptedSocket(), None, io.BytesIO(b"x"), Path("f.bin"), noise_psk=""
+        )
+
+
+def test_recv_exact_closed_mid_frame() -> None:
+    wrapper = _wrapper(_frame(b"partial")[:5])
+    with pytest.raises(OSError, match="closed inside a noise frame"):
+        wrapper._recv_frame()
+
+
+def test_recv_serves_buffered_plaintext_without_new_frame() -> None:
+    """A second recv drains the decrypted buffer without reading another frame."""
+    wrapper = _wrapper(_frame(b"ciphertext"))
+    wrapper._decrypt = Mock(decrypt=Mock(return_value=b"AB"))
+    assert wrapper.recv(1) == b"A"  # reads and decrypts one frame
+    assert wrapper.recv(1) == b"B"  # served from the buffer, no new frame
+    wrapper._decrypt.decrypt.assert_called_once()
diff --git a/tests/unit_tests/test_main.py b/tests/unit_tests/test_main.py
index a40341e194..38b94d2bb3 100644
--- a/tests/unit_tests/test_main.py
+++ b/tests/unit_tests/test_main.py
@@ -80,7 +80,9 @@ from esphome.const import (
     CONF_BAUD_RATE,
     CONF_BROKER,
     CONF_DISABLED,
+    CONF_ENCRYPTION,
     CONF_ESPHOME,
+    CONF_KEY,
     CONF_LEVEL,
     CONF_LOG,
     CONF_LOG_TOPIC,
@@ -2101,10 +2103,65 @@ def test_upload_program_ota_success(
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
     )
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.100"], 3232, "secret", expected_firmware, OTA_TYPE_UPDATE_APP
+        ["192.168.1.100"], 3232, "secret", expected_firmware, OTA_TYPE_UPDATE_APP, None
     )
 
 
+def test_upload_program_ota_encryption_key(
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """The resolved encryption key is passed through to run_ota."""
+    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")
+
+    key = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
+    config = {
+        CONF_OTA: [
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+                CONF_ENCRYPTION: {CONF_KEY: key},
+            }
+        ]
+    }
+    exit_code, host = upload_program(config, MockArgs(), ["192.168.1.100"])
+
+    assert exit_code == 0
+    assert host == "192.168.1.100"
+    expected_firmware = (
+        tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
+    )
+    mock_run_ota.assert_called_once_with(
+        ["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, key
+    )
+
+
+def test_upload_program_ota_encryption_without_key_fails_closed(
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """An encryption block with no resolved key must never upload plaintext."""
+    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,
+                CONF_ENCRYPTION: {},
+            }
+        ]
+    }
+    with pytest.raises(EsphomeError, match="no key was resolved"):
+        upload_program(config, MockArgs(), ["192.168.1.100"])
+    mock_run_ota.assert_not_called()
+
+
 def test_upload_program_ota_with_file_arg(
     mock_run_ota: Mock,
     mock_get_port_type: Mock,
@@ -2132,7 +2189,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, None, Path("custom.bin"), OTA_TYPE_UPDATE_APP
+        ["192.168.1.100"], 3232, None, Path("custom.bin"), OTA_TYPE_UPDATE_APP, None
     )
 
 
@@ -2187,6 +2244,7 @@ def test_upload_program_ota_partition_table_with_file_arg(
         None,
         partition_file,
         OTA_TYPE_UPDATE_PARTITION_TABLE,
+        None,
     )
 
 
@@ -2248,6 +2306,7 @@ def test_upload_program_ota_partition_table_mqttip(
         None,
         partition_file,
         OTA_TYPE_UPDATE_PARTITION_TABLE,
+        None,
     )
 
 
@@ -2435,6 +2494,7 @@ def test_upload_program_ota_bootloader_with_file_arg(
         None,
         bootloader_file,
         OTA_TYPE_UPDATE_BOOTLOADER,
+        None,
     )
 
 
@@ -2597,6 +2657,42 @@ def test_has_web_server_logging_respects_log_disabled() -> None:
     assert has_web_server_logging() is False
 
 
+def test_upload_program_web_server_warns_when_encryption_configured(
+    mock_run_web_server_ota: Mock,
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Explicitly picking web_server OTA on an encrypted config warns about
+    the plaintext upload path."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+    mock_run_web_server_ota.return_value = (0, "192.168.1.100")
+
+    config = {
+        CONF_OTA: [
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 3232,
+                CONF_ENCRYPTION: {CONF_KEY: "test_key"},
+            },
+            {CONF_PLATFORM: CONF_WEB_SERVER},
+        ],
+        CONF_WEB_SERVER: {
+            CONF_PORT: 80,
+            CONF_AUTH: {CONF_USERNAME: "admin", CONF_PASSWORD: "pw"},
+        },
+    }
+    args = MockArgs(ota_platform=CONF_WEB_SERVER)
+    with caplog.at_level(logging.WARNING):
+        exit_code, _ = upload_program(config, args, ["192.168.1.100"])
+
+    assert exit_code == 0
+    assert any("plaintext HTTP" in record.message for record in caplog.records)
+    mock_run_ota.assert_not_called()
+
+
 def test_upload_program_web_server_only_auto_dispatches(
     mock_run_web_server_ota: Mock,
     mock_run_ota: Mock,
@@ -2887,7 +2983,7 @@ def test_upload_program_ota_with_mqtt_resolution(
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
     )
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP
+        ["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, None
     )
 
 
@@ -2937,7 +3033,7 @@ def test_upload_program_ota_with_mqtt_empty_broker(
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
     )
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.50"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP
+        ["192.168.1.50"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, None
     )
     # Verify warning was logged
     assert "MQTT IP discovery failed" in caplog.text
@@ -5112,6 +5208,7 @@ def test_upload_program_ota_static_ip_with_mqttip(
         None,
         expected_firmware,
         OTA_TYPE_UPDATE_APP,
+        None,
     )
 
 
@@ -5161,6 +5258,7 @@ def test_upload_program_ota_multiple_mqttip_resolves_once(
         None,
         expected_firmware,
         OTA_TYPE_UPDATE_APP,
+        None,
     )
 
 
@@ -5338,7 +5436,7 @@ def test_upload_program_ota_mqtt_timeout_fallback(
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
     )
     mock_run_ota.assert_called_once_with(
-        ["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP
+        ["192.168.1.100"], 3232, None, expected_firmware, OTA_TYPE_UPDATE_APP, None
     )