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