Add api to the network integration to get ipv4 broadcast addresses (#56722)

This commit is contained in:
J. Nick Koston 2021-09-27 23:43:39 -05:00 committed by GitHub
parent aea754df5d
commit f93539ef4c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 72 additions and 1 deletions

View file

@ -1,7 +1,7 @@
"""The Network Configuration integration.""" """The Network Configuration integration."""
from __future__ import annotations from __future__ import annotations
from ipaddress import IPv4Address, IPv6Address from ipaddress import IPv4Address, IPv6Address, ip_interface
import logging import logging
import voluptuous as vol import voluptuous as vol
@ -17,6 +17,7 @@ from .const import (
ATTR_ADAPTERS, ATTR_ADAPTERS,
ATTR_CONFIGURED_ADAPTERS, ATTR_CONFIGURED_ADAPTERS,
DOMAIN, DOMAIN,
IPV4_BROADCAST_ADDR,
NETWORK_CONFIG_SCHEMA, NETWORK_CONFIG_SCHEMA,
) )
from .models import Adapter from .models import Adapter
@ -75,6 +76,26 @@ def async_only_default_interface_enabled(adapters: list[Adapter]) -> bool:
) )
@bind_hass
async def async_get_ipv4_broadcast_addresses(hass: HomeAssistant) -> set[IPv4Address]:
"""Return a set of broadcast addresses."""
broadcast_addresses: set[IPv4Address] = {IPv4Address(IPV4_BROADCAST_ADDR)}
adapters = await async_get_adapters(hass)
if async_only_default_interface_enabled(adapters):
return broadcast_addresses
for adapter in adapters:
if not adapter["enabled"]:
continue
for ip_info in adapter["ipv4"]:
interface = ip_interface(
f"{ip_info['address']}/{ip_info['network_prefix']}"
)
broadcast_addresses.add(
IPv4Address(interface.network.broadcast_address.exploded)
)
return broadcast_addresses
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up network for Home Assistant.""" """Set up network for Home Assistant."""

View file

@ -17,6 +17,7 @@ DEFAULT_CONFIGURED_ADAPTERS: list[str] = []
MDNS_TARGET_IP: Final = "224.0.0.251" MDNS_TARGET_IP: Final = "224.0.0.251"
PUBLIC_TARGET_IP: Final = "8.8.8.8" PUBLIC_TARGET_IP: Final = "8.8.8.8"
IPV4_BROADCAST_ADDR: Final = "255.255.255.255"
NETWORK_CONFIG_SCHEMA = vol.Schema( NETWORK_CONFIG_SCHEMA = vol.Schema(
{ {

View file

@ -1,4 +1,5 @@
"""Test the Network Configuration.""" """Test the Network Configuration."""
from ipaddress import IPv4Address
from unittest.mock import MagicMock, Mock, patch from unittest.mock import MagicMock, Mock, patch
import ifaddr import ifaddr
@ -552,3 +553,51 @@ async def test_async_get_source_ip_cannot_determine_target(hass, hass_storage):
await hass.async_block_till_done() await hass.async_block_till_done()
assert await network.async_get_source_ip(hass, MDNS_TARGET_IP) == "192.168.1.5" assert await network.async_get_source_ip(hass, MDNS_TARGET_IP) == "192.168.1.5"
async def test_async_get_ipv4_broadcast_addresses_default(hass, hass_storage):
"""Test getting ipv4 broadcast addresses when only the default address is enabled."""
hass_storage[STORAGE_KEY] = {
"version": STORAGE_VERSION,
"key": STORAGE_KEY,
"data": {ATTR_CONFIGURED_ADAPTERS: ["eth1"]},
}
with patch(
"homeassistant.components.network.util.socket.socket",
return_value=_mock_socket(["192.168.1.5"]),
), patch(
"homeassistant.components.network.util.ifaddr.get_adapters",
return_value=_generate_mock_adapters(),
):
assert await async_setup_component(hass, network.DOMAIN, {network.DOMAIN: {}})
await hass.async_block_till_done()
assert await network.async_get_ipv4_broadcast_addresses(hass) == {
IPv4Address("255.255.255.255")
}
async def test_async_get_ipv4_broadcast_addresses_multiple(hass, hass_storage):
"""Test getting ipv4 broadcast addresses when multiple adapters are enabled."""
hass_storage[STORAGE_KEY] = {
"version": STORAGE_VERSION,
"key": STORAGE_KEY,
"data": {ATTR_CONFIGURED_ADAPTERS: ["eth1", "vtun0"]},
}
with patch(
"homeassistant.components.network.util.socket.socket",
return_value=_mock_socket([_LOOPBACK_IPADDR]),
), patch(
"homeassistant.components.network.util.ifaddr.get_adapters",
return_value=_generate_mock_adapters(),
):
assert await async_setup_component(hass, network.DOMAIN, {network.DOMAIN: {}})
await hass.async_block_till_done()
assert await network.async_get_ipv4_broadcast_addresses(hass) == {
IPv4Address("255.255.255.255"),
IPv4Address("192.168.1.255"),
IPv4Address("169.254.255.255"),
}