[socket] Add an IPv4 allow list (#20025)

Co-authored-by: J. Nick Koston <nick@home-assistant.io>

tests/components/socket/test_ipv4_allow_host.cpp

@@ -0,0 +1,75 @@
+#include <gtest/gtest.h>
+
+#include <iterator>
+
+#include "esphome/components/socket/ipv4_allow.h"
+#include "esphome/components/socket/socket.h"
+
+#ifdef USE_HOST
+
+namespace esphome::socket::testing {
+
+// The size_t count packs into the pointer's padding; no RAM over a uint8_t.
+static_assert(sizeof(Ipv4Allow) == 2 * sizeof(void *), "unexpected padding in Ipv4Allow");
+
+// 192.168.175.20/32 and 192.168.175.0/24, network order, host bits cleared,
+// mirroring what add_ipv4_allow emits.
+static const Ipv4AllowEntry ENTRIES[] = {
+    {htonl(0xC0A8AF14), htonl(0xFFFFFFFF)},
+    {htonl(0xC0A8AF00), htonl(0xFFFFFF00)},
+};
+
+// Runs the peer through the same parser production addresses go through.
+static bool allows_peer(const Ipv4Allow &list, const char *ip) {
+  struct sockaddr_storage peer {};
+  EXPECT_NE(set_sockaddr(reinterpret_cast<struct sockaddr *>(&peer), sizeof(peer), ip, 0), 0);
+  return list.allows(reinterpret_cast<const struct sockaddr *>(&peer));
+}
+
+TEST(Ipv4Allow, EmptyAllowsEveryPeer) {
+  Ipv4Allow list;
+  EXPECT_TRUE(list.allows(htonl(0xC0A8AF01)));
+  EXPECT_TRUE(allows_peer(list, "10.0.0.1"));
+  EXPECT_TRUE(allows_peer(list, "fe80::1"));
+}
+
+TEST(Ipv4Allow, MatchesHostAndNetworkEntries) {
+  Ipv4Allow list;
+  list.set(ENTRIES, std::size(ENTRIES));
+  EXPECT_TRUE(list.allows(htonl(0xC0A8AF14)));
+  EXPECT_TRUE(list.allows(htonl(0xC0A8AF01)));
+  EXPECT_TRUE(list.allows(htonl(0xC0A8AFFF)));
+  EXPECT_FALSE(list.allows(htonl(0xC0A8B001)));
+}
+
+TEST(Ipv4Allow, ChecksTheV4PeerInsideASockaddr) {
+  Ipv4Allow list;
+  list.set(ENTRIES, std::size(ENTRIES));
+  EXPECT_TRUE(allows_peer(list, "192.168.175.66"));
+  EXPECT_FALSE(allows_peer(list, "10.0.0.1"));
+}
+
+TEST(Ipv4Allow, UnwrapsAV4MappedIpv6Peer) {
+  Ipv4Allow list;
+  list.set(ENTRIES, std::size(ENTRIES));
+  EXPECT_TRUE(allows_peer(list, "::ffff:192.168.175.66"));
+  // A native IPv6 peer cannot match an IPv4 list.
+  EXPECT_FALSE(allows_peer(list, "fe80::1"));
+}
+
+TEST(Ipv4Allow, InstancesKeepIndependentLists) {
+  // One bridge per allow list; each instance points at its own entries.
+  static const Ipv4AllowEntry OTHER[] = {{htonl(0x0A000000), htonl(0xFF000000)}};
+  Ipv4Allow first;
+  Ipv4Allow second;
+  first.set(ENTRIES, std::size(ENTRIES));
+  second.set(OTHER, std::size(OTHER));
+  EXPECT_TRUE(first.allows(htonl(0xC0A8AF14)));
+  EXPECT_FALSE(second.allows(htonl(0xC0A8AF14)));
+  EXPECT_TRUE(second.allows(htonl(0x0A00002A)));
+  EXPECT_FALSE(first.allows(htonl(0x0A00002A)));
+}
+
+}  // namespace esphome::socket::testing
+
+#endif

tests/unit_tests/components/socket/test_socket_ipv4_allow.py

@@ -0,0 +1,46 @@
+"""Tests for the socket component's IPv4 allow list codegen helper."""
+
+from ipaddress import IPv4Address, IPv4Network
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from esphome.components import socket
+import esphome.config_validation as cv
+
+
+def test_network_order_swaps_to_sockaddr_value() -> None:
+    """The emitted uint32 must equal s_addr on the little endian targets."""
+    assert socket._network_order(IPv4Address("192.168.175.20")) == 0x14AFA8C0
+    assert socket._network_order(IPv4Address("255.255.255.0")) == 0x00FFFFFF
+    assert socket._network_order(IPv4Address("0.0.0.0")) == 0
+
+
+def test_add_ipv4_allow_emits_nothing_for_an_empty_list() -> None:
+    setter = MagicMock()
+    with patch.object(socket.cg, "add") as add:
+        socket.add_ipv4_allow(setter, [], "bridge")
+    add.assert_not_called()
+    setter.assert_not_called()
+
+
+def test_add_ipv4_allow_wires_the_setter_with_cleared_host_bits() -> None:
+    setter = MagicMock()
+    networks = [IPv4Network("192.168.175.33/24", strict=False)]
+    with (
+        patch.object(socket.cg, "add") as add,
+        patch.object(socket.cg, "progmem_array") as array,
+    ):
+        socket.add_ipv4_allow(setter, networks, "bridge")
+    rendered = str(array.call_args.args[1])
+    assert str(socket._network_order(IPv4Address("192.168.175.0"))) in rendered
+    assert str(socket._network_order(IPv4Address("255.255.255.0"))) in rendered
+    setter.assert_called_once_with(array.return_value, 1)
+    add.assert_called_once()
+
+
+def test_schema_caps_the_list_length() -> None:
+    """The sanity cap rejects a list past 255 entries."""
+    assert len(socket.IPV4_ALLOW_SCHEMA([f"10.0.{i}.0/24" for i in range(255)])) == 255
+    with pytest.raises(cv.Invalid):
+        socket.IPV4_ALLOW_SCHEMA([f"10.0.{i}.0/24" for i in range(256)])