1 import sys
2 import xml.dom.minidom
3 from collections.abc import Iterable
4 from enum import Enum
5 from typing import Any, TypeVar, cast
6 from xml.dom.minidom import parseString
7
8 import pcs.cli.constraint_order.command as order_command
9 from pcs import utils
10 from pcs.cli.common import parse_args
11 from pcs.cli.common.errors import CmdLineInputError, raise_command_replaced
12 from pcs.cli.common.output import INDENT_STEP, lines_to_str
13 from pcs.cli.constraint.location.command import (
14 RESOURCE_TYPE_REGEXP,
15 RESOURCE_TYPE_RESOURCE,
16 )
17 from pcs.cli.constraint.output import (
18 CibConstraintLocationAnyDto,
19 filter_constraints_by_rule_expired_status,
20 location,
21 print_config,
22 )
23 from pcs.cli.reports import process_library_reports
24 from pcs.cli.reports.output import deprecation_warning, print_to_stderr, warn
25 from pcs.common import const, pacemaker, reports
26 from pcs.common.pacemaker.constraint import (
27 CibConstraintColocationSetDto,
28 CibConstraintLocationSetDto,
29 CibConstraintOrderSetDto,
30 CibConstraintsDto,
31 CibConstraintTicketSetDto,
32 get_all_constraints_ids,
33 )
34 from pcs.common.pacemaker.resource.list import CibResourcesDto
35 from pcs.common.pacemaker.types import CibResourceDiscovery
36 from pcs.common.reports import ReportItem
37 from pcs.common.str_tools import format_list, indent
38 from pcs.common.types import StringCollection, StringIterable, StringSequence
39 from pcs.lib.cib.constraint.order import ATTRIB as order_attrib
40 from pcs.lib.node import get_existing_nodes_names
41 from pcs.lib.pacemaker.values import SCORE_INFINITY, is_true, sanitize_id
42
43 DEFAULT_ACTION = const.PCMK_ACTION_START
44 DEFAULT_ROLE = const.PCMK_ROLE_STARTED
45
46 OPTIONS_SYMMETRICAL = order_attrib["symmetrical"]
47
48 LOCATION_NODE_VALIDATION_SKIP_MSG = (
49 "Validation for node existence in the cluster will be skipped"
50 )
51 STANDALONE_SCORE_MSG = (
52 "Specifying score as a standalone value is deprecated and "
53 "might be removed in a future release, use score=value instead"
54 )
55
56
57 class CrmRuleReturnCode(Enum):
58 IN_EFFECT = 0
59 EXPIRED = 110
60 TO_BE_IN_EFFECT = 111
61
62
63 def constraint_order_cmd(lib, argv, modifiers):
64 sub_cmd = "config" if not argv else argv.pop(0)
65
66 try:
67 if sub_cmd == "set":
68 order_command.create_with_set(lib, argv, modifiers)
69 elif sub_cmd in ["remove", "delete"]:
70 order_rm(lib, argv, modifiers)
71 elif sub_cmd == "show":
72 raise_command_replaced(
73 ["pcs constraint order config"], pcs_version="0.12"
74 )
75 elif sub_cmd == "config":
76 order_command.config_cmd(lib, argv, modifiers)
77 else:
78 order_start(lib, [sub_cmd] + argv, modifiers)
79 except CmdLineInputError as e:
80 utils.exit_on_cmdline_input_error(e, "constraint", ["order", sub_cmd])
81
82
83 def config_cmd(
84 lib: Any, argv: list[str], modifiers: parse_args.InputModifiers
85 ) -> None:
86 modifiers.ensure_only_supported("-f", "--output-format", "--full", "--all")
87 if argv:
88 raise CmdLineInputError()
89
90 print_config(
91 cast(
92 CibConstraintsDto,
93 lib.constraint.get_config(evaluate_rules=True),
94 ),
95 modifiers,
96 )
97
98
99 def _validate_constraint_resource(cib_dom, resource_id):
100 (
101 resource_valid,
102 resource_error,
103 dummy_correct_id,
104 ) = utils.validate_constraint_resource(cib_dom, resource_id)
105 if not resource_valid:
106 utils.err(resource_error)
107
108
109 def _validate_resources_not_in_same_group(cib_dom, resource1, resource2):
110 if not utils.validate_resources_not_in_same_group(
111 cib_dom, resource1, resource2
112 ):
113 utils.err(
114 "Cannot create an order constraint for resources in the same group"
115 )
116
117
118 # Syntax: colocation add [role] <src> with [role] <tgt> [score] [options]
119 # possible commands:
120 # <src> with <tgt> [score] [options]
121 # <src> with <role> <tgt> [score] [options]
122 # <role> <src> with <tgt> [score] [options]
123 # <role> <src> with <role> <tgt> [score] [options]
124 # Specifying score as a single argument is deprecated, though. The correct way
125 # is score=value in options.
126 def colocation_add(lib, argv, modifiers): # noqa: PLR0912, PLR0915
127 """
128 Options:
129 * -f - CIB file
130 * --force - allow constraint on any resource, allow duplicate constraints
131 """
132
133 def _parse_score_options(argv):
134 # When passed an array of arguments if the first argument doesn't have
135 # an '=' then it's the score, otherwise they're all arguments. Return a
136 # tuple with the score and array of name,value pairs
137 """
138 Commandline options: no options
139 """
140 if not argv:
141 return None, []
142 score = None
143 if "=" not in argv[0]:
144 score = argv.pop(0)
145 # TODO added to pcs in the first 0.12.x version
146 deprecation_warning(STANDALONE_SCORE_MSG)
147
148 # create a list of 2-tuples (name, value)
149 arg_array = [
150 parse_args.split_option(arg, allow_empty_value=False)
151 for arg in argv
152 ]
153 return score, arg_array
154
155 def _validate_and_prepare_role(new_roles_supported, role):
156 if role is None:
157 return ""
158 role_cleaned = role.lower().capitalize()
159 if role_cleaned not in const.PCMK_ROLES:
160 utils.err(
161 "invalid role value '{0}', allowed values are: {1}".format(
162 role, format_list(const.PCMK_ROLES)
163 )
164 )
165 return pacemaker.role.get_value_for_cib(
166 role_cleaned, new_roles_supported
167 )
168
169 del lib
170 modifiers.ensure_only_supported("-f", "--force")
171 if len(argv) < 3:
172 raise CmdLineInputError()
173
174 role1_candidate = None
175 role2_candidate = None
176
177 if argv[2] == "with":
178 role1_candidate = argv.pop(0)
179 resource1 = argv.pop(0)
180 elif argv[1] == "with":
181 resource1 = argv.pop(0)
182 else:
183 raise CmdLineInputError()
184
185 if argv.pop(0) != "with":
186 raise CmdLineInputError()
187 if "with" in argv:
188 raise CmdLineInputError(
189 message="Multiple 'with's cannot be specified.",
190 hint=(
191 "Use the 'pcs constraint colocation set' command if you want "
192 "to create a constraint for more than two resources."
193 ),
194 show_both_usage_and_message=True,
195 )
196
197 if not argv:
198 raise CmdLineInputError()
199 if len(argv) == 1 or utils.is_score_or_opt(argv[1]):
200 resource2 = argv.pop(0)
201 else:
202 role2_candidate = argv.pop(0)
203 resource2 = argv.pop(0)
204
205 score, nv_pairs = _parse_score_options(argv)
206 influence_attr_set = any(name == "influence" for name, _ in nv_pairs)
207
208 cib_dom = (
209 utils.cluster_upgrade_to_version(
210 const.PCMK_COLOCATION_INFLUENCE_CIB_VERSION
211 )
212 if influence_attr_set
213 else utils.get_cib_dom()
214 )
215 new_roles_supported = utils.isCibVersionSatisfied(
216 cib_dom, const.PCMK_NEW_ROLES_CIB_VERSION
217 )
218
219 role1 = _validate_and_prepare_role(new_roles_supported, role1_candidate)
220 role2 = _validate_and_prepare_role(new_roles_supported, role2_candidate)
221 _validate_constraint_resource(cib_dom, resource1)
222 _validate_constraint_resource(cib_dom, resource2)
223
224 id_in_nvpairs = None
225 for name, value in nv_pairs:
226 if name == "id":
227 id_valid, id_error = utils.validate_xml_id(value, "constraint id")
228 if not id_valid:
229 utils.err(id_error)
230 if utils.does_id_exist(cib_dom, value):
231 utils.err(
232 "id '%s' is already in use, please specify another one"
233 % value
234 )
235 id_in_nvpairs = True
236 elif name == "score":
237 score = value
238 if score is None:
239 score = SCORE_INFINITY
240 if not id_in_nvpairs:
241 nv_pairs.append(
242 (
243 "id",
244 utils.find_unique_id(
245 cib_dom,
246 "colocation-%s-%s-%s" % (resource1, resource2, score),
247 ),
248 )
249 )
250
251 (dom, constraintsElement) = getCurrentConstraints(cib_dom)
252
253 # If one role is specified, the other should default to "started"
254 if role1 != "" and role2 == "":
255 role2 = DEFAULT_ROLE
256 if role2 != "" and role1 == "":
257 role1 = DEFAULT_ROLE
258 element = dom.createElement("rsc_colocation")
259 element.setAttribute("rsc", resource1)
260 element.setAttribute("with-rsc", resource2)
261 element.setAttribute("score", score)
262 if role1 != "":
263 element.setAttribute("rsc-role", role1)
264 if role2 != "":
265 element.setAttribute("with-rsc-role", role2)
266 for nv_pair in nv_pairs:
267 element.setAttribute(nv_pair[0], nv_pair[1])
268 if not modifiers.get("--force"):
269
270 def _constraint_export(constraint_info):
271 options_dict = constraint_info["options"]
272 co_resource1 = options_dict.get("rsc", "")
273 co_resource2 = options_dict.get("with-rsc", "")
274 co_id = options_dict.get("id", "")
275 co_score = options_dict.get("score", "")
276 score_text = "(score:" + co_score + ")"
277 console_option_list = [
278 f"({option[0]}:{option[1]})"
279 for option in sorted(options_dict.items())
280 if option[0] not in ("rsc", "with-rsc", "id", "score")
281 ]
282 console_option_list.append(f"(id:{co_id})")
283 return " ".join(
284 [co_resource1, "with", co_resource2, score_text]
285 + console_option_list
286 )
287
288 duplicates = colocation_find_duplicates(constraintsElement, element)
289 if duplicates:
290 utils.err(
291 "duplicate constraint already exists, use --force to override\n"
292 + "\n".join(
293 [
294 " "
295 + _constraint_export(
296 {"options": dict(dup.attributes.items())}
297 )
298 for dup in duplicates
299 ]
300 )
301 )
302 constraintsElement.appendChild(element)
303 utils.replace_cib_configuration(dom)
304
305
306 def colocation_find_duplicates(dom, constraint_el):
307 """
308 Commandline options: no options
309 """
310 new_roles_supported = utils.isCibVersionSatisfied(
311 dom, const.PCMK_NEW_ROLES_CIB_VERSION
312 )
313
314 def normalize(const_el):
315 return (
316 const_el.getAttribute("rsc"),
317 const_el.getAttribute("with-rsc"),
318 pacemaker.role.get_value_for_cib(
319 const_el.getAttribute("rsc-role").capitalize() or DEFAULT_ROLE,
320 new_roles_supported,
321 ),
322 pacemaker.role.get_value_for_cib(
323 const_el.getAttribute("with-rsc-role").capitalize()
324 or DEFAULT_ROLE,
325 new_roles_supported,
326 ),
327 )
328
329 normalized_el = normalize(constraint_el)
330 return [
331 other_el
332 for other_el in dom.getElementsByTagName("rsc_colocation")
333 if not other_el.getElementsByTagName("resource_set")
334 and constraint_el is not other_el
335 and normalized_el == normalize(other_el)
336 ]
337
338
339 def order_rm(lib, argv, modifiers):
340 """
341 Options:
342 * -f - CIB file
343 """
344 del lib
345 modifiers.ensure_only_supported("-f")
346 if not argv:
347 raise CmdLineInputError()
348
349 elementFound = False
350 (dom, constraintsElement) = getCurrentConstraints()
351
352 for resource in argv:
353 for ord_loc in constraintsElement.getElementsByTagName("rsc_order")[:]:
354 if (
355 ord_loc.getAttribute("first") == resource
356 or ord_loc.getAttribute("then") == resource
357 ):
358 constraintsElement.removeChild(ord_loc)
359 elementFound = True
360
361 resource_refs_to_remove = []
362 for ord_set in constraintsElement.getElementsByTagName("resource_ref"):
363 if ord_set.getAttribute("id") == resource:
364 resource_refs_to_remove.append(ord_set)
365 elementFound = True
366
367 for res_ref in resource_refs_to_remove:
368 res_set = res_ref.parentNode
369 res_order = res_set.parentNode
370
371 res_ref.parentNode.removeChild(res_ref)
372 if not res_set.getElementsByTagName("resource_ref"):
373 res_set.parentNode.removeChild(res_set)
374 if not res_order.getElementsByTagName("resource_set"):
375 res_order.parentNode.removeChild(res_order)
376
377 if elementFound:
378 utils.replace_cib_configuration(dom)
379 else:
380 utils.err("No matching resources found in ordering list")
381
382
383 def order_start(lib, argv, modifiers):
384 """
385 Options:
386 * -f - CIB file
387 * --force - allow constraint for any resource, allow duplicate constraints
388 """
389 del lib
390 modifiers.ensure_only_supported("-f", "--force")
391 if len(argv) < 3:
392 raise CmdLineInputError()
393
394 first_action = DEFAULT_ACTION
395 then_action = DEFAULT_ACTION
396 action = argv[0]
397 if action in const.PCMK_ACTIONS:
398 first_action = action
399 argv.pop(0)
400
401 resource1 = argv.pop(0)
402 if argv.pop(0) != "then":
403 raise CmdLineInputError()
404
405 if not argv:
406 raise CmdLineInputError()
407
408 action = argv[0]
409 if action in const.PCMK_ACTIONS:
410 then_action = action
411 argv.pop(0)
412
413 if not argv:
414 raise CmdLineInputError()
415 resource2 = argv.pop(0)
416
417 order_options = []
418 if argv:
419 order_options = order_options + argv[:]
420 if "then" in order_options:
421 raise CmdLineInputError(
422 message="Multiple 'then's cannot be specified.",
423 hint=(
424 "Use the 'pcs constraint order set' command if you want to "
425 "create a constraint for more than two resources."
426 ),
427 show_both_usage_and_message=True,
428 )
429
430 order_options.append("first-action=" + first_action)
431 order_options.append("then-action=" + then_action)
432 _order_add(resource1, resource2, order_options, modifiers)
433
434
435 def _order_add(resource1, resource2, options_list, modifiers): # noqa: PLR0912, PLR0915
436 """
437 Commandline options:
438 * -f - CIB file
439 * --force - allow constraint for any resource, allow duplicate constraints
440 """
441 cib_dom = utils.get_cib_dom()
442 _validate_constraint_resource(cib_dom, resource1)
443 _validate_constraint_resource(cib_dom, resource2)
444
445 _validate_resources_not_in_same_group(cib_dom, resource1, resource2)
446
447 order_options = []
448 id_specified = False
449 sym = None
450 for arg in options_list:
451 if arg == "symmetrical":
452 sym = "true"
453 elif arg == "nonsymmetrical":
454 sym = "false"
455 else:
456 name, value = parse_args.split_option(arg, allow_empty_value=False)
457 if name == "id":
458 id_valid, id_error = utils.validate_xml_id(
459 value, "constraint id"
460 )
461 if not id_valid:
462 utils.err(id_error)
463 if utils.does_id_exist(cib_dom, value):
464 utils.err(
465 "id '%s' is already in use, please specify another one"
466 % value
467 )
468 id_specified = True
469 order_options.append((name, value))
470 elif name == "symmetrical":
471 if value.lower() in OPTIONS_SYMMETRICAL:
472 sym = value.lower()
473 else:
474 utils.err(
475 "invalid symmetrical value '%s', allowed values are: %s"
476 % (value, ", ".join(OPTIONS_SYMMETRICAL))
477 )
478 else:
479 order_options.append((name, value))
480 if sym:
481 order_options.append(("symmetrical", sym))
482
483 options = ""
484 if order_options:
485 options = " (Options: %s)" % " ".join(
486 [
487 "%s=%s" % (name, value)
488 for name, value in order_options
489 if name not in ("kind", "score")
490 ]
491 )
492
493 scorekind = "kind: Mandatory"
494 id_suffix = "mandatory"
495 for opt in order_options:
496 if opt[0] == "score":
497 scorekind = "score: " + opt[1]
498 id_suffix = opt[1]
499 # TODO deprecated in pacemaker 2, to be removed in pacemaker 3
500 # added to pcs after 0.11.7
501 deprecation_warning(
502 reports.messages.DeprecatedOption(opt[0], []).message
503 )
504 break
505 if opt[0] == "kind":
506 scorekind = "kind: " + opt[1]
507 id_suffix = opt[1]
508 break
509
510 if not id_specified:
511 order_id = "order-" + resource1 + "-" + resource2 + "-" + id_suffix
512 order_id = utils.find_unique_id(cib_dom, order_id)
513 order_options.append(("id", order_id))
514
515 (dom, constraintsElement) = getCurrentConstraints()
516 element = dom.createElement("rsc_order")
517 element.setAttribute("first", resource1)
518 element.setAttribute("then", resource2)
519 for order_opt in order_options:
520 element.setAttribute(order_opt[0], order_opt[1])
521 constraintsElement.appendChild(element)
522 if not modifiers.get("--force"):
523
524 def _constraint_export(constraint_info):
525 options = constraint_info["options"]
526 oc_resource1 = options.get("first", "")
527 oc_resource2 = options.get("then", "")
528 first_action = options.get("first-action", "")
529 then_action = options.get("then-action", "")
530 oc_id = options.get("id", "")
531 oc_score = options.get("score", "")
532 oc_kind = options.get("kind", "")
533 oc_sym = ""
534 oc_id_out = ""
535 oc_options = ""
536 if "symmetrical" in options and not is_true(
537 options.get("symmetrical", "false")
538 ):
539 oc_sym = "(non-symmetrical)"
540 if oc_kind != "":
541 score_text = "(kind:" + oc_kind + ")"
542 elif oc_kind == "" and oc_score == "":
543 score_text = "(kind:Mandatory)"
544 else:
545 score_text = "(score:" + oc_score + ")"
546 oc_id_out = "(id:" + oc_id + ")"
547 already_processed_options = (
548 "first",
549 "then",
550 "first-action",
551 "then-action",
552 "id",
553 "score",
554 "kind",
555 "symmetrical",
556 )
557 oc_options = " ".join(
558 [
559 f"{name}={value}"
560 for name, value in options.items()
561 if name not in already_processed_options
562 ]
563 )
564 if oc_options:
565 oc_options = "(Options: " + oc_options + ")"
566 return " ".join(
567 [
568 arg
569 for arg in [
570 first_action,
571 oc_resource1,
572 "then",
573 then_action,
574 oc_resource2,
575 score_text,
576 oc_sym,
577 oc_options,
578 oc_id_out,
579 ]
580 if arg
581 ]
582 )
583
584 duplicates = order_find_duplicates(constraintsElement, element)
585 if duplicates:
586 utils.err(
587 "duplicate constraint already exists, use --force to override\n"
588 + "\n".join(
589 [
590 " "
591 + _constraint_export(
592 {"options": dict(dup.attributes.items())}
593 )
594 for dup in duplicates
595 ]
596 )
597 )
598 print_to_stderr(f"Adding {resource1} {resource2} ({scorekind}){options}")
599 utils.replace_cib_configuration(dom)
600
601
602 def order_find_duplicates(dom, constraint_el):
603 """
604 Commandline options: no options
605 """
606
607 def normalize(constraint_el):
608 return (
609 constraint_el.getAttribute("first"),
610 constraint_el.getAttribute("then"),
611 constraint_el.getAttribute("first-action").lower()
612 or DEFAULT_ACTION,
613 constraint_el.getAttribute("then-action").lower() or DEFAULT_ACTION,
614 )
615
616 normalized_el = normalize(constraint_el)
617 return [
618 other_el
619 for other_el in dom.getElementsByTagName("rsc_order")
620 if not other_el.getElementsByTagName("resource_set")
621 and constraint_el is not other_el
622 and normalized_el == normalize(other_el)
623 ]
624
625
626 _SetConstraint = TypeVar(
627 "_SetConstraint",
628 CibConstraintLocationSetDto,
629 CibConstraintColocationSetDto,
630 CibConstraintOrderSetDto,
631 CibConstraintTicketSetDto,
632 )
633
634
635 def _filter_set_constraints_by_resources(
636 constraints_dto: Iterable[_SetConstraint], resources: set[str]
637 ) -> list[_SetConstraint]:
638 return [
639 constraint_set_dto
640 for constraint_set_dto in constraints_dto
641 if any(
642 set(resource_set.resources_ids) & resources
643 for resource_set in constraint_set_dto.resource_sets
644 )
645 ]
646
647
648 def _filter_constraints_by_resources(
649 constraints_dto: CibConstraintsDto,
650 resources: StringIterable,
651 patterns: StringIterable,
652 ) -> CibConstraintsDto:
653 required_resources_set = set(resources)
654 required_patterns_set = set(patterns)
655 return CibConstraintsDto(
656 location=[
657 constraint_dto
658 for constraint_dto in constraints_dto.location
659 if (
660 constraint_dto.resource_id is not None
661 and constraint_dto.resource_id in required_resources_set
662 )
663 or (
664 constraint_dto.resource_pattern is not None
665 and constraint_dto.resource_pattern in required_patterns_set
666 )
667 ],
668 location_set=_filter_set_constraints_by_resources(
669 constraints_dto.location_set, required_resources_set
670 ),
671 colocation=[
672 constraint_dto
673 for constraint_dto in constraints_dto.colocation
674 if {constraint_dto.resource_id, constraint_dto.with_resource_id}
675 & required_resources_set
676 ],
677 colocation_set=_filter_set_constraints_by_resources(
678 constraints_dto.colocation_set, required_resources_set
679 ),
680 order=[
681 constraint_dto
682 for constraint_dto in constraints_dto.order
683 if {
684 constraint_dto.first_resource_id,
685 constraint_dto.then_resource_id,
686 }
687 & required_resources_set
688 ],
689 order_set=_filter_set_constraints_by_resources(
690 constraints_dto.order_set, required_resources_set
691 ),
692 ticket=[
693 constraint_dto
694 for constraint_dto in constraints_dto.ticket
695 if constraint_dto.resource_id in required_resources_set
696 ],
697 ticket_set=_filter_set_constraints_by_resources(
698 constraints_dto.ticket_set, required_resources_set
699 ),
700 )
701
702
703 def _filter_location_by_node_base(
704 constraint_dtos: Iterable[CibConstraintLocationAnyDto],
705 nodes: StringCollection,
706 ) -> list[CibConstraintLocationAnyDto]:
707 return [
708 constraint_dto
709 for constraint_dto in constraint_dtos
710 if constraint_dto.attributes.node is not None
711 and constraint_dto.attributes.node in nodes
712 ]
713
714
715 def location_config_cmd(
716 lib: Any, argv: parse_args.Argv, modifiers: parse_args.InputModifiers
717 ) -> None:
718 """
719 Options:
720 * --all - print expired constraints
721 * --full - print all details
722 * -f - CIB file
723 """
724 modifiers.ensure_only_supported("-f", "--output-format", "--full", "--all")
725 filter_type: str | None = None
726 filter_items: parse_args.Argv = []
727 if argv:
728 filter_type, *filter_items = argv
729 allowed_types = ("resources", "nodes")
730 if filter_type not in allowed_types:
731 raise CmdLineInputError(
732 f"Unknown keyword '{filter_type}'. Allowed keywords: "
733 f"{format_list(allowed_types)}"
734 )
735 if modifiers.get_output_format() != parse_args.OUTPUT_FORMAT_VALUE_TEXT:
736 raise CmdLineInputError(
737 "Output formats other than 'text' are not supported together "
738 "with grouping and filtering by nodes or resources"
739 )
740
741 constraints_dto = filter_constraints_by_rule_expired_status(
742 lib.constraint.get_config(evaluate_rules=True),
743 modifiers.is_specified("--all"),
744 )
745
746 constraints_dto = CibConstraintsDto(
747 location=constraints_dto.location,
748 location_set=constraints_dto.location_set,
749 )
750
751 def _print_lines(lines: StringSequence) -> None:
752 if lines:
753 print("Location Constraints:")
754 print(lines_to_str(indent(lines, indent_step=INDENT_STEP)))
755
756 if filter_type == "resources":
757 if filter_items:
758 resources = []
759 patterns = []
760 for item in filter_items:
761 item_type, item_value = parse_args.parse_typed_arg(
762 item,
763 [RESOURCE_TYPE_RESOURCE, RESOURCE_TYPE_REGEXP],
764 RESOURCE_TYPE_RESOURCE,
765 )
766 if item_type == RESOURCE_TYPE_RESOURCE:
767 resources.append(item_value)
768 elif item_type == RESOURCE_TYPE_REGEXP:
769 patterns.append(item_value)
770 constraints_dto = _filter_constraints_by_resources(
771 constraints_dto, resources, patterns
772 )
773 _print_lines(
774 location.constraints_to_grouped_by_resource_text(
775 constraints_dto.location,
776 modifiers.is_specified("--full"),
777 )
778 )
779 return
780 if filter_type == "nodes":
781 if filter_items:
782 constraints_dto = CibConstraintsDto(
783 location=_filter_location_by_node_base(
784 constraints_dto.location, filter_items
785 ),
786 location_set=_filter_location_by_node_base(
787 constraints_dto.location_set, filter_items
788 ),
789 )
790 _print_lines(
791 location.constraints_to_grouped_by_node_text(
792 constraints_dto.location,
793 modifiers.is_specified("--full"),
794 )
795 )
796 return
797
798 print_config(constraints_dto, modifiers)
799
800
801 def _verify_node_name(node, existing_nodes):
802 report_list = []
803 if node not in existing_nodes:
804 report_list.append(
805 ReportItem.error(
806 reports.messages.NodeNotFound(node),
807 force_code=reports.codes.FORCE,
808 )
809 )
810 return report_list
811
812
813 def _verify_score(score):
814 if not utils.is_score(score):
815 utils.err(
816 "invalid score '%s', use integer or INFINITY or -INFINITY" % score
817 )
818
819
820 def location_prefer( # noqa: PLR0912
821 lib: Any, argv: parse_args.Argv, modifiers: parse_args.InputModifiers
822 ) -> None:
823 """
824 Options:
825 * --force - allow unknown options, allow constraint for any resource type
826 * -f - CIB file
827 """
828 modifiers.ensure_only_supported("--force", "-f")
829 rsc = argv.pop(0)
830 prefer_option = argv.pop(0)
831
832 dummy_rsc_type, rsc_value = parse_args.parse_typed_arg(
833 rsc,
834 [RESOURCE_TYPE_RESOURCE, RESOURCE_TYPE_REGEXP],
835 RESOURCE_TYPE_RESOURCE,
836 )
837
838 if prefer_option == "prefers":
839 prefer = True
840 elif prefer_option == "avoids":
841 prefer = False
842 else:
843 raise CmdLineInputError()
844
845 skip_node_check = False
846 existing_nodes: list[str] = []
847 if modifiers.is_specified("-f") or modifiers.get("--force"):
848 skip_node_check = True
849 warn(LOCATION_NODE_VALIDATION_SKIP_MSG)
850 else:
851 lib_env = utils.get_lib_env()
852 existing_nodes, report_list = get_existing_nodes_names(
853 corosync_conf=lib_env.get_corosync_conf(),
854 cib=lib_env.get_cib(),
855 )
856 if report_list:
857 process_library_reports(report_list)
858
859 report_list = []
860 parameters_list = []
861 for nodeconf in argv:
862 nodeconf_a = nodeconf.split("=", 1)
863 node = nodeconf_a[0]
864 if not skip_node_check:
865 report_list += _verify_node_name(node, existing_nodes)
866 if len(nodeconf_a) == 1:
867 score = "INFINITY" if prefer else "-INFINITY"
868 else:
869 score = nodeconf_a[1]
870 _verify_score(score)
871 if not prefer:
872 score = score[1:] if score[0] == "-" else "-" + score
873
874 parameters_list.append(
875 [
876 sanitize_id(f"location-{rsc_value}-{node}-{score}"),
877 rsc,
878 node,
879 f"score={score}",
880 ]
881 )
882
883 if report_list:
884 process_library_reports(report_list)
885
886 modifiers = modifiers.get_subset("--force", "-f")
887
888 for parameters in parameters_list:
889 location_add(lib, parameters, modifiers, skip_score_and_node_check=True)
890
891
892 def location_add( # noqa: PLR0912, PLR0915
893 lib: Any,
894 argv: parse_args.Argv,
895 modifiers: parse_args.InputModifiers,
896 skip_score_and_node_check: bool = False,
897 ) -> None:
898 """
899 Options:
900 * --force - allow unknown options, allow constraint for any resource type
901 * -f - CIB file
902 """
903 del lib
904 modifiers.ensure_only_supported("--force", "-f")
905 if len(argv) < 4:
906 raise CmdLineInputError()
907
908 constraint_id = argv.pop(0)
909 rsc_type, rsc_value = parse_args.parse_typed_arg(
910 argv.pop(0),
911 [RESOURCE_TYPE_RESOURCE, RESOURCE_TYPE_REGEXP],
912 RESOURCE_TYPE_RESOURCE,
913 )
914 node = argv.pop(0)
915 score = None
916 if "=" not in argv[0]:
917 score = argv.pop(0)
918 # TODO added to pcs in the first 0.12.x version
919 deprecation_warning(STANDALONE_SCORE_MSG)
920 options = []
921 # For now we only allow setting resource-discovery and score
922 for arg in argv:
923 name, value = parse_args.split_option(arg, allow_empty_value=False)
924 if name == "score":
925 score = value
926 elif name == "resource-discovery":
927 if not modifiers.get("--force"):
928 allowed_discovery = list(
929 map(
930 str,
931 [
932 CibResourceDiscovery.ALWAYS,
933 CibResourceDiscovery.EXCLUSIVE,
934 CibResourceDiscovery.NEVER,
935 ],
936 )
937 )
938 if value not in allowed_discovery:
939 utils.err(
940 (
941 "invalid {0} value '{1}', allowed values are: {2}"
942 ", use --force to override"
943 ).format(name, value, format_list(allowed_discovery))
944 )
945 options.append([name, value])
946 elif modifiers.get("--force"):
947 options.append([name, value])
948 else:
949 utils.err("bad option '%s', use --force to override" % name)
950 if score is None:
951 score = "INFINITY"
952
953 # Verify that specified node exists in the cluster and score is valid
954 if not skip_score_and_node_check:
955 if modifiers.is_specified("-f") or modifiers.get("--force"):
956 warn(LOCATION_NODE_VALIDATION_SKIP_MSG)
957 else:
958 lib_env = utils.get_lib_env()
959 existing_nodes, report_list = get_existing_nodes_names(
960 corosync_conf=lib_env.get_corosync_conf(),
961 cib=lib_env.get_cib(),
962 )
963 report_list += _verify_node_name(node, existing_nodes)
964 if report_list:
965 process_library_reports(report_list)
966 _verify_score(score)
967
968 id_valid, id_error = utils.validate_xml_id(constraint_id, "constraint id")
969 if not id_valid:
970 utils.err(id_error)
971
972 dom = utils.get_cib_dom()
973
974 if rsc_type == RESOURCE_TYPE_RESOURCE:
975 (
976 rsc_valid,
977 rsc_error,
978 dummy_correct_id,
979 ) = utils.validate_constraint_resource(dom, rsc_value)
980 if not rsc_valid:
981 utils.err(rsc_error)
982
983 # Verify current constraint doesn't already exist
984 # If it does we replace it with the new constraint
985 dummy_dom, constraintsElement = getCurrentConstraints(dom)
986 # If the id matches, or the rsc & node match, then we replace/remove
987 elementsToRemove = [
988 rsc_loc
989 for rsc_loc in constraintsElement.getElementsByTagName("rsc_location")
990 if rsc_loc.getAttribute("id") == constraint_id
991 or (
992 rsc_loc.getAttribute("node") == node
993 and (
994 (
995 rsc_type == RESOURCE_TYPE_RESOURCE
996 and rsc_loc.getAttribute("rsc") == rsc_value
997 )
998 or (
999 rsc_type == RESOURCE_TYPE_REGEXP
1000 and rsc_loc.getAttribute("rsc-pattern") == rsc_value
1001 )
1002 )
1003 )
1004 ]
1005 for etr in elementsToRemove:
1006 constraintsElement.removeChild(etr)
1007
1008 element = dom.createElement("rsc_location")
1009 element.setAttribute("id", constraint_id)
1010 if rsc_type == RESOURCE_TYPE_RESOURCE:
1011 element.setAttribute("rsc", rsc_value)
1012 elif rsc_type == RESOURCE_TYPE_REGEXP:
1013 element.setAttribute("rsc-pattern", rsc_value)
1014 element.setAttribute("node", node)
1015 element.setAttribute("score", score)
1016 for option in options:
1017 element.setAttribute(option[0], option[1])
1018 constraintsElement.appendChild(element)
1019
1020 utils.replace_cib_configuration(dom)
1021
1022
1023 # Grabs the current constraints and returns the dom and constraint element
1024 def getCurrentConstraints(passed_dom=None):
1025 """
1026 Commandline options:
1027 * -f - CIB file, only if passed_dom is None
1028 """
1029 if passed_dom:
1030 dom = passed_dom
1031 else:
1032 current_constraints_xml = utils.get_cib_xpath("//constraints")
1033 if current_constraints_xml == "":
1034 utils.err("unable to process cib")
1035 # Verify current constraint doesn't already exist
1036 # If it does we replace it with the new constraint
|
CID (unavailable; MK=42e4d47d6c846e21a3e7a16e47445de3) (#1 of 1): XML external entity processing enabled (SIGMA.xml_external_entity_enabled): |
|
(1) Event Sigma main event: |
The application uses Python's built in `xml` module which does not properly handle erroneous or maliciously constructed data, making the application vulnerable to one or more types of XML attacks. |
|
(2) Event remediation: |
Avoid using the `xml` module. Consider using the `defusedxml` module or similar which safely prevents all XML entity attacks. |
1037 dom = parseString(current_constraints_xml)
1038
1039 constraintsElement = dom.getElementsByTagName("constraints")[0]
1040 return (dom, constraintsElement)
1041
1042
1043 # If returnStatus is set, then we don't error out, we just print the error
1044 # and return false
1045 def constraint_rm( # noqa: PLR0912
1046 lib,
1047 argv,
1048 modifiers,
1049 returnStatus=False,
1050 constraintsElement=None,
1051 passed_dom=None,
1052 ):
1053 """
1054 Options:
1055 * -f - CIB file, effective only if passed_dom is None
1056 """
1057 if passed_dom is None:
1058 modifiers.ensure_only_supported("-f")
1059 if not argv:
1060 raise CmdLineInputError()
1061
1062 bad_constraint = False
1063 if len(argv) != 1:
1064 for arg in argv:
1065 if not constraint_rm(
1066 lib, [arg], modifiers, returnStatus=True, passed_dom=passed_dom
1067 ):
1068 bad_constraint = True
1069 if bad_constraint:
1070 sys.exit(1)
1071 return None
1072
1073 c_id = argv.pop(0)
1074 elementFound = False
1075 dom = None
1076 use_cibadmin = False
1077 if not constraintsElement:
1078 (dom, constraintsElement) = getCurrentConstraints(passed_dom)
1079 use_cibadmin = True
1080
1081 for co in constraintsElement.childNodes[:]:
1082 if co.nodeType != xml.dom.Node.ELEMENT_NODE:
1083 continue
1084 if co.getAttribute("id") == c_id:
1085 constraintsElement.removeChild(co)
1086 elementFound = True
1087
1088 if not elementFound:
1089 for rule in constraintsElement.getElementsByTagName("rule")[:]:
1090 if rule.getAttribute("id") == c_id:
1091 elementFound = True
1092 parent = rule.parentNode
1093 parent.removeChild(rule)
1094 if not parent.getElementsByTagName("rule"):
1095 parent.parentNode.removeChild(parent)
1096
1097 if elementFound:
1098 if passed_dom:
1099 return dom
1100 if use_cibadmin:
1101 utils.replace_cib_configuration(dom)
1102 if returnStatus:
1103 return True
1104 else:
1105 utils.err("Unable to find constraint - '%s'" % c_id, False)
1106 if returnStatus:
1107 return False
1108 sys.exit(1)
1109 return None
1110
1111
1112 def _split_set_constraints(
1113 constraints_dto: CibConstraintsDto,
1114 ) -> tuple[CibConstraintsDto, CibConstraintsDto]:
1115 return (
1116 CibConstraintsDto(
1117 location=constraints_dto.location,
1118 colocation=constraints_dto.colocation,
1119 order=constraints_dto.order,
1120 ticket=constraints_dto.ticket,
1121 ),
1122 CibConstraintsDto(
1123 location_set=constraints_dto.location_set,
1124 colocation_set=constraints_dto.colocation_set,
1125 order_set=constraints_dto.order_set,
1126 ticket_set=constraints_dto.ticket_set,
1127 ),
1128 )
1129
1130
1131 def _find_constraints_containing_resource(
1132 resources_dto: CibResourcesDto,
1133 constraints_dto: CibConstraintsDto,
1134 resource_id: str,
1135 ) -> CibConstraintsDto:
1136 resources_filter = [resource_id]
1137 # Original implementation only included parent resource only if resource_id
1138 # was referring to a primitive resource, ignoring groups. This may change in
1139 # the future if necessary.
1140 if any(
1141 primitive_dto.id == resource_id
1142 for primitive_dto in resources_dto.primitives
1143 ):
1144 for clone_dto in resources_dto.clones:
1145 if clone_dto.member_id == resource_id:
1146 resources_filter.append(clone_dto.id)
1147 break
1148 return _filter_constraints_by_resources(
1149 constraints_dto, resources_filter, []
1150 )
1151
1152
1153 def ref(
1154 lib: Any, argv: list[str], modifiers: parse_args.InputModifiers
1155 ) -> None:
1156 modifiers.ensure_only_supported("-f")
1157 if not argv:
1158 raise CmdLineInputError()
1159
1160 resources_dto = cast(
1161 CibResourcesDto, lib.resource.get_configured_resources()
1162 )
1163
1164 constraints_dto = cast(
1165 CibConstraintsDto,
1166 lib.constraint.get_config(evaluate_rules=False),
1167 )
1168
1169 for resource_id in sorted(set(argv)):
1170 constraint_ids = get_all_constraints_ids(
1171 _find_constraints_containing_resource(
1172 resources_dto, constraints_dto, resource_id
1173 )
1174 )
1175 print(f"Resource: {resource_id}")
1176 if constraint_ids:
1177 print(
1178 "\n".join(
1179 indent(
1180 sorted(constraint_ids),
1181 indent_step=INDENT_STEP,
1182 )
1183 )
1184 )
1185 else:
1186 print(" No Matches")
1187
1188
1189 def remove_constraints_containing(
1190 resource_id: str, output=False, constraints_element=None, passed_dom=None
1191 ):
1192 """
1193 Commandline options:
1194 * -f - CIB file, effective only if passed_dom is None
1195 """
1196 lib = utils.get_library_wrapper()
1197 modifiers = utils.get_input_modifiers()
1198 resources_dto = cast(
1199 CibResourcesDto, lib.resource.get_configured_resources()
1200 )
1201
1202 constraints_dto, set_constraints_dto = _split_set_constraints(
1203 cast(
1204 CibConstraintsDto,
1205 lib.constraint.get_config(evaluate_rules=False),
1206 )
1207 )
1208 constraints = sorted(
1209 get_all_constraints_ids(
1210 _find_constraints_containing_resource(
1211 resources_dto, constraints_dto, resource_id
1212 )
1213 )
1214 )
1215 set_constraints = sorted(
1216 get_all_constraints_ids(
1217 _find_constraints_containing_resource(
1218 resources_dto, set_constraints_dto, resource_id
1219 )
1220 )
1221 )
1222 for c in constraints:
1223 if output:
1224 print_to_stderr(f"Removing Constraint - {c}")
1225 if constraints_element is not None:
1226 constraint_rm(
1227 lib,
1228 [c],
1229 modifiers,
1230 True,
1231 constraints_element,
1232 passed_dom=passed_dom,
1233 )
1234 else:
1235 constraint_rm(lib, [c], modifiers, passed_dom=passed_dom)
1236
1237 if set_constraints:
1238 (dom, constraintsElement) = getCurrentConstraints(passed_dom)
1239 for set_c in constraintsElement.getElementsByTagName("resource_ref")[:]:
1240 # If resource id is in a set, remove it from the set, if the set
1241 # is empty, then we remove the set, if the parent of the set
1242 # is empty then we remove it
1243 if set_c.getAttribute("id") == resource_id:
1244 parent_node = set_c.parentNode
1245 parent_node.removeChild(set_c)
1246 if output:
1247 print_to_stderr(
1248 "Removing {} from set {}".format(
1249 resource_id, parent_node.getAttribute("id")
1250 )
1251 )
1252 if parent_node.getElementsByTagName("resource_ref").length == 0:
1253 print_to_stderr(
1254 "Removing set {}".format(parent_node.getAttribute("id"))
1255 )
1256 parent_node_2 = parent_node.parentNode
1257 parent_node_2.removeChild(parent_node)
1258 if (
1259 parent_node_2.getElementsByTagName(
1260 "resource_set"
1261 ).length
1262 == 0
1263 ):
1264 parent_node_2.parentNode.removeChild(parent_node_2)
1265 print_to_stderr(
1266 "Removing constraint {}".format(
1267 parent_node_2.getAttribute("id")
1268 )
1269 )
1270 if passed_dom:
1271 return dom
1272 utils.replace_cib_configuration(dom)
1273 return None
1274
1275
1276 # Re-assign any constraints referencing a resource to its parent (a clone
1277 # or master)
1278 def constraint_resource_update(old_id, dom):
1279 """
1280 Commandline options: no options
1281 """
1282 new_id = None
1283 clone_ms_parent = utils.dom_get_resource_clone_ms_parent(dom, old_id)
1284 if clone_ms_parent:
1285 new_id = clone_ms_parent.getAttribute("id")
1286
1287 if new_id:
1288 constraints = dom.getElementsByTagName("rsc_location")
1289 constraints += dom.getElementsByTagName("rsc_order")
1290 constraints += dom.getElementsByTagName("rsc_colocation")
1291 attrs_to_update = ["rsc", "first", "then", "with-rsc"]
1292 for constraint in constraints:
1293 for attr in attrs_to_update:
1294 if constraint.getAttribute(attr) == old_id:
1295 constraint.setAttribute(attr, new_id)
1296 return dom
1297