1
0
mirror of https://github.com/certbot/certbot.git synced 2026-01-26 07:41:33 +03:00

[Windows] Handle file renaming when the destination path already exists (#6415)

On Linux, you can invoke os.rename(src, dst) even if dst already exists. In this case, destination file will be atomically replaced by the source file.

On Windows, this will lead to an OSError because changes are not atomic. This cause certbot renew to fail in particular, because the old certificate configuration needs to be replace by the new when a certificate is effectively renewed.

One could use the cross-platform function os.replace, but it is available only on Python >= 3.3.

This PR add a function in compat to handle correctly this case on Windows, and delegating everything else to os.rename.

* Cross platform compatible os.rename (we can use os.replace if its python 3)

* Use os.replace instead of custom non-atomic code.

* Avoid errors for lint and mypy. Add a test.
This commit is contained in:
Adrien Ferrand
2018-11-07 00:35:09 +01:00
committed by Brad Warren
parent 47062dbfbf
commit 4edfb3ef65
5 changed files with 49 additions and 3 deletions

View File

@@ -65,6 +65,30 @@ def os_geteuid():
# Windows specific
return 0
def os_rename(src, dst):
"""
Rename a file to a destination path and handles situations where the destination exists.
:param str src: The current file path.
:param str dst: The new file path.
"""
try:
os.rename(src, dst)
except OSError as err:
# Windows specific, renaming a file on an existing path is not possible.
# On Python 3, the best fallback with atomic capabilities we have is os.replace.
if err.errno != errno.EEXIST:
# Every other error is a legitimate exception.
raise
if not hasattr(os, 'replace'): # pragma: no cover
# We should never go on this line. Either we are on Linux and os.rename has succeeded,
# either we are on Windows, and only Python >= 3.4 is supported where os.replace is
# available.
raise RuntimeError('Error: tried to run os_rename on Python < 3.3. '
'Certbot supports only Python 3.4 >= on Windows.')
getattr(os, 'replace')(src, dst)
def readline_with_timeout(timeout, prompt):
"""
Read user input to return the first line entered, or raise after specified timeout.

View File

@@ -576,7 +576,7 @@ class Reverter(object):
timestamp = self._checkpoint_timestamp()
final_dir = os.path.join(self.config.backup_dir, timestamp)
try:
os.rename(self.config.in_progress_dir, final_dir)
compat.os_rename(self.config.in_progress_dir, final_dir)
return
except OSError:
logger.warning("Extreme, unexpected race condition, retrying (%s)", timestamp)

View File

@@ -14,6 +14,7 @@ import six
import certbot
from certbot import cli
from certbot import compat
from certbot import constants
from certbot import crypto_util
from certbot import errors
@@ -188,7 +189,7 @@ def update_configuration(lineagename, archive_dir, target, cli_config):
# Save only the config items that are relevant to renewal
values = relevant_values(vars(cli_config.namespace))
write_renewal_config(config_filename, temp_filename, archive_dir, target, values)
os.rename(temp_filename, config_filename)
compat.os_rename(temp_filename, config_filename)
return configobj.ConfigObj(config_filename)

View File

@@ -0,0 +1,21 @@
"""Tests for certbot.compat."""
import os
from certbot import compat
import certbot.tests.util as test_util
class OsReplaceTest(test_util.TempDirTestCase):
"""Test to ensure consistent behavior of os_rename method"""
def test_os_rename_to_existing_file(self):
"""Ensure that os_rename will effectively rename src into dst for all platforms."""
src = os.path.join(self.tempdir, 'src')
dst = os.path.join(self.tempdir, 'dst')
open(src, 'w').close()
open(dst, 'w').close()
# On Windows, a direct call to os.rename will fail because dst already exists.
compat.os_rename(src, dst)
self.assertFalse(os.path.exists(src))
self.assertTrue(os.path.exists(dst))

View File

@@ -356,7 +356,7 @@ class TestFullCheckpointsReverter(test_util.ConfigTestCase):
self.assertRaises(
errors.ReverterError, self.reverter.finalize_checkpoint, "Title")
@mock.patch("certbot.reverter.os.rename")
@mock.patch("certbot.reverter.compat.os_rename")
def test_finalize_checkpoint_no_rename_directory(self, mock_rename):
self.reverter.add_to_checkpoint(self.sets[0], "perm save")