mirror of
https://github.com/esphome/esphome.git
synced 2025-07-17 08:46:34 +00:00
Sort resolved IP addresses for dashboard (#8536)
Co-authored-by: J. Nick Koston <nick+github@koston.org>
This commit is contained in:
parent
4a65fd76b3
commit
3c7bb65a23
@ -38,7 +38,7 @@ import yaml
|
|||||||
from yaml.nodes import Node
|
from yaml.nodes import Node
|
||||||
|
|
||||||
from esphome import const, platformio_api, yaml_util
|
from esphome import const, platformio_api, yaml_util
|
||||||
from esphome.helpers import get_bool_env, mkdir_p
|
from esphome.helpers import get_bool_env, mkdir_p, sort_ip_addresses
|
||||||
from esphome.storage_json import (
|
from esphome.storage_json import (
|
||||||
StorageJSON,
|
StorageJSON,
|
||||||
archive_storage_path,
|
archive_storage_path,
|
||||||
@ -336,7 +336,7 @@ class EsphomePortCommandWebSocket(EsphomeCommandWebSocket):
|
|||||||
# Use the IP address if available but only
|
# Use the IP address if available but only
|
||||||
# if the API is loaded and the device is online
|
# if the API is loaded and the device is online
|
||||||
# since MQTT logging will not work otherwise
|
# since MQTT logging will not work otherwise
|
||||||
port = address_list[0]
|
port = sort_ip_addresses(address_list)[0]
|
||||||
elif (
|
elif (
|
||||||
entry.address
|
entry.address
|
||||||
and (
|
and (
|
||||||
@ -347,7 +347,7 @@ class EsphomePortCommandWebSocket(EsphomeCommandWebSocket):
|
|||||||
and not isinstance(address_list, Exception)
|
and not isinstance(address_list, Exception)
|
||||||
):
|
):
|
||||||
# If mdns is not available, try to use the DNS cache
|
# If mdns is not available, try to use the DNS cache
|
||||||
port = address_list[0]
|
port = sort_ip_addresses(address_list)[0]
|
||||||
|
|
||||||
return [
|
return [
|
||||||
*DASHBOARD_COMMAND,
|
*DASHBOARD_COMMAND,
|
||||||
|
@ -200,6 +200,45 @@ def resolve_ip_address(host, port):
|
|||||||
return res
|
return res
|
||||||
|
|
||||||
|
|
||||||
|
def sort_ip_addresses(address_list: list[str]) -> list[str]:
|
||||||
|
"""Takes a list of IP addresses in string form, e.g. from mDNS or MQTT,
|
||||||
|
and sorts them into the best order to actually try connecting to them.
|
||||||
|
|
||||||
|
This is roughly based on RFC6724 but a lot simpler: First we choose
|
||||||
|
IPv6 addresses, then Legacy IP addresses, and lowest priority is
|
||||||
|
link-local IPv6 addresses that don't have a link specified (which
|
||||||
|
are useless, but mDNS does provide them in that form). Addresses
|
||||||
|
which cannot be parsed are silently dropped.
|
||||||
|
"""
|
||||||
|
import socket
|
||||||
|
|
||||||
|
# First "resolve" all the IP addresses to getaddrinfo() tuples of the form
|
||||||
|
# (family, type, proto, canonname, sockaddr)
|
||||||
|
res: list[
|
||||||
|
tuple[
|
||||||
|
int,
|
||||||
|
int,
|
||||||
|
int,
|
||||||
|
Union[str, None],
|
||||||
|
Union[tuple[str, int], tuple[str, int, int, int]],
|
||||||
|
]
|
||||||
|
] = []
|
||||||
|
for addr in address_list:
|
||||||
|
# This should always work as these are supposed to be IP addresses
|
||||||
|
try:
|
||||||
|
res += socket.getaddrinfo(
|
||||||
|
addr, 0, proto=socket.IPPROTO_TCP, flags=socket.AI_NUMERICHOST
|
||||||
|
)
|
||||||
|
except OSError:
|
||||||
|
_LOGGER.info("Failed to parse IP address '%s'", addr)
|
||||||
|
|
||||||
|
# Now use that information to sort them.
|
||||||
|
res.sort(key=addr_preference_)
|
||||||
|
|
||||||
|
# Finally, turn the getaddrinfo() tuples back into plain hostnames.
|
||||||
|
return [socket.getnameinfo(r[4], socket.NI_NUMERICHOST)[0] for r in res]
|
||||||
|
|
||||||
|
|
||||||
def get_bool_env(var, default=False):
|
def get_bool_env(var, default=False):
|
||||||
value = os.getenv(var, default)
|
value = os.getenv(var, default)
|
||||||
if isinstance(value, str):
|
if isinstance(value, str):
|
||||||
|
@ -267,3 +267,13 @@ def test_sanitize(text, expected):
|
|||||||
actual = helpers.sanitize(text)
|
actual = helpers.sanitize(text)
|
||||||
|
|
||||||
assert actual == expected
|
assert actual == expected
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"text, expected",
|
||||||
|
((["127.0.0.1", "fe80::1", "2001::2"], ["2001::2", "127.0.0.1", "fe80::1"]),),
|
||||||
|
)
|
||||||
|
def test_sort_ip_addresses(text: list[str], expected: list[str]) -> None:
|
||||||
|
actual = helpers.sort_ip_addresses(text)
|
||||||
|
|
||||||
|
assert actual == expected
|
||||||
|
Loading…
x
Reference in New Issue
Block a user