show: add new show tool written in python3

The reason for this is:
1) We want to use the same fetch tool for all operational data, to
   avoid having duplicated xpaths scattered across the repo.
2) We want to call this directly from the CLI, which means it need to
   be escape safe. A unprivileged user shall be able to use it but not
   escape into a shell or do other malicious stuff.

Signed-off-by: Richard Alpe <richard@bit42.se>
This commit is contained in:
Richard Alpe
2025-04-07 09:06:26 +02:00
parent 827dc098e4
commit c584af356b
13 changed files with 255 additions and 0 deletions
View File
+1
View File
@@ -139,6 +139,7 @@ BR2_PACKAGE_CURIOS_HTTPD=y
BR2_PACKAGE_CURIOS_NFTABLES=y
BR2_PACKAGE_GENCERT=y
BR2_PACKAGE_STATD=y
BR2_PACKAGE_SHOW=y
BR2_PACKAGE_FACTORY=y
BR2_PACKAGE_FINIT_PLUGIN_HOTPLUG=y
BR2_PACKAGE_FINIT_PLUGIN_HOOK_SCRIPTS=y
+1
View File
@@ -119,6 +119,7 @@ BR2_PACKAGE_CONFD=y
BR2_PACKAGE_CONFD_TEST_MODE=y
BR2_PACKAGE_GENCERT=y
BR2_PACKAGE_STATD=y
BR2_PACKAGE_SHOW=y
BR2_PACKAGE_FACTORY=y
BR2_PACKAGE_FINIT_PLUGIN_HOTPLUG=y
BR2_PACKAGE_FINIT_PLUGIN_HOOK_SCRIPTS=y
+1
View File
@@ -176,6 +176,7 @@ INFIX_SUPPORT="mailto:kernelkit@googlegroups.com"
BR2_PACKAGE_CONFD=y
BR2_PACKAGE_GENCERT=y
BR2_PACKAGE_STATD=y
BR2_PACKAGE_SHOW=y
BR2_PACKAGE_FACTORY=y
BR2_PACKAGE_FINIT_PLUGIN_HOTPLUG=y
BR2_PACKAGE_FINIT_PLUGIN_HOOK_SCRIPTS=y
+1
View File
@@ -169,6 +169,7 @@ BR2_PACKAGE_CONFD=y
# BR2_PACKAGE_CONFD_TEST_MODE is not set
BR2_PACKAGE_GENCERT=y
BR2_PACKAGE_STATD=y
BR2_PACKAGE_SHOW=y
BR2_PACKAGE_FACTORY=y
BR2_PACKAGE_FINIT_PLUGIN_HOTPLUG=y
BR2_PACKAGE_FINIT_PLUGIN_HOOK_SCRIPTS=y
+1
View File
@@ -141,6 +141,7 @@ BR2_PACKAGE_CURIOS_HTTPD=y
BR2_PACKAGE_CURIOS_NFTABLES=y
BR2_PACKAGE_GENCERT=y
BR2_PACKAGE_STATD=y
BR2_PACKAGE_SHOW=y
BR2_PACKAGE_FACTORY=y
BR2_PACKAGE_FINIT_PLUGIN_HOTPLUG=y
BR2_PACKAGE_FINIT_PLUGIN_HOOK_SCRIPTS=y
+1
View File
@@ -123,6 +123,7 @@ BR2_PACKAGE_CONFD=y
BR2_PACKAGE_CONFD_TEST_MODE=y
BR2_PACKAGE_GENCERT=y
BR2_PACKAGE_STATD=y
BR2_PACKAGE_SHOW=y
BR2_PACKAGE_FACTORY=y
BR2_PACKAGE_FINIT_PLUGIN_HOTPLUG=y
BR2_PACKAGE_FINIT_PLUGIN_HOOK_SCRIPTS=y
+1
View File
@@ -34,6 +34,7 @@ source "$BR2_EXTERNAL_INFIX_PATH/package/python-libyang/Config.in"
source "$BR2_EXTERNAL_INFIX_PATH/package/python-statd/Config.in"
source "$BR2_EXTERNAL_INFIX_PATH/package/python-yangdoc/Config.in"
source "$BR2_EXTERNAL_INFIX_PATH/package/skeleton-init-finit/Config.in"
source "$BR2_EXTERNAL_INFIX_PATH/package/show/Config.in"
source "$BR2_EXTERNAL_INFIX_PATH/package/tetris/Config.in"
source "$BR2_EXTERNAL_INFIX_PATH/package/libyang-cpp/Config.in"
source "$BR2_EXTERNAL_INFIX_PATH/package/sysrepo-cpp/Config.in"
+6
View File
@@ -0,0 +1,6 @@
config BR2_PACKAGE_SHOW
bool "show"
help
Tool to retrieve and present operational data from sysrepo.
https://github.com/kernelkit/infix
+19
View File
@@ -0,0 +1,19 @@
################################################################################
#
# show
#
################################################################################
SHOW_VERSION = 1.0
SHOW_LICENSE = MIT
SHOW_LICENSE_FILES = LICENSE
SHOW_SITE_METHOD = local
SHOW_SITE = $(BR2_EXTERNAL_INFIX_PATH)/src/show
SHOW_REDISTRIBUTE = NO
define SHOW_INSTALL_TARGET_CMDS
$(TARGET_MAKE_ENV) $(TARGET_CONFIGURE_OPTS) $(MAKE) -C $(@D) \
DESTDIR="$(TARGET_DIR)" install
endef
$(eval $(generic-package))
+21
View File
@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2025 Richard Alpe
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
+10
View File
@@ -0,0 +1,10 @@
.PHONY: all clean distclean install
all:
clean:
distclean: clean
install:
install -D show.py $(DESTDIR)/bin/show
+192
View File
@@ -0,0 +1,192 @@
#!/usr/bin/env python3
import re
import subprocess
import json
from typing import List
import os
import shlex
import argparse
RAW_OUTPUT = False
def run_sysrepocfg(xpath: str) -> dict:
if not isinstance(xpath, str) or not xpath.startswith("/"):
print("Invalid XPATH. It must be a valid string starting with '/'.")
return {}
safe_xpath = shlex.quote(xpath)
try:
result = subprocess.run([
"sysrepocfg", "-f", "json", "-X", "-d", "operational", "-x", safe_xpath
], capture_output=True, text=True, check=True)
json_data = json.loads(result.stdout)
return json_data
except subprocess.CalledProcessError as e:
print(f"Error running sysrepocfg: {e}")
return {}
except json.JSONDecodeError as e:
print(f"Error parsing JSON output: {e}")
return {}
def cli_pretty(json_data: dict, command: str, *args: str):
if not command or not all(isinstance(arg, str) for arg in args):
print("Invalid command or arguments. All arguments must be strings.")
return
safe_args = [shlex.quote(arg) for arg in args]
try:
json_input = json.dumps(json_data) # Keep as string, not bytes
result = subprocess.run([
"/usr/libexec/statd/cli-pretty", command, *safe_args
], input=json_input, capture_output=True, text=True, check=True)
print(result.stdout, end="")
except subprocess.CalledProcessError as e:
print(f"Error running cli-pretty: {e}")
def dhcp(args: List[str]) -> None:
data = run_sysrepocfg("/infix-dhcp-server:dhcp-server")
if not data:
print("No interface data retrieved.")
return
if RAW_OUTPUT:
print(json.dumps(data, indent=2))
return
cli_pretty(data, "show-dhcp-server")
def hardware(args: List[str]) -> None:
data = run_sysrepocfg("/ietf-hardware:hardware")
if not data:
print("No hardware data retrieved.")
return
if RAW_OUTPUT:
print(json.dumps(data, indent=2))
return
cli_pretty(data, "show-hardware")
def ntp(args: List[str]) -> None:
data = run_sysrepocfg("/system-state/ntp")
if not data:
print("No ntp data retrieved.")
return
if RAW_OUTPUT:
print(json.dumps(data, indent=2))
return
cli_pretty(data, "show-ntp")
def is_valid_interface_name(interface_name: str) -> bool:
"""
Validates a Linux network interface name.
"""
if len(interface_name) > 15:
return False
pattern = r'^[a-zA-Z0-9._-]+$'
return bool(re.match(pattern, interface_name))
def interface(args: List[str]) -> None:
data = run_sysrepocfg("/ietf-interfaces:interfaces")
if not data:
print("No interface data retrieved.")
return
if RAW_OUTPUT:
print(json.dumps(data, indent=2))
return
if len(args) == 0:
cli_pretty(data, "show-interfaces")
elif len(args) == 1:
iface = args[0]
if is_valid_interface_name(iface):
cli_pretty(data, f"show-interfaces", "-n", iface)
else:
print(f"Invalid interface name: {iface}")
else:
print("Too many arguments provided. Only one interface name is expected.")
def stp(args: List[str]) -> None:
data = run_sysrepocfg("/ietf-interfaces:interfaces")
if not data:
print("No interface data retrieved.")
return
if RAW_OUTPUT:
print(json.dumps(data, indent=2))
return
cli_pretty(data, "show-bridge-stp")
def software(args: List[str]) -> None:
data = run_sysrepocfg("/ietf-system:system-state/infix-system:software")
if not data:
print("No software data retrieved.")
return
if RAW_OUTPUT:
print(json.dumps(data, indent=2))
return
if len(args) == 0:
cli_pretty(data, "show-software")
print(".....")
elif len(args) == 1:
name = args[0]
if name not in ("primary", "secondary"):
print(f"Invalid software name: {name}")
return
cli_pretty(data, f"show-software", "-n", name)
else:
print("Too many arguments provided. Only one name is expected.")
def routes(args: List[str]):
ip_version = args[0] if args and args[0] in ["ipv4", "ipv6"] else "ipv4"
data = run_sysrepocfg("/ietf-routing:routing/ribs")
if not data:
print("No route data retrieved.")
return
if RAW_OUTPUT:
print(json.dumps(data, indent=2))
return
cli_pretty(data, "show-routing-table", "-i", ip_version)
def execute_command(command: str, args: List[str]):
command_mapping = {
'dhcp': dhcp,
'hardware': hardware,
'interface': interface,
'ntp': ntp,
'routes': routes,
'software' : software,
'stp': stp,
}
if command in command_mapping:
command_mapping[command](args)
else:
print(f"Unknown command: {command}")
def main():
global RAW_OUTPUT
parser = argparse.ArgumentParser(description="Show operational data")
parser.add_argument('command', help="Command to execute")
parser.add_argument('args', nargs=argparse.REMAINDER, help="Additional arguments for the command")
parser.add_argument('-r', '--raw', action='store_true', help="Print raw JSON output from Sysrepo")
args = parser.parse_args()
RAW_OUTPUT = args.raw
execute_command(args.command, args.args)
if __name__ == "__main__":
main()