vyos.utils: T8977: let a caller accept a non-local address as assigned

Whether an address may be bound is not the same question as whether it is
currently configured somewhere: the kernel permits binding a foreign address
once the non-local bind sysctl is set, which is how a failover address is
commonly handed over after the commit. Callers can now opt into that, the
default is unchanged.

A lookup no longer skips the VRF device itself, which carries the addresses
of its own L3 domain, and an interface name can be retrieved for an address
without the caller asking a yes/no question in the first place.

Address and range validation is split into separate checks, so a caller no
longer has to accept ranges where only an address is meaningful. The two
duplicated version checks are dropped in favour of the existing ones.

Co-authored-by: Christian Breunig <christian@breunig.cc>

1 files changed, 133 insertions, 1 deletions

diff --git a/src/tests/test_utils.py b/src/tests/test_utils.py
index 432672f06..8e6f320ee 100644
--- a/src/tests/test_utils.py
+++ b/src/tests/test_utils.py
@@ -13,7 +13,16 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 from unittest import TestCase
-from unittest.mock import patch
+from unittest.mock import patch, MagicMock
+
+
+def _make_psutil_net_conn(ip, port, pid):
+    c = MagicMock()
+    c.laddr.ip = ip
+    c.laddr.port = port
+    c.pid = pid
+    return c
+
 
 class TestVyOSUtils(TestCase):
     def test_cmdl_basic(self):
@@ -117,3 +126,126 @@ class TestVyOSUtils(TestCase):
         # duplicated values
         self.assertEqual(list_to_range_str([1, 1, 2, 2, 3, 3]), '1-3')
         self.assertEqual(list_to_range_str([5, 1, 2, 2, 3, 5]), '1-3,5')
+
+    def test_is_ip_range(self):
+        """A range is exactly two plain host addresses."""
+        from vyos.utils.network import is_ipv4_range
+        from vyos.utils.network import is_ipv6_range
+
+        self.assertTrue(is_ipv4_range('192.0.2.1-192.0.2.9'))
+        self.assertTrue(is_ipv6_range('2001:db8::1-2001:db8::9'))
+        # a nested range or a prefix as endpoint is not a range
+        self.assertFalse(is_ipv4_range('192.0.2.1-192.0.2.2-192.0.2.3'))
+        self.assertFalse(is_ipv4_range('192.0.2.0/24-192.0.2.9'))
+        self.assertFalse(is_ipv6_range('2001:db8::/64-2001:db8::9'))
+        self.assertFalse(is_ipv4_range('192.0.2.1'))
+
+    def test_is_addr_assigned_nonlocal_bind_ipv4(self):
+        """When net.ipv4.ip_nonlocal_bind=1, any valid IPv4 must return True."""
+        from vyos.utils.network import is_addr_assigned
+
+        with patch('vyos.utils.system.sysctl_read', return_value='1'):
+            self.assertTrue(is_addr_assigned('192.0.2.99', allow_nonlocal=True))
+
+    def test_is_addr_assigned_nonlocal_bind_ipv6(self):
+        from vyos.utils.network import is_addr_assigned
+
+        with patch('vyos.utils.system.sysctl_read', return_value='1'):
+            self.assertTrue(is_addr_assigned('2001:db8::1', allow_nonlocal=True))
+
+    def test_is_addr_assigned_nonlocal_bind_ipv4_off_unassigned(self):
+        """When sysctl is 0 and address is not assigned, must return False."""
+        from vyos.utils.network import is_addr_assigned
+
+        with patch('vyos.utils.system.sysctl_read', return_value='0'), patch(
+            'netifaces.interfaces', return_value=[]
+        ):
+            self.assertFalse(is_addr_assigned('192.0.2.99', allow_nonlocal=True))
+
+    def test_is_addr_assigned_nonlocal_bind_ipv6_off_unassigned(self):
+        """When sysctl is 0 and address is not assigned, must return False."""
+        from vyos.utils.network import is_addr_assigned
+
+        with patch('vyos.utils.system.sysctl_read', return_value='0'), patch(
+            'netifaces.interfaces', return_value=[]
+        ):
+            self.assertFalse(is_addr_assigned('2001:db8::1', allow_nonlocal=True))
+
+    def test_is_addr_assigned_vrf_filter(self):
+        """Address on a VRF slave should only match when vrf matches."""
+        from vyos.utils.network import is_addr_assigned
+
+        iface_cfg_red = {'master': 'red'}
+        iface_cfg_lo = {'master': None}  # loopback has no master
+
+        with patch('vyos.utils.system.sysctl_read', return_value='0'), patch(
+            'netifaces.interfaces', return_value=['lo', 'eth0']
+        ), patch(
+            'vyos.utils.network.get_interface_config',
+            side_effect=[iface_cfg_lo, iface_cfg_red],
+        ), patch(
+            'vyos.utils.network.is_intf_addr_assigned', return_value=True
+        ) as mock_is_assigned:
+            # vrf='red' → should find it on eth0
+            self.assertTrue(
+                is_addr_assigned('10.0.0.1', vrf='red', allow_nonlocal=True)
+            )
+            mock_is_assigned.assert_called_once_with('eth0', '10.0.0.1')
+
+    def test_is_addr_assigned_wrong_vrf(self):
+        """Address on VRF 'red' must not match when querying vrf='blue'."""
+        from vyos.utils.network import is_addr_assigned
+
+        iface_cfg = {'master': 'red'}
+        with patch('vyos.utils.system.sysctl_read', return_value='0'), patch(
+            'netifaces.interfaces', return_value=['eth0']
+        ), patch(
+            'vyos.utils.network.get_interface_config', return_value=iface_cfg
+        ), patch(
+            'vyos.utils.network.is_intf_addr_assigned', return_value=True
+        ):
+            self.assertFalse(
+                is_addr_assigned('10.0.0.1', vrf='blue', allow_nonlocal=True)
+            )
+
+    def test_is_listen_port_wildcard_ipv4(self):
+        """Service on 0.0.0.0 must match even when a specific address filter is given."""
+        from vyos.utils.network import is_listen_port_bind_service
+
+        conn = _make_psutil_net_conn('0.0.0.0', 443, 1234)
+        with patch('psutil.net_connections', return_value=[conn]), patch(
+            'psutil.Process'
+        ) as MockProc:
+            MockProc.return_value.name.return_value = 'haproxy'
+            self.assertTrue(
+                is_listen_port_bind_service(443, 'haproxy', address='194.0.2.1')
+            )
+
+    def test_is_listen_port_wildcard_ipv6(self):
+        """Service on :: must match even when an address filter is given."""
+        from vyos.utils.network import is_listen_port_bind_service
+
+        conn = _make_psutil_net_conn('::', 443, 1234)
+        with patch('psutil.net_connections', return_value=[conn]), patch(
+            'psutil.Process'
+        ) as MockProc:
+            MockProc.return_value.name.return_value = 'haproxy'
+            self.assertTrue(
+                is_listen_port_bind_service(443, 'haproxy', address='2001:db8::1')
+            )
+
+    def test_is_listen_port_specific_address_match(self):
+        """Service on specific IP must match only the exact address."""
+        from vyos.utils.network import is_listen_port_bind_service
+
+        conn = _make_psutil_net_conn('194.0.2.1', 443, 1234)
+        with patch('psutil.net_connections', return_value=[conn]), patch(
+            'psutil.Process'
+        ) as MockProc:
+            MockProc.return_value.name.return_value = 'haproxy'
+            self.assertTrue(
+                is_listen_port_bind_service(443, 'haproxy', address='194.0.2.1')
+            )
+            self.assertFalse(
+                is_listen_port_bind_service(443, 'haproxy', address='10.0.0.1')
+            )