[config] fix interface IPv6 address removal. (sonic-net#1819)

Current code takes an input from the user, converts the IPv6 string to lower letters and removes lower case IPv6 string from CONFIG DB. This is a bug since, according to the schema CONFIG DB is case insensitive for IPv6 address.

- What I did
Fixed CLI for removing IPv6 address.
Issue is that below command does not work if IP address is written in upper case in CONFIG DB, like this: FC00::1/64.
'config interface ip remove Ethernet0 FC00::1/64'

- How I did it
Make it case insensitive
Relaxed the validation of IP address, a built-in validation from ipaddress package in python is used.
Refactored interface_ipaddr_dependent_on_interface -> get_interface_ipaddresses
Separated some functions (has_static_routes_attached, flush_ip_neigh_in_kernel, can_remove_router_interface, remove_router_interface_ip_address, remove_router_interface, is_management_interface)

- How to verify it
Run UT. Try to reproduce the scenario described above.

Signed-off-by: Stepan Blyschak <stepanb@nvidia.com>

config/main.py

@@ -343,18 +343,22 @@ def interface_name_to_alias(config_db, interface_name):
 
  return None
 
-def interface_ipaddr_dependent_on_interface(config_db, interface_name):
- """Get table keys including ipaddress
+def get_interface_ipaddresses(config_db, interface_name):
+ """Get IP addresses attached to interface
  """
- data = []
+ ipaddresses = set()
  table_name = get_interface_table_name(interface_name)
- if table_name == "":
- return data
+ if not table_name:
+ return ipaddresses
+
  keys = config_db.get_keys(table_name)
  for key in keys:
- if interface_name in key and len(key) == 2:
- data.append(key)
- return data
+ if isinstance(key, tuple) and len(key) == 2:
+ iface, interface_ip = key
+ if iface == interface_name:
+ ipaddresses.add(ipaddress.ip_interface(interface_ip))
+
+ return ipaddresses
 
 def is_interface_bind_to_vrf(config_db, interface_name):
  """Get interface if bind to vrf or not
@@ -448,9 +452,9 @@ def del_interface_bind_to_vrf(config_db, vrf_name):
  if interface_dict:
  for interface_name in interface_dict:
  if 'vrf_name' in interface_dict[interface_name] and vrf_name == interface_dict[interface_name]['vrf_name']:
- interface_dependent = interface_ipaddr_dependent_on_interface(config_db, interface_name)
- for interface_del in interface_dependent:
- config_db.set_entry(table_name, interface_del, None)
+ interface_ipaddresses = get_interface_ipaddresses(config_db, interface_name)
+ for ipaddress in interface_ipaddresses:
+ remove_router_interface_ip_address(config_db, interface_name, ipaddress)
  config_db.set_entry(table_name, interface_name, None)
 
 def set_interface_naming_mode(mode):
@@ -854,27 +858,6 @@ def validate_mirror_session_config(config_db, session_name, dst_port, src_port,
 
  return True
 
-def is_valid_ip_interface(ctx, ip_addr):
- split_ip_mask = ip_addr.split("/")
- if len(split_ip_mask) < 2:
- return False
-
- # Check if the IP address is correct or if there are leading zeros.
- ip_obj = ipaddress.ip_address(split_ip_mask[0])
-
- if isinstance(ip_obj, ipaddress.IPv4Address):
- # Since the IP address is used as a part of a key in Redis DB,
- # do not tolerate extra zeros in IPv4.
- if str(ip_obj) != split_ip_mask[0]:
- return False
-
- # Check if the mask is correct
- net = ipaddress.ip_network(ip_addr, strict=False)
- if str(net.prefixlen) != split_ip_mask[1] or net.prefixlen == 0:
- return False
-
- return True
-
 def cli_sroute_to_config(ctx, command_str, strict_nh = True):
  if len(command_str) < 2 or len(command_str) > 9:
  ctx.fail("argument is not in pattern prefix [vrf <vrf_name>] <A.B.C.D/M> nexthop <[vrf <vrf_name>] <A.B.C.D>>|<dev <dev_name>>!")
@@ -985,6 +968,20 @@ def cache_arp_entries():
  open(restore_flag_file, 'w').close()
  return success
 
+def remove_router_interface_ip_address(config_db, interface_name, ipaddress_to_remove):
+ table_name = get_interface_table_name(interface_name)
+ keys = config_db.get_keys(table_name)
+
+ for key in keys:
+ if not isinstance(key, tuple) or len(key) != 2:
+ continue
+
+ iface, ipaddress_string = key
+ if iface != interface_name:
+ continue
+
+ if ipaddress.ip_interface(ipaddress_string) == ipaddress_to_remove:
+ config_db.set_entry(table_name, (interface_name, ipaddress_string), None)
 
 def validate_ipv4_address(ctx, param, ip_addr):
  """Helper function to validate ipv4 address
@@ -3749,50 +3746,44 @@ def add(ctx, interface_name, ip_addr, gw):
  return
 
  try:
- net = ipaddress.ip_network(ip_addr, strict=False)
- if '/' not in ip_addr:
- ip_addr += '/' + str(net.prefixlen)
-
- if not is_valid_ip_interface(ctx, ip_addr):
- raise ValueError('')
-
- if interface_name == 'eth0':
-
- # Configuring more than 1 IPv4 or more than 1 IPv6 address fails.
- # Allow only one IPv4 and only one IPv6 address to be configured for IPv6.
- # If a row already exist, overwrite it (by doing delete and add).
- mgmtintf_key_list = _get_all_mgmtinterface_keys()
-
- for key in mgmtintf_key_list:
- # For loop runs for max 2 rows, once for IPv4 and once for IPv6.
- # No need to capture the exception since the ip_addr is already validated earlier
- ip_input = ipaddress.ip_interface(ip_addr)
- current_ip = ipaddress.ip_interface(key[1])
- if (ip_input.version == current_ip.version):
- # If user has configured IPv4/v6 address and the already available row is also IPv4/v6, delete it here.
- config_db.set_entry("MGMT_INTERFACE", ("eth0", key[1]), None)
-
- # Set the new row with new value
- if not gw:
- config_db.set_entry("MGMT_INTERFACE", (interface_name, ip_addr), {"NULL": "NULL"})
- else:
- config_db.set_entry("MGMT_INTERFACE", (interface_name, ip_addr), {"gwaddr": gw})
- mgmt_ip_restart_services()
+ ip_address = ipaddress.ip_interface(ip_addr)
+ except ValueError as err:
+ ctx.fail("IP address is not valid: {}".format(err))
+
+ if interface_name == 'eth0':
+
+ # Configuring more than 1 IPv4 or more than 1 IPv6 address fails.
+ # Allow only one IPv4 and only one IPv6 address to be configured for IPv6.
+ # If a row already exist, overwrite it (by doing delete and add).
+ mgmtintf_key_list = _get_all_mgmtinterface_keys()
+
+ for key in mgmtintf_key_list:
+ # For loop runs for max 2 rows, once for IPv4 and once for IPv6.
+ # No need to capture the exception since the ip_addr is already validated earlier
+ current_ip = ipaddress.ip_interface(key[1])
+ if (ip_address.version == current_ip.version):
+ # If user has configured IPv4/v6 address and the already available row is also IPv4/v6, delete it here.
+ config_db.set_entry("MGMT_INTERFACE", ("eth0", key[1]), None)
+
+ # Set the new row with new value
+ if not gw:
+ config_db.set_entry("MGMT_INTERFACE", (interface_name, str(ip_address)), {"NULL": "NULL"})
+ else:
+ config_db.set_entry("MGMT_INTERFACE", (interface_name, str(ip_address)), {"gwaddr": gw})
+ mgmt_ip_restart_services()
 
-  return
+ return
 
- table_name = get_interface_table_name(interface_name)
- if table_name == "":
- ctx.fail("'interface_name' is not valid. Valid names [Ethernet/PortChannel/Vlan/Loopback]")
- interface_entry = config_db.get_entry(table_name, interface_name)
- if len(interface_entry) == 0:
- if table_name == "VLAN_SUB_INTERFACE":
- config_db.set_entry(table_name, interface_name, {"admin_status": "up"})
- else:
- config_db.set_entry(table_name, interface_name, {"NULL": "NULL"})
- config_db.set_entry(table_name, (interface_name, ip_addr), {"NULL": "NULL"})
- except ValueError:
- ctx.fail("ip address or mask is not valid.")
+ table_name = get_interface_table_name(interface_name)
+ if table_name == "":
+ ctx.fail("'interface_name' is not valid. Valid names [Ethernet/PortChannel/Vlan/Loopback]")
+ interface_entry = config_db.get_entry(table_name, interface_name)
+ if len(interface_entry) == 0:
+ if table_name == "VLAN_SUB_INTERFACE":
+ config_db.set_entry(table_name, interface_name, {"admin_status": "up"})
+ else:
+ config_db.set_entry(table_name, interface_name, {"NULL": "NULL"})
+ config_db.set_entry(table_name, (interface_name, str(ip_address)), {"NULL": "NULL"})
 
 #
 # 'del' subcommand
@@ -3813,52 +3804,47 @@ def remove(ctx, interface_name, ip_addr):
  ctx.fail("'interface_name' is None!")
 
  try:
- net = ipaddress.ip_network(ip_addr, strict=False)
-  if '/' not in ip_addr:
-  ip_addr += '/' + str(net.prefixlen)
+ ip_address = ipaddress.ip_interface(ip_addr)
+ except ValueError as err:
+ ctx.fail("IP address is not valid: {}".format(err))
 
- if not is_valid_ip_interface(ctx, ip_addr):
- raise ValueError('')
+ if interface_name == 'eth0':
+ config_db.set_entry("MGMT_INTERFACE", (interface_name, str(ip_address)), None)
+ mgmt_ip_restart_services()
+ return
 
- if interface_name == 'eth0':
- config_db.set_entry("MGMT_INTERFACE", (interface_name, ip_addr), None)
- mgmt_ip_restart_services()
- return
+ table_name = get_interface_table_name(interface_name)
+ if table_name == "":
+ ctx.fail("'interface_name' is not valid. Valid names [Ethernet/PortChannel/Vlan/Loopback]")
+ interface_addresses = get_interface_ipaddresses(config_db, interface_name)
+ # If we deleting the last IP entry of the interface, check whether a static route present for the RIF
+ # before deleting the entry and also the RIF.
+ if interface_addresses == {ip_address}:
+ # Check both IPv4 and IPv6 routes.
+ ip_versions = [ "ip", "ipv6"]
+ for ip_ver in ip_versions:
+ # Compete the command and ask Zebra to return the routes.
+ # Scopes of all VRFs will be checked.
+ cmd = "show {} route vrf all static".format(ip_ver)
+ if multi_asic.is_multi_asic():
+ output = bgp_util.run_bgp_command(cmd, ctx.obj['namespace'])
+ else:
+ output = bgp_util.run_bgp_command(cmd)
+ # If there is output data, check is there a static route,
+ # bound to the interface.
+ if output != "":
+ if any(interface_name in output_line for output_line in output.splitlines()):
+ ctx.fail("Cannot remove the last IP entry of interface {}. A static {} route is still bound to the RIF.".format(interface_name, ip_ver))
+ remove_router_interface_ip_address(config_db, interface_name, ip_address)
+ interface_addresses = get_interface_ipaddresses(config_db, interface_name)
+ if len(interface_addresses) == 0 and is_interface_bind_to_vrf(config_db, interface_name) is False and get_intf_ipv6_link_local_mode(ctx, interface_name, table_name) != "enable":
+ config_db.set_entry(table_name, interface_name, None)
 
- table_name = get_interface_table_name(interface_name)
- if table_name == "":
- ctx.fail("'interface_name' is not valid. Valid names [Ethernet/PortChannel/Vlan/Loopback]")
- interface_dependent = interface_ipaddr_dependent_on_interface(config_db, interface_name)
- # If we deleting the last IP entry of the interface, check whether a static route present for the RIF
- # before deleting the entry and also the RIF.
- if len(interface_dependent) == 1 and interface_dependent[0][1] == ip_addr:
- # Check both IPv4 and IPv6 routes.
- ip_versions = [ "ip", "ipv6"]
- for ip_ver in ip_versions:
- # Compete the command and ask Zebra to return the routes.
- # Scopes of all VRFs will be checked.
- cmd = "show {} route vrf all static".format(ip_ver)
- if multi_asic.is_multi_asic():
- output = bgp_util.run_bgp_command(cmd, ctx.obj['namespace'])
- else:
- output = bgp_util.run_bgp_command(cmd)
- # If there is output data, check is there a static route,
- # bound to the interface.
- if output != "":
- if any(interface_name in output_line for output_line in output.splitlines()):
- ctx.fail("Cannot remove the last IP entry of interface {}. A static {} route is still bound to the RIF.".format(interface_name, ip_ver))
- config_db.set_entry(table_name, (interface_name, ip_addr), None)
- interface_dependent = interface_ipaddr_dependent_on_interface(config_db, interface_name)
- if len(interface_dependent) == 0 and is_interface_bind_to_vrf(config_db, interface_name) is False and get_intf_ipv6_link_local_mode(ctx, interface_name, table_name) != "enable":
- config_db.set_entry(table_name, interface_name, None)
-
- if multi_asic.is_multi_asic():
- command = "sudo ip netns exec {} ip neigh flush dev {} {}".format(ctx.obj['namespace'], interface_name, ip_addr)
- else:
- command = "ip neigh flush dev {} {}".format(interface_name, ip_addr)
- clicommon.run_command(command)
- except ValueError:
- ctx.fail("ip address or mask is not valid.")
+ if multi_asic.is_multi_asic():
+ command = "sudo ip netns exec {} ip neigh flush dev {} {}".format(ctx.obj['namespace'], interface_name, str(ip_address))
+ else:
+ command = "ip neigh flush dev {} {}".format(interface_name, str(ip_address))
+ clicommon.run_command(command)
 
 
 #
@@ -4243,9 +4229,9 @@ def bind(ctx, interface_name, vrf_name):
  config_db.get_entry(table_name, interface_name).get('vrf_name') == vrf_name:
  return
  # Clean ip addresses if interface configured
- interface_dependent = interface_ipaddr_dependent_on_interface(config_db, interface_name)
- for interface_del in interface_dependent:
- config_db.set_entry(table_name, interface_del, None)
+ interface_addresses = get_interface_ipaddresses(config_db, interface_name)
+ for ipaddress in interface_addresses:
+ remove_router_interface_ip_address(config_db, interface_name, ipaddress)
  config_db.set_entry(table_name, interface_name, None)
  # When config_db del entry and then add entry with same key, the DEL will lost.
  if ctx.obj['namespace'] is DEFAULT_NAMESPACE:
@@ -4281,9 +4267,9 @@ def unbind(ctx, interface_name):
  ctx.fail("'interface_name' is not valid. Valid names [Ethernet/PortChannel/Vlan/Loopback]")
  if is_interface_bind_to_vrf(config_db, interface_name) is False:
  return
- interface_dependent = interface_ipaddr_dependent_on_interface(config_db, interface_name)
- for interface_del in interface_dependent:
- config_db.set_entry(table_name, interface_del, None)
+ interface_ipaddresses = get_interface_ipaddresses(config_db, interface_name)
+ for ipaddress in interface_ipaddresses:
+ remove_router_interface_ip_address(config_db, interface_name, ipaddress)
  config_db.set_entry(table_name, interface_name, None)
 
 

tests/ip_config_test.py

@@ -8,9 +8,7 @@
 import show.main as show
 from utilities_common.db import Db
 
-ERROR_MSG = '''
-Error: ip address or mask is not valid.
-'''
+ERROR_MSG = "Error: IP address is not valid"
 
 class TestConfigIP(object):
  @classmethod
@@ -65,7 +63,7 @@ def test_add_interface_ipv4_with_leading_zeros(self):
  obj = {'config_db':db.cfgdb}
 
  # config int ip add Ethernet68 10.10.10.002/24
- result = runner.invoke(config.config.commands["interface"].commands["ip"].commands["add"], ["Ethernet68", "10.10.10.002/24"], obj=obj) 
+ result = runner.invoke(config.config.commands["interface"].commands["ip"].commands["add"], ["Ethernet68", "10.10.10.0002/24"], obj=obj) 
  print(result.exit_code, result.output)
  assert result.exit_code != 0
  assert ERROR_MSG in result.output
@@ -88,7 +86,21 @@ def test_add_del_interface_valid_ipv6(self):
  print(result.exit_code, result.output)
  assert result.exit_code != 0
  assert ('Ethernet72', '2001:1db8:11a3:19d7:1f34:8a2e:17a0:765d/34') not in db.cfgdb.get_table('INTERFACE')
-
+
+ def test_del_interface_case_sensitive_ipv6(self):
+ db = Db()
+ runner = CliRunner()
+ obj = {'config_db':db.cfgdb}
+
+ obj['config_db'].set_entry('INTERFACE', ('Ethernet72', 'FC00::1/126'), {})
+ assert ('Ethernet72', 'FC00::1/126') in db.cfgdb.get_table('INTERFACE')
+
+ # config int ip remove Ethernet72 FC00::1/126
+ result = runner.invoke(config.config.commands["interface"].commands["ip"].commands["remove"], ["Ethernet72", "FC00::1/126"], obj=obj)
+ print(result.exit_code, result.output)
+ assert result.exit_code != 0
+ assert ('Ethernet72', 'FC00::1/126') not in db.cfgdb.get_table('INTERFACE')
+
  def test_add_interface_invalid_ipv6(self):
  db = Db()
  runner = CliRunner()
@@ -120,14 +132,14 @@ def test_add_del_interface_ipv6_with_leading_zeros(self):
  result = runner.invoke(config.config.commands["interface"].commands["ip"].commands["add"], ["Ethernet68", "2001:0db8:11a3:09d7:1f34:8a2e:07a0:765d/34"], obj=obj)
  print(result.exit_code, result.output)
  assert result.exit_code == 0
- assert ('Ethernet68', '2001:0db8:11a3:09d7:1f34:8a2e:07a0:765d/34') in db.cfgdb.get_table('INTERFACE')
- 
+ assert ('Ethernet68', '2001:db8:11a3:9d7:1f34:8a2e:7a0:765d/34') in db.cfgdb.get_table('INTERFACE')
+
  # config int ip remove Ethernet68 2001:0db8:11a3:09d7:1f34:8a2e:07a0:765d/34
  result = runner.invoke(config.config.commands["interface"].commands["ip"].commands["remove"], ["Ethernet68", "2001:0db8:11a3:09d7:1f34:8a2e:07a0:765d/34"], obj=obj)
  print(result.exit_code, result.output)
  assert result.exit_code != 0
- assert ('Ethernet68', '2001:0db8:11a3:09d7:1f34:8a2e:07a0:765d/34') not in db.cfgdb.get_table('INTERFACE')
- 
+ assert ('Ethernet68', '2001:db8:11a3:9d7:1f34:8a2e:7a0:765d/34') not in db.cfgdb.get_table('INTERFACE')
+
  def test_add_del_interface_shortened_ipv6_with_leading_zeros(self):
  db = Db()
  runner = CliRunner()
@@ -137,14 +149,14 @@ def test_add_del_interface_shortened_ipv6_with_leading_zeros(self):
  result = runner.invoke(config.config.commands["interface"].commands["ip"].commands["add"], ["Ethernet68", "3000::001/64"], obj=obj) 
  print(result.exit_code, result.output)
  assert result.exit_code == 0
- assert ('Ethernet68', '3000::001/64') in db.cfgdb.get_table('INTERFACE')
- 
+ assert ('Ethernet68', '3000::1/64') in db.cfgdb.get_table('INTERFACE')
+
  # config int ip remove Ethernet68 3000::001/64
  result = runner.invoke(config.config.commands["interface"].commands["ip"].commands["remove"], ["Ethernet68", "3000::001/64"], obj=obj)
  print(result.exit_code, result.output)
  assert result.exit_code != 0
- assert ('Ethernet68', '3000::001/64') not in db.cfgdb.get_table('INTERFACE')
- 
+ assert ('Ethernet68', '3000::1/64') not in db.cfgdb.get_table('INTERFACE')
+
  @classmethod
  def teardown_class(cls):
  os.environ['UTILITIES_UNIT_TESTING'] = "0"