mirror of
https://github.com/esphome/esphome.git
synced 2026-09-11 15:27:33 +00:00
[usb_host] Validate USB configuration more strictly
This commit is contained in:
@@ -58,6 +58,41 @@ def usb_device_schema(
|
||||
)
|
||||
|
||||
|
||||
_validate_filters_complete = cv.has_none_or_all_keys(CONF_MANUFACTURER, CONF_PRODUCT)
|
||||
|
||||
|
||||
def validate_usb_clients(configs: list[ConfigType]) -> list[ConfigType]:
|
||||
"""Reject invalid USB configuration."""
|
||||
for config in configs:
|
||||
_validate_filters_complete(config)
|
||||
for index, first in enumerate(configs):
|
||||
# Ensure matching logic does not overlap between entries
|
||||
for second in configs[index + 1 :]:
|
||||
if (
|
||||
not (first[CONF_VID] == 0 and first[CONF_PID] == 0)
|
||||
and not (second[CONF_VID] == 0 and second[CONF_PID] == 0)
|
||||
and (
|
||||
first[CONF_VID] != second[CONF_VID]
|
||||
or first[CONF_PID] != second[CONF_PID]
|
||||
)
|
||||
):
|
||||
continue
|
||||
|
||||
# An unset filter constrains nothing, so only a differing value separates them
|
||||
if not all(
|
||||
(a := first.get(key)) is None
|
||||
or (b := second.get(key)) is None
|
||||
or a == b
|
||||
for key in (CONF_MANUFACTURER, CONF_PRODUCT)
|
||||
):
|
||||
continue
|
||||
|
||||
raise cv.Invalid(
|
||||
f"USB configs overlap: {first[CONF_ID]!r}, {second[CONF_ID]!r}"
|
||||
)
|
||||
return configs
|
||||
|
||||
|
||||
def _set_max_packet_size(config: dict) -> dict:
|
||||
CORE.data.setdefault(DOMAIN, {})[CONF_MAX_PACKET_SIZE] = config[
|
||||
CONF_MAX_PACKET_SIZE
|
||||
@@ -80,7 +115,9 @@ CONFIG_SCHEMA = cv.All(
|
||||
cv.Optional(CONF_MAX_PACKET_SIZE, default=64): cv.one_of(
|
||||
64, 128, 256, 512, 1024, int=True
|
||||
),
|
||||
cv.Optional(CONF_DEVICES): cv.ensure_list(usb_device_schema()),
|
||||
cv.Optional(CONF_DEVICES): cv.All(
|
||||
cv.ensure_list(usb_device_schema()), validate_usb_clients
|
||||
),
|
||||
}
|
||||
),
|
||||
only_on_variant(
|
||||
|
||||
@@ -6,6 +6,7 @@ from esphome.components.usb_host import (
|
||||
get_max_packet_size,
|
||||
register_usb_client,
|
||||
usb_device_schema,
|
||||
validate_usb_clients,
|
||||
)
|
||||
import esphome.config_validation as cv
|
||||
from esphome.const import (
|
||||
@@ -154,16 +155,19 @@ def channel_schema(type_: "Type") -> cv.Schema:
|
||||
)
|
||||
|
||||
|
||||
CONFIG_SCHEMA = cv.ensure_list(
|
||||
cv.typed_schema(
|
||||
{
|
||||
it.name: usb_device_schema(it.cls, it.vid, it.pid).extend(
|
||||
channel_schema(it)
|
||||
)
|
||||
for it in uart_types
|
||||
},
|
||||
upper=True,
|
||||
)
|
||||
CONFIG_SCHEMA = cv.All(
|
||||
cv.ensure_list(
|
||||
cv.typed_schema(
|
||||
{
|
||||
it.name: usb_device_schema(it.cls, it.vid, it.pid).extend(
|
||||
channel_schema(it)
|
||||
)
|
||||
for it in uart_types
|
||||
},
|
||||
upper=True,
|
||||
)
|
||||
),
|
||||
validate_usb_clients,
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -35,8 +35,10 @@ usb_uart:
|
||||
debug: true
|
||||
dummy_receiver: true
|
||||
debug_prefix: "[ESP_JTAG] "
|
||||
# A CP2105, so it does not share 10C4:EA60 with uart_1 above
|
||||
- id: uart_5
|
||||
type: cp210x
|
||||
pid: 0xEA70
|
||||
channels:
|
||||
- id: channel_5_1
|
||||
baud_rate: 9600
|
||||
|
||||
@@ -0,0 +1,199 @@
|
||||
"""Tests for usb_host device matching validation."""
|
||||
|
||||
import pytest
|
||||
|
||||
from esphome.components.usb_host import validate_usb_clients
|
||||
import esphome.config_validation as cv
|
||||
from esphome.types import ConfigType
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("first", "second"),
|
||||
[
|
||||
pytest.param(
|
||||
{
|
||||
"id": "a",
|
||||
"vid": 0x303A,
|
||||
"pid": 0x4001,
|
||||
},
|
||||
{
|
||||
"id": "b",
|
||||
"vid": 0x303A,
|
||||
"pid": 0x4002,
|
||||
},
|
||||
id="different_pid",
|
||||
),
|
||||
pytest.param(
|
||||
{
|
||||
"id": "a",
|
||||
"vid": 0x303A,
|
||||
"pid": 0x4001,
|
||||
},
|
||||
{
|
||||
"id": "b",
|
||||
"vid": 0x1A86,
|
||||
"pid": 0x4001,
|
||||
},
|
||||
id="different_vid",
|
||||
),
|
||||
pytest.param(
|
||||
{
|
||||
"id": "a",
|
||||
"vid": 0x303A,
|
||||
"pid": 0x4001,
|
||||
"manufacturer": "Nabu Casa",
|
||||
"product": "ZBT-2",
|
||||
},
|
||||
{
|
||||
"id": "b",
|
||||
"vid": 0x303A,
|
||||
"pid": 0x4001,
|
||||
"manufacturer": "Nabu Casa",
|
||||
"product": "ZWA-2",
|
||||
},
|
||||
id="different_product",
|
||||
),
|
||||
pytest.param(
|
||||
{
|
||||
"id": "a",
|
||||
"vid": 0x303A,
|
||||
"pid": 0x4001,
|
||||
"manufacturer": "Nabu Casa",
|
||||
"product": "ZBT-2",
|
||||
},
|
||||
{
|
||||
"id": "b",
|
||||
"vid": 0x303A,
|
||||
"pid": 0x4001,
|
||||
"manufacturer": "Espressif",
|
||||
"product": "ZBT-2",
|
||||
},
|
||||
id="different_manufacturer",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_disjoint_clients_are_accepted(first: ConfigType, second: ConfigType) -> None:
|
||||
configs = [first, second]
|
||||
assert validate_usb_clients(configs) is configs
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("first", "second"),
|
||||
[
|
||||
pytest.param(
|
||||
{
|
||||
"id": "a",
|
||||
"vid": 0x303A,
|
||||
"pid": 0x4001,
|
||||
},
|
||||
{
|
||||
"id": "b",
|
||||
"vid": 0x303A,
|
||||
"pid": 0x4001,
|
||||
},
|
||||
id="exact_duplicate",
|
||||
),
|
||||
pytest.param(
|
||||
{
|
||||
"id": "a",
|
||||
"vid": 0x303A,
|
||||
"pid": 0x4001,
|
||||
},
|
||||
{
|
||||
"id": "b",
|
||||
"vid": 0x303A,
|
||||
"pid": 0x4001,
|
||||
"manufacturer": "Nabu Casa",
|
||||
"product": "ZBT-2",
|
||||
},
|
||||
id="unfiltered_shadows_filtered",
|
||||
),
|
||||
pytest.param(
|
||||
{
|
||||
"id": "a",
|
||||
"vid": 0x303A,
|
||||
"pid": 0x4001,
|
||||
"manufacturer": "Nabu Casa",
|
||||
"product": "ZBT-2",
|
||||
},
|
||||
{
|
||||
"id": "b",
|
||||
"vid": 0x303A,
|
||||
"pid": 0x4001,
|
||||
"manufacturer": "Nabu Casa",
|
||||
"product": "ZBT-2",
|
||||
},
|
||||
id="identical_filters",
|
||||
),
|
||||
pytest.param(
|
||||
{
|
||||
"id": "a",
|
||||
"vid": 0,
|
||||
"pid": 0,
|
||||
},
|
||||
{
|
||||
"id": "b",
|
||||
"vid": 0x303A,
|
||||
"pid": 0x4001,
|
||||
"manufacturer": "Nabu Casa",
|
||||
"product": "ZBT-2",
|
||||
},
|
||||
id="zero_ids_match_every_device",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_overlapping_clients_are_rejected(
|
||||
first: ConfigType, second: ConfigType
|
||||
) -> None:
|
||||
with pytest.raises(cv.Invalid, match="overlap"):
|
||||
validate_usb_clients([first, second])
|
||||
|
||||
|
||||
def test_every_pair_is_compared_not_just_neighbours() -> None:
|
||||
configs = [
|
||||
{
|
||||
"id": "a",
|
||||
"vid": 0x303A,
|
||||
"pid": 0x4001,
|
||||
"manufacturer": "Nabu Casa",
|
||||
"product": "ZBT-2",
|
||||
},
|
||||
{
|
||||
"id": "b",
|
||||
"vid": 0x303A,
|
||||
"pid": 0x4002,
|
||||
},
|
||||
{
|
||||
"id": "c",
|
||||
"vid": 0x303A,
|
||||
"pid": 0x4001,
|
||||
"manufacturer": "Nabu Casa",
|
||||
"product": "ZBT-2",
|
||||
},
|
||||
]
|
||||
with pytest.raises(cv.Invalid, match="'a', 'c'"):
|
||||
validate_usb_clients(configs)
|
||||
|
||||
|
||||
def test_incomplete_filter_is_rejected() -> None:
|
||||
configs = [
|
||||
{
|
||||
"id": "a",
|
||||
"vid": 0x303A,
|
||||
"pid": 0x4001,
|
||||
"product": "ZBT-2",
|
||||
}
|
||||
]
|
||||
with pytest.raises(cv.Invalid, match="none or all"):
|
||||
validate_usb_clients(configs)
|
||||
|
||||
|
||||
def test_single_client_is_always_valid() -> None:
|
||||
configs = [
|
||||
{
|
||||
"id": "a",
|
||||
"vid": 0x303A,
|
||||
"pid": 0x4001,
|
||||
}
|
||||
]
|
||||
assert validate_usb_clients(configs) is configs
|
||||
Reference in New Issue
Block a user