From 812a4fc3f3063741da0fa01cbbbf17dead66a664 Mon Sep 17 00:00:00 2001
From: Daniil Baturin <daniil@vyos.io>
Date: Thu, 9 Jun 2022 09:16:53 -0400
Subject: T2719: prototype of an op mode command runner based on type hints and
introspection
---
op-mode-definitions/ipv6-route.xml.in | 6 +-
op-mode-definitions/show-arp.xml.in | 4 +-
op-mode-definitions/show-ip.xml.in | 6 +-
python/vyos/opmode.py | 123 ++++++++++++++++++++++++++++++++++
src/op_mode/neighbor.py | 104 ++++++++++++++++++++++++++++
src/op_mode/show_neigh.py | 102 ----------------------------
6 files changed, 235 insertions(+), 110 deletions(-)
create mode 100644 python/vyos/opmode.py
create mode 100755 src/op_mode/neighbor.py
delete mode 100755 src/op_mode/show_neigh.py
diff --git a/op-mode-definitions/ipv6-route.xml.in b/op-mode-definitions/ipv6-route.xml.in
index 4f8792f9f..50f34dd56 100644
--- a/op-mode-definitions/ipv6-route.xml.in
+++ b/op-mode-definitions/ipv6-route.xml.in
@@ -20,7 +20,7 @@
<properties>
<help>Show IPv6 neighbor (NDP) table</help>
</properties>
- <command>${vyos_op_scripts_dir}/show_neigh.py --family inet6</command>
+ <command>${vyos_op_scripts_dir}/neighbor.py --family inet6</command>
<children>
<tagNode name="interface">
<properties>
@@ -29,7 +29,7 @@
<script>${vyos_completion_dir}/list_interfaces.py -b</script>
</completionHelp>
</properties>
- <command>${vyos_op_scripts_dir}/show_neigh.py --family inet6 --interface "$5"</command>
+ <command>${vyos_op_scripts_dir}/neighbor.py --family inet6 --interface "$5"</command>
</tagNode>
<tagNode name="state">
<properties>
@@ -38,7 +38,7 @@
<list>reachable stale failed permanent</list>
</completionHelp>
</properties>
- <command>${vyos_op_scripts_dir}/show_neigh.py --family inet6 --state "$5"</command>
+ <command>${vyos_op_scripts_dir}/neighbor.py --family inet6 --state "$5"</command>
</tagNode>
</children>
</node>
diff --git a/op-mode-definitions/show-arp.xml.in b/op-mode-definitions/show-arp.xml.in
index 58cc6e45e..11ac6a252 100644
--- a/op-mode-definitions/show-arp.xml.in
+++ b/op-mode-definitions/show-arp.xml.in
@@ -6,7 +6,7 @@
<properties>
<help>Show Address Resolution Protocol (ARP) information</help>
</properties>
- <command>${vyos_op_scripts_dir}/show_neigh.py --family inet</command>
+ <command>${vyos_op_scripts_dir}/neighbor.py --family inet</command>
<children>
<tagNode name="interface">
<properties>
@@ -15,7 +15,7 @@
<script>${vyos_completion_dir}/list_interfaces.py -b</script>
</completionHelp>
</properties>
- <command>${vyos_op_scripts_dir}/show_neigh.py --family inet --interface "$4"</command>
+ <command>${vyos_op_scripts_dir}/neighbor.py --family inet --interface "$4"</command>
</tagNode>
</children>
</node>
diff --git a/op-mode-definitions/show-ip.xml.in b/op-mode-definitions/show-ip.xml.in
index d342ac192..1903aca05 100644
--- a/op-mode-definitions/show-ip.xml.in
+++ b/op-mode-definitions/show-ip.xml.in
@@ -11,7 +11,7 @@
<properties>
<help>Show IPv4 neighbor (ARP) table</help>
</properties>
- <command>${vyos_op_scripts_dir}/show_neigh.py --family inet</command>
+ <command>${vyos_op_scripts_dir}/neighbor.py --family inet</command>
<children>
<tagNode name="interface">
<properties>
@@ -20,7 +20,7 @@
<script>${vyos_completion_dir}/list_interfaces.py -b</script>
</completionHelp>
</properties>
- <command>${vyos_op_scripts_dir}/show_neigh.py --family inet --interface "$5"</command>
+ <command>${vyos_op_scripts_dir}/neighbor.py --family inet --interface "$5"</command>
</tagNode>
<tagNode name="state">
<properties>
@@ -29,7 +29,7 @@
<list>reachable stale failed permanent</list>
</completionHelp>
</properties>
- <command>${vyos_op_scripts_dir}/show_neigh.py --family inet --state "$5"</command>
+ <command>${vyos_op_scripts_dir}/neighbor.py --family inet --state "$5"</command>
</tagNode>
</children>
</node>
diff --git a/python/vyos/opmode.py b/python/vyos/opmode.py
new file mode 100644
index 000000000..c0812c7fd
--- /dev/null
+++ b/python/vyos/opmode.py
@@ -0,0 +1,123 @@
+# Copyright 2022 VyOS maintainers and contributors <maintainers@vyos.io>
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library. If not, see <http://www.gnu.org/licenses/>.
+
+import typing
+
+
+def _is_op_mode_function_name(name):
+ from re import match
+
+ if match(r"^(show|clear|reset|restart)", name):
+ return True
+ else:
+ return False
+
+def _is_show(name):
+ from re import match
+
+ if match(r"^show", name):
+ return True
+ else:
+ return False
+
+def _get_op_mode_functions(module):
+ from inspect import getmembers, isfunction
+
+ # Get all functions in that module
+ funcs = getmembers(module, isfunction)
+
+ # getmembers returns (name, func) tuples
+ funcs = list(filter(lambda ft: _is_op_mode_function_name(ft[0]), funcs))
+
+ funcs_dict = {}
+ for (name, thunk) in funcs:
+ funcs_dict[name] = thunk
+
+ return funcs_dict
+
+def _is_optional_type(t):
+ # Optional[t] is internally an alias for Union[t, NoneType]
+ # and there's no easy way to get union members it seems
+ if (type(t) == typing._UnionGenericAlias):
+ if (len(t.__args__) == 2):
+ if t.__args__[1] == type(None):
+ return True
+
+ return False
+
+def _get_arg_type(t):
+ """ Returns the type itself if it's a primitive type,
+ or the "real" type of typing.Optional
+
+ Doesn't work with anything else at the moment!
+ """
+ if _is_optional_type(t):
+ t.__args__[0]
+ else:
+ return t
+
+def run(module):
+ from argparse import ArgumentParser
+
+ functions = _get_op_mode_functions(module)
+
+ parser = ArgumentParser()
+ subparsers = parser.add_subparsers(dest="subcommand")
+
+ for function_name in functions:
+ subparser = subparsers.add_parser(function_name, help=functions[function_name].__doc__)
+
+ type_hints = typing.get_type_hints(functions[function_name])
+ for opt in type_hints:
+ th = type_hints[opt]
+
+ # Show commands require an option to choose between raw JSON and human-readable
+ # formatted output.
+ # For interactive use, they default to formatted output.
+ if (function_name == "show") and (opt == "raw"):
+ subparser.add_argument(f"--raw", action='store_true')
+ elif _is_optional_type(th):
+ subparser.add_argument(f"--{opt}", type=_get_arg_type(th), default=None)
+ else:
+ subparser.add_argument(f"--{opt}", type=_get_arg_type(th), required=True)
+
+ # Get options as a dict rather than a namespace,
+ # so that we can modify it and pack for passing to functions
+ args = vars(parser.parse_args())
+
+ func = functions[args["subcommand"]]
+
+ # Remove the subcommand from the arguments,
+ # it would cause an extra argument error when we pass the dict to a function
+ del args["subcommand"]
+
+ if "raw" not in args:
+ args["raw"] = False
+
+ if function_name == "show":
+ # Show commands are slightly special:
+ # they may return human-formatted output
+ # or a raw dict that we need to serialize in JSON for printing
+ res = func(**args)
+ if not args["raw"]:
+ return res
+ else:
+ from json import dumps
+ return dumps(res, indent=4)
+ else:
+ # Other functions should not return anything,
+ # although they may print their own warnings or status messages
+ func(**args)
+
diff --git a/src/op_mode/neighbor.py b/src/op_mode/neighbor.py
new file mode 100755
index 000000000..b50e75d91
--- /dev/null
+++ b/src/op_mode/neighbor.py
@@ -0,0 +1,104 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2022 VyOS maintainers and contributors
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 or later as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+# Sample output of `ip --json neigh list`:
+#
+# [
+# {
+# "dst": "192.168.1.1",
+# "dev": "eth0", # Missing if `dev ...` option is used
+# "lladdr": "00:aa:bb:cc:dd:ee", # May be missing for failed entries
+# "state": [
+# "REACHABLE"
+# ]
+# },
+# ]
+
+import sys
+import typing
+
+import opmode
+
+
+def get_raw_data(family, interface=None, state=None):
+ from json import loads
+ from vyos.util import cmd
+
+ if interface:
+ interface = f"dev {interface}"
+ else:
+ interface = ""
+
+ if state:
+ state = f"nud {state}"
+ else:
+ state = ""
+
+ neigh_cmd = f"ip --family {family} --json neighbor list {interface} {state}"
+
+ data = loads(cmd(neigh_cmd))
+
+ return data
+
+def format_neighbors(neighs, interface=None):
+ from tabulate import tabulate
+
+ def entry_to_list(e, intf=None):
+ dst = e["dst"]
+
+ # State is always a list in the iproute2 output
+ state = ", ".join(e["state"])
+
+ # Link layer address is absent from e.g. FAILED entries
+ if "lladdr" in e:
+ lladdr = e["lladdr"]
+ else:
+ lladdr = None
+
+ # Device field is absent from outputs of `ip neigh list dev ...`
+ if "dev" in e:
+ dev = e["dev"]
+ elif interface:
+ dev = interface
+ else:
+ raise ValueError("interface is not defined")
+
+ return [dst, dev, lladdr, state]
+
+ neighs = map(entry_to_list, neighs)
+
+ headers = ["Address", "Interface", "Link layer address", "State"]
+ return tabulate(neighs, headers)
+
+def show(raw: bool, family: str, interface: typing.Optional[str], state: typing.Optional[str]):
+ """ Display neighbor table contents """
+ data = get_raw_data(family, interface, state=state)
+
+ if raw:
+ return data
+ else:
+ return format_neighbors(data, interface)
+
+
+if __name__ == '__main__':
+ from argparse import ArgumentParser
+
+ try:
+ print(opmode.run(sys.modules[__name__]))
+ except ValueError as e:
+ print(e)
+ sys.exit(1)
+
diff --git a/src/op_mode/show_neigh.py b/src/op_mode/show_neigh.py
deleted file mode 100755
index d874bd544..000000000
--- a/src/op_mode/show_neigh.py
+++ /dev/null
@@ -1,102 +0,0 @@
-#!/usr/bin/env python3
-#
-# Copyright (C) 2022 VyOS maintainers and contributors
-#
-# This program is free software; you can redistribute it and/or modify
-# it under the terms of the GNU General Public License version 2 or later as
-# published by the Free Software Foundation.
-#
-# This program is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU General Public License for more details.
-#
-# You should have received a copy of the GNU General Public License
-# along with this program. If not, see <http://www.gnu.org/licenses/>.
-
-# Sample output of `ip --json neigh list`:
-#
-# [
-# {
-# "dst": "192.168.1.1",
-# "dev": "eth0", # Missing if `dev ...` option is used
-# "lladdr": "00:aa:bb:cc:dd:ee", # May be missing for failed entries
-# "state": [
-# "REACHABLE"
-# ]
-# },
-# ]
-
-import sys
-
-
-def get_raw_data(family, device=None, state=None):
- from json import loads
- from vyos.util import cmd
-
- if device:
- device = f"dev {device}"
- else:
- device = ""
-
- if state:
- state = f"nud {state}"
- else:
- state = ""
-
- neigh_cmd = f"ip --family {family} --json neighbor list {device} {state}"
-
- data = loads(cmd(neigh_cmd))
-
- return data
-
-def get_formatted_output(family, device=None, state=None):
- from tabulate import tabulate
-
- def entry_to_list(e, intf=None):
- dst = e["dst"]
-
- # State is always a list in the iproute2 output
- state = ", ".join(e["state"])
-
- # Link layer address is absent from e.g. FAILED entries
- if "lladdr" in e:
- lladdr = e["lladdr"]
- else:
- lladdr = None
-
- # Device field is absent from outputs of `ip neigh list dev ...`
- if "dev" in e:
- dev = e["dev"]
- elif device:
- dev = device
- else:
- raise ValueError("interface is not defined")
-
- return [dst, dev, lladdr, state]
-
- neighs = get_raw_data(family, device=device, state=state)
- neighs = map(entry_to_list, neighs)
-
- headers = ["Address", "Interface", "Link layer address", "State"]
- return tabulate(neighs, headers)
-
-if __name__ == '__main__':
- from argparse import ArgumentParser
-
- parser = ArgumentParser()
- parser.add_argument("-f", "--family", type=str, default="inet", help="Address family")
- parser.add_argument("-i", "--interface", type=str, help="Network interface")
- parser.add_argument("-s", "--state", type=str, help="Neighbor table entry state")
-
- args = parser.parse_args()
-
- if args.state:
- if args.state not in ["reachable", "failed", "stale", "permanent"]:
- raise ValueError(f"""Incorrect state "{args.state}"! Must be one of: reachable, stale, failed, permanent""")
-
- try:
- print(get_formatted_output(args.family, device=args.interface, state=args.state))
- except ValueError as e:
- print(e)
- sys.exit(1)
--
cgit v1.2.3