1 from collections import Counter
2 from collections.abc import Mapping
3 from functools import partial
4 from typing import Final
5
6 from pcs.cli.common.errors import SEE_MAN_CHANGES, CmdLineInputError
7 from pcs.common.const import INFINITY
8 from pcs.common.str_tools import (
9 format_list,
10 format_list_custom_last_separator,
11 format_plural,
12 )
13 from pcs.common.tools import timeout_to_seconds
14 from pcs.common.types import StringCollection, StringIterable, StringSequence
15
16 # sys.argv always returns a list, we don't need StringSequence in here
17 Argv = list[str]
18 ModifierValueType = None | bool | str
19
20 _FUTURE_OPTION_STR: Final = "future"
21 FUTURE_OPTION: Final = f"--{_FUTURE_OPTION_STR}"
22 _OUTPUT_FORMAT_OPTION_STR: Final = "output-format"
23 OUTPUT_FORMAT_OPTION: Final = f"--{_OUTPUT_FORMAT_OPTION_STR}"
24 OUTPUT_FORMAT_VALUE_CMD: Final = "cmd"
25 OUTPUT_FORMAT_VALUE_JSON: Final = "json"
26 OUTPUT_FORMAT_VALUE_TEXT: Final = "text"
27 OUTPUT_FORMAT_VALUES: Final = frozenset(
28 (
29 OUTPUT_FORMAT_VALUE_CMD,
30 OUTPUT_FORMAT_VALUE_JSON,
31 OUTPUT_FORMAT_VALUE_TEXT,
32 )
33 )
34
35 MODIFIER_OPTIONS_BOOL: Final[frozenset[str]] = frozenset(
36 (
37 "--all",
38 "--agent-validation",
39 # TODO remove
40 # used only in acl commands and it is deprecated there
41 "--autodelete",
42 "--brief",
43 "--config",
44 "--corosync",
45 "--debug",
46 "--defaults",
47 "--disabled",
48 "--enable",
49 "--expired",
50 "--force",
51 "--full",
52 "--quiet",
53 FUTURE_OPTION,
54 # TODO remove
55 # used only in deprecated 'pcs resource|stonith show'
56 "--groups",
57 "--hide-inactive",
58 "--local",
59 "--monitor",
60 "--no-default-ops",
61 "--nodesc",
62 "--no-expire-check",
63 "--no-cluster-uuid",
64 "--no-keys-sync",
65 "--no-stop",
66 "--no-strict",
67 "--no-watchdog-validation",
68 "--off",
69 "--overwrite",
70 "--pacemaker",
71 "--promoted",
72 "--safe",
73 "--show-secrets",
74 "--simulate",
75 "--skip-offline",
76 "--start",
77 "--strict",
78 "--yes",
79 )
80 )
81
82 MODIFIER_OPTIONS_VAL: Final[frozenset[str]] = frozenset(
83 (
84 "--after",
85 "--before",
86 "--booth-conf",
87 "--booth-key",
88 "--corosync_conf",
89 "--from",
90 # TODO remove
91 # used in resource create and stonith create, deprecated in both
92 "--group",
93 "--name",
94 "--node",
95 "--request-timeout",
96 "--to",
97 "--token",
98 "-f",
99 "-p",
100 "-u",
101 )
102 )
103
104 ARG_TYPE_DELIMITER: Final = "%"
105
106 # h = help, f = file,
107 # p = password (cluster auth), u = user (cluster auth),
108 PCS_SHORT_OPTIONS: Final = "hf:p:u:"
109 PCS_LONG_OPTIONS: Final = [
110 "debug",
111 "version",
112 "help",
113 "fullhelp",
114 "force",
115 "skip-offline",
116 "autodelete",
117 "simulate",
118 "all",
119 "full",
120 "local",
121 "wait",
122 "config",
123 "start",
124 "enable",
125 "disabled",
126 "off",
127 "request-timeout=",
128 "brief",
129 _FUTURE_OPTION_STR,
130 # resource (safe-)disable
131 "safe",
132 "no-strict",
133 # resource cleanup | refresh
134 "strict",
135 "pacemaker",
136 "corosync",
137 "no-default-ops",
138 "defaults",
139 "nodesc",
140 "promoted",
141 "name=",
142 "group=",
143 "node=",
144 "from=",
145 "to=",
146 "after=",
147 "before=",
148 "corosync_conf=",
149 "booth-conf=",
150 "booth-key=",
151 # do not stop resources in resource delete command
152 "no-stop",
153 "no-watchdog-validation",
154 # pcs cluster setup
155 "no-cluster-uuid",
156 "no-keys-sync",
157 # in pcs status - do not display resource status on inactive node
158 "hide-inactive",
159 # pcs resource (un)manage - enable or disable monitor operations
160 "monitor",
161 # TODO remove
162 # used only in deprecated 'pcs resource|stonith show'
163 "groups",
164 # "pcs resource clear --expired" - only clear expired moves and bans
165 "expired",
166 # disable evaluating whether rules are expired
167 "no-expire-check",
168 # allow overwriting existing files, currently meant for / used in CLI only
169 "overwrite",
170 # output format of commands, e.g: json, cmd, text, ...
171 f"{_OUTPUT_FORMAT_OPTION_STR}=",
172 # auth token
173 "token=",
174 # enable agent self validation
175 "agent-validation",
176 # disable text output in query commands
177 "quiet",
178 # proceed with dangerous actions, meant for / used in CLI only
179 "yes",
180 # retrieve and display cibsecret values / used in CLI only
181 "show-secrets",
182 ]
183
184
185 def split_list(arg_list: Argv, separator: str) -> list[Argv]:
186 """
187 split a list of arguments to several lists using separator as a delimiter
188
189 arg_list -- list of command line arguments to split
190 separator -- delimiter
191 """
192 separator_indexes = [i for i, x in enumerate(arg_list) if x == separator]
193 bounds = zip(
194 [0] + [i + 1 for i in separator_indexes],
195 separator_indexes + [None],
196 strict=False,
197 )
198 return [arg_list[i:j] for i, j in bounds]
199
200
201 def split_list_by_any_keywords(
202 arg_list: Argv, keyword_label: str
203 ) -> dict[str, Argv]:
204 """
205 split a list of arguments using any argument not containing = as a delimiter
206
207 arg_list -- list of command line arguments to split
208 keyword_label -- description of all keywords
209 """
210 groups: dict[str, Argv] = {}
211 if not arg_list:
212 return groups
213
214 if "=" in arg_list[0]:
215 raise CmdLineInputError(
216 f"Invalid character '=' in {keyword_label} '{arg_list[0]}'"
217 )
218
219 current_keyword = arg_list[0]
220 groups[current_keyword] = []
221 for arg in arg_list[1:]:
222 if "=" in arg:
223 groups[current_keyword].append(arg)
224 else:
225 current_keyword = arg
226 if current_keyword in groups:
227 raise CmdLineInputError(
228 "{} '{}' defined multiple times".format(
229 keyword_label.capitalize(), current_keyword
230 )
231 )
232 groups[current_keyword] = []
233 return groups
234
235
236 def split_option(arg: str, allow_empty_value: bool = True) -> tuple[str, str]:
237 """
238 Get (key, value) from a key=value commandline argument.
239
240 Split the argument by the first = and return resulting parts. Raise
241 CmdLineInputError if the argument cannot be split.
242
243 arg -- commandline argument to split
244 allow_empty_value -- if False, raise CmdLineInputError on empty value
245 """
246 if "=" not in arg:
247 raise CmdLineInputError(f"missing value of '{arg}' option")
248 if arg.startswith("="):
249 raise CmdLineInputError(f"missing key in '{arg}' option")
250 key, value = arg.split("=", 1)
251 if not (value or allow_empty_value):
252 raise CmdLineInputError(f"value of '{key}' option is empty")
253 return key, value
254
255
256 def ensure_unique_args(cmdline_args: Argv) -> None:
257 """
258 Raises in case there are duplicate args
259 """
260 duplicities = [
261 item for item, count in Counter(cmdline_args).items() if count > 1
262 ]
263 if duplicities:
264 argument_pl = format_plural(duplicities, "argument")
265 duplicities_list = format_list(duplicities)
266 raise CmdLineInputError(f"duplicate {argument_pl}: {duplicities_list}")
267
268
269 class KeyValueParser:
270 """
271 Parse and check key=value options
272 """
273
274 def __init__(self, arg_list: Argv, repeatable: StringCollection = ()):
275 """
276 arg_list -- commandline arguments to be parsed
277 repeatable -- keys that are allowed to be specified several times
278 """
279 self._repeatable_keys = repeatable
280 self._key_value_map: dict[str, list[str]] = {}
281 for arg in arg_list:
282 name, value = split_option(arg)
283 if name not in self._key_value_map:
284 self._key_value_map[name] = [value]
285 else:
286 self._key_value_map[name].append(value)
287
288 def check_allowed_keys(self, allowed_keys: StringCollection) -> None:
289 """
290 Check that only allowed keys were specified
291
292 allowed_keys -- list of allowed keys
293 """
294 unknown_options = set(self._key_value_map.keys()) - set(allowed_keys)
295 if unknown_options:
296 raise CmdLineInputError(
297 "Unknown option{s} '{options}'".format(
298 s=("s" if len(unknown_options) > 1 else ""),
299 options="', '".join(sorted(unknown_options)),
300 )
301 )
302
303 def get_unique(self) -> dict[str, str]:
304 """
305 Get all non-repeatable keys and their values; raise if a key has more values
306 """
307 result: dict[str, str] = {}
308 for key, values in self._key_value_map.items():
309 if key in self._repeatable_keys:
310 continue
311 values_uniq = set(values)
312 if len(values_uniq) > 1:
313 raise CmdLineInputError(
314 f"duplicate option '{key}' with different values "
315 f"{format_list_custom_last_separator(values_uniq, ' and ')}"
316 )
317 result[key] = values[0]
318 return result
319
320 def get_repeatable(self) -> dict[str, list[str]]:
321 """
322 Get all repeatable keys and their values
323 """
324 return {
325 key: self._key_value_map[key]
326 for key in self._repeatable_keys
327 if key in self._key_value_map
328 }
329
330
331 class ArgsByKeywords:
332 def __init__(self, groups: Mapping[str, list[Argv]]):
333 self._groups = groups
334 self._flat_cache: dict[str, Argv] = {}
335
336 def allow_repetition_only_for(self, keyword_set: StringCollection) -> None:
337 """
338 Raise CmdLineInputError if a keyword has been repeated when not allowed
339
340 keyword_set -- repetition is allowed for these keywords
341 """
342 for keyword, arg_groups in self._groups.items():
343 if len(arg_groups) > 1 and keyword not in keyword_set:
344 raise CmdLineInputError(
345 f"'{keyword}' cannot be used more than once"
346 )
347
348 def ensure_unique_keywords(self) -> None:
349 """
350 Raise CmdLineInputError if any keyword has been repeated
351 """
352 return self.allow_repetition_only_for(set())
353
354 def is_empty(self) -> bool:
355 """
356 Check if any args have been specified
357 """
358 return not self._groups
359
360 def has_keyword(self, keyword: str) -> bool:
361 """
362 Check if a keyword has been specified
363
364 keyword -- a keyword to check
365 """
366 return keyword in self._groups
367
368 def has_empty_keyword(self, keyword: str) -> bool:
369 """
370 Check if a keyword has been specified without any following args
371
372 keyword -- a keyword to check
373 """
374 return self.has_keyword(keyword) and not self.get_args_flat(keyword)
375
376 def get_args_flat(self, keyword: str) -> Argv:
377 """
378 Get arguments of a keyword in one sequence
379 """
380 if keyword in self._groups:
381 if keyword not in self._flat_cache:
382 self._flat_cache[keyword] = [
383 arg
384 for one_group in self._groups[keyword]
385 for arg in one_group
386 ]
387 return self._flat_cache[keyword]
388 return []
389
390 def get_args_groups(self, keyword: str) -> list[Argv]:
391 """
392 Get arguments of a keyword, one group for each keyword occurrence
393 """
394 if keyword in self._groups:
395 return self._groups[keyword]
396 return []
397
398
399 def group_by_keywords(
400 arg_list: Argv,
401 keyword_set: StringCollection,
402 implicit_first_keyword: str | None = None,
403 ) -> ArgsByKeywords:
404 """
405 Separate argv into groups delimited by specified keywords
406
407 arg_list -- commandline arguments containing keywords
408 keyword_set -- all expected keywords
409 implicit_first_keyword -- key for capturing args before the first keyword
410 """
411 args_by_keywords: dict[str, list[Argv]] = {}
412
413 def new_keyword(keyword: str) -> None:
414 if keyword not in args_by_keywords:
415 args_by_keywords[keyword] = []
416 args_by_keywords[keyword].append([])
417
418 if arg_list:
419 if arg_list[0] not in keyword_set:
420 if not implicit_first_keyword:
421 raise CmdLineInputError()
422 current_keyword = implicit_first_keyword
423 new_keyword(current_keyword)
424
425 for arg in arg_list:
426 if arg in keyword_set:
427 current_keyword = arg
428 new_keyword(current_keyword)
429 else:
430 args_by_keywords[current_keyword][-1].append(arg)
431
432 return ArgsByKeywords(args_by_keywords)
433
434
435 def parse_typed_arg(
436 arg: str, allowed_types: StringSequence, default_type: str
437 ) -> tuple[str, str]:
438 """
439 Get (type, value) from a typed commandline argument.
440
441 Split the argument by the type separator and return the type and the value.
442 Raise CmdLineInputError in the argument format or type is not valid.
443 string arg -- commandline argument
444 Iterable allowed_types -- list of allowed argument types
445 string default_type -- type to return if the argument doesn't specify a type
446 """
447 if ARG_TYPE_DELIMITER not in arg:
448 return default_type, arg
449 arg_type, arg_value = arg.split(ARG_TYPE_DELIMITER, 1)
450 if not arg_type:
451 return default_type, arg_value
452 if arg_type not in allowed_types:
453 raise CmdLineInputError(
454 (
455 "'{arg_type}' is not an allowed type for '{arg_full}', use "
456 "{hint}"
457 ).format(
458 arg_type=arg_type,
459 arg_full=arg,
460 hint=", ".join(sorted(allowed_types)),
461 )
462 )
463 return arg_type, arg_value
464
465
466 def _is_num(arg: str) -> bool:
467 if arg.lower() == INFINITY.lower():
468 return True
469 try:
470 int(arg)
471 return True
472 except ValueError:
473 return False
474
475
476 def _is_float(arg: str) -> bool:
477 try:
478 float(arg)
479 return True
480 except ValueError:
481 return False
482
483
484 def _is_negative_num(arg: str) -> bool:
485 return arg.startswith("-") and (_is_num(arg[1:]) or _is_float(arg))
486
487
488 def is_short_option_expecting_value(arg: str) -> bool:
489 return len(arg) == 2 and arg[0] == "-" and f"{arg[1]}:" in PCS_SHORT_OPTIONS
490
491
492 def is_long_option_expecting_value(arg: str) -> bool:
493 return (
494 len(arg) > 2 and arg[0:2] == "--" and f"{arg[2:]}=" in PCS_LONG_OPTIONS
495 )
496
497
498 def is_option_expecting_value(arg: str) -> bool:
499 return is_short_option_expecting_value(
500 arg
501 ) or is_long_option_expecting_value(arg)
502
503
504 # DEPRECATED
505 # TODO remove
506 # This function is called only by deprecated code for parsing argv containing
507 # negative numbers without -- prepending them.
508 def filter_out_non_option_negative_numbers(arg_list: Argv) -> tuple[Argv, Argv]:
509 """
510 Return arg_list without non-option negative numbers.
511 Negative numbers following the option expecting value are kept.
512
513 There is the problematic legacy:
514 Argument "--" has special meaning: it can be used to signal that no more
515 options will follow. This would solve the problem with negative numbers in
516 a standard way: there would be no special approach to negative numbers,
517 everything would be left in the hands of users.
518
519 We cannot use "--" as it would be a backward incompatible change:
520 * "pcs ... -infinity" would not work any more, users would have to switch
521 to "pcs ... -- ... -infinity"
522 * previously, position of some --options mattered, for example
523 "--clone <clone options>", this syntax would not be possible with the "--"
524 in place
525
526 Currently used --options, which may be problematic when switching to "--":
527 * --group <group name>, --before | --after <resource id>
528 * pcs resource | stonith create, pcs resource group add, pcs tag update
529 * They have a single argument, so they would work even with --. But the
530 command may look weird:
531 pcs resource create --group G --after R2 -- R3 ocf:pacemaker:Dummy
532 vs. current command
533 pcs resource create R3 ocf:pacemaker:Dummy --group G --after R2
534
535 list arg_list contains command line arguments
536 """
537 args_without_negative_nums = []
538 args_filtered_out = []
539 for i, arg in enumerate(arg_list):
540 prev_arg = arg_list[i - 1] if i > 0 else ""
541 if not _is_negative_num(arg) or is_option_expecting_value(prev_arg):
542 args_without_negative_nums.append(arg)
543 else:
544 args_filtered_out.append(arg)
545
546 return args_without_negative_nums, args_filtered_out
547
548
549 # DEPRECATED
550 # TODO remove
551 # This function is called only by deprecated code for parsing argv containing
552 # negative numbers without -- prepending them.
553 def filter_out_options(arg_list: Argv) -> Argv:
554 """
555 Return arg_list without options and negative numbers
556
557 See a comment in filter_out_non_option_negative_numbers.
558
559 arg_list -- command line arguments
560 """
561 args_without_options = []
562 for i, arg in enumerate(arg_list):
563 prev_arg = arg_list[i - 1] if i > 0 else ""
564 if not is_option_expecting_value(prev_arg) and (
565 not arg.startswith("-") or arg == "-" or _is_negative_num(arg)
566 ):
567 args_without_options.append(arg)
568 return args_without_options
569
570
571 def wait_to_timeout(wait: bool | str | None) -> int:
572 if wait is False:
573 return -1
574 if wait is None:
575 return 0
576 timeout = timeout_to_seconds(wait)
577 if timeout is None:
578 raise CmdLineInputError(f"'{wait}' is not a valid interval value")
579 return timeout
580
581
582 class InputModifiers:
583 def __init__(self, options: Mapping[str, ModifierValueType]):
584 self._defined_options = set(options.keys())
585 self._options = dict(options)
586 self._options.update(
587 {opt: opt in options for opt in MODIFIER_OPTIONS_BOOL}
588 )
589 self._options.update(
590 {opt: options.get(opt, None) for opt in MODIFIER_OPTIONS_VAL}
591 )
592 self._options.update(
593 {
594 OUTPUT_FORMAT_OPTION: options.get(
595 OUTPUT_FORMAT_OPTION, OUTPUT_FORMAT_VALUE_TEXT
596 ),
597 "--wait": options.get("--wait", False),
598 }
599 )
600
601 def get_subset(
602 self, *options: str, **custom_options: ModifierValueType
603 ) -> "InputModifiers":
604 opt_dict = {
605 opt: self.get(opt) for opt in options if self.is_specified(opt)
606 }
607 opt_dict.update(custom_options)
608 return InputModifiers(opt_dict)
609
610 def ensure_only_supported(
611 self,
612 *supported_options: str,
613 hint_syntax_changed: str | None = None,
614 output_format_supported: bool = False,
615 ) -> None:
616 # --debug is supported in all commands
617 supported_options_set = set(supported_options) | {"--debug"}
618 if output_format_supported:
619 supported_options_set.add(OUTPUT_FORMAT_OPTION)
620 unsupported_options = self._defined_options - supported_options_set
621 if unsupported_options:
622 pluralize = partial(format_plural, unsupported_options)
623 raise CmdLineInputError(
624 "Specified {option} {option_list} {_is} not supported in this "
625 "command".format(
626 option=pluralize("option"),
627 option_list=format_list(sorted(unsupported_options)),
628 _is=pluralize("is"),
629 ),
630 hint=(
631 "Syntax has changed from previous version. {}".format(
632 SEE_MAN_CHANGES.format(hint_syntax_changed)
633 )
634 if hint_syntax_changed
635 else None
636 ),
637 )
638
639 def ensure_not_mutually_exclusive(self, *mutually_exclusive: str) -> None:
640 """
641 Raise CmdLineInputError if several exclusive options were specified
642
643 mutually_exclusive -- mutually exclusive options
644 """
645 options_to_report = self._defined_options & set(mutually_exclusive)
646 if len(options_to_report) > 1:
647 raise CmdLineInputError(
648 "Only one of {} can be used".format(
649 format_list(sorted(options_to_report))
650 )
651 )
652
653 def ensure_not_incompatible(
654 self, checked: str, incompatible: StringCollection
655 ) -> None:
656 """
657 Raise CmdLineInputError if both the checked and an incompatible option
658 were specified
659
660 checked -- option incompatible with any of incompatible options
661 incompatible -- set of options incompatible with checked
662 """
663 if checked not in self._defined_options:
664 return
665 disallowed = self._defined_options & set(incompatible)
666 if disallowed:
667 raise CmdLineInputError(
668 "'{}' cannot be used with {}".format(
669 checked, format_list(sorted(disallowed))
670 )
671 )
672
673 def ensure_dependency_satisfied(
674 self, main_option: str, dependent_options: StringCollection
675 ) -> None:
676 """
677 Raise CmdLineInputError if any of dependent_options is present and
678 main_option is not present.
679
680 main_option -- option on which dependent_options depend
681 dependent_options -- none of these options can be specified if
682 main_option is not specified
683 """
684 if main_option in self._defined_options:
685 return
686 disallowed = self._defined_options & set(dependent_options)
687 if disallowed:
688 raise CmdLineInputError(
689 "{} cannot be used without '{}'".format(
690 format_list(sorted(disallowed)), main_option
691 )
692 )
693
694 def is_specified(self, option: str) -> bool:
695 return option in self._defined_options
696
697 def is_specified_any(self, option_list: StringIterable) -> bool:
698 return any(self.is_specified(option) for option in option_list)
699
700 def get(
701 self, option: str, default: ModifierValueType = None
702 ) -> ModifierValueType:
703 if option in self._defined_options:
|
CID (unavailable; MK=489eaaa87afbb746a3d63c1bcfb03702) (#1 of 1): Copy-paste error (COPY_PASTE_ERROR): |
|
(2) Event copy_paste_error: |
"_options" in "self._options[option]" looks like a copy-paste error. |
|
(3) Event remediation: |
Should it say "_defined_options" instead? |
| Also see events: |
[original] |
704 return self._options[option]
705 if default is not None:
706 return default
707 if option in self._options:
708 return self._options[option]
709 raise AssertionError(f"Non existing default value for '{option}'")
710
711 def get_output_format(
712 self,
713 supported_formats: StringCollection = OUTPUT_FORMAT_VALUES,
714 ) -> str:
715 output_format = self.get(OUTPUT_FORMAT_OPTION)
716 if output_format in supported_formats:
717 return str(output_format)
718 raise CmdLineInputError(
719 (
720 "Unknown value '{value}' for '{option}' option. Supported "
721 "{value_pl} {is_pl}: {supported}"
722 ).format(
723 value=output_format,
724 option=OUTPUT_FORMAT_OPTION,
725 value_pl=format_plural(supported_formats, "value"),
726 is_pl=format_plural(supported_formats, "is"),
727 supported=format_list(list(supported_formats)),
728 )
729 )
730
731
732 def get_rule_str(argv: Argv) -> str | None:
733 if argv:
734 if len(argv) > 1:
735 raise CmdLineInputError()
736 return argv[0]
737 return None
738