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