evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Relocate async_get_announce_addresses from zeroconf to network (#94816)

Browse source
This commit is contained in:
J. Nick Koston 2023-06-21 10:29:04 +01:00 • committed by GitHub
parent c47543c9dd
commit 605c4db142
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 147 additions and 73 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
homeassistant/components/network/__init__.py
View file

@ -119,6 +119,32 @@ async def async_get_ipv4_broadcast_addresses(hass: HomeAssistant) -> set[IPv4Add
return broadcast_addresses
async def async_get_announce_addresses(hass: HomeAssistant) -> list[str]:
"""Return a list of IP addresses to announce/use via zeroconf/ssdp/etc.
The default ip address is always returned first if available.
"""
adapters = await async_get_adapters(hass)
addresses: list[str] = []
default_ip: str | None = None
for adapter in adapters:
if not adapter["enabled"]:
continue
for ips in adapter["ipv4"]:
addresses.append(str(IPv4Address(ips["address"])))
for ips in adapter["ipv6"]:
addresses.append(str(IPv6Address(ips["address"])))
# Puts the default IPv4 address first in the list to preserve compatibility,
# because some mDNS implementations ignores anything but the first announced
# address.
if default_ip := await async_get_source_ip(hass, target_ip=MDNS_TARGET_IP):
if default_ip in addresses:
addresses.remove(default_ip)
return [default_ip] + list(addresses)
return list(addresses)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up network for Home Assistant."""
# Avoid circular issue: http->network->websocket_api->http

44
homeassistant/components/zeroconf/__init__.py
View file

@ -7,10 +7,9 @@ from contextlib import suppress
from dataclasses import dataclass
from fnmatch import translate
from functools import lru_cache
from ipaddress import IPv4Address, IPv6Address, ip_address
from ipaddress import IPv4Address, IPv6Address
import logging
import re
import socket
import sys
from typing import Any, Final, cast
@ -25,8 +24,6 @@ from zeroconf.asyncio import AsyncServiceInfo
from homeassistant import config_entries
from homeassistant.components import network
from homeassistant.components.network import MDNS_TARGET_IP, async_get_source_ip
from homeassistant.components.network.models import Adapter
from homeassistant.const import EVENT_HOMEASSISTANT_STOP, __version__
from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.data_entry_flow import BaseServiceInfo
@ -243,32 +240,6 @@ def _build_homekit_model_lookups(
return homekit_model_lookup, homekit_model_matchers
def _get_announced_addresses(
adapters: list[Adapter],
first_ip: bytes | None = None,
) -> list[bytes]:
"""Return a list of IP addresses to announce via zeroconf.
If first_ip is not None, it will be the first address in the list.
"""
addresses = {
addr.packed
for addr in [
ip_address(ip["address"])
for adapter in adapters
if adapter["enabled"]
for ip in cast(list, adapter["ipv6"]) + cast(list, adapter["ipv4"])
]
if not (addr.is_unspecified or addr.is_loopback)
}
if first_ip:
address_list = [first_ip]
address_list.extend(addresses - set({first_ip}))
else:
address_list = list(addresses)
return address_list
def _filter_disallowed_characters(name: str) -> str:
"""Filter disallowed characters from a string.
@ -307,24 +278,13 @@ async def _async_register_hass_zc_service(
# Set old base URL based on external or internal
params["base_url"] = params["external_url"] or params["internal_url"]
adapters = await network.async_get_adapters(hass)
# Puts the default IPv4 address first in the list to preserve compatibility,
# because some mDNS implementations ignores anything but the first announced
# address.
host_ip = await async_get_source_ip(hass, target_ip=MDNS_TARGET_IP)
host_ip_pton = None
if host_ip:
host_ip_pton = socket.inet_pton(socket.AF_INET, host_ip)
address_list = _get_announced_addresses(adapters, host_ip_pton)
_suppress_invalid_properties(params)
info = AsyncServiceInfo(
ZEROCONF_TYPE,
name=f"{valid_location_name}.{ZEROCONF_TYPE}",
server=f"{uuid}.local.",
addresses=address_list,
parsed_addresses=await network.async_get_announce_addresses(hass),
port=hass.http.server_port,
properties=params,
)

3
tests/components/local_ip/test_init.py
View file

@ -3,8 +3,7 @@ from __future__ import annotations
from homeassistant import config_entries
from homeassistant.components.local_ip import DOMAIN
from homeassistant.components.network import async_get_source_ip
from homeassistant.components.zeroconf import MDNS_TARGET_IP
from homeassistant.components.network import MDNS_TARGET_IP, async_get_source_ip
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry

117
tests/components/network/test_init.py
View file

@ -712,3 +712,120 @@ async def test_async_get_source_ip_no_ip_loopback(
await hass.async_block_till_done()
assert await network.async_get_source_ip(hass) == "127.0.0.1"
_ADAPTERS_WITH_MANUAL_CONFIG = [
{
"auto": True,
"index": 1,
"default": False,
"enabled": True,
"ipv4": [],
"ipv6": [
{
"address": "2001:db8::",
"network_prefix": 64,
"flowinfo": 1,
"scope_id": 1,
},
{
"address": "fe80::1234:5678:9abc:def0",
"network_prefix": 64,
"flowinfo": 1,
"scope_id": 1,
},
],
"name": "eth0",
},
{
"auto": True,
"index": 2,
"default": False,
"enabled": True,
"ipv4": [{"address": "192.168.1.5", "network_prefix": 23}],
"ipv6": [],
"name": "eth1",
},
{
"auto": True,
"index": 3,
"default": False,
"enabled": True,
"ipv4": [{"address": "172.16.1.5", "network_prefix": 23}],
"ipv6": [
{
"address": "fe80::dead:beef:dead:beef",
"network_prefix": 64,
"flowinfo": 1,
"scope_id": 3,
}
],
"name": "eth2",
},
{
"auto": False,
"index": 4,
"default": False,
"enabled": False,
"ipv4": [{"address": "169.254.3.2", "network_prefix": 16}],
"ipv6": [],
"name": "vtun0",
},
]
async def test_async_get_announce_addresses(hass: HomeAssistant) -> None:
"""Test addresses for mDNS/etc announcement."""
first_ip = "172.16.1.5"
with patch(
"homeassistant.components.network.async_get_source_ip",
return_value=first_ip,
), patch(
"homeassistant.components.network.async_get_adapters",
return_value=_ADAPTERS_WITH_MANUAL_CONFIG,
):
actual = await network.async_get_announce_addresses(hass)
assert actual[0] == first_ip and actual == [
first_ip,
"2001:db8::",
"fe80::1234:5678:9abc:def0",
"192.168.1.5",
"fe80::dead:beef:dead:beef",
]
first_ip = "192.168.1.5"
with patch(
"homeassistant.components.network.async_get_source_ip",
return_value=first_ip,
), patch(
"homeassistant.components.network.async_get_adapters",
return_value=_ADAPTERS_WITH_MANUAL_CONFIG,
):
actual = await network.async_get_announce_addresses(hass)
assert actual[0] == first_ip and actual == [
first_ip,
"2001:db8::",
"fe80::1234:5678:9abc:def0",
"172.16.1.5",
"fe80::dead:beef:dead:beef",
]
async def test_async_get_announce_addresses_no_source_ip(hass: HomeAssistant) -> None:
"""Test addresses for mDNS/etc announcement without source ip."""
with patch(
"homeassistant.components.network.async_get_source_ip",
return_value=None,
), patch(
"homeassistant.components.network.async_get_adapters",
return_value=_ADAPTERS_WITH_MANUAL_CONFIG,
):
actual = await network.async_get_announce_addresses(hass)
assert actual == [
"2001:db8::",
"fe80::1234:5678:9abc:def0",
"192.168.1.5",
"172.16.1.5",
"fe80::dead:beef:dead:beef",
]

30
tests/components/zeroconf/test_init.py
View file

@ -1,5 +1,4 @@
"""Test Zeroconf component setup process."""
from ipaddress import ip_address
from typing import Any
from unittest.mock import call, patch
@ -13,11 +12,7 @@ from zeroconf import (
from zeroconf.asyncio import AsyncServiceInfo
from homeassistant.components import zeroconf
from homeassistant.components.zeroconf import (
CONF_DEFAULT_INTERFACE,
CONF_IPV6,
_get_announced_addresses,
)
from homeassistant.components.zeroconf import CONF_DEFAULT_INTERFACE, CONF_IPV6
from homeassistant.const import (
EVENT_COMPONENT_LOADED,
EVENT_HOMEASSISTANT_START,
@ -1202,29 +1197,6 @@ async def test_async_detect_interfaces_setting_empty_route_freebsd(
)
async def test_get_announced_addresses(
hass: HomeAssistant, mock_async_zeroconf: None
) -> None:
"""Test addresses for mDNS announcement."""
expected = {
ip_address(ip).packed
for ip in [
"fe80::1234:5678:9abc:def0",
"2001:db8::",
"192.168.1.5",
"fe80::dead:beef:dead:beef",
"172.16.1.5",
]
}
first_ip = ip_address("172.16.1.5").packed
actual = _get_announced_addresses(_ADAPTERS_WITH_MANUAL_CONFIG, first_ip)
assert actual[0] == first_ip and set(actual) == expected
first_ip = ip_address("192.168.1.5").packed
actual = _get_announced_addresses(_ADAPTERS_WITH_MANUAL_CONFIG, first_ip)
assert actual[0] == first_ip and set(actual) == expected
_ADAPTER_WITH_DEFAULT_ENABLED_AND_IPV6 = [
{
"auto": True,