Merge "Use netaddr IPNetwork instead of parsing IP strings"

ovn_bgp_agent/privileged/linux_net.py

@@ -192,10 +192,7 @@ def rule_delete(rule):
 @ovn_bgp_agent.privileged.default.entrypoint
 def delete_ip_rules(ip_rules):
     for rule_ip, rule_info in ip_rules.items():
-        rule = {'dst': rule_ip.split("/")[0],
-                'dst_len': int(rule_ip.split("/")[1]),
-                'table': int(rule_info['table']),
-                'family': rule_info['family']}
+        rule = l_net.create_rule_from_ip(rule_ip, int(rule_info['table']))
         _run_iproute_rule('del', **rule)
 
 
@@ -260,11 +257,12 @@ def add_ip_nei(ip, lladdr, dev):
 
 @ovn_bgp_agent.privileged.default.entrypoint
 def del_ip_nei(ip, lladdr, dev):
-    ip_version = l_net.get_ip_version(ip)
-    family = common_utils.IP_VERSION_FAMILY_MAP[ip_version]
+    ip_network = netaddr.IPNetwork(ip)
+    family = common_utils.IP_VERSION_FAMILY_MAP[ip_network.version]
+
     _run_iproute_neigh('del',
                        dev,
-                       dst=ip.split("/")[0],
+                       dst=str(ip_network.ip),
                        lladdr=lladdr,
                        family=family,
                        state=ndmsg.states['permanent'])

ovn_bgp_agent/tests/unit/utils/test_linux_net.py

@@ -514,7 +514,8 @@ class TestLinuxNet(test_base.TestCase):
         linux_net.add_ip_rule(
             self.ip, 7, dev=self.dev, lladdr=self.mac)
 
-        expected_args = {'dst': self.ip, 'table': 7, 'dst_len': 32}
+        expected_args = {'dst': self.ip, 'table': 7,
+                         'dst_len': 32, 'family': constants.AF_INET}
         mock_rule_create.assert_called_once_with(expected_args)
         mock_add_ip_nei.assert_called_once_with(self.ip, self.mac, self.dev)
 
@@ -545,7 +546,8 @@ class TestLinuxNet(test_base.TestCase):
     def test_del_ip_rule(self, mock_rule_delete, mock_del_ip_nei):
         linux_net.del_ip_rule(self.ip, 7, dev=self.dev, lladdr=self.mac)
 
-        expected_args = {'dst': self.ip, 'table': 7, 'dst_len': 32}
+        expected_args = {'dst': self.ip, 'table': 7,
+                         'dst_len': 32, 'family': constants.AF_INET}
         mock_rule_delete.assert_called_once_with(expected_args)
         mock_del_ip_nei.assert_called_once_with(self.ip, self.mac, self.dev)
 

ovn_bgp_agent/utils/linux_net.py

@@ -17,6 +17,7 @@ import random
 import re
 import sys
 
+import netaddr
 from oslo_log import log as logging
 import pyroute2
 from pyroute2.netlink import exceptions as netlink_exceptions
@@ -25,12 +26,15 @@ import tenacity
 from ovn_bgp_agent import constants
 from ovn_bgp_agent import exceptions as agent_exc
 import ovn_bgp_agent.privileged.linux_net
+from ovn_bgp_agent.utils import common as common_utils
 
 LOG = logging.getLogger(__name__)
 
 
 def get_ip_version(ip):
-    return ipaddress.ip_address(ip.split('/')[0]).version
+    # IP network can consume both an IP address and a network with cidr
+    # notation
+    return netaddr.IPNetwork(ip).version
 
 
 @tenacity.retry(
@@ -586,22 +590,23 @@ def del_ips_from_dev(nic, ips):
         ovn_bgp_agent.privileged.linux_net.del_ip_from_dev(ip, nic)
 
 
-def add_ip_rule(ip, table, dev=None, lladdr=None):
-    ip_version = get_ip_version(ip)
-    ip_info = ip.split("/")
-
-    if len(ip_info) == 1:
-        rule = {'dst': ip_info[0], 'table': table, 'dst_len': 32}
-        if ip_version == constants.IP_VERSION_6:
-            rule['dst_len'] = 128
-            rule['family'] = constants.AF_INET6
-    elif len(ip_info) == 2:
-        rule = {'dst': ip_info[0], 'table': table, 'dst_len': int(ip_info[1])}
-        if ip_version == constants.IP_VERSION_6:
-            rule['family'] = constants.AF_INET6
-    else:
+def create_rule_from_ip(ip, table):
+    try:
+        ip_network = netaddr.IPNetwork(ip)
+    except (netaddr.AddrFormatError, ValueError):
         raise agent_exc.InvalidPortIP(ip=ip)
 
+    return {
+        'dst': str(ip_network.ip),
+        'table': table,
+        'dst_len': ip_network.prefixlen,
+        'family': common_utils.IP_VERSION_FAMILY_MAP[ip_network.version],
+    }
+
+
+def add_ip_rule(ip, table, dev=None, lladdr=None):
+    rule = create_rule_from_ip(ip, table)
+
     ovn_bgp_agent.privileged.linux_net.rule_create(rule)
 
     if lladdr:
@@ -619,20 +624,7 @@ def add_ip_nei(ip, lladdr, dev):
 
 
 def del_ip_rule(ip, table, dev=None, lladdr=None):
-    ip_version = get_ip_version(ip)
-    ip_info = ip.split("/")
-
-    if len(ip_info) == 1:
-        rule = {'dst': ip_info[0], 'table': table, 'dst_len': 32}
-        if ip_version == constants.IP_VERSION_6:
-            rule['dst_len'] = 128
-            rule['family'] = constants.AF_INET6
-    elif len(ip_info) == 2:
-        rule = {'dst': ip_info[0], 'table': table, 'dst_len': int(ip_info[1])}
-        if ip_version == constants.IP_VERSION_6:
-            rule['family'] = constants.AF_INET6
-    else:
-        raise agent_exc.InvalidPortIP(ip=ip)
+    rule = create_rule_from_ip(ip, table)
 
     ovn_bgp_agent.privileged.linux_net.rule_delete(rule)