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.output import print_to_stderr, warn
30 from pcs.cli.reports.processor import report_and_exit_on_error
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 modifiers.ensure_only_supported("--local", "--request-timeout")
285 if len(argv) > 1:
286 raise CmdLineInputError()
287
288 infile_name = infile_obj = None
289 if argv:
290 infile_name = argv[0]
291 if not infile_name:
292 # in python3 stdin returns str so we need to use buffer
293 infile_obj = BytesIO(sys.stdin.buffer.read())
294
295 if os.getuid() == 0:
296 if modifiers.get("--local"):
297 config_restore_local(infile_name, infile_obj)
298 else:
299 config_restore_remote(
300 lib.env.report_processor, infile_name, infile_obj
301 )
302 else:
303 new_argv = ["config", "restore"]
304 options = []
305 new_stdin = None
306 if modifiers.get("--local"):
307 options.append("--local")
308 if infile_name:
309 new_argv.append(os.path.abspath(infile_name))
310 else:
311 new_stdin = infile_obj.read()
312 err_msgs, exitcode, std_out, std_err = utils.call_local_pcsd(
313 new_argv, options, new_stdin
314 )
315 if err_msgs:
316 for msg in err_msgs:
317 utils.err(msg, False)
318 sys.exit(1)
319 print(std_out)
320 sys.stderr.write(std_err)
321 sys.exit(exitcode)
322
323
324 def config_restore_remote(report_processor, infile_name, infile_obj): # noqa: PLR0912
325 """
326 Commandline options:
327 * --request-timeout - timeout for HTTP requests
328 """
329 extracted = {
330 "version.txt": "",
331 "corosync.conf": "",
332 }
333 try:
334 with tarfile.open(infile_name, "r|*", infile_obj) as tarball:
335 while True:
336 # next(tarball) does not work in python2.6
337 tar_member_info = tarball.next()
338 if tar_member_info is None:
339 break
340 if tar_member_info.name in extracted:
341 tar_member = tarball.extractfile(tar_member_info)
342 extracted[tar_member_info.name] = tar_member.read()
343 tar_member.close()
344 except (tarfile.TarError, OSError) as e:
345 utils.err("unable to read the tarball: %s" % e)
346
347 config_backup_check_version(extracted["version.txt"])
348
349 node_list, report_list = get_existing_nodes_names(
350 utils.get_corosync_conf_facade(
351 conf_text=extracted["corosync.conf"].decode("utf-8")
352 )
353 )
354 if report_list:
355 report_and_exit_on_error(report_processor, report_list)
356 if not node_list:
357 utils.err("no nodes found in the tarball")
358
359 err_msgs = []
360 for node in node_list:
361 try:
362 retval, output = utils.checkStatus(node)
363 if retval != 0:
364 err_msgs.append(output)
365 continue
366 _status = json.loads(output)
367 if any(
368 _status["node"]["services"][service_name]["running"]
369 for service_name in (
370 "corosync",
371 "pacemaker",
372 "pacemaker_remote",
373 )
374 ):
375 err_msgs.append(
376 "Cluster is currently running on node %s. You need to stop "
377 "the cluster in order to restore the configuration." % node
378 )
379 continue
380 except (ValueError, NameError, LookupError):
381 err_msgs.append("unable to determine status of the node %s" % node)
382 if err_msgs:
383 for msg in err_msgs:
384 utils.err(msg, False)
385 sys.exit(1)
386
387 # Temporarily disable config files syncing thread in pcsd so it will not
388 # rewrite restored files. 10 minutes should be enough time to restore.
389 # If node returns HTTP 404 it does not support config syncing at all.
390 for node in node_list:
391 retval, output = utils.pauseConfigSyncing(node, 10 * 60)
392 if not (retval == 0 or "(HTTP error: 404)" in output):
393 utils.err(output)
394
395 if infile_obj:
396 infile_obj.seek(0)
397 tarball_data = infile_obj.read()
398 else:
399 with open(infile_name, "rb") as tarball:
400 tarball_data = tarball.read()
401
402 error_list = []
403 for node in node_list:
404 retval, error = utils.restoreConfig(node, tarball_data)
405 if retval != 0:
406 error_list.append(error)
407 if error_list:
408 utils.err("unable to restore all nodes\n" + "\n".join(error_list))
409
410
411 def config_restore_local(infile_name, infile_obj): # noqa: PLR0912, PLR0915
412 """
413 Commandline options: no options
414 """
415 service_manager = utils.get_service_manager()
416 if (
417 service_manager.is_running("corosync")
418 or service_manager.is_running("pacemaker")
419 or service_manager.is_running("pacemaker_remote")
420 ):
421 utils.err(
422 "Cluster is currently running on this node. You need to stop "
423 "the cluster in order to restore the configuration."
424 )
425
426 file_list = config_backup_path_list(with_uid_gid=True)
427 tarball_file_list = []
428 version = None
429 tmp_dir = None
430 try:
431 with tarfile.open(infile_name, "r|*", infile_obj) as tarball:
432 while True:
433 # next(tarball) does not work in python2.6
434 tar_member_info = tarball.next()
435 if tar_member_info is None:
436 break
437 if tar_member_info.name == "version.txt":
438 version_data = tarball.extractfile(tar_member_info)
439 version = version_data.read()
440 version_data.close()
441 continue
442 tarball_file_list.append(tar_member_info.name)
443
444 required_file_list = [
445 tar_path
446 for tar_path, path_info in file_list.items()
447 if path_info["required"]
448 ]
449 missing = set(required_file_list) - set(tarball_file_list)
450 if missing:
451 utils.err(
452 "unable to restore the cluster, missing files in backup: %s"
453 % ", ".join(missing)
454 )
455
456 config_backup_check_version(version)
457
458 if infile_obj:
459 infile_obj.seek(0)
460 with tarfile.open(infile_name, "r|*", infile_obj) as tarball:
461 while True:
462 # next(tarball) does not work in python2.6
463 tar_member_info = tarball.next()
464 if tar_member_info is None:
465 break
466 extract_info = None
467 path = tar_member_info.name
468 while path:
469 if path in file_list:
470 extract_info = file_list[path]
471 break
472 path = os.path.dirname(path)
473 if not extract_info:
474 continue
475 path_full = None
476 if callable(extract_info.get("pre_store_call")):
477 extract_info["pre_store_call"]()
478 if "rename" in extract_info and extract_info["rename"]:
479 if tmp_dir is None:
480 tmp_dir = tempfile.mkdtemp()
481 if hasattr(tarfile, "data_filter"):
482 # Safe way of extraction is available since Python 3.12,
483 # hasattr above checks if it's available.
484 # It's also backported to 3.11.4, 3.10.12, 3.9.17.
485 # It may be backported to older versions in downstream.
486 tarball.extractall(
487 tmp_dir, [tar_member_info], filter="data"
488 )
489 else:
490 # Unsafe way of extraction
491 # Remove once we don't support Python 3.8 and older
492 tarball.extractall(tmp_dir, [tar_member_info])
493 path_full = extract_info["path"]
494 shutil.move(
495 os.path.join(tmp_dir, tar_member_info.name), path_full
496 )
497 else:
498 dir_path = os.path.dirname(extract_info["path"])
499 if hasattr(tarfile, "data_filter"):
500 # Safe way of extraction is available since Python 3.12,
501 # hasattr above checks if it's available.
502 # It's also backported to 3.11.4, 3.10.12, 3.9.17.
503 # It may be backported to older versions in downstream.
504 tarball.extractall(
505 dir_path, [tar_member_info], filter="data"
506 )
507 else:
508 # Unsafe way of extracting
509 # Remove once we don't support Python 3.8 and older
510 tarball.extractall(dir_path, [tar_member_info])
511 path_full = os.path.join(dir_path, tar_member_info.name)
512 file_attrs = extract_info["attrs"]
513 os.chmod(path_full, file_attrs["mode"])
514 os.chown(path_full, file_attrs["uid"], file_attrs["gid"])
515 except (tarfile.TarError, OSError) as e:
516 utils.err("unable to restore the cluster: %s" % e)
517 finally:
518 if tmp_dir:
519 shutil.rmtree(tmp_dir, ignore_errors=True)
520
521 try:
522 sig_path = os.path.join(settings.cib_dir, "cib.xml.sig")
523 if os.path.exists(sig_path):
524 os.remove(sig_path)
525 except OSError as e:
526 utils.err("unable to remove %s: %s" % (sig_path, e))
527
528
529 def config_backup_path_list(with_uid_gid=False):
530 """
531 Commandline options: no option
532 NOTE: corosync.conf path may be altered using --corosync_conf
533 """
534 corosync_attrs = {
535 "mtime": int(time.time()),
536 "mode": 0o644,
537 "uname": "root",
538 "gname": "root",
539 "uid": 0,
540 "gid": 0,
541 }
542 corosync_authkey_attrs = dict(corosync_attrs)
543 corosync_authkey_attrs["mode"] = 0o400
544 cib_attrs = {
545 "mtime": int(time.time()),
546 "mode": 0o600,
547 "uname": settings.pacemaker_uname,
548 "gname": settings.pacemaker_gname,
549 }
550 if with_uid_gid:
551 cib_attrs["uid"] = _get_uid(cib_attrs["uname"])
552 cib_attrs["gid"] = _get_gid(cib_attrs["gname"])
553
554 pcmk_authkey_attrs = dict(cib_attrs)
555 pcmk_authkey_attrs["mode"] = 0o440
556 return {
557 "cib.xml": {
558 "path": os.path.join(settings.cib_dir, "cib.xml"),
559 "required": True,
560 "attrs": dict(cib_attrs),
561 },
562 "corosync_authkey": {
563 "path": settings.corosync_authkey_file,
564 "required": False,
565 "attrs": corosync_authkey_attrs,
566 "restore_procedure": None,
567 "rename": True,
568 },
569 "pacemaker_authkey": {
570 "path": settings.pacemaker_authkey_file,
571 "required": False,
572 "attrs": pcmk_authkey_attrs,
573 "restore_procedure": None,
574 "rename": True,
575 "pre_store_call": _ensure_etc_pacemaker_exists,
576 },
577 "corosync.conf": {
578 "path": settings.corosync_conf_file,
579 "required": True,
580 "attrs": dict(corosync_attrs),
581 },
582 "uidgid.d": {
583 "path": settings.corosync_uidgid_dir,
584 "required": False,
585 "attrs": dict(corosync_attrs),
586 },
587 "pcs_settings.conf": {
588 "path": settings.pcsd_settings_conf_location,
589 "required": False,
590 "attrs": {
591 "mtime": int(time.time()),
592 "mode": 0o644,
593 "uname": "root",
594 "gname": "root",
595 "uid": 0,
596 "gid": 0,
597 },
598 },
599 }
600
601
602 def _get_uid(user_name):
603 """
604 Commandline options: no options
605 """
606 try:
607 return pwd.getpwnam(user_name).pw_uid
608 except KeyError:
609 return utils.err(
610 "Unable to determine uid of user '{0}'".format(user_name)
611 )
612
613
614 def _get_gid(group_name):
615 """
616 Commandline options: no options
617 """
618 try:
619 return grp.getgrnam(group_name).gr_gid
620 except KeyError:
621 return utils.err(
622 "Unable to determine gid of group '{0}'".format(group_name)
623 )
624
625
626 def _ensure_etc_pacemaker_exists():
627 """
628 Commandline options: no options
629 """
630 dir_name = os.path.dirname(settings.pacemaker_authkey_file)
631 if not os.path.exists(dir_name):
632 os.mkdir(dir_name)
633 os.chmod(dir_name, 0o750)
634 os.chown(
635 dir_name,
636 _get_uid(settings.pacemaker_uname),
637 _get_gid(settings.pacemaker_gname),
638 )
639
640
641 def config_backup_check_version(version):
642 """
643 Commandline options: no options
644 """
645 try:
646 version_number = int(version)
647 supported_version = config_backup_version()
648 if version_number > supported_version:
649 utils.err(
650 f"Unsupported version of the backup, supported version is "
651 f"{supported_version}, backup version is {version_number}"
652 )
653 if version_number < supported_version:
654 warn(
655 f"Restoring from the backup version {version_number}, current "
656 f"supported version is {supported_version}"
657 )
658 except TypeError:
659 utils.err("Cannot determine version of the backup")
660
661
662 def config_backup_add_version_to_tarball(tarball, version=None):
663 """
664 Commandline options: no options
665 """
666 ver = version if version is not None else str(config_backup_version())
667 return utils.tar_add_file_data(tarball, ver.encode("utf-8"), "version.txt")
668
669
670 def config_backup_version():
671 """
672 Commandline options: no options
673 """
674 return 1
675
676
677 def config_checkpoint_list(lib, argv, modifiers):
678 """
679 Options: no options
680 """
681 del lib
682 modifiers.ensure_only_supported()
683 if argv:
684 raise CmdLineInputError()
685 try:
686 file_list = os.listdir(settings.cib_dir)
687 except OSError as e:
688 utils.err("unable to list checkpoints: %s" % e)
689 cib_list = []
690 cib_name_re = re.compile(r"^cib-(\d+)\.raw$")
691 for filename in file_list:
692 match = cib_name_re.match(filename)
693 if not match:
694 continue
695 file_path = os.path.join(settings.cib_dir, filename)
696 try:
697 if os.path.isfile(file_path):
698 cib_list.append(
699 (float(os.path.getmtime(file_path)), match.group(1))
700 )
701 except OSError:
702 pass
703 cib_list.sort()
704 if not cib_list:
705 print_to_stderr("No checkpoints available")
706 return
707 for cib_info in cib_list:
708 print(
709 "checkpoint %s: date %s"
710 % (cib_info[1], datetime.datetime.fromtimestamp(round(cib_info[0])))
711 )
712
713
714 def _checkpoint_to_lines(lib, checkpoint_number):
715 # backup current settings
716 orig_usefile = utils.usefile
717 orig_filename = utils.filename
718 orig_middleware = lib.middleware_factory
719 orig_env = lib.env
720 # configure old code to read the CIB from a file
721 utils.usefile = True
722 utils.filename = os.path.join(
723 settings.cib_dir, "cib-%s.raw" % checkpoint_number
724 )
725 # configure new code to read the CIB from a file
726 lib.middleware_factory = orig_middleware._replace(
727 cib=middleware.cib(utils.filename, utils.touch_cib_file)
728 )
729 lib.env = utils.get_cli_env()
730 # export the CIB to text
731 result = False, []
732 if os.path.isfile(utils.filename):
733 result = True, _config_show_cib_lines(lib)
734 # restore original settings
735 utils.usefile = orig_usefile
736 utils.filename = orig_filename
737 lib.middleware_factory = orig_middleware
738 lib.env = orig_env
739 return result
740
741
742 def config_checkpoint_view(lib, argv, modifiers):
743 """
744 Options: no options
745 """
746 modifiers.ensure_only_supported()
747 if len(argv) != 1:
748 print_to_stderr(usage.config(["checkpoint view"]))
749 sys.exit(1)
750
751 loaded, lines = _checkpoint_to_lines(lib, argv[0])
752 if not loaded:
753 utils.err("unable to read the checkpoint")
754 print("\n".join(lines))
755
756
757 def config_checkpoint_diff(lib, argv, modifiers):
758 """
759 Commandline options:
760 * -f - CIB file
761 """
762 modifiers.ensure_only_supported("-f")
763 if len(argv) != 2:
764 print_to_stderr(usage.config(["checkpoint diff"]))
765 sys.exit(1)
766
767 if argv[0] == argv[1]:
768 utils.err("cannot diff a checkpoint against itself")
769
770 errors = []
771 checkpoints_lines = []
772 for checkpoint in argv:
773 if checkpoint == "live":
774 lines = _config_show_cib_lines(lib)
775 if not lines:
776 errors.append("unable to read live configuration")
777 else:
778 checkpoints_lines.append(lines)
779 else:
780 loaded, lines = _checkpoint_to_lines(lib, checkpoint)
781 if not loaded:
782 errors.append(
783 "unable to read checkpoint '{0}'".format(checkpoint)
784 )
785 else:
786 checkpoints_lines.append(lines)
787
788 if errors:
789 utils.err("\n".join(errors))
790
791 print(
792 "Differences between {0} (-) and {1} (+):".format(
793 *[
794 (
795 "live configuration"
796 if label == "live"
797 else f"checkpoint {label}"
798 )
799 for label in argv
800 ]
801 )
802 )
803 print(
804 "\n".join(
805 [
806 line.rstrip()
807 for line in difflib.Differ().compare(
808 checkpoints_lines[0], checkpoints_lines[1]
809 )
810 ]
811 )
812 )
813
814
815 def config_checkpoint_restore(lib, argv, modifiers):
816 """
817 Options:
818 * -f - CIB file, a checkpoint will be restored into a specified file
819 """
820 del lib
821 modifiers.ensure_only_supported("-f")
822 if len(argv) != 1:
823 print_to_stderr(usage.config(["checkpoint restore"]))
824 sys.exit(1)
825
826 cib_path = os.path.join(settings.cib_dir, "cib-%s.raw" % argv[0])
827 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. |
828 snapshot_dom = parse(cib_path)
829 except Exception as e:
830 utils.err("unable to read the checkpoint: %s" % e)
831 utils.replace_cib_configuration(snapshot_dom)
832