mirror of
https://github.com/esphome/esphome.git
synced 2026-10-07 11:26:39 +00:00
[socket][tcp_uart][uart_tcp] Reject an IPv6 host at config time (#20165)
This commit is contained in:
@@ -149,6 +149,21 @@ def require_wake_loop_threadsafe() -> None:
|
||||
IPV4_ALLOW_SCHEMA = cv.All(cv.ensure_list(cv.ipv4network), cv.Length(max=255))
|
||||
|
||||
|
||||
_HOST = cv.Any(cv.domain, cv.hostname)
|
||||
|
||||
|
||||
def ipv4_host(value: object) -> str:
|
||||
"""Validate an IPv4 address or a hostname; the resolver behind it is IPv4 only."""
|
||||
value = cv.string(value)
|
||||
try:
|
||||
cv.ipv6address(value)
|
||||
except cv.Invalid:
|
||||
return _HOST(value)
|
||||
raise cv.Invalid(
|
||||
"IPv6 addresses are not supported, use an IPv4 address or a hostname"
|
||||
)
|
||||
|
||||
|
||||
def _network_order(addr: IPv4Address) -> int:
|
||||
"""The s_addr value for addr on the little endian targets."""
|
||||
return int.from_bytes(addr.packed, "little")
|
||||
|
||||
@@ -54,7 +54,7 @@ CONFIG_SCHEMA = cv.All(
|
||||
{
|
||||
"client": BASE_SCHEMA.extend(
|
||||
{
|
||||
cv.Required(CONF_HOST): cv.string,
|
||||
cv.Required(CONF_HOST): socket.ipv4_host,
|
||||
}
|
||||
),
|
||||
"server": BASE_SCHEMA.extend(
|
||||
|
||||
@@ -44,7 +44,7 @@ BASE_SCHEMA = cv.Schema(
|
||||
CONFIG_SCHEMA = cv.All(
|
||||
cv.typed_schema(
|
||||
{
|
||||
"client": BASE_SCHEMA.extend({cv.Required(CONF_HOST): cv.string}),
|
||||
"client": BASE_SCHEMA.extend({cv.Required(CONF_HOST): socket.ipv4_host}),
|
||||
"server": BASE_SCHEMA.extend(
|
||||
{cv.Optional(CONF_ALLOWED_IPS): socket.IPV4_ALLOW_SCHEMA}
|
||||
),
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
"""Tests for the socket component's IPv4 host validator."""
|
||||
|
||||
import pytest
|
||||
|
||||
from esphome.components import socket
|
||||
import esphome.config_validation as cv
|
||||
|
||||
|
||||
@pytest.mark.parametrize("value", ["192.0.2.10", "bridge.local", "bridge"])
|
||||
def test_ipv4_host_accepts_ipv4_address_and_hostname(value: str) -> None:
|
||||
assert socket.ipv4_host(value) == value
|
||||
|
||||
|
||||
@pytest.mark.parametrize("value", ["2001:db8::10", "::1", "::ffff:192.0.2.10"])
|
||||
def test_ipv4_host_rejects_ipv6_address(value: str) -> None:
|
||||
with pytest.raises(cv.Invalid, match="IPv6 addresses are not supported"):
|
||||
socket.ipv4_host(value)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("value", ["bridge host", ""])
|
||||
def test_ipv4_host_rejects_invalid_host(value: str) -> None:
|
||||
with pytest.raises(cv.Invalid):
|
||||
socket.ipv4_host(value)
|
||||
Reference in New Issue
Block a user