1#!/usr/bin/env python3 2# SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause 3 4import argparse 5import json 6import os 7import pathlib 8import pprint 9import sys 10 11sys.path.append(pathlib.Path(__file__).resolve().parent.as_posix()) 12from lib import YnlFamily, Netlink, NlError 13 14sys_schema_dir='/usr/share/ynl' 15relative_schema_dir='../../../../Documentation/netlink' 16 17def schema_dir(): 18 script_dir = os.path.dirname(os.path.abspath(__file__)) 19 schema_dir = os.path.abspath(f"{script_dir}/{relative_schema_dir}") 20 if not os.path.isdir(schema_dir): 21 schema_dir = sys_schema_dir 22 if not os.path.isdir(schema_dir): 23 raise Exception(f"Schema directory {schema_dir} does not exist") 24 return schema_dir 25 26def spec_dir(): 27 spec_dir = schema_dir() + '/specs' 28 if not os.path.isdir(spec_dir): 29 raise Exception(f"Spec directory {spec_dir} does not exist") 30 return spec_dir 31 32 33class YnlEncoder(json.JSONEncoder): 34 def default(self, obj): 35 if isinstance(obj, bytes): 36 return bytes.hex(obj) 37 if isinstance(obj, set): 38 return list(obj) 39 return json.JSONEncoder.default(self, obj) 40 41 42def main(): 43 description = """ 44 YNL CLI utility - a general purpose netlink utility that uses YAML 45 specs to drive protocol encoding and decoding. 46 """ 47 epilog = """ 48 The --multi option can be repeated to include several do operations 49 in the same netlink payload. 50 """ 51 52 parser = argparse.ArgumentParser(description=description, 53 epilog=epilog) 54 spec_group = parser.add_mutually_exclusive_group(required=True) 55 spec_group.add_argument('--family', dest='family', type=str, 56 help='name of the netlink FAMILY') 57 spec_group.add_argument('--list-families', action='store_true', 58 help='list all netlink families supported by YNL (has spec)') 59 spec_group.add_argument('--spec', dest='spec', type=str, 60 help='choose the family by SPEC file path') 61 62 parser.add_argument('--schema', dest='schema', type=str) 63 parser.add_argument('--no-schema', action='store_true') 64 parser.add_argument('--json', dest='json_text', type=str) 65 66 group = parser.add_mutually_exclusive_group() 67 group.add_argument('--do', dest='do', metavar='DO-OPERATION', type=str) 68 group.add_argument('--multi', dest='multi', nargs=2, action='append', 69 metavar=('DO-OPERATION', 'JSON_TEXT'), type=str) 70 group.add_argument('--dump', dest='dump', metavar='DUMP-OPERATION', type=str) 71 group.add_argument('--list-ops', action='store_true') 72 group.add_argument('--list-msgs', action='store_true') 73 74 parser.add_argument('--duration', dest='duration', type=int, 75 help='when subscribed, watch for DURATION seconds') 76 parser.add_argument('--sleep', dest='duration', type=int, 77 help='alias for duration') 78 parser.add_argument('--subscribe', dest='ntf', type=str) 79 parser.add_argument('--replace', dest='flags', action='append_const', 80 const=Netlink.NLM_F_REPLACE) 81 parser.add_argument('--excl', dest='flags', action='append_const', 82 const=Netlink.NLM_F_EXCL) 83 parser.add_argument('--create', dest='flags', action='append_const', 84 const=Netlink.NLM_F_CREATE) 85 parser.add_argument('--append', dest='flags', action='append_const', 86 const=Netlink.NLM_F_APPEND) 87 parser.add_argument('--process-unknown', action=argparse.BooleanOptionalAction) 88 parser.add_argument('--output-json', action='store_true') 89 parser.add_argument('--dbg-small-recv', default=0, const=4000, 90 action='store', nargs='?', type=int) 91 args = parser.parse_args() 92 93 def output(msg): 94 if args.output_json: 95 print(json.dumps(msg, cls=YnlEncoder)) 96 else: 97 pprint.PrettyPrinter().pprint(msg) 98 99 if args.list_families: 100 for filename in sorted(os.listdir(spec_dir())): 101 if filename.endswith('.yaml'): 102 print(filename.removesuffix('.yaml')) 103 return 104 105 if args.no_schema: 106 args.schema = '' 107 108 attrs = {} 109 if args.json_text: 110 attrs = json.loads(args.json_text) 111 112 if args.family: 113 spec = f"{spec_dir()}/{args.family}.yaml" 114 if args.schema is None and spec.startswith(sys_schema_dir): 115 args.schema = '' # disable schema validation when installed 116 else: 117 spec = args.spec 118 if not os.path.isfile(spec): 119 raise Exception(f"Spec file {spec} does not exist") 120 121 ynl = YnlFamily(spec, args.schema, args.process_unknown, 122 recv_size=args.dbg_small_recv) 123 if args.dbg_small_recv: 124 ynl.set_recv_dbg(True) 125 126 if args.ntf: 127 ynl.ntf_subscribe(args.ntf) 128 129 if args.list_ops: 130 for op_name, op in ynl.ops.items(): 131 print(op_name, " [", ", ".join(op.modes), "]") 132 if args.list_msgs: 133 for op_name, op in ynl.msgs.items(): 134 print(op_name, " [", ", ".join(op.modes), "]") 135 136 try: 137 if args.do: 138 reply = ynl.do(args.do, attrs, args.flags) 139 output(reply) 140 if args.dump: 141 reply = ynl.dump(args.dump, attrs) 142 output(reply) 143 if args.multi: 144 ops = [ (item[0], json.loads(item[1]), args.flags or []) for item in args.multi ] 145 reply = ynl.do_multi(ops) 146 output(reply) 147 except NlError as e: 148 print(e) 149 exit(1) 150 151 if args.ntf: 152 try: 153 for msg in ynl.poll_ntf(duration=args.duration): 154 output(msg) 155 except KeyboardInterrupt: 156 pass 157 158 159if __name__ == "__main__": 160 main() 161