T8454: fix VRF-bind port availability check in service_https

When service https is configured with a listen-address on a VRF
interface, adding or changing the vrf option causes the commit to
fail with "TCP port 443 is used by another service!" even though
no conflict exists.

Root cause: check_port_availability() performs a socket.bind() in
the default network namespace. When the listen-address belongs to
a VRF interface the address is unreachable from the default
namespace, so the bind fails with OSError which is misinterpreted
as "port in use". The secondary check via is_listen_port_bind_service()
also fails because psutil cannot see sockets bound inside a VRF.

Fix: add an optional vrf parameter to check_port_availability().
When set, the test socket is bound to the VRF master device via
SO_BINDTODEVICE before the bind() call, so that the address is
resolved in the correct L3 domain. This is done in-process (no
subprocess) and works for both IPv4 and IPv6.

service_https verify() now passes the configured VRF (if any) to
check_port_availability(). Non-VRF configurations are unaffected.

Add smoke test for HTTPS with listen-address inside a VRF.

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

3 files changed, 53 insertions, 10 deletions

diff --git a/python/vyos/utils/network.py b/python/vyos/utils/network.py
index 8985224f3..8544b2163 100644
--- a/python/vyos/utils/network.py
+++ b/python/vyos/utils/network.py
@@ -305,7 +305,9 @@ def mac2eui64(mac, prefix=None):
         except:  # pylint: disable=bare-except
             return
 
-def check_port_availability(address: str=None, port: int=0, protocol: str='tcp') -> bool:
+
+def check_port_availability(address: str = None, port: int = 0,
+                            protocol: str = 'tcp', vrf: str = None) -> bool:
     """
     Check if given port is available and not used by any service.
 
@@ -315,7 +317,9 @@ def check_port_availability(address: str=None, port: int=0, protocol: str='tcp')
     Args:
       address: IPv4 or IPv6 address - if None, checks on all interfaces
       port:  TCP/UDP port number.
-
+      vrf:   VRF name to test the bind in - when set, the socket is bound
+             to the VRF master device via SO_BINDTODEVICE so that the
+             port check runs in the correct L3 domain.
 
     Returns:
       False if a port is busy or IP address does not exists
@@ -348,12 +352,15 @@ def check_port_availability(address: str=None, port: int=0, protocol: str='tcp')
         try:
             with socket.socket(family, socktype, proto) as s:
                 s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+                # When a VRF is specified, bind the socket to the VRF master
+                # device so the address is resolved in that VRF's L3 domain.
+                if vrf:
+                    s.setsockopt(socket.SOL_SOCKET, socket.SO_BINDTODEVICE,
+                                 (vrf + '\0').encode())
                 s.bind(sockaddr)
-                # port is free to use
-                return True
+                return True # port is free to use
         except OSError:
-            # port is already in use
-            return False
+            return False # port is already in use
 
     # if we reach this point, no socket was tested and we assume the port is
     # already in use - better safe then sorry
diff --git a/smoketest/scripts/cli/test_service_https.py b/smoketest/scripts/cli/test_service_https.py
index 4585bb0bd..b29be9eda 100755
--- a/smoketest/scripts/cli/test_service_https.py
+++ b/smoketest/scripts/cli/test_service_https.py
@@ -28,6 +28,7 @@ from base_vyostest_shim import ignore_warning
 from vyos.utils.file import read_file
 from vyos.utils.file import write_file
 from vyos.utils.process import call
+from vyos.utils.process import cmd
 from vyos.utils.process import process_named_running
 from vyos.xml_ref import default_value
 
@@ -177,6 +178,35 @@ class TestHTTPSService(VyOSUnitTestSHIM.TestCase):
 
         self.assertEqual(res, set(test_addr))
 
+    def test_listen_address_vrf(self):
+        # Verify that HTTPS service can be configured with a listen-address
+        # inside a VRF.  Regression test: the port availability check used to
+        # fail because it ran in the default namespace where the VRF address
+        # is unreachable.
+        vrf = 'mgmt'
+        vrf_table = '1337'
+        test_addr = '192.0.2.1'
+        test_prefix = f'{test_addr}/26'
+        interface = 'dum0'
+
+        self.cli_set(['interfaces', 'dummy', interface, 'address', test_prefix])
+        self.cli_set(['interfaces', 'dummy', interface, 'vrf', vrf])
+        self.cli_set(['vrf', 'name', vrf, 'table', vrf_table])
+
+        self.cli_set(
+            base_path + ['api', 'keys', 'id', 'key-01', 'key', 'MySuperSecretVyOS']
+        )
+        self.cli_set(base_path + ['listen-address', test_addr])
+        self.cli_set(base_path + ['vrf', vrf])
+        self.cli_commit()
+
+        # Verify nginx is running inside the VRF
+        tmp = cmd(f'ip vrf pids {vrf}')
+        self.assertIn(PROCESS_NAME, tmp)
+
+        self.cli_delete(['interfaces', 'dummy', interface])
+        self.cli_delete(['vrf', 'name', vrf])
+
     def test_certificate(self):
         cert_name = 'test_https'
         dh_name = 'dh-test'
diff --git a/src/conf_mode/service_https.py b/src/conf_mode/service_https.py
index 59fb90bd1..13a4930fd 100755
--- a/src/conf_mode/service_https.py
+++ b/src/conf_mode/service_https.py
@@ -113,12 +113,18 @@ def verify(https):
     if 'listen_address' in https:
         listen_address = https['listen_address']
 
-    for address in listen_address:
-        if not check_port_availability(address, port, 'tcp') and not is_listen_port_bind_service(port, 'nginx'):
-            raise ConfigError(f'TCP port "{port}" is used by another service!')
-
     verify_vrf(https)
 
+    vrf = https.get('vrf', None)
+    for address in listen_address:
+        if (not check_port_availability(address, port, 'tcp', vrf=vrf)
+            and not is_listen_port_bind_service(port, 'nginx')):
+            vrf_error_msg = ''
+            if vrf:
+                vrf_error_msg = f' in vrf "{vrf}"'
+            raise ConfigError(f'TCP port "{port}"{vrf_error_msg} is already ' \
+                               'used by another service!')
+
     # Verify API server settings, if present
     if 'api' in https:
         keys = dict_search('api.keys.id', https)