Vyos

ntc_rosetta/__init__.py

@@ -1,6 +1,6 @@
 from typing import Dict, Type
 
-from ntc_rosetta.drivers import ios, junos
+from ntc_rosetta.drivers import ios, junos, vyos
 from ntc_rosetta.drivers.base import Driver
 
 
@@ -15,7 +15,12 @@ def get_driver(driver: str, model: str = "openconfig") -> Type[Driver]:
         "openconfig": {
             "ios": ios.IOSDriverOpenconfig,
             "junos": junos.JunosDriverOpenconfig,
+            "vyos": vyos.VyOSDriverOpenconfig,
+        },
+        "ntc": {
+            "ios": ios.IOSDriverNTC,
+            "junos": junos.JunosDriverNTC,
+            "vyos": vyos.VyOSDriverNTC,
         },
-        "ntc": {"ios": ios.IOSDriverNTC, "junos": junos.JunosDriverNTC},
     }
     return mapping[model][driver]

ntc_rosetta/drivers/vyos.py

@@ -0,0 +1,17 @@
+from ntc_rosetta.drivers.base import Driver
+from ntc_rosetta.parsers.ntc.vyos import VyOSParser as NTCVyOSParser
+from ntc_rosetta.parsers.openconfig.vyos import VyOSParser as OCVyOSParser
+from ntc_rosetta.translators.ntc.vyos import VyOSTranslator as NTCVyOSTranslator
+from ntc_rosetta.translators.openconfig.vyos import VyOSTranslator as OCVyOSTranslator
+
+
+class VyOSDriverOpenconfig(Driver):
+    parser = OCVyOSParser
+    translator = OCVyOSTranslator
+    datamodel_name = "openconfig"
+
+
+class VyOSDriverNTC(Driver):
+    parser = NTCVyOSParser
+    translator = NTCVyOSTranslator
+    datamodel_name = "ntc"

ntc_rosetta/helpers/request_list.py

@@ -0,0 +1,30 @@
+import json
+from typing import List, Dict, Any
+
+
+class RequestList:
+    def __init__(self) -> None:
+        self.requests: List[Dict[str, Any]] = []
+        self.pre_path: List[str] = []
+
+    def set(self, path: List[str], value: str, use_pre_path: bool = False) -> None:
+        if use_pre_path:
+            path = self.pre_path + path
+        self.requests.append({"op": "set", "path": path, "value": value})
+
+    def delete(self, path: List[str], value: str, use_pre_path: bool = False) -> None:
+        if use_pre_path:
+            path = self.pre_path + path
+        self.requests.append({"op": "delete", "path": path, "value": value})
+
+    # This is in preparation for possible move to using HTTP/API instead SSH for commands
+    def to_json(self) -> str:
+        return json.dumps(self.requests, indent=2)
+
+    def to_set(self) -> str:
+        result = ""
+        for request in self.requests:
+            result += (
+                f'{request["op"]} {" ".join(request["path"])} \'{request["value"]}\'\n'
+            )
+        return result

ntc_rosetta/helpers/vyos.py

@@ -0,0 +1,77 @@
+"""VyOS helper consts and methods"""
+from typing import Dict, Any, List
+
+SYSLOG_FACILITY = {
+    "all": "ALL",
+    "auth": "AUTH",
+    "authpriv": "AUTHPRIV",
+    "cron": None,
+    "daemon": "SYSTEM_DAEMON",
+    "kern": "KERNEL",
+    "lpr": None,
+    "mail": "MAIL",
+    "mark": None,
+    "news": None,
+    "protocols": "LOCAL7",
+    "security": "AUTH",
+    "syslog": "SYSLOG",
+    "user": "USER",
+    "uucp": None,
+    "local0": "LOCAL0",
+    "local1": "LOCAL1",
+    "local2": "LOCAL2",
+    "local3": "LOCAL3",
+    "local4": "LOCAL4",
+    "local5": "LOCAL5",
+    "local6": "LOCAL6",
+    "local7": "LOCAL7",
+}
+SYSLOG_SEVERITY = {
+    "emerg": "EMERGENCY",
+    "alert": "ALERT",
+    "crit": "CRITICAL",
+    "err": "ERROR",
+    "warning": "WARNING",
+    "notice": "NOTICE",
+    "info": "INFORMATIONAL",
+    "debug": "DEBUG",
+    "all": "DEBUG",
+}
+
+
+def parse_config_tree(config: List[str]) -> Dict[str, Any]:
+    parsed: Dict[str, Any] = {}
+    while True:
+        if not config:
+            break
+        line = config.pop(0).rstrip()
+        if line.lstrip().startswith("/*"):
+            continue
+        nodes = line.lstrip().split()
+        if nodes[-1] == "{":
+            if len(nodes) > 2:
+                parsed.setdefault(nodes[0], {}).update(
+                    {nodes[1]: parse_config_tree(config)}
+                )
+            else:
+                parsed[nodes[0]] = parse_config_tree(config)
+        elif nodes[-1] == "}":
+            break
+        else:
+            if len(nodes) == 1:
+                parsed[nodes[0]] = ""
+            else:
+                if nodes[1].startswith('"'):
+                    nodes[1] = nodes[1].lstrip('"')
+                    nodes[-1] = nodes[-1].rstrip('"')
+                    value = " ".join(nodes[1:])
+                else:
+                    value = str(nodes[1])
+                if nodes[0] in parsed:
+                    if isinstance(parsed[nodes[0]], list):
+                        parsed[nodes[0]].append(value)
+                    else:
+                        parsed[nodes[0]] = [parsed[nodes[0]], value]
+                else:
+                    parsed[nodes[0]] = value
+    return parsed

ntc_rosetta/parsers/__init__.py

@@ -1,4 +1,5 @@
 from ntc_rosetta.parsers.openconfig.ios import IOSParser
 from ntc_rosetta.parsers.openconfig.junos import JunosParser
+from ntc_rosetta.parsers.openconfig.vyos import VyOSParser
 
-__all__ = ("IOSParser", "JunosParser")
+__all__ = ("IOSParser", "JunosParser", "VyOSParser")

ntc_rosetta/parsers/ntc/vyos/__init__.py

@@ -0,0 +1,5 @@
+from yangify import parser
+
+
+class VyOSParser(parser.RootParser):
+    pass

ntc_rosetta/parsers/openconfig/vyos/__init__.py

@@ -0,0 +1,29 @@
+from yangify import parser
+
+from ntc_rosetta.helpers.vyos import parse_config_tree
+from ntc_rosetta.parsers.openconfig.vyos.openconfig_interfaces.interfaces import (
+    Interfaces,
+)
+from ntc_rosetta.parsers.openconfig.vyos.openconfig_network_instance.network_instances import (
+    NetworkInstances,
+)
+from ntc_rosetta.parsers.openconfig.vyos.openconfig_system.system import System
+
+
+class VyOSParser(parser.RootParser):
+    """
+    VyOSParser expects as native data a dictionary where the `dev_conf`
+    key is reserved for the device configuration.
+    """
+
+    class Yangify(parser.ParserData):
+        def init(self) -> None:
+            self.root_native["dev_conf"] = parse_config_tree(
+                self.root_native["dev_conf"].splitlines()
+            )
+            # self.root_native["dev_conf"] = json.loads(self.root_native["dev_conf"])
+            self.native["dev_conf"] = self.root_native["dev_conf"]
+
+    interfaces = Interfaces
+    network_instances = NetworkInstances
+    system = System

ntc_rosetta/parsers/openconfig/vyos/openconfig_if_ethernet/ethernet.py

@@ -0,0 +1,8 @@
+from yangify.parser import Parser, ParserData
+
+
+class Ethernet(Parser):
+    class Yangify(ParserData):
+        path = (
+            "openconfig-interfaces:interfaces/interface/openconfig-if-ethernet:ethernet"
+        )

ntc_rosetta/parsers/openconfig/vyos/openconfig_interfaces/interfaces.py

@@ -0,0 +1,121 @@
+from typing import Any, Dict, Iterator, Optional, Tuple, cast
+
+from ntc_rosetta.parsers.openconfig.vyos.openconfig_if_ethernet import ethernet
+from ntc_rosetta.parsers.openconfig.vyos.openconfig_vlan import vlan
+
+from yangify.parser import Parser, ParserData
+
+
+class SubinterfaceConfig(Parser):
+    class Yangify(ParserData):
+        path = "/openconfig-interfaces:interfaces/interface/subinterfaces/subinterface/config"
+
+    def description(self) -> Optional[str]:
+        return self.yy.native.get("description")
+
+    def enabled(self) -> bool:
+        return "disable" not in self.yy.native
+
+    def index(self) -> int:
+        if isinstance(self.yy.key, str):
+            return int(self.yy.key)
+        vlan_s = self.yy.key[0]
+        vlan_c = self.yy.key[1]
+        prepends = 4 - len(vlan_c)
+        vlan_c = ("0" * prepends) + vlan_c
+        return int(vlan_s + vlan_c)
+
+
+class Subinterface(Parser):
+    class Yangify(ParserData):
+        path = "/openconfig-interfaces:interfaces/interface/subinterfaces/subinterface"
+
+        def extract_elements(self) -> Iterator[Tuple[str, Any]]:
+            for k, v in self.native.get("vif", {}).items():
+                yield k, v
+            for k_s, v_s in self.native.get("vif-s", {}).items():
+                # yield k_s, v_s
+                for k_c, v_c in v_s.get("vif-c", {}).items():
+                    yield (k_s, k_c), v_c
+
+    config = SubinterfaceConfig
+    vlan = vlan.Vlan
+
+    def index(self) -> int:
+        if isinstance(self.yy.key, str):
+            return int(self.yy.key)
+        vlan_s = self.yy.key[0]
+        vlan_c = self.yy.key[1]
+        prepends = 4 - len(vlan_c)
+        vlan_c = ("0" * prepends) + vlan_c
+        return int(vlan_s + vlan_c)
+
+
+class Subinterfaces(Parser):
+    class Yangify(ParserData):
+        path = "/openconfig-interfaces:interfaces/interface/subinterfaces"
+
+    subinterface = Subinterface
+
+
+class InterfaceConfig(Parser):
+    class Yangify(ParserData):
+        path = "/openconfig-interfaces:interfaces/interface/config"
+
+    def description(self) -> Optional[str]:
+        return self.yy.native.get("description")
+
+    def enabled(self) -> bool:
+        return "disable" not in self.yy.native
+
+    def name(self) -> str:
+        return cast(str, self.yy.key)
+
+    def type(self) -> Optional[str]:
+        if self.yy.key.startswith("eth"):
+            return "iana-if-type:ethernetCsmacd"
+        elif self.yy.key.startswith("lo"):
+            return "iana-if-type:softwareLoopback"
+        elif self.yy.key.startswith("bond"):
+            return "iana-if-type:ieee8023adLag"
+        elif self.yy.key.startswith("br"):
+            return "iana-if-type:bridge"
+        elif self.yy.key.startswith("macsec"):
+            return "iana-if-type:macSecControlledIF"
+        elif self.yy.key.startswith("tunnel"):
+            return "iana-if-type:tunnel"
+        else:
+            raise Exception(f"don't know the type for {self.yy.key}")
+
+    def mtu(self) -> Optional[int]:
+        mtu = self.yy.native.get("mtu")
+        return int(mtu) if mtu else None
+
+
+class Interface(Parser):
+    class Yangify(ParserData):
+        path = "/openconfig-interfaces:interfaces/interface"
+
+        def extract_elements(self) -> Iterator[Tuple[str, Any]]:
+            if "interfaces" in self.native:
+                for interfaces in self.native["interfaces"].values():
+                    for k, v in interfaces.items():
+                        yield k, v
+
+    config = InterfaceConfig
+    subinterfaces = Subinterfaces
+    ethernet = ethernet.Ethernet
+
+    def name(self) -> str:
+        return cast(str, self.yy.key)
+
+
+class Interfaces(Parser):
+    interface = Interface
+
+    class Yangify(ParserData):
+        path = "/openconfig-interfaces:interfaces"
+        metadata = {"key": "dev_conf", "rpc": "showConfig"}
+
+        def pre_process(self) -> None:
+            self.native: Dict[str, Any] = self.root_native["dev_conf"]

ntc_rosetta/parsers/openconfig/vyos/openconfig_network_instance/network_instances.py

@@ -0,0 +1,41 @@
+from typing import Any, Dict, Iterator, Tuple, cast
+
+from yangify.parser import Parser, ParserData
+
+
+class NetworkInstanceConfig(Parser):
+    class Yangify(ParserData):
+        path = "/openconfig-network-instance:network-instances/network-instance/config"
+
+    def name(self) -> str:
+        return cast(str, self.yy.key)
+
+    def description(self) -> str:
+        return self.yy.native.get("description")
+
+
+class NetworkInstance(Parser):
+    class Yangify(ParserData):
+        path = "/openconfig-network-instance:network-instances/network-instance"
+
+        def extract_elements(self) -> Iterator[Tuple[str, Dict[str, Any]]]:
+            yield "default", self.root_native["dev_conf"]
+            if "vrf" in self.native:
+                for k, v in self.native["vrf"]["name"].items():
+                    yield k, v
+
+    config = NetworkInstanceConfig
+
+    def name(self) -> str:
+        return cast(str, self.yy.key)
+
+
+class NetworkInstances(Parser):
+    class Yangify(ParserData):
+        path = "/openconfig-network-instance:network-instances"
+        metadata = {"key": "dev_conf", "rpc": "showConfig"}
+
+        def pre_process(self) -> None:
+            self.native: Dict[str, Any] = self.root_native["dev_conf"]
+
+    network_instance = NetworkInstance