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