ifupdown-ng NMSA: Add logic for adding addresses

1 files changed, 42 insertions, 9 deletions

diff --git a/ncserver/module/nms_ifupdownng.py b/ncserver/module/nms_ifupdownng.py
index 7c1746e..16183ea 100644
--- a/ncserver/module/nms_ifupdownng.py
+++ b/ncserver/module/nms_ifupdownng.py
@@ -14,12 +14,14 @@ from contextlib import contextmanager
 
 import ipaddress
 import logging
-import netifaces
 import pathlib
 import socket
 import subprocess
 import yaml
 
+# pylint: disable=I1101
+import netifaces
+
 from ncserver.base.util import _
 
 
@@ -165,6 +167,12 @@ def _find_many(iface: str, key: str) -> list:
     return ret
 
 
+def _add_one_to_list(iface: str, key: str, value: str):
+    """Add +value+ to list +key+ for +iface+."""
+    _CONFIG[iface].append({key: value})
+    _save()
+
+
 def _replace_one(iface: str, key: str, value: str):
     """Replace a single instance of +key+ for +iface+ with +value+."""
     iface_cfg = _CONFIG[iface]
@@ -189,6 +197,17 @@ def _remove_one(iface: str, key: str):
             return
 
 
+def _remove_one_from_list(iface: str, key: str, value: str):
+    """Remove +value+ from list +key+ for +iface+."""
+    iface_cfg = _CONFIG[iface]
+    for item in iface_cfg:
+        if key in item.keys():
+            if item[key] == value:
+                iface_cfg.remove(item)
+                _save()
+                return
+
+
 def _iface_path(iface: str) -> pathlib.Path:
     """Retrieve the system device path for the specified interface."""
     return pathlib.Path('/sys/class/net/' + iface)
@@ -695,15 +714,15 @@ def live_addresses(iface: str) -> list:
         return addresses
 
     raw = netifaces.ifaddresses(iface)
-    for v4 in raw.get(socket.AF_INET, tuple()):
-        addr = v4['addr']
-        mask = v4['netmask']
+    for ipv4 in raw.get(socket.AF_INET, tuple()):
+        addr = ipv4['addr']
+        mask = ipv4['netmask']
         iface = ipaddress.IPv4Interface("{a}/{m}".format(a=addr, m=mask))
         addresses.append(iface)
 
-    for v6 in raw.get(socket.AF_INET6, tuple()):
-        addr = v6['addr'].split('%')[0]
-        mask = v6['netmask'][v6['netmask'].find('/') + 1:]
+    for ipv6 in raw.get(socket.AF_INET6, tuple()):
+        addr = ipv6['addr'].split('%')[0]
+        mask = ipv6['netmask'][ipv6['netmask'].find('/') + 1:]
         iface = ipaddress.IPv6Interface("{a}/{m}".format(a=addr, m=mask))
         addresses.append(iface)
 
@@ -724,8 +743,22 @@ def add_address(iface: str, _type, addr: str, prefix):
         LOGGER.error(_("unknown address type %r"), _type)
         return
 
-    # implement this.
-    raise NotImplementedError
+    addr = None
+    iface = ipaddress.IPv6Interface
+    if _type == socket.AF_INET:
+        iface = ipaddress.IPv4Interface
+
+    try:
+        addr = iface("{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")
+
+    _add_one_to_list(iface, 'address', s_addr)
 
 
 def remove_address(iface: str, addr: str):