From b4ec63029b8917e1b41b4b95252703c13197f53c Mon Sep 17 00:00:00 2001 From: "A. Wilcox" Date: Mon, 14 Dec 2020 15:38:08 -0600 Subject: ifupdown-ng NMSA: Add logic for adding addresses --- ncserver/module/nms_ifupdownng.py | 51 ++++++++++++++++++++++++++++++++------- 1 file 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): -- cgit v1.2.3-60-g2f50