Validate interface name length in CLI

config/main.py

@@ -34,7 +34,8 @@
 from sonic_yang_cfg_generator import SonicYangCfgDbGenerator
 from utilities_common import util_base
 from swsscommon import swsscommon
-from swsscommon.swsscommon import SonicV2Connector, ConfigDBConnector, ConfigDBPipeConnector
+from swsscommon.swsscommon import SonicV2Connector, ConfigDBConnector, ConfigDBPipeConnector, \
+                                isInterfaceNameValid, IFACE_NAME_MAX_LEN
 from utilities_common.db import Db
 from utilities_common.intf_filter import parse_interface_in_filter
 from utilities_common import bgp_util
@@ -106,7 +107,6 @@
 
 CFG_PORTCHANNEL_PREFIX = "PortChannel"
 CFG_PORTCHANNEL_PREFIX_LEN = 11
-CFG_PORTCHANNEL_NAME_TOTAL_LEN_MAX = 15
 CFG_PORTCHANNEL_MAX_VAL = 9999
 CFG_PORTCHANNEL_NO="<0-9999>"
 
@@ -439,7 +439,7 @@ def is_portchannel_name_valid(portchannel_name):
     if (portchannel_name[CFG_PORTCHANNEL_PREFIX_LEN:].isdigit() is False or
           int(portchannel_name[CFG_PORTCHANNEL_PREFIX_LEN:]) > CFG_PORTCHANNEL_MAX_VAL) :
         return False
-    if len(portchannel_name) > CFG_PORTCHANNEL_NAME_TOTAL_LEN_MAX:
+    if not isInterfaceNameValid(portchannel_name):
         return False
     return True
 
@@ -2481,8 +2481,8 @@ def add_portchannel(ctx, portchannel_name, min_links, fallback, fast_rate):
     db = ValidatedConfigDBConnector(ctx.obj['db'])
     if ADHOC_VALIDATION:
         if is_portchannel_name_valid(portchannel_name) != True:
-            ctx.fail("{} is invalid!, name should have prefix '{}' and suffix '{}'"
-                    .format(portchannel_name, CFG_PORTCHANNEL_PREFIX, CFG_PORTCHANNEL_NO))
+            ctx.fail("{} is invalid!, name should have prefix '{}' and suffix '{}' and its length should not exceed {} characters"
+                    .format(portchannel_name, CFG_PORTCHANNEL_PREFIX, CFG_PORTCHANNEL_NO, IFACE_NAME_MAX_LEN))
         if is_portchannel_present_in_db(db, portchannel_name):
             ctx.fail("{} already exists!".format(portchannel_name)) # TODO: MISSING CONSTRAINT IN YANG MODEL
 
@@ -6031,8 +6031,8 @@ def add_vrf(ctx, vrf_name):
     config_db = ValidatedConfigDBConnector(ctx.obj['config_db'])
     if not vrf_name.startswith("Vrf") and not (vrf_name == 'mgmt') and not (vrf_name == 'management'):
         ctx.fail("'vrf_name' must begin with 'Vrf' or named 'mgmt'/'management' in case of ManagementVRF.")
-    if len(vrf_name) > 15:
-        ctx.fail("'vrf_name' is too long!")
+    if not isInterfaceNameValid(vrf_name):
+        ctx.fail("'vrf_name' length should not exceed {} characters".format(IFACE_NAME_MAX_LEN))
     if is_vrf_exists(config_db, vrf_name):
         ctx.fail("VRF {} already exists!".format(vrf_name))
     elif (vrf_name == 'mgmt' or vrf_name == 'management'):
@@ -6051,8 +6051,8 @@ def del_vrf(ctx, vrf_name):
     config_db = ValidatedConfigDBConnector(ctx.obj['config_db'])
     if not vrf_name.startswith("Vrf") and not (vrf_name == 'mgmt') and not (vrf_name == 'management'):
         ctx.fail("'vrf_name' must begin with 'Vrf' or named 'mgmt'/'management' in case of ManagementVRF.")
-    if len(vrf_name) > 15:
-        ctx.fail("'vrf_name' is too long!")
+    if not isInterfaceNameValid(vrf_name):
+        ctx.fail("'vrf_name' length should not exceed {} characters".format((IFACE_NAME_MAX_LEN)))
     syslog_table = config_db.get_table("SYSLOG_SERVER")
     syslog_vrf_dev = "mgmt" if vrf_name == "management" else vrf_name
     for syslog_entry, syslog_data in syslog_table.items():
@@ -7082,8 +7082,8 @@ def add_loopback(ctx, loopback_name):
     config_db = ValidatedConfigDBConnector(ctx.obj['db'])
     if ADHOC_VALIDATION:
         if is_loopback_name_valid(loopback_name) is False:
-            ctx.fail("{} is invalid, name should have prefix '{}' and suffix '{}' "
-                    .format(loopback_name, CFG_LOOPBACK_PREFIX, CFG_LOOPBACK_NO))
+            ctx.fail("{} is invalid, name should have prefix '{}' and suffix '{}' and should not exceed {} characters"
+                    .format(loopback_name, CFG_LOOPBACK_PREFIX, CFG_LOOPBACK_NO, IFACE_NAME_MAX_LEN))
 
         lo_intfs = [k for k, v in config_db.get_table('LOOPBACK_INTERFACE').items() if type(k) != tuple]
         if loopback_name in lo_intfs:
@@ -7830,6 +7830,8 @@ def add_subinterface(ctx, subinterface_name, vid):
 
         if interface_alias is None:
             ctx.fail("{} invalid subinterface".format(interface_alias))
+        if not isInterfaceNameValid(interface_alias):
+            ctx.fail("Subinterface name length should not exceed {} characters".format(IFACE_NAME_MAX_LEN))
 
         if interface_alias.startswith("Po") is True:
             intf_table_name = CFG_PORTCHANNEL_PREFIX

config/vxlan.py

@@ -3,6 +3,7 @@
 
 from jsonpatch import JsonPatchConflict
 from .validated_config_db_connector import ValidatedConfigDBConnector
+from swsscommon.swsscommon import isInterfaceNameValid, IFACE_NAME_MAX_LEN
 
 ADHOC_VALIDATION = True
 #
@@ -23,7 +24,9 @@ def add_vxlan(db, vxlan_name, src_ip):
 
     if ADHOC_VALIDATION:
         if not clicommon.is_ipaddress(src_ip):
-            ctx.fail("{} invalid src ip address".format(src_ip))  
+            ctx.fail("{} invalid src ip address".format(src_ip))
+        if not isInterfaceNameValid(vxlan_name):
+            ctx.fail("'vxlan_name' length should not exceed {} characters".format(IFACE_NAME_MAX_LEN))
 
     vxlan_keys = db.cfgdb.get_keys('VXLAN_TUNNEL')
     if not vxlan_keys:
@@ -32,7 +35,7 @@ def add_vxlan(db, vxlan_name, src_ip):
       vxlan_count = len(vxlan_keys)
 
     if(vxlan_count > 0):
-        ctx.fail("VTEP already configured.")  
+        ctx.fail("VTEP already configured.")
 
     fvs = {'src_ip': src_ip}
     try:
@@ -59,7 +62,7 @@ def del_vxlan(db, vxlan_name):
       vxlan_count = len(vxlan_keys)
 
     if(vxlan_count > 0):
-        ctx.fail("Please delete the EVPN NVO configuration.")  
+        ctx.fail("Please delete the EVPN NVO configuration.")
 
     vxlan_keys = db.cfgdb.get_keys("VXLAN_TUNNEL_MAP|*")
     if not vxlan_keys:
@@ -68,7 +71,7 @@ def del_vxlan(db, vxlan_name):
       vxlan_count = len(vxlan_keys)
 
     if(vxlan_count > 0):
-        ctx.fail("Please delete all VLAN VNI mappings.")  
+        ctx.fail("Please delete all VLAN VNI mappings.")
 
     vnet_table = db.cfgdb.get_table('VNET')
     vnet_keys = vnet_table.keys()
@@ -100,7 +103,7 @@ def add_vxlan_evpn_nvo(db, nvo_name, vxlan_name):
       vxlan_count = len(vxlan_keys)
 
     if(vxlan_count > 0):
-        ctx.fail("EVPN NVO already configured")  
+        ctx.fail("EVPN NVO already configured")
 
     if len(db.cfgdb.get_entry('VXLAN_TUNNEL', vxlan_name)) == 0:
         ctx.fail("VTEP {} not configured".format(vxlan_name))
@@ -317,4 +320,3 @@ def del_vxlan_map_range(db, vxlan_name, vlan_start, vlan_end, vni_start):
            config_db.set_entry('VXLAN_TUNNEL_MAP', mapname, None)
        except JsonPatchConflict as e:
            ctx.fail("Invalid ConfigDB. Error: {}".format(e))
-

tests/config_test.py

@@ -2847,6 +2847,13 @@ def test_add_loopback_with_invalid_name_adhoc_validation(self):
         assert result.exit_code != 0
         assert "Error: Loopbax1 is invalid, name should have prefix 'Loopback' and suffix '<0-999>'" in result.output
 
+        result = runner.invoke(config.config.commands["loopback"].commands["add"], ["Loopback0000"], obj=obj)
+        print(result.exit_code)
+        print(result.output)
+        assert result.exit_code != 0
+        assert "Error: Loopback0000 is invalid, name should have prefix 'Loopback' and suffix '<0-999>' and " \
+            "should not exceed 15 characters" in result.output
+
     def test_del_nonexistent_loopback_adhoc_validation(self):
         config.ADHOC_VALIDATION = True
         runner = CliRunner()

tests/portchannel_test.py

@@ -34,7 +34,7 @@ def test_add_portchannel_with_invalid_name_yang_validation(self):
         print(result.output)
         assert result.exit_code != 0
         assert "Error: PortChan005 is invalid!, name should have prefix 'PortChannel' and suffix '<0-9999>'" in result.output
-    
+
     def test_add_portchannel_with_invalid_name_adhoc_validation(self):
         config.ADHOC_VALIDATION = True
         runner = CliRunner()
@@ -46,7 +46,15 @@ def test_add_portchannel_with_invalid_name_adhoc_validation(self):
         print(result.exit_code)
         print(result.output)
         assert result.exit_code != 0
-        assert "Error: PortChan005 is invalid!, name should have prefix 'PortChannel' and suffix '<0-9999>'" in result.output
+        assert "Error: PortChan005 is invalid!, name should have prefix 'PortChannel' and suffix '<0-9999>' " \
+            "and its length should not exceed 15 characters" in result.output
+
+        result = runner.invoke(config.config.commands["portchannel"].commands["add"], ["PortChanl00000"], obj=obj)
+        print(result.exit_code)
+        print(result.output)
+        assert result.exit_code != 0
+        assert "Error: PortChanl00000 is invalid!, name should have prefix 'PortChannel' and suffix '<0-9999>' and " \
+            "its length should not exceed 15 characters" in result.output
 
     @patch("config.validated_config_db_connector.ValidatedConfigDBConnector.validated_set_entry", mock.Mock(side_effect=JsonPatchConflict))
     @patch("validated_config_db_connector.device_info.is_yang_config_validation_enabled", mock.Mock(return_value=True))
@@ -119,7 +127,7 @@ def test_add_portchannel_with_invalid_fast_rate(self, fast_rate):
         runner = CliRunner()
         db = Db()
         obj = {'db':db.cfgdb}
-        
+
         # add a portchannel with invalid fats rate
         result = runner.invoke(config.config.commands["portchannel"].commands["add"], ["PortChannel0005", "--fast-rate", fast_rate], obj=obj)
         print(result.exit_code)

tests/subintf_test.py

@@ -24,7 +24,7 @@ def test_add_del_subintf_short_name(self):
         runner = CliRunner()
         db = Db()
         obj = {'db':db.cfgdb}
-        
+
         result = runner.invoke(config.config.commands["subinterface"].commands["add"], ["Eth0.102", "1002"], obj=obj)
         print(result.exit_code, result.output)
         assert result.exit_code == 0
@@ -53,35 +53,23 @@ def test_add_del_subintf_with_long_name(self):
         runner = CliRunner()
         db = Db()
         obj = {'db':db.cfgdb}
-        
+
         result = runner.invoke(config.config.commands["subinterface"].commands["add"], ["Ethernet0.102"], obj=obj)
         print(result.exit_code, result.output)
         assert result.exit_code == 0
         assert ('Ethernet0.102') in db.cfgdb.get_table('VLAN_SUB_INTERFACE')
         assert db.cfgdb.get_table('VLAN_SUB_INTERFACE')['Ethernet0.102']['admin_status'] == 'up'
 
-        result = runner.invoke(config.config.commands["subinterface"].commands["add"], ["PortChannel0004.104"], obj=obj)
-        print(result.exit_code, result.output)
-        assert result.exit_code == 0
-        assert ('PortChannel0004.104') in db.cfgdb.get_table('VLAN_SUB_INTERFACE')
-        assert db.cfgdb.get_table('VLAN_SUB_INTERFACE')['PortChannel0004.104']['admin_status'] == 'up'
-
         result = runner.invoke(config.config.commands["subinterface"].commands["del"], ["Ethernet0.102"], obj=obj)
         print(result.exit_code, result.output)
         assert result.exit_code == 0
         assert ('Ethernet0.102') not in db.cfgdb.get_table('VLAN_SUB_INTERFACE')
 
-        result = runner.invoke(config.config.commands["subinterface"].commands["del"], ["PortChannel0004.104"], obj=obj)
-        print(result.exit_code, result.output)
-        assert result.exit_code == 0
-        assert ('PortChannel0004.104') not in db.cfgdb.get_table('VLAN_SUB_INTERFACE')
-
-
     def test_add_existing_subintf_again(self):
         runner = CliRunner()
         db = Db()
         obj = {'db':db.cfgdb}
-        
+
         result = runner.invoke(config.config.commands["subinterface"].commands["add"], ["Ethernet0.102"], obj=obj)
         print(result.exit_code, result.output)
         assert result.exit_code == 0
@@ -104,7 +92,7 @@ def test_delete_non_existing_subintf(self):
         runner = CliRunner()
         db = Db()
         obj = {'db':db.cfgdb}
-        
+
         result = runner.invoke(config.config.commands["subinterface"].commands["del"], ["Ethernet0.102"], obj=obj)
         print(result.exit_code, result.output)
         assert result.exit_code != 0

tests/vlan_test.py

@@ -574,6 +574,15 @@ def test_config_vlan_del_member_with_invalid_port(self):
         assert result.exit_code != 0
         assert "Error: Invalid VLAN ID 4097 (2-4094)" in result.output
 
+    def test_config_vlan_add_member_with_invalid_long_name(self):
+        runner = CliRunner()
+        result = runner.invoke(config.config.commands["vlan"].commands["member"].commands["add"],
+                               ["123456789012", "Ethernet4"])
+        print(result.exit_code)
+        print(result.output)
+        assert result.exit_code != 0
+        assert "Error: Invalid VLAN ID 123456789012 (2-4094)" in result.output
+
     def test_config_vlan_add_member_with_nonexist_vlanid(self):
         runner = CliRunner()
         result = runner.invoke(config.config.commands["vlan"].commands["member"].commands["add"], ["1001", "Ethernet4"])

tests/vrf_test.py

@@ -41,7 +41,7 @@ def test_vrf_show(self):
         Loopback0
         Po0002.101
 """
-       
+
         result = runner.invoke(show.cli.commands['vrf'], [], obj=db)
         dbconnector.dedicated_dbs = {}
         assert result.exit_code == 0
@@ -65,7 +65,7 @@ def test_vrf_bind_unbind(self):
         Loopback0
         Po0002.101
 """
-       
+
         result = runner.invoke(show.cli.commands['vrf'], [], obj=db)
         dbconnector.dedicated_dbs = {}
         assert result.exit_code == 0
@@ -81,7 +81,7 @@ def test_vrf_bind_unbind(self):
         assert result.exit_code == 0
         assert 'Ethernet4' not in db.cfgdb.get_table('INTERFACE')
         assert result.output == expected_output_unbind
-        
+
         expected_output_unbind = "Interface Loopback0 IP disabled and address(es) removed due to unbinding VRF.\n"
 
         result = runner.invoke(config.config.commands["interface"].commands["vrf"].commands["unbind"], ["Loopback0"], obj=vrf_obj)
@@ -108,7 +108,7 @@ def test_vrf_bind_unbind(self):
         assert result.exit_code == 0
         assert 'PortChannel002' not in db.cfgdb.get_table('PORTCHANNEL_INTERFACE')
         assert result.output == expected_output_unbind
-        
+
         vrf_obj = {'config_db':db.cfgdb, 'namespace':DEFAULT_NAMESPACE}
         state_db = SonicV2Connector(use_unix_socket_path=True, namespace='')
         state_db.connect(state_db.STATE_DB, False)
@@ -117,7 +117,7 @@ def test_vrf_bind_unbind(self):
         vrf_obj['state_db'] = state_db
 
         expected_output_unbind = "Interface Eth36.10 IP disabled and address(es) removed due to unbinding VRF.\n"
-        T1 = threading.Thread( target = self.update_statedb, args = (state_db, db.db.STATE_DB, _hash))  
+        T1 = threading.Thread( target = self.update_statedb, args = (state_db, db.db.STATE_DB, _hash))
         T1.start()
         result = runner.invoke(config.config.commands["interface"].commands["vrf"].commands["unbind"], ["Eth36.10"], obj=vrf_obj)
         T1.join()
@@ -203,7 +203,7 @@ def test_vrf_bind_unbind(self):
         Loopback0
         Po0002.101
 """
-       
+
         result = runner.invoke(show.cli.commands['vrf'], [], obj=db)
         dbconnector.dedicated_dbs = {}
         assert result.exit_code == 0
@@ -213,24 +213,24 @@ def test_vrf_add_del(self):
         runner = CliRunner()
         db = Db()
         vrf_obj = {'config_db':db.cfgdb, 'namespace':db.db.namespace}
-        
+
         result = runner.invoke(config.config.commands["vrf"].commands["add"], ["Vrf100"], obj=vrf_obj)
         assert ('Vrf100') in db.cfgdb.get_table('VRF')
         assert result.exit_code == 0
-        
+
         result = runner.invoke(config.config.commands["vrf"].commands["add"], ["Vrf1"], obj=vrf_obj)
         assert "VRF Vrf1 already exists!" in result.output
         assert ('Vrf1') in db.cfgdb.get_table('VRF')
         assert result.exit_code != 0
-        
+
         expected_output_del = "VRF Vrf1 deleted and all associated IP addresses removed.\n"
         result = runner.invoke(config.config.commands["vrf"].commands["del"], ["Vrf1"], obj=vrf_obj)
         assert result.exit_code == 0
         assert result.output == expected_output_del
         assert ('Vrf1') not in db.cfgdb.get_table('VRF')
 
         result = runner.invoke(config.config.commands["vrf"].commands["del"], ["Vrf200"], obj=vrf_obj)
-        assert result.exit_code != 0       
+        assert result.exit_code != 0
         assert ('Vrf200') not in db.cfgdb.get_table('VRF')
         assert "VRF Vrf200 does not exist!" in result.output
 
@@ -245,25 +245,33 @@ def test_invalid_vrf_name(self):
         assert result.exit_code != 0
         assert ('vrf-blue') not in db.cfgdb.get_table('VRF')
         assert expected_output in result.output
-        
+
         result = runner.invoke(config.config.commands["vrf"].commands["add"], ["VRF2"], obj=obj)
         assert result.exit_code != 0
         assert ('VRF2') not in db.cfgdb.get_table('VRF')
         assert expected_output in result.output
-        
+
         result = runner.invoke(config.config.commands["vrf"].commands["add"], ["VrF10"], obj=obj)
         assert result.exit_code != 0
         assert ('VrF10') not in db.cfgdb.get_table('VRF')
         assert expected_output in result.output
-        
+
         result = runner.invoke(config.config.commands["vrf"].commands["del"], ["vrf-blue"], obj=obj)
         assert result.exit_code != 0
         assert expected_output in result.output
-        
+
         result = runner.invoke(config.config.commands["vrf"].commands["del"], ["VRF2"], obj=obj)
         assert result.exit_code != 0
         assert expected_output in result.output
-        
+
         result = runner.invoke(config.config.commands["vrf"].commands["del"], ["VrF10"], obj=obj)
         assert result.exit_code != 0
         assert expected_output in result.output
+
+        expected_output = """\
+Error: 'vrf_name' length should not exceed 15 characters
+"""
+        result = runner.invoke(config.config.commands["vrf"].commands["add"], ["VrfNameTooLong!!!"], obj=obj)
+        assert result.exit_code != 0
+        assert ('VrfNameTooLong!!!') not in db.cfgdb.get_table('VRF')
+        assert expected_output in result.output