From ffb4ca997ef970d496bfb0e6eb546e76e0568668 Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Wed, 9 Sep 2026 21:26:54 +0000 Subject: [PATCH] [usb_host] Validate USB configuration more strictly --- esphome/components/usb_host/__init__.py | 39 +++- esphome/components/usb_uart/__init__.py | 24 ++- tests/components/usb_uart/common.yaml | 2 + tests/unit_tests/components/test_usb_host.py | 199 +++++++++++++++++++ 4 files changed, 253 insertions(+), 11 deletions(-) create mode 100644 tests/unit_tests/components/test_usb_host.py diff --git a/esphome/components/usb_host/__init__.py b/esphome/components/usb_host/__init__.py index 1263120c4e..5aa3763f03 100644 --- a/esphome/components/usb_host/__init__.py +++ b/esphome/components/usb_host/__init__.py @@ -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( diff --git a/esphome/components/usb_uart/__init__.py b/esphome/components/usb_uart/__init__.py index b696ca0352..a56306e79d 100644 --- a/esphome/components/usb_uart/__init__.py +++ b/esphome/components/usb_uart/__init__.py @@ -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, ) diff --git a/tests/components/usb_uart/common.yaml b/tests/components/usb_uart/common.yaml index 704418ff61..2586acfae6 100644 --- a/tests/components/usb_uart/common.yaml +++ b/tests/components/usb_uart/common.yaml @@ -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 diff --git a/tests/unit_tests/components/test_usb_host.py b/tests/unit_tests/components/test_usb_host.py new file mode 100644 index 0000000000..3d6b8dc1c0 --- /dev/null +++ b/tests/unit_tests/components/test_usb_host.py @@ -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