1 import datetime
2 import difflib
3 import grp
4 import json
5 import os
6 import os.path
7 import pwd
8 import re
9 import shutil
10 import sys
11 import tarfile
12 import tempfile
13 import time
14 from io import BytesIO
15 from typing import cast
16 from xml.dom.minidom import parse
17
18 from pcs import cluster, quorum, settings, status, usage, utils
19 from pcs.cli.alert.output import config_dto_to_lines as alerts_to_lines
20 from pcs.cli.cluster_property.output import (
21 PropertyConfigurationFacade,
22 properties_to_text,
23 )
24 from pcs.cli.common import middleware
25 from pcs.cli.common.errors import CmdLineInputError
26 from pcs.cli.common.output import INDENT_STEP, smart_wrap_text
27 from pcs.cli.constraint.output import constraints_to_text
28 from pcs.cli.nvset import nvset_dto_list_to_lines
29 from pcs.cli.reports import process_library_reports
30 from pcs.cli.reports.output import print_to_stderr, warn
31 from pcs.cli.resource.output import (
32 ResourcesConfigurationFacade,
33 resources_to_text,
34 )
35 from pcs.cli.stonith.levels.output import stonith_level_config_to_text
36 from pcs.cli.tag.output import tags_to_text
37 from pcs.common.interface import dto
38 from pcs.common.pacemaker.constraint import CibConstraintsDto
39 from pcs.common.str_tools import indent
40 from pcs.lib.errors import LibraryError
41 from pcs.lib.node import get_existing_nodes_names
42
43
44 def config_show(lib, argv, modifiers):
45 """
46 Options:
47 * -f - CIB file, when getting cluster name on remote node (corosync.conf
48 doesn't exist)
49 * --corosync_conf - corosync.conf file
50 """
51 modifiers.ensure_only_supported("-f", "--corosync_conf", "--show-secrets")
52 if argv:
53 raise CmdLineInputError()
54
55 corosync_conf_dto = None
56 cluster_name = ""
57 properties_facade = PropertyConfigurationFacade.from_properties_config(
58 lib.cluster_property.get_properties(),
59 )
60 try:
61 corosync_conf_dto = lib.cluster.get_corosync_conf_struct()
62 cluster_name = corosync_conf_dto.cluster_name
63 except LibraryError:
64 # there is no corosync.conf on remote nodes, we can try to
65 # get cluster name from pacemaker
66 pass
67 if not cluster_name:
68 cluster_name = properties_facade.get_property_value("cluster-name", "")
69 print("Cluster Name: %s" % cluster_name)
70
71 status.nodes_status(lib, ["config"], modifiers.get_subset("-f"))
72 cib_lines = _config_show_cib_lines(lib, properties_facade=properties_facade)
73 if cib_lines:
74 print()
75 print("\n".join(cib_lines))
76 if (
77 utils.hasCorosyncConf()
78 and not modifiers.is_specified("-f")
79 and not modifiers.is_specified("--corosync_conf")
80 ):
81 cluster.cluster_uidgid(
82 lib, [], modifiers.get_subset(), silent_list=True
83 )
84 if corosync_conf_dto:
85 quorum_device_dict = {}
86 if corosync_conf_dto.quorum_device:
87 quorum_device_dict = dto.to_dict(corosync_conf_dto.quorum_device)
88 config = dict(
89 options=corosync_conf_dto.quorum_options,
90 device=quorum_device_dict,
91 )
92 quorum_lines = quorum.quorum_config_to_str(config)
93 if quorum_lines:
94 print()
95 print("Quorum:")
96 print("\n".join(indent(quorum_lines)))
97
98
99 def _config_show_cib_lines(lib, properties_facade=None): # noqa: PLR0912, PLR0915
100 """
101 Commandline options:
102 * -f - CIB file
103 """
104
105 # update of pcs_options will change output of constraint show and
106 # displaying resources and operations defaults
107 utils.pcs_options["--full"] = 1
108 # get latest modifiers object after updating pcs_options
109 modifiers = utils.get_input_modifiers()
110
111 resources_facade = ResourcesConfigurationFacade.from_resources_dto(
112 lib.resource.get_configured_resources()
113 )
114 resources_only_facade = resources_facade.filter_stonith(False)
115 stonith_only_facade = resources_facade.filter_stonith(True)
116 if modifiers.is_specified("--show-secrets"):
117 for facade in [resources_only_facade, stonith_only_facade]:
118 queries = facade.get_secrets_queries()
119 if queries:
120 facade.update_secrets_values(
121 lib.resource.get_cibsecrets(queries)
122 )
123
124 all_lines = []
125
126 resources_lines = smart_wrap_text(
127 indent(
128 resources_to_text(resources_only_facade), indent_step=INDENT_STEP
129 )
130 )
131 if resources_lines:
132 all_lines.append("Resources:")
133 all_lines.extend(resources_lines)
134
135 stonith_lines = smart_wrap_text(
136 indent(resources_to_text(stonith_only_facade), indent_step=INDENT_STEP)
137 )
138 if stonith_lines:
139 if all_lines:
140 all_lines.append("")
141 all_lines.append("Stonith Devices:")
142 all_lines.extend(stonith_lines)
143
144 levels_lines = stonith_level_config_to_text(
145 lib.fencing_topology.get_config_dto()
146 )
147 if levels_lines:
148 if all_lines:
149 all_lines.append("")
150 all_lines.append("Fencing Levels:")
151 all_lines.extend(indent(levels_lines, indent_step=2))
152
153 constraints_lines = smart_wrap_text(
154 constraints_to_text(
155 cast(
156 CibConstraintsDto,
157 lib.constraint.get_config(evaluate_rules=False),
158 ),
159 modifiers.is_specified("--full"),
160 )
161 )
162 if constraints_lines:
163 if all_lines:
164 all_lines.append("")
165 all_lines.extend(constraints_lines)
166
167 alert_lines = indent(alerts_to_lines(lib.alert.get_config_dto()))
168 if alert_lines:
169 if all_lines:
170 all_lines.append("")
171 all_lines.append("Alerts:")
172 all_lines.extend(alert_lines)
173
174 resources_defaults_lines = indent(
175 nvset_dto_list_to_lines(
176 lib.cib_options.resource_defaults_config(
177 evaluate_expired=False
178 ).meta_attributes,
179 nvset_label="Meta Attrs",
180 with_ids=modifiers.get("--full"),
181 )
182 )
183 if resources_defaults_lines:
184 if all_lines:
185 all_lines.append("")
186 all_lines.append("Resources Defaults:")
187 all_lines.extend(resources_defaults_lines)
188
189 operations_defaults_lines = indent(
190 nvset_dto_list_to_lines(
191 lib.cib_options.operation_defaults_config(
192 evaluate_expired=False
193 ).meta_attributes,
194 nvset_label="Meta Attrs",
195 with_ids=modifiers.get("--full"),
196 )
197 )
198 if operations_defaults_lines:
199 if all_lines:
200 all_lines.append("")
201 all_lines.append("Operations Defaults:")
202 all_lines.extend(operations_defaults_lines)
203
204 if not properties_facade:
205 properties_facade = PropertyConfigurationFacade.from_properties_config(
206 lib.cluster_property.get_properties()
207 )
208 properties_lines = properties_to_text(properties_facade)
209 if properties_lines:
210 if all_lines:
211 all_lines.append("")
212 all_lines.extend(properties_lines)
213
214 tag_lines = smart_wrap_text(tags_to_text(lib.tag.get_config_dto([])))
215 if tag_lines:
216 if all_lines:
217 all_lines.append("")
218 all_lines.append("Tags:")
219 all_lines.extend(indent(tag_lines, indent_step=1))
220
221 return all_lines
222
223
224 def config_backup(lib, argv, modifiers):
225 """
226 Options:
227 * --force - overwrite file if already exists
228 """
229 del lib
230 modifiers.ensure_only_supported("--force")
231 if len(argv) > 1:
232 raise CmdLineInputError()
233
234 outfile_name = None
235 if argv:
236 outfile_name = argv[0]
237 if not outfile_name.endswith(".tar.bz2"):
238 outfile_name += ".tar.bz2"
239
240 tar_data = config_backup_local()
241 if outfile_name:
242 ok, message = utils.write_file(
243 outfile_name, tar_data, permissions=0o600, binary=True
244 )
245 if not ok:
246 utils.err(message)
247 else:
248 # in python3 stdout accepts str so we need to use buffer
249 sys.stdout.buffer.write(tar_data)
250
251
252 def config_backup_local():
253 """
254 Commandline options: no options
255 """
256 file_list = config_backup_path_list()
257 tar_data = BytesIO()
258
259 try:
260 with tarfile.open(fileobj=tar_data, mode="w|bz2") as tarball:
261 config_backup_add_version_to_tarball(tarball)
262 for tar_path, path_info in file_list.items():
263 if (
264 not os.path.exists(path_info["path"])
265 and not path_info["required"]
266 ):
267 continue
268 tarball.add(path_info["path"], tar_path)
269 except (tarfile.TarError, OSError) as e:
270 utils.err("unable to create tarball: %s" % e)
271
272 tar = tar_data.getvalue()
273 tar_data.close()
274 return tar
275
276
277 def config_restore(lib, argv, modifiers):
278 """
279 Options:
280 * --local - restore config only on local node
281 * --request-timeout - timeout for HTTP requests, used only if --local was
282 not defined or user is not root
283 """
284 del lib
285 modifiers.ensure_only_supported("--local", "--request-timeout")
286 if len(argv) > 1:
287 raise CmdLineInputError()
288
289 infile_name = infile_obj = None
290 if argv:
291 infile_name = argv[0]
292 if not infile_name:
293 # in python3 stdin returns str so we need to use buffer
294 infile_obj = BytesIO(sys.stdin.buffer.read())
295
296 if os.getuid() == 0:
297 if modifiers.get("--local"):
298 config_restore_local(infile_name, infile_obj)
299 else:
300 config_restore_remote(infile_name, infile_obj)
301 else:
302 new_argv = ["config", "restore"]
303 options = []
304 new_stdin = None
305 if modifiers.get("--local"):
306 options.append("--local")
307 if infile_name:
308 new_argv.append(os.path.abspath(infile_name))
309 else:
310 new_stdin = infile_obj.read()
311 err_msgs, exitcode, std_out, std_err = utils.call_local_pcsd(
312 new_argv, options, new_stdin
313 )
314 if err_msgs:
315 for msg in err_msgs:
316 utils.err(msg, False)
317 sys.exit(1)
318 print(std_out)
319 sys.stderr.write(std_err)
320 sys.exit(exitcode)
321
322
323 def config_restore_remote(infile_name, infile_obj): # noqa: PLR0912
324 """
325 Commandline options:
326 * --request-timeout - timeout for HTTP requests
327 """
328 extracted = {
329 "version.txt": "",
330 "corosync.conf": "",
331 }
332 try:
333 with tarfile.open(infile_name, "r|*", infile_obj) as tarball:
334 while True:
335 # next(tarball) does not work in python2.6
336 tar_member_info = tarball.next()
337 if tar_member_info is None:
338 break
339 if tar_member_info.name in extracted:
340 tar_member = tarball.extractfile(tar_member_info)
341 extracted[tar_member_info.name] = tar_member.read()
342 tar_member.close()
343 except (tarfile.TarError, OSError) as e:
344 utils.err("unable to read the tarball: %s" % e)
345
346 config_backup_check_version(extracted["version.txt"])
347
348 node_list, report_list = get_existing_nodes_names(
349 utils.get_corosync_conf_facade(
350 conf_text=extracted["corosync.conf"].decode("utf-8")
351 )
352 )
353 if report_list:
354 process_library_reports(report_list)
355 if not node_list:
356 utils.err("no nodes found in the tarball")
357
358 err_msgs = []
359 for node in node_list:
360 try:
361 retval, output = utils.checkStatus(node)
362 if retval != 0:
363 err_msgs.append(output)
364 continue
365 _status = json.loads(output)
366 if any(
367 _status["node"]["services"][service_name]["running"]
368 for service_name in (
369 "corosync",
370 "pacemaker",
371 "pacemaker_remote",
372 )
373 ):
374 err_msgs.append(
375 "Cluster is currently running on node %s. You need to stop "
376 "the cluster in order to restore the configuration." % node
377 )
378 continue
379 except (ValueError, NameError, LookupError):
380 err_msgs.append("unable to determine status of the node %s" % node)
381 if err_msgs:
382 for msg in err_msgs:
383 utils.err(msg, False)
384 sys.exit(1)
385
386 # Temporarily disable config files syncing thread in pcsd so it will not
387 # rewrite restored files. 10 minutes should be enough time to restore.
388 # If node returns HTTP 404 it does not support config syncing at all.
389 for node in node_list:
390 retval, output = utils.pauseConfigSyncing(node, 10 * 60)
391 if not (retval == 0 or "(HTTP error: 404)" in output):
392 utils.err(output)
393
394 if infile_obj:
395 infile_obj.seek(0)
396 tarball_data = infile_obj.read()
397 else:
398 with open(infile_name, "rb") as tarball:
399 tarball_data = tarball.read()
400
401 error_list = []
402 for node in node_list:
403 retval, error = utils.restoreConfig(node, tarball_data)
404 if retval != 0:
405 error_list.append(error)
406 if error_list:
407 utils.err("unable to restore all nodes\n" + "\n".join(error_list))
408
409
410 def config_restore_local(infile_name, infile_obj): # noqa: PLR0912, PLR0915
411 """
412 Commandline options: no options
413 """
414 service_manager = utils.get_service_manager()
415 if (
416 service_manager.is_running("corosync")
417 or service_manager.is_running("pacemaker")
418 or service_manager.is_running("pacemaker_remote")
419 ):
420 utils.err(
421 "Cluster is currently running on this node. You need to stop "
422 "the cluster in order to restore the configuration."
423 )
424
425 file_list = config_backup_path_list(with_uid_gid=True)
426 tarball_file_list = []
427 version = None
428 tmp_dir = None
429 try:
430 with tarfile.open(infile_name, "r|*", infile_obj) as tarball:
431 while True:
432 # next(tarball) does not work in python2.6
433 tar_member_info = tarball.next()
434 if tar_member_info is None:
435 break
436 if tar_member_info.name == "version.txt":
437 version_data = tarball.extractfile(tar_member_info)
438 version = version_data.read()
439 version_data.close()
440 continue
441 tarball_file_list.append(tar_member_info.name)
442
443 required_file_list = [
444 tar_path
445 for tar_path, path_info in file_list.items()
446 if path_info["required"]
447 ]
448 missing = set(required_file_list) - set(tarball_file_list)
449 if missing:
450 utils.err(
451 "unable to restore the cluster, missing files in backup: %s"
452 % ", ".join(missing)
453 )
454
455 config_backup_check_version(version)
456
457 if infile_obj:
458 infile_obj.seek(0)
459 with tarfile.open(infile_name, "r|*", infile_obj) as tarball:
460 while True:
461 # next(tarball) does not work in python2.6
462 tar_member_info = tarball.next()
463 if tar_member_info is None:
464 break
465 extract_info = None
466 path = tar_member_info.name
467 while path:
468 if path in file_list:
469 extract_info = file_list[path]
470 break
471 path = os.path.dirname(path)
472 if not extract_info:
473 continue
474 path_full = None
475 if callable(extract_info.get("pre_store_call")):
476 extract_info["pre_store_call"]()
477 if "rename" in extract_info and extract_info["rename"]:
478 if tmp_dir is None:
479 tmp_dir = tempfile.mkdtemp()
480 if hasattr(tarfile, "data_filter"):
481 # Safe way of extraction is available since Python 3.12,
482 # hasattr above checks if it's available.
483 # It's also backported to 3.11.4, 3.10.12, 3.9.17.
484 # It may be backported to older versions in downstream.
485 tarball.extractall(
486 tmp_dir, [tar_member_info], filter="data"
487 )
488 else:
489 # Unsafe way of extraction
490 # Remove once we don't support Python 3.8 and older
491 tarball.extractall(tmp_dir, [tar_member_info])
492 path_full = extract_info["path"]
493 shutil.move(
494 os.path.join(tmp_dir, tar_member_info.name), path_full
495 )
496 else:
497 dir_path = os.path.dirname(extract_info["path"])
498 if hasattr(tarfile, "data_filter"):
499 # Safe way of extraction is available since Python 3.12,
500 # hasattr above checks if it's available.
501 # It's also backported to 3.11.4, 3.10.12, 3.9.17.
502 # It may be backported to older versions in downstream.
503 tarball.extractall(
504 dir_path, [tar_member_info], filter="data"
505 )
506 else:
507 # Unsafe way of extracting
508 # Remove once we don't support Python 3.8 and older
509 tarball.extractall(dir_path, [tar_member_info])
510 path_full = os.path.join(dir_path, tar_member_info.name)
511 file_attrs = extract_info["attrs"]
512 os.chmod(path_full, file_attrs["mode"])
513 os.chown(path_full, file_attrs["uid"], file_attrs["gid"])
514 except (tarfile.TarError, OSError) as e:
515 utils.err("unable to restore the cluster: %s" % e)
516 finally:
517 if tmp_dir:
518 shutil.rmtree(tmp_dir, ignore_errors=True)
519
520 try:
521 sig_path = os.path.join(settings.cib_dir, "cib.xml.sig")
522 if os.path.exists(sig_path):
523 os.remove(sig_path)
524 except OSError as e:
525 utils.err("unable to remove %s: %s" % (sig_path, e))
526
527
528 def config_backup_path_list(with_uid_gid=False):
529 """
530 Commandline options: no option
531 NOTE: corosync.conf path may be altered using --corosync_conf
532 """
533 corosync_attrs = {
534 "mtime": int(time.time()),
535 "mode": 0o644,
536 "uname": "root",
537 "gname": "root",
538 "uid": 0,
539 "gid": 0,
540 }
541 corosync_authkey_attrs = dict(corosync_attrs)
542 corosync_authkey_attrs["mode"] = 0o400
543 cib_attrs = {
544 "mtime": int(time.time()),
545 "mode": 0o600,
546 "uname": settings.pacemaker_uname,
547 "gname": settings.pacemaker_gname,
548 }
549 if with_uid_gid:
550 cib_attrs["uid"] = _get_uid(cib_attrs["uname"])
551 cib_attrs["gid"] = _get_gid(cib_attrs["gname"])
552
553 pcmk_authkey_attrs = dict(cib_attrs)
554 pcmk_authkey_attrs["mode"] = 0o440
555 return {
556 "cib.xml": {
557 "path": os.path.join(settings.cib_dir, "cib.xml"),
558 "required": True,
559 "attrs": dict(cib_attrs),
560 },
561 "corosync_authkey": {
562 "path": settings.corosync_authkey_file,
563 "required": False,
564 "attrs": corosync_authkey_attrs,
565 "restore_procedure": None,
566 "rename": True,
567 },
568 "pacemaker_authkey": {
569 "path": settings.pacemaker_authkey_file,
570 "required": False,
571 "attrs": pcmk_authkey_attrs,
572 "restore_procedure": None,
573 "rename": True,
574 "pre_store_call": _ensure_etc_pacemaker_exists,
575 },
576 "corosync.conf": {
577 "path": settings.corosync_conf_file,
578 "required": True,
579 "attrs": dict(corosync_attrs),
580 },
581 "uidgid.d": {
582 "path": settings.corosync_uidgid_dir,
583 "required": False,
584 "attrs": dict(corosync_attrs),
585 },
586 "pcs_settings.conf": {
587 "path": settings.pcsd_settings_conf_location,
588 "required": False,
589 "attrs": {
590 "mtime": int(time.time()),
591 "mode": 0o644,
592 "uname": "root",
593 "gname": "root",
594 "uid": 0,
595 "gid": 0,
596 },
597 },
598 }
599
600
601 def _get_uid(user_name):
602 """
603 Commandline options: no options
604 """
605 try:
606 return pwd.getpwnam(user_name).pw_uid
607 except KeyError:
608 return utils.err(
609 "Unable to determine uid of user '{0}'".format(user_name)
610 )
611
612
613 def _get_gid(group_name):
614 """
615 Commandline options: no options
616 """
617 try:
618 return grp.getgrnam(group_name).gr_gid
619 except KeyError:
620 return utils.err(
621 "Unable to determine gid of group '{0}'".format(group_name)
622 )
623
624
625 def _ensure_etc_pacemaker_exists():
626 """
627 Commandline options: no options
628 """
629 dir_name = os.path.dirname(settings.pacemaker_authkey_file)
630 if not os.path.exists(dir_name):
631 os.mkdir(dir_name)
632 os.chmod(dir_name, 0o750)
633 os.chown(
634 dir_name,
635 _get_uid(settings.pacemaker_uname),
636 _get_gid(settings.pacemaker_gname),
637 )
638
639
640 def config_backup_check_version(version):
641 """
642 Commandline options: no options
643 """
644 try:
645 version_number = int(version)
646 supported_version = config_backup_version()
647 if version_number > supported_version:
648 utils.err(
649 f"Unsupported version of the backup, supported version is "
650 f"{supported_version}, backup version is {version_number}"
651 )
652 if version_number < supported_version:
653 warn(
654 f"Restoring from the backup version {version_number}, current "
655 f"supported version is {supported_version}"
656 )
657 except TypeError:
658 utils.err("Cannot determine version of the backup")
659
660
661 def config_backup_add_version_to_tarball(tarball, version=None):
662 """
663 Commandline options: no options
664 """
665 ver = version if version is not None else str(config_backup_version())
666 return utils.tar_add_file_data(tarball, ver.encode("utf-8"), "version.txt")
667
668
669 def config_backup_version():
670 """
671 Commandline options: no options
672 """
673 return 1
674
675
676 def config_checkpoint_list(lib, argv, modifiers):
677 """
678 Options: no options
679 """
680 del lib
681 modifiers.ensure_only_supported()
682 if argv:
683 raise CmdLineInputError()
684 try:
685 file_list = os.listdir(settings.cib_dir)
686 except OSError as e:
687 utils.err("unable to list checkpoints: %s" % e)
688 cib_list = []
689 cib_name_re = re.compile(r"^cib-(\d+)\.raw$")
690 for filename in file_list:
691 match = cib_name_re.match(filename)
692 if not match:
693 continue
694 file_path = os.path.join(settings.cib_dir, filename)
695 try:
696 if os.path.isfile(file_path):
697 cib_list.append(
698 (float(os.path.getmtime(file_path)), match.group(1))
699 )
700 except OSError:
701 pass
702 cib_list.sort()
703 if not cib_list:
704 print_to_stderr("No checkpoints available")
705 return
706 for cib_info in cib_list:
707 print(
708 "checkpoint %s: date %s"
709 % (cib_info[1], datetime.datetime.fromtimestamp(round(cib_info[0])))
710 )
711
712
713 def _checkpoint_to_lines(lib, checkpoint_number):
714 # backup current settings
715 orig_usefile = utils.usefile
716 orig_filename = utils.filename
717 orig_middleware = lib.middleware_factory
718 orig_env = lib.env
719 # configure old code to read the CIB from a file
720 utils.usefile = True
721 utils.filename = os.path.join(
722 settings.cib_dir, "cib-%s.raw" % checkpoint_number
723 )
724 # configure new code to read the CIB from a file
725 lib.middleware_factory = orig_middleware._replace(
726 cib=middleware.cib(utils.filename, utils.touch_cib_file)
727 )
728 lib.env = utils.get_cli_env()
729 # export the CIB to text
730 result = False, []
731 if os.path.isfile(utils.filename):
732 result = True, _config_show_cib_lines(lib)
733 # restore original settings
734 utils.usefile = orig_usefile
735 utils.filename = orig_filename
736 lib.middleware_factory = orig_middleware
737 lib.env = orig_env
738 return result
739
740
741 def config_checkpoint_view(lib, argv, modifiers):
742 """
743 Options: no options
744 """
745 modifiers.ensure_only_supported()
746 if len(argv) != 1:
747 print_to_stderr(usage.config(["checkpoint view"]))
748 sys.exit(1)
749
750 loaded, lines = _checkpoint_to_lines(lib, argv[0])
751 if not loaded:
752 utils.err("unable to read the checkpoint")
753 print("\n".join(lines))
754
755
756 def config_checkpoint_diff(lib, argv, modifiers):
757 """
758 Commandline options:
759 * -f - CIB file
760 """
761 modifiers.ensure_only_supported("-f")
762 if len(argv) != 2:
763 print_to_stderr(usage.config(["checkpoint diff"]))
764 sys.exit(1)
765
766 if argv[0] == argv[1]:
767 utils.err("cannot diff a checkpoint against itself")
768
769 errors = []
770 checkpoints_lines = []
771 for checkpoint in argv:
772 if checkpoint == "live":
773 lines = _config_show_cib_lines(lib)
774 if not lines:
775 errors.append("unable to read live configuration")
776 else:
777 checkpoints_lines.append(lines)
778 else:
779 loaded, lines = _checkpoint_to_lines(lib, checkpoint)
780 if not loaded:
781 errors.append(
782 "unable to read checkpoint '{0}'".format(checkpoint)
783 )
784 else:
785 checkpoints_lines.append(lines)
786
787 if errors:
788 utils.err("\n".join(errors))
789
790 print(
791 "Differences between {0} (-) and {1} (+):".format(
792 *[
793 (
794 "live configuration"
795 if label == "live"
796 else f"checkpoint {label}"
797 )
798 for label in argv
799 ]
800 )
801 )
802 print(
803 "\n".join(
804 [
805 line.rstrip()
806 for line in difflib.Differ().compare(
807 checkpoints_lines[0], checkpoints_lines[1]
808 )
809 ]
810 )
811 )
812
813
814 def config_checkpoint_restore(lib, argv, modifiers):
815 """
816 Options:
817 * -f - CIB file, a checkpoint will be restored into a specified file
818 """
819 del lib
820 modifiers.ensure_only_supported("-f")
821 if len(argv) != 1:
822 print_to_stderr(usage.config(["checkpoint restore"]))
823 sys.exit(1)
824
825 cib_path = os.path.join(settings.cib_dir, "cib-%s.raw" % argv[0])
826 try:
|
CID (unavailable; MK=9a5c3687cf169763ce9a245b2f141b8d) (#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. |
827 snapshot_dom = parse(cib_path)
828 except Exception as e:
829 utils.err("unable to read the checkpoint: %s" % e)
830 utils.replace_cib_configuration(snapshot_dom)
831