Validate interface name length in CLI

config/main.py

@@ -33,7 +33,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, \
+ validate_interface_name_length, iface_name_max_length
 from utilities_common.db import Db
 from utilities_common.intf_filter import parse_interface_in_filter
 from utilities_common import bgp_util
@@ -105,7 +106,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>"
 
@@ -438,7 +438,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 validate_interface_name_length(portchannel_name):
  return False
  return True
 
@@ -2440,8 +2440,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_length()))
  if is_portchannel_present_in_db(db, portchannel_name):
  ctx.fail("{} already exists!".format(portchannel_name)) # TODO: MISSING CONSTRAINT IN YANG MODEL
 
@@ -5875,8 +5875,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 validate_interface_name_length(vrf_name):
+ ctx.fail("'vrf_name' length should not exceed {} characters".format(iface_name_max_length()))
  if is_vrf_exists(config_db, vrf_name):
  ctx.fail("VRF {} already exists!".format(vrf_name))
  elif (vrf_name == 'mgmt' or vrf_name == 'management'):
@@ -5895,8 +5895,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 validate_interface_name_length(vrf_name):
+ ctx.fail("'vrf_name' length should not exceed {} characters".format((iface_name_max_length())))
  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():
@@ -6903,8 +6903,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_length()))
 
  lo_intfs = [k for k, v in config_db.get_table('LOOPBACK_INTERFACE').items() if type(k) != tuple]
  if loopback_name in lo_intfs:
@@ -7651,6 +7651,8 @@ def add_subinterface(ctx, subinterface_name, vid):
 
  if interface_alias is None:
  ctx.fail("{} invalid subinterface".format(interface_alias))
+ if not validate_interface_name_length(interface_alias):
+ ctx.fail("Subinterface name length should not exceed {} characters".format(iface_name_max_length()))
 
  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 validate_interface_name_length, iface_name_max_length
 
 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 validate_interface_name_length(vxlan_name):
+ ctx.fail("'vxlan_name' length should not exceed {} characters".format(iface_name_max_length()))
 
  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))

tests/config_test.py

@@ -2772,6 +2772,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 11 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 (1-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)
@@ -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 16 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