mirror of
https://github.com/esphome/esphome.git
synced 2026-09-03 19:46:02 +00:00
Merge all hardening branches into integration
Resolve conflict in web_server_idf.cpp: keep constant-time compare logic adapted to stack-allocated digest buffer from integration.
This commit is contained in:
@@ -354,7 +354,26 @@ bool AsyncWebServerRequest::authenticate(const char *username, const char *passw
|
||||
esp_crypto_base64_encode(reinterpret_cast<uint8_t *>(digest), max_digest_len, &out,
|
||||
reinterpret_cast<const uint8_t *>(user_info), user_info_len);
|
||||
|
||||
return strcmp(digest, auth_str + auth_prefix_len) == 0;
|
||||
// Constant-time comparison to avoid timing side channels.
|
||||
// No early return on length mismatch — the length difference is folded
|
||||
// into the accumulator so any mismatch is rejected.
|
||||
const char *provided = auth_str + auth_prefix_len;
|
||||
size_t digest_len = out; // length from esp_crypto_base64_encode
|
||||
// Derive provided_len from the already-sized std::string rather than
|
||||
// rescanning with strlen (avoids attacker-controlled scan length).
|
||||
size_t provided_len = auth.value().size() - auth_prefix_len;
|
||||
// Use full-width XOR so any bit difference in the lengths is preserved
|
||||
// (uint8_t truncation would miss differences in higher bytes, e.g.
|
||||
// digest_len vs digest_len + 256).
|
||||
volatile size_t result = digest_len ^ provided_len;
|
||||
// Iterate over the expected digest length only — the full-width length
|
||||
// XOR above already rejects any length mismatch, and bounding the loop
|
||||
// prevents a long Authorization header from forcing extra work.
|
||||
for (size_t i = 0; i < digest_len; i++) {
|
||||
char provided_ch = (i < provided_len) ? provided[i] : 0;
|
||||
result |= static_cast<uint8_t>(digest[i] ^ provided_ch);
|
||||
}
|
||||
return result == 0;
|
||||
}
|
||||
|
||||
void AsyncWebServerRequest::requestAuthentication(const char *realm) const {
|
||||
|
||||
@@ -32,7 +32,7 @@ class DashboardSettings:
|
||||
def __init__(self) -> None:
|
||||
"""Initialize the dashboard settings."""
|
||||
self.config_dir: Path = None
|
||||
self.password_hash: str = ""
|
||||
self.password_hash: bytes = b""
|
||||
self.username: str = ""
|
||||
self.using_password: bool = False
|
||||
self.on_ha_addon: bool = False
|
||||
@@ -84,11 +84,14 @@ class DashboardSettings:
|
||||
def check_password(self, username: str, password: str) -> bool:
|
||||
if not self.using_auth:
|
||||
return True
|
||||
if username != self.username:
|
||||
return False
|
||||
|
||||
# Compare password in constant running time (to prevent timing attacks)
|
||||
return hmac.compare_digest(self.password_hash, password_hash(password))
|
||||
# Compare in constant running time (to prevent timing attacks)
|
||||
username_matches = hmac.compare_digest(
|
||||
username.encode("utf-8"), self.username.encode("utf-8")
|
||||
)
|
||||
password_matches = hmac.compare_digest(
|
||||
self.password_hash, password_hash(password)
|
||||
)
|
||||
return username_matches and password_matches
|
||||
|
||||
def rel_path(self, *args: Any) -> Path:
|
||||
"""Return a path relative to the ESPHome config folder."""
|
||||
|
||||
@@ -120,8 +120,11 @@ def is_authenticated(handler: BaseHandler) -> bool:
|
||||
if auth_header := handler.request.headers.get("Authorization"):
|
||||
assert isinstance(auth_header, str)
|
||||
if auth_header.startswith("Basic "):
|
||||
auth_decoded = base64.b64decode(auth_header[6:]).decode()
|
||||
username, password = auth_decoded.split(":", 1)
|
||||
try:
|
||||
auth_decoded = base64.b64decode(auth_header[6:]).decode()
|
||||
username, password = auth_decoded.split(":", 1)
|
||||
except (binascii.Error, ValueError, UnicodeDecodeError):
|
||||
return False
|
||||
return settings.check_password(username, password)
|
||||
return handler.get_secure_cookie(AUTH_COOKIE_NAME) == COOKIE_AUTHENTICATED_YES
|
||||
|
||||
@@ -1054,17 +1057,26 @@ class DownloadBinaryRequestHandler(BaseHandler):
|
||||
# fallback to type=, but prioritize file=
|
||||
file_name = self.get_argument("type", None)
|
||||
file_name = self.get_argument("file", file_name)
|
||||
if file_name is None:
|
||||
if file_name is None or not file_name.strip():
|
||||
self.send_error(400)
|
||||
return
|
||||
file_name = file_name.replace("..", "").lstrip("/")
|
||||
# get requested download name, or build it based on filename
|
||||
download_name = self.get_argument(
|
||||
"download",
|
||||
f"{storage_json.name}-{file_name}",
|
||||
)
|
||||
|
||||
path = storage_json.firmware_bin_path.parent.joinpath(file_name)
|
||||
if storage_json.firmware_bin_path is None:
|
||||
self.send_error(404)
|
||||
return
|
||||
|
||||
base_dir = storage_json.firmware_bin_path.parent.resolve()
|
||||
path = base_dir.joinpath(file_name).resolve()
|
||||
try:
|
||||
path.relative_to(base_dir)
|
||||
except ValueError:
|
||||
self.send_error(403)
|
||||
return
|
||||
|
||||
if not path.is_file():
|
||||
args = ["esphome", "idedata", settings.rel_path(configuration)]
|
||||
@@ -1078,7 +1090,7 @@ class DownloadBinaryRequestHandler(BaseHandler):
|
||||
|
||||
found = False
|
||||
for image in idedata.extra_flash_images:
|
||||
if image.path.endswith(file_name):
|
||||
if image.path.as_posix().endswith(file_name):
|
||||
path = image.path
|
||||
download_name = file_name
|
||||
found = True
|
||||
|
||||
+3
-3
@@ -6,7 +6,7 @@ import hashlib
|
||||
import io
|
||||
import logging
|
||||
from pathlib import Path
|
||||
import random
|
||||
import secrets
|
||||
import socket
|
||||
import sys
|
||||
import time
|
||||
@@ -300,8 +300,8 @@ def perform_ota(
|
||||
nonce = nonce_bytes.decode()
|
||||
_LOGGER.debug("Auth: %s Nonce is %s", hash_name, nonce)
|
||||
|
||||
# Generate cnonce
|
||||
cnonce = hash_func(str(random.random()).encode()).hexdigest()
|
||||
# Generate cnonce matching the hash algorithm's digest size
|
||||
cnonce = secrets.token_hex(nonce_size // 2)
|
||||
_LOGGER.debug("Auth: %s CNonce is %s", hash_name, cnonce)
|
||||
|
||||
send_check(sock, cnonce, "auth cnonce")
|
||||
|
||||
+1
-2
@@ -1,6 +1,5 @@
|
||||
import base64
|
||||
from pathlib import Path
|
||||
import random
|
||||
import secrets
|
||||
import string
|
||||
from typing import Literal, NotRequired, TypedDict, Unpack
|
||||
@@ -130,7 +129,7 @@ def wizard_file(**kwargs: Unpack[WizardFileKwargs]) -> str:
|
||||
if len(ap_name) > 32:
|
||||
ap_name = ap_name_base
|
||||
kwargs["fallback_name"] = ap_name
|
||||
kwargs["fallback_psk"] = "".join(random.choice(letters) for _ in range(12))
|
||||
kwargs["fallback_psk"] = "".join(secrets.choice(letters) for _ in range(12))
|
||||
|
||||
base = BASE_CONFIG_FRIENDLY if kwargs.get("friendly_name") else BASE_CONFIG
|
||||
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
"""Tests for dashboard settings Path-related functionality."""
|
||||
"""Tests for DashboardSettings (path resolution and authentication)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -10,6 +10,7 @@ import pytest
|
||||
|
||||
from esphome.core import CORE
|
||||
from esphome.dashboard.settings import DashboardSettings
|
||||
from esphome.dashboard.util.password import password_hash
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -221,3 +222,66 @@ def test_config_path_parent_resolves_to_config_dir(tmp_path: Path) -> None:
|
||||
# Verify that CORE.config_path itself uses the sentinel file
|
||||
assert CORE.config_path.name == "___DASHBOARD_SENTINEL___.yaml"
|
||||
assert not CORE.config_path.exists() # Sentinel file doesn't actually exist
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_settings(dashboard_settings: DashboardSettings) -> DashboardSettings:
|
||||
"""Create DashboardSettings with auth configured, based on dashboard_settings."""
|
||||
dashboard_settings.username = "admin"
|
||||
dashboard_settings.using_password = True
|
||||
dashboard_settings.password_hash = password_hash("correctpassword")
|
||||
return dashboard_settings
|
||||
|
||||
|
||||
def test_check_password_correct_credentials(auth_settings: DashboardSettings) -> None:
|
||||
"""Test check_password returns True for correct username and password."""
|
||||
assert auth_settings.check_password("admin", "correctpassword") is True
|
||||
|
||||
|
||||
def test_check_password_wrong_password(auth_settings: DashboardSettings) -> None:
|
||||
"""Test check_password returns False for wrong password."""
|
||||
assert auth_settings.check_password("admin", "wrongpassword") is False
|
||||
|
||||
|
||||
def test_check_password_wrong_username(auth_settings: DashboardSettings) -> None:
|
||||
"""Test check_password returns False for wrong username."""
|
||||
assert auth_settings.check_password("notadmin", "correctpassword") is False
|
||||
|
||||
|
||||
def test_check_password_both_wrong(auth_settings: DashboardSettings) -> None:
|
||||
"""Test check_password returns False when both are wrong."""
|
||||
assert auth_settings.check_password("notadmin", "wrongpassword") is False
|
||||
|
||||
|
||||
def test_check_password_no_auth(dashboard_settings: DashboardSettings) -> None:
|
||||
"""Test check_password returns True when auth is not configured."""
|
||||
assert dashboard_settings.check_password("anyone", "anything") is True
|
||||
|
||||
|
||||
def test_check_password_non_ascii_username(
|
||||
dashboard_settings: DashboardSettings,
|
||||
) -> None:
|
||||
"""Test check_password handles non-ASCII usernames without TypeError."""
|
||||
dashboard_settings.username = "\u00e9l\u00e8ve"
|
||||
dashboard_settings.using_password = True
|
||||
dashboard_settings.password_hash = password_hash("pass")
|
||||
assert dashboard_settings.check_password("\u00e9l\u00e8ve", "pass") is True
|
||||
assert dashboard_settings.check_password("\u00e9l\u00e8ve", "wrong") is False
|
||||
assert dashboard_settings.check_password("other", "pass") is False
|
||||
|
||||
|
||||
def test_check_password_ha_addon_no_password(
|
||||
dashboard_settings: DashboardSettings,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Test check_password doesn't crash in HA add-on mode without a password.
|
||||
|
||||
In HA add-on mode, using_ha_addon_auth can be True while using_password
|
||||
is False, leaving password_hash as b"". This must not raise TypeError
|
||||
in hmac.compare_digest.
|
||||
"""
|
||||
monkeypatch.delenv("DISABLE_HA_AUTHENTICATION", raising=False)
|
||||
dashboard_settings.on_ha_addon = True
|
||||
dashboard_settings.using_password = False
|
||||
# password_hash stays as default b""
|
||||
assert dashboard_settings.check_password("anyone", "anything") is False
|
||||
|
||||
@@ -2,12 +2,14 @@ from __future__ import annotations
|
||||
|
||||
from argparse import Namespace
|
||||
import asyncio
|
||||
import base64
|
||||
from collections.abc import Generator
|
||||
from contextlib import asynccontextmanager
|
||||
import gzip
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import sys
|
||||
from unittest.mock import AsyncMock, MagicMock, Mock, patch
|
||||
|
||||
import pytest
|
||||
@@ -421,7 +423,7 @@ async def test_download_binary_handler_idedata_fallback(
|
||||
|
||||
# Mock idedata response
|
||||
mock_image = Mock()
|
||||
mock_image.path = str(bootloader_file)
|
||||
mock_image.path = bootloader_file
|
||||
mock_idedata_instance = Mock()
|
||||
mock_idedata_instance.extra_flash_images = [mock_image]
|
||||
mock_idedata.return_value = mock_idedata_instance
|
||||
@@ -528,14 +530,22 @@ async def test_download_binary_handler_subdirectory_file_url_encoded(
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.usefixtures("mock_ext_storage_path")
|
||||
@pytest.mark.parametrize(
|
||||
"attack_path",
|
||||
("attack_path", "expected_code"),
|
||||
[
|
||||
pytest.param("../../../secrets.yaml", id="basic_traversal"),
|
||||
pytest.param("..%2F..%2F..%2Fsecrets.yaml", id="url_encoded"),
|
||||
pytest.param("zephyr/../../../secrets.yaml", id="traversal_with_prefix"),
|
||||
pytest.param("/etc/passwd", id="absolute_path"),
|
||||
pytest.param("//etc/passwd", id="double_slash_absolute"),
|
||||
pytest.param("....//secrets.yaml", id="multiple_dots"),
|
||||
pytest.param("../../../secrets.yaml", 403, id="basic_traversal"),
|
||||
pytest.param("..%2F..%2F..%2Fsecrets.yaml", 403, id="url_encoded"),
|
||||
pytest.param("zephyr/../../../secrets.yaml", 403, id="traversal_with_prefix"),
|
||||
pytest.param("/etc/passwd", 403, id="absolute_path"),
|
||||
pytest.param("//etc/passwd", 403, id="double_slash_absolute"),
|
||||
pytest.param(
|
||||
"....//secrets.yaml",
|
||||
# On Windows, Path.resolve() treats "..." and "...." as parent
|
||||
# traversal (like ".."), so the path escapes base_dir -> 403.
|
||||
# On Unix, "...." is a literal directory name that stays inside
|
||||
# base_dir but doesn't exist -> 404.
|
||||
403 if sys.platform == "win32" else 404,
|
||||
id="multiple_dots",
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_download_binary_handler_path_traversal_protection(
|
||||
@@ -543,11 +553,14 @@ async def test_download_binary_handler_path_traversal_protection(
|
||||
tmp_path: Path,
|
||||
mock_storage_json: MagicMock,
|
||||
attack_path: str,
|
||||
expected_code: int,
|
||||
) -> None:
|
||||
"""Test that DownloadBinaryRequestHandler prevents path traversal attacks.
|
||||
|
||||
Verifies that attempts to use '..' in file paths are sanitized to prevent
|
||||
accessing files outside the build directory. Tests multiple attack vectors.
|
||||
Verifies that attempts to escape the build directory via '..' are rejected
|
||||
using resolve()/relative_to() validation. Tests multiple attack vectors.
|
||||
Real traversals that escape the base directory get 403. Paths like '....'
|
||||
that resolve inside the base directory but don't exist get 404.
|
||||
"""
|
||||
# Create build structure
|
||||
build_dir = get_build_path(tmp_path, "test")
|
||||
@@ -565,14 +578,67 @@ async def test_download_binary_handler_path_traversal_protection(
|
||||
mock_storage.firmware_bin_path = firmware_file
|
||||
mock_storage_json.load.return_value = mock_storage
|
||||
|
||||
# Attempt path traversal attack - should be blocked
|
||||
with pytest.raises(HTTPClientError) as exc_info:
|
||||
# Mock async_run_system_command so paths that pass validation but don't exist
|
||||
# return 404 deterministically without spawning a real subprocess.
|
||||
with (
|
||||
patch(
|
||||
"esphome.dashboard.web_server.async_run_system_command",
|
||||
new_callable=AsyncMock,
|
||||
return_value=(2, "", ""),
|
||||
),
|
||||
pytest.raises(HTTPClientError) as exc_info,
|
||||
):
|
||||
await dashboard.fetch(
|
||||
f"/download.bin?configuration=test.yaml&file={attack_path}",
|
||||
method="GET",
|
||||
)
|
||||
# Should get 404 (file not found after sanitization) or 500 (idedata fails)
|
||||
assert exc_info.value.code in (404, 500)
|
||||
assert exc_info.value.code == expected_code
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.usefixtures("mock_ext_storage_path")
|
||||
async def test_download_binary_handler_no_firmware_bin_path(
|
||||
dashboard: DashboardTestHelper,
|
||||
mock_storage_json: MagicMock,
|
||||
) -> None:
|
||||
"""Test that download returns 404 when firmware_bin_path is None.
|
||||
|
||||
This covers configs created by StorageJSON.from_wizard() where no
|
||||
firmware has been compiled yet.
|
||||
"""
|
||||
mock_storage = Mock()
|
||||
mock_storage.name = "test_device"
|
||||
mock_storage.firmware_bin_path = None
|
||||
mock_storage_json.load.return_value = mock_storage
|
||||
|
||||
with pytest.raises(HTTPClientError) as exc_info:
|
||||
await dashboard.fetch(
|
||||
"/download.bin?configuration=test.yaml&file=firmware.bin",
|
||||
method="GET",
|
||||
)
|
||||
assert exc_info.value.code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.usefixtures("mock_ext_storage_path")
|
||||
@pytest.mark.parametrize("file_value", ["", "%20%20", "%20"])
|
||||
async def test_download_binary_handler_empty_file_name(
|
||||
dashboard: DashboardTestHelper,
|
||||
mock_storage_json: MagicMock,
|
||||
file_value: str,
|
||||
) -> None:
|
||||
"""Test that download returns 400 for empty or whitespace-only file names."""
|
||||
mock_storage = Mock()
|
||||
mock_storage.name = "test_device"
|
||||
mock_storage.firmware_bin_path = Path("/fake/firmware.bin")
|
||||
mock_storage_json.load.return_value = mock_storage
|
||||
|
||||
with pytest.raises(HTTPClientError) as exc_info:
|
||||
await dashboard.fetch(
|
||||
f"/download.bin?configuration=test.yaml&file={file_value}",
|
||||
method="GET",
|
||||
)
|
||||
assert exc_info.value.code == 400
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -1676,3 +1742,85 @@ def test_proc_on_exit_skips_when_already_closed() -> None:
|
||||
|
||||
handler.write_message.assert_not_called()
|
||||
handler.close.assert_not_called()
|
||||
|
||||
|
||||
def _make_auth_handler(auth_header: str | None = None) -> Mock:
|
||||
"""Create a mock handler with the given Authorization header."""
|
||||
handler = Mock()
|
||||
handler.request = Mock()
|
||||
if auth_header is not None:
|
||||
handler.request.headers = {"Authorization": auth_header}
|
||||
else:
|
||||
handler.request.headers = {}
|
||||
handler.get_secure_cookie = Mock(return_value=None)
|
||||
return handler
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_auth_settings(mock_dashboard_settings: MagicMock) -> MagicMock:
|
||||
"""Fixture to configure mock dashboard settings with auth enabled."""
|
||||
mock_dashboard_settings.using_auth = True
|
||||
mock_dashboard_settings.on_ha_addon = False
|
||||
return mock_dashboard_settings
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_auth_settings")
|
||||
def test_is_authenticated_malformed_base64() -> None:
|
||||
"""Test that invalid base64 in Authorization header returns False."""
|
||||
handler = _make_auth_handler("Basic !!!not-valid-base64!!!")
|
||||
assert web_server.is_authenticated(handler) is False
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_auth_settings")
|
||||
def test_is_authenticated_bad_base64_padding() -> None:
|
||||
"""Test that incorrect base64 padding (binascii.Error) returns False."""
|
||||
handler = _make_auth_handler("Basic abc")
|
||||
assert web_server.is_authenticated(handler) is False
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_auth_settings")
|
||||
def test_is_authenticated_invalid_utf8() -> None:
|
||||
"""Test that base64 decoding to invalid UTF-8 returns False."""
|
||||
# \xff\xfe is invalid UTF-8
|
||||
bad_payload = base64.b64encode(b"\xff\xfe").decode("ascii")
|
||||
handler = _make_auth_handler(f"Basic {bad_payload}")
|
||||
assert web_server.is_authenticated(handler) is False
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_auth_settings")
|
||||
def test_is_authenticated_no_colon() -> None:
|
||||
"""Test that base64 payload without ':' separator returns False."""
|
||||
no_colon = base64.b64encode(b"nocolonhere").decode("ascii")
|
||||
handler = _make_auth_handler(f"Basic {no_colon}")
|
||||
assert web_server.is_authenticated(handler) is False
|
||||
|
||||
|
||||
def test_is_authenticated_valid_credentials(
|
||||
mock_auth_settings: MagicMock,
|
||||
) -> None:
|
||||
"""Test that valid Basic auth credentials are checked."""
|
||||
creds = base64.b64encode(b"admin:secret").decode("ascii")
|
||||
mock_auth_settings.check_password.return_value = True
|
||||
handler = _make_auth_handler(f"Basic {creds}")
|
||||
assert web_server.is_authenticated(handler) is True
|
||||
mock_auth_settings.check_password.assert_called_once_with("admin", "secret")
|
||||
|
||||
|
||||
def test_is_authenticated_wrong_credentials(
|
||||
mock_auth_settings: MagicMock,
|
||||
) -> None:
|
||||
"""Test that valid Basic auth with wrong credentials returns False."""
|
||||
creds = base64.b64encode(b"admin:wrong").decode("ascii")
|
||||
mock_auth_settings.check_password.return_value = False
|
||||
handler = _make_auth_handler(f"Basic {creds}")
|
||||
assert web_server.is_authenticated(handler) is False
|
||||
|
||||
|
||||
def test_is_authenticated_no_auth_configured(
|
||||
mock_dashboard_settings: MagicMock,
|
||||
) -> None:
|
||||
"""Test that requests pass when auth is not configured."""
|
||||
mock_dashboard_settings.using_auth = False
|
||||
mock_dashboard_settings.on_ha_addon = False
|
||||
handler = _make_auth_handler()
|
||||
assert web_server.is_authenticated(handler) is True
|
||||
|
||||
@@ -18,8 +18,8 @@ from esphome import espota2
|
||||
from esphome.core import EsphomeError
|
||||
|
||||
# Test constants
|
||||
MOCK_RANDOM_VALUE = 0.123456
|
||||
MOCK_RANDOM_BYTES = b"0.123456"
|
||||
MOCK_MD5_CNONCE = "a" * 32 # Mock 32-char hex string from secrets.token_hex(16)
|
||||
MOCK_SHA256_CNONCE = "b" * 64 # Mock 64-char hex string from secrets.token_hex(32)
|
||||
MOCK_MD5_NONCE = b"12345678901234567890123456789012" # 32 char nonce for MD5
|
||||
MOCK_SHA256_NONCE = b"1234567890123456789012345678901234567890123456789012345678901234" # 64 char nonce for SHA256
|
||||
|
||||
@@ -55,10 +55,18 @@ def mock_time() -> Generator[None]:
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_random() -> Generator[Mock]:
|
||||
"""Mock random for predictable test values."""
|
||||
with patch("random.random", return_value=MOCK_RANDOM_VALUE) as mock_rand:
|
||||
yield mock_rand
|
||||
def mock_token_hex() -> Generator[Mock]:
|
||||
"""Mock secrets.token_hex for predictable test values."""
|
||||
|
||||
def _token_hex(nbytes: int) -> str:
|
||||
if nbytes == 16:
|
||||
return MOCK_MD5_CNONCE
|
||||
if nbytes == 32:
|
||||
return MOCK_SHA256_CNONCE
|
||||
raise ValueError(f"Unexpected nbytes for token_hex mock: {nbytes}")
|
||||
|
||||
with patch("esphome.espota2.secrets.token_hex", side_effect=_token_hex) as mock:
|
||||
yield mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -236,7 +244,7 @@ def test_send_check_socket_error(mock_socket: Mock) -> None:
|
||||
|
||||
@pytest.mark.usefixtures("mock_time")
|
||||
def test_perform_ota_successful_md5_auth(
|
||||
mock_socket: Mock, mock_file: io.BytesIO, mock_random: Mock
|
||||
mock_socket: Mock, mock_file: io.BytesIO, mock_token_hex: Mock
|
||||
) -> None:
|
||||
"""Test successful OTA with MD5 authentication."""
|
||||
# Setup socket responses for recv calls
|
||||
@@ -272,8 +280,11 @@ def test_perform_ota_successful_md5_auth(
|
||||
)
|
||||
)
|
||||
|
||||
# Verify cnonce was sent (MD5 of random.random())
|
||||
cnonce = hashlib.md5(MOCK_RANDOM_BYTES).hexdigest()
|
||||
# Verify token_hex was called with MD5 digest size
|
||||
mock_token_hex.assert_called_once_with(16)
|
||||
|
||||
# Verify cnonce was sent
|
||||
cnonce = MOCK_MD5_CNONCE
|
||||
assert mock_socket.sendall.call_args_list[2] == call(cnonce.encode())
|
||||
|
||||
# Verify auth result was computed correctly
|
||||
@@ -366,7 +377,7 @@ def test_perform_ota_auth_without_password(mock_socket: Mock) -> None:
|
||||
|
||||
@pytest.mark.usefixtures("mock_time")
|
||||
def test_perform_ota_md5_auth_wrong_password(
|
||||
mock_socket: Mock, mock_file: io.BytesIO, mock_random: Mock
|
||||
mock_socket: Mock, mock_file: io.BytesIO, mock_token_hex: Mock
|
||||
) -> None:
|
||||
"""Test OTA fails when MD5 authentication is rejected due to wrong password."""
|
||||
# Setup socket responses for recv calls
|
||||
@@ -390,7 +401,7 @@ def test_perform_ota_md5_auth_wrong_password(
|
||||
|
||||
@pytest.mark.usefixtures("mock_time")
|
||||
def test_perform_ota_sha256_auth_wrong_password(
|
||||
mock_socket: Mock, mock_file: io.BytesIO, mock_random: Mock
|
||||
mock_socket: Mock, mock_file: io.BytesIO, mock_token_hex: Mock
|
||||
) -> None:
|
||||
"""Test OTA fails when SHA256 authentication is rejected due to wrong password."""
|
||||
# Setup socket responses for recv calls
|
||||
@@ -603,7 +614,7 @@ def test_progress_bar(capsys: CaptureFixture[str]) -> None:
|
||||
# Tests for SHA256 authentication
|
||||
@pytest.mark.usefixtures("mock_time")
|
||||
def test_perform_ota_successful_sha256_auth(
|
||||
mock_socket: Mock, mock_file: io.BytesIO, mock_random: Mock
|
||||
mock_socket: Mock, mock_file: io.BytesIO, mock_token_hex: Mock
|
||||
) -> None:
|
||||
"""Test successful OTA with SHA256 authentication."""
|
||||
# Setup socket responses for recv calls
|
||||
@@ -639,8 +650,11 @@ def test_perform_ota_successful_sha256_auth(
|
||||
)
|
||||
)
|
||||
|
||||
# Verify cnonce was sent (SHA256 of random.random())
|
||||
cnonce = hashlib.sha256(MOCK_RANDOM_BYTES).hexdigest()
|
||||
# Verify token_hex was called with SHA256 digest size
|
||||
mock_token_hex.assert_called_once_with(32)
|
||||
|
||||
# Verify cnonce was sent
|
||||
cnonce = MOCK_SHA256_CNONCE
|
||||
assert mock_socket.sendall.call_args_list[2] == call(cnonce.encode())
|
||||
|
||||
# Verify auth result was computed correctly with SHA256
|
||||
@@ -654,7 +668,7 @@ def test_perform_ota_successful_sha256_auth(
|
||||
|
||||
@pytest.mark.usefixtures("mock_time")
|
||||
def test_perform_ota_sha256_fallback_to_md5(
|
||||
mock_socket: Mock, mock_file: io.BytesIO, mock_random: Mock
|
||||
mock_socket: Mock, mock_file: io.BytesIO, mock_token_hex: Mock
|
||||
) -> None:
|
||||
"""Test SHA256-capable client falls back to MD5 for compatibility."""
|
||||
# This test verifies the temporary backward compatibility
|
||||
@@ -692,7 +706,8 @@ def test_perform_ota_sha256_fallback_to_md5(
|
||||
)
|
||||
|
||||
# But authentication was done with MD5
|
||||
cnonce = hashlib.md5(MOCK_RANDOM_BYTES).hexdigest()
|
||||
mock_token_hex.assert_called_once_with(16)
|
||||
cnonce = MOCK_MD5_CNONCE
|
||||
expected_hash = hashlib.md5()
|
||||
expected_hash.update(b"testpass")
|
||||
expected_hash.update(MOCK_MD5_NONCE)
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from pytest import MonkeyPatch
|
||||
@@ -632,3 +632,14 @@ def test_wizard_accepts_rpipico_board(tmp_path: Path, monkeypatch: MonkeyPatch):
|
||||
# rpipico doesn't support WiFi, so no api_encryption_key or ota_password
|
||||
assert "api_encryption_key" not in call_kwargs
|
||||
assert "ota_password" not in call_kwargs
|
||||
|
||||
|
||||
def test_fallback_psk_uses_secrets_choice(
|
||||
default_config: dict[str, Any],
|
||||
) -> None:
|
||||
"""Test that fallback PSK is generated using secrets.choice."""
|
||||
with patch("esphome.wizard.secrets.choice", return_value="X") as mock_choice:
|
||||
config = wz.wizard_file(**default_config)
|
||||
|
||||
assert 'password: "XXXXXXXXXXXX"' in config
|
||||
assert mock_choice.call_count == 12
|
||||
|
||||
Reference in New Issue
Block a user