Merge "[OVN] Sanitize the classless-static-route DHCP option" into stable/2024.1

Author: Zuul

neutron/common/ovn/constants.py

@@ -217,6 +217,10 @@
     'wpad',
     'tftp_server']
 
+OVN_MAP_TYPE_DHCP_OPTS = [
+    'classless_static_route',
+]
+
 # Special option for disabling DHCP via extra DHCP options
 DHCP_DISABLED_OPT = 'dhcp_disabled'
 

neutron/common/ovn/utils.py

@@ -250,6 +250,10 @@ def is_dhcp_option_quoted(opt_value):
     return opt_value.startswith('"') and opt_value.endswith('"')
 
 
+def is_dhcp_option_a_map(opt_value):
+    return opt_value.startswith('{') and opt_value.endswith('}')
+
+
 def get_lsp_dhcp_opts(port, ip_version):
     # Get dhcp options from Neutron port, for setting DHCP_Options row
     # in OVN.
@@ -288,6 +292,9 @@ def get_lsp_dhcp_opts(port, ip_version):
             if (opt in constants.OVN_STR_TYPE_DHCP_OPTS and
                     not is_dhcp_option_quoted(edo['opt_value'])):
                 edo['opt_value'] = '"%s"' % edo['opt_value']
+            elif (opt in constants.OVN_MAP_TYPE_DHCP_OPTS and
+                  not is_dhcp_option_a_map(edo['opt_value'])):
+                edo['opt_value'] = '{%s}' % edo['opt_value']
             lsp_dhcp_opts[opt] = edo['opt_value']
 
     return (lsp_dhcp_disabled, lsp_dhcp_opts)

neutron/plugins/ml2/drivers/ovn/mech_driver/ovsdb/ovn_client.py

@@ -194,6 +194,18 @@ def _get_subnet_dhcp_options_for_port(self, port, ip_version):
                         return opts
             return get_opts[0]
 
+    def _merge_map_dhcp_option(self, opt, port_opts, subnet_opts):
+        """Merge a port and subnet map DHCP option.
+
+        If a DHCP option exists in both port and subnet, the port
+        should inherit the values from the subnet.
+        """
+        port_opt = port_opts[opt]
+        subnet_opt = subnet_opts.get(opt)
+        if not subnet_opt:
+            return port_opt
+        return '{%s, %s}' % (subnet_opt[1:-1], port_opt[1:-1])
+
     def _get_port_dhcp_options(self, port, ip_version):
         """Return dhcp options for port.
 
@@ -226,6 +238,12 @@ def _get_port_dhcp_options(self, port, ip_version):
         if not lsp_dhcp_opts:
             return subnet_dhcp_options
 
+        # Check for map DHCP options
+        for opt in ovn_const.OVN_MAP_TYPE_DHCP_OPTS:
+            if opt in lsp_dhcp_opts:
+                lsp_dhcp_opts[opt] = self._merge_map_dhcp_option(
+                    opt, lsp_dhcp_opts, subnet_dhcp_options['options'])
+
         # This port has extra DHCP options defined, so we will create a new
         # row in DHCP_Options table for it.
         subnet_dhcp_options['options'].update(lsp_dhcp_opts)

neutron/tests/unit/common/ovn/test_utils.py

@@ -521,6 +521,20 @@ def test_get_lsp_dhcp_opts_for_domain_search(self):
         expected_options = {'domain_search_list': '"openstack.org,ovn.org"'}
         self.assertEqual(expected_options, options)
 
+    def test_get_lsp_dhcp_opts_sanitize_map(self):
+        opt = {'opt_name': 'classless-static-route',
+               'opt_value': '128.128.128.128/32,22.2.0.2',
+               'ip_version': 4}
+        port = {portbindings.VNIC_TYPE: portbindings.VNIC_NORMAL,
+                edo_ext.EXTRADHCPOPTS: [opt]}
+        dhcp_disabled, options = utils.get_lsp_dhcp_opts(port, 4)
+        self.assertFalse(dhcp_disabled)
+        # Assert option got translated to "classless_static_route" and
+        # the value is a map (wrapped with {})
+        expected_options = {
+            'classless_static_route': '{128.128.128.128/32,22.2.0.2}'}
+        self.assertEqual(expected_options, options)
+
 
 class TestGetDhcpDnsServers(base.BaseTestCase):
 

neutron/tests/unit/plugins/ml2/drivers/ovn/mech_driver/test_mech_driver.py

@@ -4009,6 +4009,44 @@ def test__get_subnet_dhcp_options_for_port_v6_dhcp_disabled(self):
         self._test__get_subnet_dhcp_options_for_port(ip_version=6,
                                                      enable_dhcp=False)
 
+    def test_get_port_dhcp_options_classless_static_route(self):
+        port = {
+            'id': 'foo-port',
+            'device_owner': 'compute:None',
+            'fixed_ips': [{'subnet_id': 'foo-subnet',
+                           'ip_address': '10.0.0.11'}],
+            'extra_dhcp_opts': [
+                {'ip_version': 4, 'opt_name': 'classless-static-route',
+                 'opt_value': '128.128.128.128/32,22.2.0.2'}]}
+
+        self.mech_driver._ovn_client._get_subnet_dhcp_options_for_port = (
+            mock.Mock(
+                return_value=({
+                    'cidr': '10.0.0.0/24',
+                    'external_ids': {'subnet_id': 'foo-subnet'},
+                    'options': {
+                        'classless_static_route':
+                            '{169.254.169.254/32,10.0.0.2}',},
+                    'uuid': 'foo-uuid'})))
+
+        # Expect both the subnet and port classless_static_route
+        # to be merged
+        expected_routes = ('{169.254.169.254/32,10.0.0.2, '
+                           '128.128.128.128/32,22.2.0.2}')
+        expected_dhcp_options = {
+            'cidr': '10.0.0.0/24',
+            'external_ids': {'subnet_id': 'foo-subnet',
+                             'port_id': 'foo-port'},
+            'options': {'classless_static_route': expected_routes}
+        }
+
+        self.mech_driver.nb_ovn.add_dhcp_options.return_value = 'foo-val'
+        dhcp_options = self.mech_driver._ovn_client._get_port_dhcp_options(
+            port, 4)
+        self.assertEqual({'cmd': 'foo-val'}, dhcp_options)
+        self.mech_driver.nb_ovn.add_dhcp_options.assert_called_once_with(
+            'foo-subnet', port_id='foo-port', **expected_dhcp_options)
+
 
 class TestOVNMechanismDriverSecurityGroup(MechDriverSetupBase,
         test_security_group.Ml2SecurityGroupsTestCase):