1#!/usr/bin/env python3 2# SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause 3 4""" 5YNL cli tool 6""" 7 8import argparse 9import json 10import os 11import pathlib 12import pprint 13import shutil 14import sys 15import textwrap 16 17# pylint: disable=no-name-in-module,wrong-import-position 18sys.path.append(pathlib.Path(__file__).resolve().parent.as_posix()) 19from lib import YnlFamily, Netlink, NlError, SpecFamily, SpecException, YnlException 20from lib import list_families 21 22# pylint: disable=too-few-public-methods,too-many-locals 23class Colors: 24 """ANSI color and font modifier codes""" 25 RESET = '\033[0m' 26 27 BOLD = '\033[1m' 28 ITALICS = '\033[3m' 29 UNDERLINE = '\033[4m' 30 INVERT = '\033[7m' 31 32 33def color(text, modifiers): 34 """Add color to text if output is a TTY 35 36 Returns: 37 Colored text if stdout is a TTY, otherwise plain text 38 """ 39 if sys.stdout.isatty(): 40 # Join the colors if they are a list, if it's a string this a noop 41 modifiers = "".join(modifiers) 42 return f"{modifiers}{text}{Colors.RESET}" 43 return text 44 45def term_width(): 46 """ Get terminal width in columns (80 if stdout is not a terminal) """ 47 return shutil.get_terminal_size().columns 48 49class YnlEncoder(json.JSONEncoder): 50 """A custom encoder for emitting JSON with ynl-specific instance types""" 51 def default(self, o): 52 if isinstance(o, bytes): 53 return bytes.hex(o) 54 if isinstance(o, set): 55 return sorted(o) 56 return json.JSONEncoder.default(self, o) 57 58 59def print_attr_list(ynl, attr_names, attr_set, indent=2): 60 """Print a list of attributes with their types and documentation.""" 61 prefix = ' ' * indent 62 for attr_name in attr_names: 63 if attr_name in attr_set.attrs: 64 attr = attr_set.attrs[attr_name] 65 attr_info = f'{prefix}- {color(attr_name, Colors.BOLD)}: {attr.type}' 66 if 'enum' in attr.yaml: 67 enum_name = attr.yaml['enum'] 68 attr_info += f" (enum: {enum_name})" 69 # Print enum values if available 70 if enum_name in ynl.consts: 71 const = ynl.consts[enum_name] 72 enum_values = list(const.entries.keys()) 73 type_fmted = color(const.type.capitalize(), Colors.ITALICS) 74 attr_info += f"\n{prefix} {type_fmted}: {', '.join(enum_values)}" 75 76 # Show nested attributes reference and recursively display them 77 nested_set_name = None 78 if attr.type == 'nest' and 'nested-attributes' in attr.yaml: 79 nested_set_name = attr.yaml['nested-attributes'] 80 attr_info += f" -> {nested_set_name}" 81 82 if attr.yaml.get('doc'): 83 doc_prefix = prefix + ' ' * 4 84 doc_text = textwrap.fill(attr.yaml['doc'], width=term_width(), 85 initial_indent=doc_prefix, 86 subsequent_indent=doc_prefix) 87 attr_info += f"\n{doc_text}" 88 print(attr_info) 89 90 # Recursively show nested attributes 91 if nested_set_name in ynl.attr_sets: 92 nested_set = ynl.attr_sets[nested_set_name] 93 # Filter out 'unspec' and other unused attrs 94 nested_names = [n for n in nested_set.attrs.keys() 95 if nested_set.attrs[n].type != 'unused'] 96 if nested_names: 97 print_attr_list(ynl, nested_names, nested_set, indent + 4) 98 99 100def print_mode_attrs(ynl, mode, mode_spec, attr_set, consistent_dd_reply=None): 101 """Print a given mode (do/dump/event/notify).""" 102 mode_title = mode.capitalize() 103 104 if 'request' in mode_spec and 'attributes' in mode_spec['request']: 105 print(f'\n{mode_title} request attributes:') 106 print_attr_list(ynl, mode_spec['request']['attributes'], attr_set) 107 108 if 'reply' in mode_spec and 'attributes' in mode_spec['reply']: 109 if consistent_dd_reply and mode == "do": 110 title = None # Dump handling will print in combined format 111 elif consistent_dd_reply and mode == "dump": 112 title = 'Do and Dump' 113 else: 114 title = f'{mode_title}' 115 if title: 116 print(f'\n{title} reply attributes:') 117 print_attr_list(ynl, mode_spec['reply']['attributes'], attr_set) 118 119 120def do_doc(ynl, op): 121 """Handle --list-attrs $op, print the attr information to stdout""" 122 print(f'Operation: {color(op.name, Colors.BOLD)}') 123 print(op.yaml['doc']) 124 125 consistent_dd_reply = False 126 if 'do' in op.yaml and 'dump' in op.yaml and 'reply' in op.yaml['do'] and \ 127 op.yaml['do']['reply'] == op.yaml['dump'].get('reply'): 128 consistent_dd_reply = True 129 130 for mode in ['do', 'dump']: 131 if mode in op.yaml: 132 print_mode_attrs(ynl, mode, op.yaml[mode], op.attr_set, 133 consistent_dd_reply=consistent_dd_reply) 134 135 if 'attributes' in op.yaml.get('event', {}): 136 print('\nEvent attributes:') 137 print_attr_list(ynl, op.yaml['event']['attributes'], op.attr_set) 138 139 if 'notify' in op.yaml: 140 mode_spec = op.yaml['notify'] 141 ref_spec = ynl.msgs.get(mode_spec).yaml.get('do') 142 if not ref_spec: 143 ref_spec = ynl.msgs.get(mode_spec).yaml.get('dump') 144 if ref_spec: 145 print('\nNotification attributes:') 146 print_attr_list(ynl, ref_spec['reply']['attributes'], op.attr_set) 147 148 if 'mcgrp' in op.yaml: 149 print(f"\nMulticast group: {op.yaml['mcgrp']}") 150 151 152# pylint: disable=too-many-locals,too-many-branches,too-many-statements 153def main(): 154 """YNL cli tool""" 155 156 description = """ 157 YNL CLI utility - a general purpose netlink utility that uses YAML 158 specs to drive protocol encoding and decoding. 159 """ 160 epilog = """ 161 The --multi option can be repeated to include several do operations 162 in the same netlink payload. 163 """ 164 165 parser = argparse.ArgumentParser(description=description, 166 epilog=epilog, add_help=False) 167 168 gen_group = parser.add_argument_group('General options') 169 gen_group.add_argument('-h', '--help', action='help', 170 help='show this help message and exit') 171 172 spec_group = parser.add_argument_group('Netlink family selection') 173 spec_sel = spec_group.add_mutually_exclusive_group(required=True) 174 spec_sel.add_argument('--list-families', action='store_true', 175 help=('list Netlink families supported by YNL ' 176 '(which have a spec available in the standard ' 177 'system path)')) 178 spec_sel.add_argument('--family', dest='family', type=str, 179 help='name of the Netlink FAMILY to use') 180 spec_sel.add_argument('--spec', dest='spec', type=str, 181 help='full file path to the YAML spec file') 182 183 ops_group = parser.add_argument_group('Operations') 184 ops = ops_group.add_mutually_exclusive_group() 185 ops.add_argument('--do', dest='do', metavar='DO-OPERATION', type=str) 186 ops.add_argument('--dump', dest='dump', metavar='DUMP-OPERATION', type=str) 187 ops.add_argument('--multi', dest='multi', nargs=2, action='append', 188 metavar=('DO-OPERATION', 'JSON_TEXT'), type=str, 189 help="Multi-message operation sequence (for nftables)") 190 ops.add_argument('--list-ops', action='store_true', 191 help="List available --do and --dump operations") 192 ops.add_argument('--list-msgs', action='store_true', 193 help="List all messages of the family (incl. notifications)") 194 ops.add_argument('--list-attrs', '--doc', dest='list_attrs', metavar='MSG', 195 type=str, help='List attributes for a message / operation') 196 ops.add_argument('--validate', action='store_true', 197 help="Validate the spec against schema and exit") 198 199 io_group = parser.add_argument_group('Input / Output') 200 io_group.add_argument('--json', dest='json_text', type=str, 201 help=('Specify attributes of the message to send ' 202 'to the kernel in JSON format. Can be left out ' 203 'if the message is expected to be empty.')) 204 io_group.add_argument('--output-json', action='store_true', 205 help='Format output as JSON') 206 207 ntf_group = parser.add_argument_group('Notifications') 208 ntf_group.add_argument('--subscribe', dest='ntf', type=str) 209 ntf_group.add_argument('--duration', dest='duration', type=int, 210 help='when subscribed, watch for DURATION seconds') 211 ntf_group.add_argument('--sleep', dest='duration', type=int, 212 help='alias for duration') 213 214 nlflags = parser.add_argument_group('Netlink message flags (NLM_F_*)', 215 ('Extra flags to set in nlmsg_flags of ' 216 'the request, used mostly by older ' 217 'Classic Netlink families.')) 218 nlflags.add_argument('--replace', dest='flags', action='append_const', 219 const=Netlink.NLM_F_REPLACE) 220 nlflags.add_argument('--excl', dest='flags', action='append_const', 221 const=Netlink.NLM_F_EXCL) 222 nlflags.add_argument('--create', dest='flags', action='append_const', 223 const=Netlink.NLM_F_CREATE) 224 nlflags.add_argument('--append', dest='flags', action='append_const', 225 const=Netlink.NLM_F_APPEND) 226 227 schema_group = parser.add_argument_group('Development options') 228 schema_group.add_argument('--schema', dest='schema', type=str, 229 help="JSON schema to validate the spec") 230 schema_group.add_argument('--no-schema', action='store_true') 231 232 dbg_group = parser.add_argument_group('Debug options') 233 io_group.add_argument('--policy', action='store_true', 234 help='Query kernel policy for the operation instead of executing it') 235 dbg_group.add_argument('--dbg-small-recv', default=0, const=4000, 236 action='store', nargs='?', type=int, metavar='INT', 237 help="Length of buffers used for recv()") 238 dbg_group.add_argument('--process-unknown', action=argparse.BooleanOptionalAction) 239 240 args = parser.parse_args() 241 242 def output(msg): 243 if args.output_json: 244 print(json.dumps(msg, cls=YnlEncoder)) 245 else: 246 pprint.pprint(msg, width=term_width(), compact=True) 247 248 if args.list_families: 249 for family in list_families(): 250 print(family) 251 return 252 253 if args.no_schema: 254 args.schema = '' 255 256 attrs = {} 257 if args.json_text: 258 attrs = json.loads(args.json_text) 259 260 if args.spec and not os.path.isfile(args.spec): 261 raise YnlException(f"Spec file {args.spec} does not exist") 262 263 # Spec/YnlFamily will raise if both or neither spec and family are given 264 if args.validate: 265 # Force validation even for installed specs (schema=True), unless the 266 # user explicitly picked a schema or opted out with --no-schema. 267 schema = True if args.schema is None else args.schema 268 try: 269 SpecFamily(args.spec, schema_path=schema, family=args.family) 270 except SpecException as error: 271 print(error) 272 sys.exit(1) 273 return 274 275 ynl = YnlFamily(args.spec, schema=args.schema, family=args.family, 276 process_unknown=args.process_unknown, 277 recv_size=args.dbg_small_recv) 278 if args.dbg_small_recv: 279 ynl.set_recv_dbg(True) 280 281 if args.policy: 282 if args.do: 283 pol = ynl.get_policy(args.do, 'do') 284 output(pol.to_dict() if pol else None) 285 args.do = None 286 if args.dump: 287 pol = ynl.get_policy(args.dump, 'dump') 288 output(pol.to_dict() if pol else None) 289 args.dump = None 290 291 if args.ntf: 292 ynl.ntf_subscribe(args.ntf) 293 294 if args.list_ops: 295 for op_name, op in ynl.ops.items(): 296 print(op_name, " [", ", ".join(op.modes), "]") 297 if args.list_msgs: 298 for op_name, op in ynl.msgs.items(): 299 print(op_name, " [", ", ".join(op.modes), "]") 300 301 if args.list_attrs: 302 op = ynl.msgs.get(args.list_attrs) 303 if not op: 304 print(f'Operation {args.list_attrs} not found') 305 sys.exit(1) 306 307 do_doc(ynl, op) 308 309 try: 310 if args.do: 311 reply = ynl.do(args.do, attrs, args.flags) 312 output(reply) 313 if args.dump: 314 reply = ynl.dump(args.dump, attrs) 315 output(reply) 316 if args.multi: 317 ops = [ (item[0], json.loads(item[1]), args.flags or []) for item in args.multi ] 318 reply = ynl.do_multi(ops) 319 output(reply) 320 321 if args.ntf: 322 for msg in ynl.poll_ntf(duration=args.duration): 323 output(msg) 324 except NlError as e: 325 print(e) 326 sys.exit(1) 327 except KeyboardInterrupt: 328 pass 329 except BrokenPipeError: 330 pass 331 332 333if __name__ == "__main__": 334 main() 335