mirror of
https://github.com/certbot/certbot.git
synced 2026-08-01 19:02:52 +02:00
[Windows] Security model for files permissions - STEP 3b (#6965)
* Modifications for misc * Add some types * Use os_rename * Move rename into filesystem * Use our os package * Rename filesystem.rename to filesystem.replace * Disable globally function redefined lint in os module
This commit is contained in:
committed by
Brad Warren
parent
72e5d89e95
commit
d75908c645
@@ -0,0 +1,20 @@
|
|||||||
|
"""Compat module to handle files security on Windows and Linux"""
|
||||||
|
from __future__ import absolute_import
|
||||||
|
|
||||||
|
import os # pylint: disable=os-module-forbidden
|
||||||
|
|
||||||
|
|
||||||
|
def replace(src, dst):
|
||||||
|
# type: (str, str) -> None
|
||||||
|
"""
|
||||||
|
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.
|
||||||
|
"""
|
||||||
|
if hasattr(os, 'replace'):
|
||||||
|
# Use replace if possible. On Windows, only Python >= 3.4 is supported
|
||||||
|
# so we can assume that os.replace() is always available for this platform.
|
||||||
|
getattr(os, 'replace')(src, dst)
|
||||||
|
else:
|
||||||
|
# Otherwise, use os.rename() that behaves like os.replace() on Linux.
|
||||||
|
os.rename(src, dst)
|
||||||
+15
-48
@@ -2,42 +2,31 @@
|
|||||||
This compat module handles various platform specific calls that do not fall into one
|
This compat module handles various platform specific calls that do not fall into one
|
||||||
particular category.
|
particular category.
|
||||||
"""
|
"""
|
||||||
import ctypes
|
from __future__ import absolute_import
|
||||||
import errno
|
|
||||||
import select
|
import select
|
||||||
import stat
|
import stat
|
||||||
import sys
|
import sys
|
||||||
|
|
||||||
|
try:
|
||||||
|
from win32com.shell import shell as shellwin32 # pylint: disable=import-error
|
||||||
|
POSIX_MODE = False
|
||||||
|
except ImportError: # pragma: no cover
|
||||||
|
POSIX_MODE = True
|
||||||
|
|
||||||
from certbot import errors
|
from certbot import errors
|
||||||
from certbot.compat import os
|
from certbot.compat import os
|
||||||
|
|
||||||
UNPRIVILEGED_SUBCOMMANDS_ALLOWED = [
|
def raise_for_non_administrative_windows_rights():
|
||||||
'certificates', 'enhance', 'revoke', 'delete',
|
# type: () -> None
|
||||||
'register', 'unregister', 'config_changes', 'plugins']
|
|
||||||
|
|
||||||
|
|
||||||
def raise_for_non_administrative_windows_rights(subcommand):
|
|
||||||
"""
|
"""
|
||||||
On Windows, raise if current shell does not have the administrative rights.
|
On Windows, raise if current shell does not have the administrative rights.
|
||||||
Do nothing on Linux.
|
Do nothing on Linux.
|
||||||
|
|
||||||
:param str subcommand: The subcommand (like 'certonly') passed to the certbot client.
|
:raises .errors.Error: If the current shell does not have administrative rights on Windows.
|
||||||
|
|
||||||
:raises .errors.Error: If the provided subcommand must be run on a shell with
|
|
||||||
administrative rights, and current shell does not have these rights.
|
|
||||||
|
|
||||||
"""
|
"""
|
||||||
# Why not simply try ctypes.windll.shell32.IsUserAnAdmin() and catch AttributeError ?
|
if not POSIX_MODE and shellwin32.IsUserAnAdmin() == 0: # pragma: no cover
|
||||||
# Because windll exists only on a Windows runtime, and static code analysis engines
|
raise errors.Error('Error, certbot must be run on a shell with administrative rights.')
|
||||||
# do not like at all non existent objects when run from Linux (even if we handle properly
|
|
||||||
# all the cases in the code).
|
|
||||||
# So we access windll only by reflection to trick these engines.
|
|
||||||
if hasattr(ctypes, 'windll') and subcommand not in UNPRIVILEGED_SUBCOMMANDS_ALLOWED:
|
|
||||||
windll = getattr(ctypes, 'windll')
|
|
||||||
if windll.shell32.IsUserAnAdmin() == 0:
|
|
||||||
raise errors.Error(
|
|
||||||
'Error, "{0}" subcommand must be run on a shell with administrative rights.'
|
|
||||||
.format(subcommand))
|
|
||||||
|
|
||||||
|
|
||||||
def os_geteuid():
|
def os_geteuid():
|
||||||
@@ -56,31 +45,8 @@ def os_geteuid():
|
|||||||
return 0
|
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,
|
|
||||||
# or 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):
|
def readline_with_timeout(timeout, prompt):
|
||||||
|
# type: (float, str) -> str
|
||||||
"""
|
"""
|
||||||
Read user input to return the first line entered, or raise after specified timeout.
|
Read user input to return the first line entered, or raise after specified timeout.
|
||||||
|
|
||||||
@@ -132,6 +98,7 @@ LINUX_DEFAULT_FOLDERS = {
|
|||||||
|
|
||||||
|
|
||||||
def get_default_folder(folder_type):
|
def get_default_folder(folder_type):
|
||||||
|
# type: (str) -> str
|
||||||
"""
|
"""
|
||||||
Return the relevant default folder for the current OS
|
Return the relevant default folder for the current OS
|
||||||
|
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ This compat modules is a wrapper of the core os module that forbids usage of spe
|
|||||||
(e.g. chown, chmod, getuid) that would be harmful to the Windows file security model of Certbot.
|
(e.g. chown, chmod, getuid) that would be harmful to the Windows file security model of Certbot.
|
||||||
This module is intended to replace standard os module throughout certbot projects (except acme).
|
This module is intended to replace standard os module throughout certbot projects (except acme).
|
||||||
"""
|
"""
|
||||||
|
# pylint: disable=function-redefined
|
||||||
from __future__ import absolute_import
|
from __future__ import absolute_import
|
||||||
|
|
||||||
# First round of wrapping: we import statically all public attributes exposed by the os module
|
# First round of wrapping: we import statically all public attributes exposed by the os module
|
||||||
@@ -29,3 +30,19 @@ std_sys.modules[__name__ + '.path'] = path
|
|||||||
|
|
||||||
# Clean all remaining importables that are not from the core os module.
|
# Clean all remaining importables that are not from the core os module.
|
||||||
del ourselves, std_os, std_sys
|
del ourselves, std_os, std_sys
|
||||||
|
|
||||||
|
|
||||||
|
# Because of the blocking strategy on file handlers on Windows, rename does not behave as expected
|
||||||
|
# with POSIX systems: an exception will be raised if dst already exists.
|
||||||
|
def rename(*unused_args, **unused_kwargs):
|
||||||
|
"""Method os.rename() is forbidden"""
|
||||||
|
raise RuntimeError('Usage of os.rename() is forbidden. '
|
||||||
|
'Use certbot.compat.filesystem.replace() instead.')
|
||||||
|
|
||||||
|
|
||||||
|
# Behavior of os.replace is consistent between Windows and Linux. However, it is not supported on
|
||||||
|
# Python 2.x. So, as for os.rename, we forbid it in favor of filesystem.replace.
|
||||||
|
def replace(*unused_args, **unused_kwargs):
|
||||||
|
"""Method os.replace() is forbidden"""
|
||||||
|
raise RuntimeError('Usage of os.replace() is forbidden. '
|
||||||
|
'Use certbot.compat.filesystem.replace() instead.')
|
||||||
|
|||||||
+1
-1
@@ -1359,7 +1359,7 @@ def main(cli_args=None):
|
|||||||
|
|
||||||
# On windows, shell without administrative right cannot create symlinks required by certbot.
|
# On windows, shell without administrative right cannot create symlinks required by certbot.
|
||||||
# So we check the rights before continuing.
|
# So we check the rights before continuing.
|
||||||
misc.raise_for_non_administrative_windows_rights(config.verb)
|
misc.raise_for_non_administrative_windows_rights()
|
||||||
|
|
||||||
try:
|
try:
|
||||||
log.post_arg_parse_setup(config)
|
log.post_arg_parse_setup(config)
|
||||||
|
|||||||
+2
-1
@@ -16,6 +16,7 @@ from certbot import interfaces
|
|||||||
from certbot import util
|
from certbot import util
|
||||||
from certbot.compat import misc
|
from certbot.compat import misc
|
||||||
from certbot.compat import os
|
from certbot.compat import os
|
||||||
|
from certbot.compat import filesystem
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
@@ -583,7 +584,7 @@ class Reverter(object):
|
|||||||
timestamp = self._checkpoint_timestamp()
|
timestamp = self._checkpoint_timestamp()
|
||||||
final_dir = os.path.join(self.config.backup_dir, timestamp)
|
final_dir = os.path.join(self.config.backup_dir, timestamp)
|
||||||
try:
|
try:
|
||||||
misc.os_rename(self.config.in_progress_dir, final_dir)
|
filesystem.replace(self.config.in_progress_dir, final_dir)
|
||||||
return
|
return
|
||||||
except OSError:
|
except OSError:
|
||||||
logger.warning("Extreme, unexpected race condition, retrying (%s)", timestamp)
|
logger.warning("Extreme, unexpected race condition, retrying (%s)", timestamp)
|
||||||
|
|||||||
+3
-3
@@ -18,8 +18,8 @@ from certbot import crypto_util
|
|||||||
from certbot import error_handler
|
from certbot import error_handler
|
||||||
from certbot import errors
|
from certbot import errors
|
||||||
from certbot import util
|
from certbot import util
|
||||||
from certbot.compat import misc
|
|
||||||
from certbot.compat import os
|
from certbot.compat import os
|
||||||
|
from certbot.compat import filesystem
|
||||||
from certbot.plugins import common as plugins_common
|
from certbot.plugins import common as plugins_common
|
||||||
from certbot.plugins import disco as plugins_disco
|
from certbot.plugins import disco as plugins_disco
|
||||||
|
|
||||||
@@ -162,7 +162,7 @@ def rename_renewal_config(prev_name, new_name, cli_config):
|
|||||||
raise errors.ConfigurationError("The new certificate name "
|
raise errors.ConfigurationError("The new certificate name "
|
||||||
"is already in use.")
|
"is already in use.")
|
||||||
try:
|
try:
|
||||||
os.rename(prev_filename, new_filename)
|
filesystem.replace(prev_filename, new_filename)
|
||||||
except OSError:
|
except OSError:
|
||||||
raise errors.ConfigurationError("Please specify a valid filename "
|
raise errors.ConfigurationError("Please specify a valid filename "
|
||||||
"for the new certificate name.")
|
"for the new certificate name.")
|
||||||
@@ -191,7 +191,7 @@ def update_configuration(lineagename, archive_dir, target, cli_config):
|
|||||||
# Save only the config items that are relevant to renewal
|
# Save only the config items that are relevant to renewal
|
||||||
values = relevant_values(vars(cli_config.namespace))
|
values = relevant_values(vars(cli_config.namespace))
|
||||||
write_renewal_config(config_filename, temp_filename, archive_dir, target, values)
|
write_renewal_config(config_filename, temp_filename, archive_dir, target, values)
|
||||||
misc.os_rename(temp_filename, config_filename)
|
filesystem.replace(temp_filename, config_filename)
|
||||||
|
|
||||||
return configobj.ConfigObj(config_filename)
|
return configobj.ConfigObj(config_filename)
|
||||||
|
|
||||||
|
|||||||
@@ -1,21 +0,0 @@
|
|||||||
"""Tests for certbot.compat."""
|
|
||||||
import certbot.tests.util as test_util
|
|
||||||
from certbot.compat import misc
|
|
||||||
from certbot.compat import os
|
|
||||||
|
|
||||||
|
|
||||||
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.
|
|
||||||
misc.os_rename(src, dst)
|
|
||||||
|
|
||||||
self.assertFalse(os.path.exists(src))
|
|
||||||
self.assertTrue(os.path.exists(dst))
|
|
||||||
@@ -0,0 +1,21 @@
|
|||||||
|
"""Tests for certbot.compat.filesystem"""
|
||||||
|
import certbot.tests.util as test_util
|
||||||
|
from certbot.compat import os
|
||||||
|
from certbot.compat import filesystem
|
||||||
|
|
||||||
|
|
||||||
|
class OsReplaceTest(test_util.TempDirTestCase):
|
||||||
|
"""Test to ensure consistent behavior of rename method"""
|
||||||
|
|
||||||
|
def test_os_replace_to_existing_file(self):
|
||||||
|
"""Ensure that replace 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 would fail because dst already exists.
|
||||||
|
filesystem.replace(src, dst)
|
||||||
|
|
||||||
|
self.assertFalse(os.path.exists(src))
|
||||||
|
self.assertTrue(os.path.exists(dst))
|
||||||
@@ -347,11 +347,11 @@ class TestFullCheckpointsReverter(test_util.ConfigTestCase):
|
|||||||
self.assertRaises(
|
self.assertRaises(
|
||||||
errors.ReverterError, self.reverter.finalize_checkpoint, "Title")
|
errors.ReverterError, self.reverter.finalize_checkpoint, "Title")
|
||||||
|
|
||||||
@mock.patch("certbot.reverter.misc.os_rename")
|
@mock.patch("certbot.reverter.filesystem.replace")
|
||||||
def test_finalize_checkpoint_no_rename_directory(self, mock_rename):
|
def test_finalize_checkpoint_no_rename_directory(self, mock_replace):
|
||||||
|
|
||||||
self.reverter.add_to_checkpoint(self.sets[0], "perm save")
|
self.reverter.add_to_checkpoint(self.sets[0], "perm save")
|
||||||
mock_rename.side_effect = OSError
|
mock_replace.side_effect = OSError
|
||||||
|
|
||||||
self.assertRaises(
|
self.assertRaises(
|
||||||
errors.ReverterError, self.reverter.finalize_checkpoint, "Title")
|
errors.ReverterError, self.reverter.finalize_checkpoint, "Title")
|
||||||
|
|||||||
Reference in New Issue
Block a user