Merge "Add ethertype validator for custom ethertype validation"

neutron_lib/api/validators/__init__.py

@@ -16,6 +16,7 @@ import inspect
 import re
 
 import netaddr
+from oslo_config import cfg
 from oslo_log import log as logging
 from oslo_utils import netutils
 from oslo_utils import strutils
@@ -1122,11 +1123,44 @@ def validate_subnet_service_types(service_types, valid_values=None):
                 raise n_exc.InvalidSubnetServiceType(service_type=service_type)
 
 
+def validate_ethertype(ethertype, valid_values=None):
+    """Validates ethertype is a valid ethertype.
+
+    :param ethertype: Ethertype to validate
+    :returns: None if data is a valid ethertype.  Otherwise a human-readable
+        message indicating that the data is not a valid ethertype.
+    """
+    if cfg.CONF.sg_filter_ethertypes:
+        try:
+            if isinstance(ethertype, six.string_types):
+                ethertype = int(ethertype, 16)
+            if (isinstance(ethertype, six.integer_types) and
+                    constants.ETHERTYPE_MIN <= ethertype and
+                    ethertype <= constants.ETHERTYPE_MAX):
+                return None
+        except ValueError:
+            pass
+        msg = _("Ethertype %s is not a two octet hexadecimal "
+                "number.") % ethertype
+        LOG.debug(msg)
+        return msg
+    else:
+        if ethertype in constants.VALID_ETHERTYPES:
+            return None
+        valids = ','.join(map(str, constants.VALID_ETHERTYPES))
+        msg = _("Ethertype %(ethertype)s is not a valid ethertype, must be "
+                "one of %(valid_ethertypes)s.") % {'ethertype': ethertype,
+                                                   'valid_ethertypes': valids}
+        LOG.debug(msg)
+        return msg
+
+
 # Dictionary that maintains a list of validation functions
 validators = {'type:dict': validate_dict,
               'type:dict_or_none': validate_dict_or_none,
               'type:dict_or_empty': validate_dict_or_empty,
               'type:dict_or_nodata': validate_dict_or_nodata,
+              'type:ethertype': validate_ethertype,
               'type:fixed_ips': validate_fixed_ips,
               'type:hostroutes': validate_hostroutes,
               'type:ip_address': validate_ip_address,

neutron_lib/constants.py

@@ -563,6 +563,9 @@ IP_ALLOWED_VERSIONS = [IP_VERSION_4, IP_VERSION_6]
 PORT_RANGE_MIN = 1
 PORT_RANGE_MAX = 65535
 
+ETHERTYPE_MIN = 0
+ETHERTYPE_MAX = 65535
+
 DHCPV6_CLIENT_PORT = 546
 
 # Configuration values for accept_ra sysctl, copied from linux kernel

neutron_lib/tests/unit/api/validators/test_validators.py

@@ -1107,6 +1107,50 @@ class TestAttributeValidation(base.BaseTestCase):
         ]
         self.assertIsNone(validators.validate_subports(body))
 
+    @mock.patch('oslo_config.cfg.CONF')
+    def test_validate_ethertype_valid_string(self, CONF):
+        CONF.sg_filter_ethertypes = False
+        self.assertIsNone(validators.validate_ethertype('IPv4'))
+        self.assertIsNone(validators.validate_ethertype('IPv6'))
+
+    @mock.patch('oslo_config.cfg.CONF')
+    def test_validate_ethertype_invalid_string(self, CONF):
+        CONF.sg_filter_ethertypes = False
+        self.assertEqual(('Ethertype 0x4008 is not a valid ethertype, must be '
+                          'one of IPv4,IPv6.'),
+                         validators.validate_ethertype('0x4008'))
+
+    @mock.patch('oslo_config.cfg.CONF')
+    def test_validate_ethertype_extended_valid_string(self, CONF):
+        CONF.sg_filter_ethertypes = True
+        self.assertIsNone(validators.validate_ethertype('0x4008'))
+
+    @mock.patch('oslo_config.cfg.CONF')
+    def test_validate_ethertype_extended_valid_hexint(self, CONF):
+        CONF.sg_filter_ethertypes = True
+        self.assertIsNone(validators.validate_ethertype(0x4008))
+
+    @mock.patch('oslo_config.cfg.CONF')
+    def test_validate_ethertype_extended_invalid_negative(self, CONF):
+        CONF.sg_filter_ethertypes = True
+        self.assertEqual(("Ethertype -16392 is not a two octet "
+                          "hexadecimal number."),
+                         validators.validate_ethertype('-0x4008'))
+
+    @mock.patch('oslo_config.cfg.CONF')
+    def test_validate_ethertype_extended_invalid_nonhex(self, CONF):
+        CONF.sg_filter_ethertypes = True
+        self.assertEqual(("Ethertype invalid is not a two octet "
+                          "hexadecimal number."),
+                         validators.validate_ethertype('invalid'))
+
+    @mock.patch('oslo_config.cfg.CONF')
+    def test_validate_ethertype_extended_invalid_toobig(self, CONF):
+        CONF.sg_filter_ethertypes = True
+        self.assertEqual(("Ethertype 3735928559 is not a two octet "
+                          "hexadecimal number."),
+                         validators.validate_ethertype('0xdeadbeef'))
+
 
 class TestValidateIPSubnetNone(base.BaseTestCase):
 

releasenotes/notes/ethertype_validator-2d608a46c237e214.yaml

@@ -0,0 +1,5 @@
+---
+features:
+  - A new API validation type ``type:ethertype`` has been added and validates
+    ethertypes either as a valid two byte octet or as 'IPv4' and 'IPv6' based on
+    the sg_filter_ethertypes configuration setting.