tools/net/ynl/pyynl/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 os
   7 import pathlib
   8 import pprint
   9 import sys
  10
  11 sys.path.append(pathlib.Path(__file__).resolve().parent.as_posix())
  12 from lib import YnlFamily, Netlink, NlError
  13
  14 sys_schema_dir='/usr/share/ynl'
  15 relative_schema_dir='../../../../Documentation/netlink'
  16
  17 def 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
  26 def 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
  33 class 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
  42 def 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
 159 if __name__ == "__main__":
 160     main()