projects/linux·blob master · refs

334 lines · 13.0 KiB

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()