1#!/usr/bin/env python3 2# SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause 3 4import argparse 5import json 6import pathlib 7import pprint 8import sys 9 10sys.path.append(pathlib.Path(__file__).resolve().parent.as_posix()) 11from lib import YnlFamily, Netlink, NlError 12 13 14class YnlEncoder(json.JSONEncoder): 15 def default(self, obj): 16 if isinstance(obj, bytes): 17 return bytes.hex(obj) 18 if isinstance(obj, set): 19 return list(obj) 20 return json.JSONEncoder.default(self, obj) 21 22 23def main(): 24 description = """ 25 YNL CLI utility - a general purpose netlink utility that uses YAML 26 specs to drive protocol encoding and decoding. 27 """ 28 epilog = """ 29 The --multi option can be repeated to include several do operations 30 in the same netlink payload. 31 """ 32 33 parser = argparse.ArgumentParser(description=description, 34 epilog=epilog) 35 parser.add_argument('--spec', dest='spec', type=str, required=True) 36 parser.add_argument('--schema', dest='schema', type=str) 37 parser.add_argument('--no-schema', action='store_true') 38 parser.add_argument('--json', dest='json_text', type=str) 39 40 group = parser.add_mutually_exclusive_group() 41 group.add_argument('--do', dest='do', metavar='DO-OPERATION', type=str) 42 group.add_argument('--multi', dest='multi', nargs=2, action='append', 43 metavar=('DO-OPERATION', 'JSON_TEXT'), type=str) 44 group.add_argument('--dump', dest='dump', metavar='DUMP-OPERATION', type=str) 45 group.add_argument('--list-ops', action='store_true') 46 group.add_argument('--list-msgs', action='store_true') 47 48 parser.add_argument('--duration', dest='duration', type=int, 49 help='when subscribed, watch for DURATION seconds') 50 parser.add_argument('--sleep', dest='duration', type=int, 51 help='alias for duration') 52 parser.add_argument('--subscribe', dest='ntf', type=str) 53 parser.add_argument('--replace', dest='flags', action='append_const', 54 const=Netlink.NLM_F_REPLACE) 55 parser.add_argument('--excl', dest='flags', action='append_const', 56 const=Netlink.NLM_F_EXCL) 57 parser.add_argument('--create', dest='flags', action='append_const', 58 const=Netlink.NLM_F_CREATE) 59 parser.add_argument('--append', dest='flags', action='append_const', 60 const=Netlink.NLM_F_APPEND) 61 parser.add_argument('--process-unknown', action=argparse.BooleanOptionalAction) 62 parser.add_argument('--output-json', action='store_true') 63 parser.add_argument('--dbg-small-recv', default=0, const=4000, 64 action='store', nargs='?', type=int) 65 args = parser.parse_args() 66 67 def output(msg): 68 if args.output_json: 69 print(json.dumps(msg, cls=YnlEncoder)) 70 else: 71 pprint.PrettyPrinter().pprint(msg) 72 73 if args.no_schema: 74 args.schema = '' 75 76 attrs = {} 77 if args.json_text: 78 attrs = json.loads(args.json_text) 79 80 ynl = YnlFamily(args.spec, args.schema, args.process_unknown, 81 recv_size=args.dbg_small_recv) 82 if args.dbg_small_recv: 83 ynl.set_recv_dbg(True) 84 85 if args.ntf: 86 ynl.ntf_subscribe(args.ntf) 87 88 if args.list_ops: 89 for op_name, op in ynl.ops.items(): 90 print(op_name, " [", ", ".join(op.modes), "]") 91 if args.list_msgs: 92 for op_name, op in ynl.msgs.items(): 93 print(op_name, " [", ", ".join(op.modes), "]") 94 95 try: 96 if args.do: 97 reply = ynl.do(args.do, attrs, args.flags) 98 output(reply) 99 if args.dump: 100 reply = ynl.dump(args.dump, attrs) 101 output(reply) 102 if args.multi: 103 ops = [ (item[0], json.loads(item[1]), args.flags or []) for item in args.multi ] 104 reply = ynl.do_multi(ops) 105 output(reply) 106 except NlError as e: 107 print(e) 108 exit(1) 109 110 if args.ntf: 111 try: 112 for msg in ynl.poll_ntf(duration=args.duration): 113 output(msg) 114 except KeyboardInterrupt: 115 pass 116 117 118if __name__ == "__main__": 119 main() 120