Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
cp_dir,
os.path.basename(path) + "_" + str(idx)), path)
except (IOError, OSError):
# This file is required in all checkpoints.
logger.error("Unable to recover files from %s", cp_dir)
raise errors.ReverterError(
"Unable to recover files from %s" % cp_dir)
# Remove any newly added files if they exist
self._remove_contained_files(os.path.join(cp_dir, "NEW_FILES"))
try:
shutil.rmtree(cp_dir)
except OSError:
logger.error("Unable to remove directory: %s", cp_dir)
raise errors.ReverterError(
"Unable to remove directory: %s" % cp_dir)
with open(file_list, "r") as list_fd:
filepaths = list_fd.read().splitlines()
for path in filepaths:
# Files are registered before they are added... so
# check to see if file exists first
if os.path.lexists(path):
os.remove(path)
else:
logger.warning(
"File: %s - Could not be found to be deleted\n"
" - Certbot probably shut down unexpectedly",
path)
except (IOError, OSError):
logger.critical(
"Unable to remove filepaths contained within %s", file_list)
raise errors.ReverterError(
"Unable to remove filepaths contained within "
"{0}".format(file_list))
return True
"""Reload users original configuration files after a temporary save.
This function should reinstall the users original configuration files
for all saves with temporary=True
:raises .ReverterError: when unable to revert config
"""
if os.path.isdir(self.config.temp_checkpoint_dir):
try:
self._recover_checkpoint(self.config.temp_checkpoint_dir)
except errors.ReverterError:
# We have a partial or incomplete recovery
logger.fatal("Incomplete or failed recovery for %s",
self.config.temp_checkpoint_dir)
raise errors.ReverterError("Unable to revert temporary config")
file will be cleaned up if the program exits unexpectedly.
(Before a save occurs)
:param bool temporary: If the file creation registry is for
a temp or permanent save.
:param \*files: file paths (str) to be registered
:raises certbot.errors.ReverterError: If
call does not contain necessary parameters or if the file creation
is unable to be registered.
"""
# Make sure some files are provided... as this is an error
# Made this mistake in my initial implementation of apache.dvsni.py
if not files:
raise errors.ReverterError("Forgot to provide files to registration call")
cp_dir = self._get_cp_dir(temporary)
# Append all new files (that aren't already registered)
new_fd = None
try:
new_fd, ex_files = self._read_and_append(os.path.join(cp_dir, "NEW_FILES"))
for path in files:
if path not in ex_files:
new_fd.write("{0}\n".format(path))
except (IOError, OSError):
logger.error("Unable to register file creation(s) - %s", files)
raise errors.ReverterError(
"Unable to register file creation(s) - {0}".format(files))
finally:
command_file = None
# It is strongly advised to set newline = '' on Python 3 with CSV,
# and it fixes problems on Windows.
kwargs = {'newline': ''} if sys.version_info[0] > 2 else {}
try:
if os.path.isfile(commands_fp):
command_file = open(commands_fp, "a", **kwargs) # type: ignore
else:
command_file = open(commands_fp, "w", **kwargs) # type: ignore
csvwriter = csv.writer(command_file)
csvwriter.writerow(command)
except (IOError, OSError):
logger.error("Unable to register undo command")
raise errors.ReverterError(
"Unable to register undo command.")
finally:
if command_file is not None:
command_file.close()
def rollback_checkpoints(self, rollback=1):
"""Revert 'rollback' number of configuration checkpoints.
:param int rollback: Number of checkpoints to reverse. A str num will be
cast to an integer. So "2" is also acceptable.
:raises .ReverterError:
if there is a problem with the input or if the function is
unable to correctly revert the configuration checkpoints
"""
try:
rollback = int(rollback)
except ValueError:
logger.error("Rollback argument must be a positive integer")
raise errors.ReverterError("Invalid Input")
# Sanity check input
if rollback < 0:
logger.error("Rollback argument must be a positive integer")
raise errors.ReverterError("Invalid Input")
backups = os.listdir(self.config.backup_dir)
backups.sort()
if not backups:
logger.warning(
"Certbot hasn't modified your configuration, so rollback "
"isn't available.")
elif len(backups) < rollback:
logger.warning("Unable to rollback %d checkpoints, only %d exist",
rollback, len(backups))
with open(changes_since_path, 'w') as f:
f.write("No changes\n")
# Add title to self.config.in_progress_dir CHANGES_SINCE
try:
with open(changes_since_tmp_path, "w") as changes_tmp:
changes_tmp.write("-- %s --\n" % title)
with open(changes_since_path, "r") as changes_orig:
changes_tmp.write(changes_orig.read())
# Move self.config.in_progress_dir to Backups directory
shutil.move(changes_since_tmp_path, changes_since_path)
except (IOError, OSError):
logger.error("Unable to finalize checkpoint - adding title")
logger.debug("Exception was:\n%s", traceback.format_exc())
raise errors.ReverterError("Unable to add title")
# rename the directory as a timestamp
self._timestamp_progress_dir()
This function should reinstall the users original configuration files
for all saves with temporary=True
:raises .ReverterError: when unable to revert config
"""
if os.path.isdir(self.config.temp_checkpoint_dir):
try:
self._recover_checkpoint(self.config.temp_checkpoint_dir)
except errors.ReverterError:
# We have a partial or incomplete recovery
logger.critical(
"Incomplete or failed recovery for %s",
self.config.temp_checkpoint_dir,
)
raise errors.ReverterError("Unable to revert temporary config")
self.reverter.add_to_temp_checkpoint(
save_files, self.save_notes)
else:
self.reverter.add_to_checkpoint(save_files,
self.save_notes)
except errors.ReverterError as err:
raise errors.PluginError(str(err))
self.aug.set("/augeas/save", save_state)
self.save_notes = ""
self.aug.save()
if title and not temporary:
try:
self.reverter.finalize_checkpoint(title)
except errors.ReverterError as err:
raise errors.PluginError(str(err))