ietf-ip: Add ability to manipulate IP addressesHEADcurrent

2 files changed, 54 insertions, 12 deletions

diff --git a/ncserver/module/ip.py b/ncserver/module/ip.py
index 84eab13..024ce4f 100644
--- a/ncserver/module/ip.py
+++ b/ncserver/module/ip.py
@@ -10,6 +10,8 @@ with this source distribution for more information.
 SPDX-License-Identifier: NCSA
 """
 
+from socket import AF_INET, AF_INET6
+
 import ipaddress
 import logging
 
@@ -267,6 +269,47 @@ def _clear_ipv4(iface: str):
     nmsa.unset_param(iface, 'ipv4_mtu')
 
 
+def _edit_address(session, rpc, node, operation, iface: str, _type):
+    """Edit an IP address."""
+    addr_node = node.find('{'+M_NS+'}ip')
+    if addr_node is None:
+        raise error.MissingElementAppError(rpc, node)
+    address = addr_node.text
+
+    nmsa = get_nmsa()
+
+    if operation in ('delete', 'remove'):
+        try:
+            nmsa.remove_address(iface, address)
+        except KeyError as key_e:
+            if operation == 'delete':
+                raise error.DataMissingAppError(rpc) from key_e
+        log_config_change(session, "[ietf-ip %s]" % iface,
+                          "%s IP %s" % (operation, address))
+        return
+
+    pref_node = node.find('{'+M_NS+'}prefix-length')
+    if pref_node is None:
+        raise error.MissingElementAppError(rpc, node)
+    prefix = pref_node.text
+
+    if operation not in ('create', 'merge', 'replace'):
+        raise error.OperationNotSupportedAppError(rpc)
+
+    try:
+        nmsa.add_address(iface, _type, address, prefix)
+    except ValueError as val_e:
+        raise error.InvalidValueAppError(rpc, info="invalid IP") from val_e
+    except RuntimeError as run_e:
+        if operation == 'create':
+            raise error.DataExistsAppError(rpc) from run_e
+        else:
+            nmsa.modify_prefix(iface, address, prefix)
+    else:
+        log_config_change(session, "[ietf-ip %s]" % iface,
+                          "%s IP %s/%s" % (operation, address, prefix))
+
+
 def _edit_ipv4(session, rpc, node, def_op, iface: str):
     """Edit IPv4 configuration for a given interface."""
     _params = {'enabled': 'ipv4_enabled', 'forwarding': 'ipv4_forwarding',
@@ -292,8 +335,7 @@ def _edit_ipv4(session, rpc, node, def_op, iface: str):
                               "IPv4 %s: -> %s" % (param, xparam.text))
             _edit_param(iface, param, operation, rpc, xparam)
         elif qparam.localname == 'address':
-            # Oh no.
-            raise NotImplementedError
+            _edit_address(session, rpc, xparam, operation, iface, AF_INET)
         elif qparam.localname == 'neighbor':
             # Oh *no*!
             raise NotImplementedError
@@ -372,8 +414,7 @@ def _edit_ipv6(session, rpc, node, def_op, iface: str):
                               "IPv6 %s: -> %s" % (param, xparam.text))
             _edit_param(iface, param, p_op, rpc, xparam)
         elif qparam.localname == 'address':
-            # Oh no.
-            raise NotImplementedError
+            _edit_address(session, rpc, xparam, operation, iface, AF_INET6)
         elif qparam.localname == 'neighbor':
             # Oh *no*!
             raise NotImplementedError
diff --git a/ncserver/module/nms_ifupdownng.py b/ncserver/module/nms_ifupdownng.py
index a607215..76f7c32 100644
--- a/ncserver/module/nms_ifupdownng.py
+++ b/ncserver/module/nms_ifupdownng.py
@@ -756,21 +756,22 @@ def add_address(iface: str, _type, addr: str, prefix):
         LOGGER.error(_("unknown address type %r"), _type)
         return
 
-    addr = None
-    iface = ipaddress.IPv6Interface
+    ip_addr = None
+    ctor = ipaddress.IPv6Interface
     if _type == socket.AF_INET:
-        iface = ipaddress.IPv4Interface
+        ctor = ipaddress.IPv4Interface
 
     try:
-        addr = iface("{a}/{p}".format(a=addr, p=prefix))
+        ip_addr = ctor("{a}/{p}".format(a=addr, p=prefix))
     except Exception as err:
         raise ValueError("IP address is not valid") from err
 
-    s_addr = str(addr)
-
-    if s_addr in list_addresses(iface):
-        raise RuntimeError("Duplicate address attempt")
+    for candidate in list_addresses(iface):
+        cand_addr = candidate.split('/')[0]
+        if addr == cand_addr:
+            raise RuntimeError("Duplicate address attempt")
 
+    s_addr = str(ip_addr)
     _add_one_to_list(iface, 'address', s_addr)