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