tools/net/ynl/cli.py

   1 #!/usr/bin/env python3
   2 # SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause
   3
   4 import argparse
   5 import json
   6 import pathlib
   7 import pprint
   8 import sys
   9
  10 sys.path.append(pathlib.Path(__file__).resolve().parent.as_posix())
  11 from lib import YnlFamily, Netlink, NlError
  12
  13
  14 class 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
  23 def 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
 118 if __name__ == "__main__":
 119     main()