Merge pull request #1875 from letsencrypt/webroot-permissions

Webroot permissions

letsencrypt/plugins/webroot.py

@@ -2,7 +2,6 @@
 import errno
 import logging
 import os
-import stat
 
 import zope.interface
 
@@ -59,24 +58,38 @@ def prepare(self):  # pylint: disable=missing-docstring
 
             logger.debug("Creating root challenges validation dir at %s",
                          self.full_roots[name])
+
+            # Change the permissions to be writable (GH #1389)
+            # Umask is used instead of chmod to ensure the client can also
+            # run as non-root (GH #1795)
+            old_umask = os.umask(0o022)
+
             try:
-                os.makedirs(self.full_roots[name])
-                # Set permissions as parent directory (GH #1389)
-                # We don't use the parameters in makedirs because it
-                # may not always work
+                # This is coupled with the "umask" call above because
+                # os.makedirs's "mode" parameter may not always work:
                 # https://stackoverflow.com/questions/5231901/permission-problems-when-creating-a-dir-with-os-makedirs-python
-                stat_path = os.stat(path)
-                filemode = stat.S_IMODE(stat_path.st_mode)
-                os.chmod(self.full_roots[name], filemode)
-                # Set owner and group, too
-                os.chown(self.full_roots[name], stat_path.st_uid,
-                          stat_path.st_gid)
+                os.makedirs(self.full_roots[name], 0o0755)
+
+                # Set owner as parent directory if possible
+                try:
+                    stat_path = os.stat(path)
+                    os.chown(self.full_roots[name], stat_path.st_uid,
+                             stat_path.st_gid)
+                except OSError as exception:
+                    if exception.errno == errno.EACCES:
+                        logger.debug("Insufficient permissions to change owner and uid - ignoring")
+                    else:
+                        raise errors.PluginError(
+                            "Couldn't create root for {0} http-01 "
+                            "challenge responses: {1}", name, exception)
 
             except OSError as exception:
                 if exception.errno != errno.EEXIST:
                     raise errors.PluginError(
                         "Couldn't create root for {0} http-01 "
                         "challenge responses: {1}", name, exception)
+            finally:
+                os.umask(old_umask)
 
     def perform(self, achalls):  # pylint: disable=missing-docstring
         assert self.full_roots, "Webroot plugin appears to be missing webroot map"
@@ -87,26 +100,26 @@ def _path_for_achall(self, achall):
             path = self.full_roots[achall.domain]
         except IndexError:
             raise errors.PluginError("Missing --webroot-path for domain: {1}"
-                        .format(achall.domain))
+                                     .format(achall.domain))
         if not os.path.exists(path):
             raise errors.PluginError("Mysteriously missing path {0} for domain: {1}"
-                        .format(path, achall.domain))
+                                     .format(path, achall.domain))
         return os.path.join(path, achall.chall.encode("token"))
 
     def _perform_single(self, achall):
         response, validation = achall.response_and_validation()
+
         path = self._path_for_achall(achall)
         logger.debug("Attempting to save validation to %s", path)
-        with open(path, "w") as validation_file:
-            validation_file.write(validation.encode())
-
-        # Set permissions as parent directory (GH #1389)
-        parent_path = self.full_roots[achall.domain]
-        stat_parent_path = os.stat(parent_path)
-        filemode = stat.S_IMODE(stat_parent_path.st_mode)
-        # Remove execution bit (not needed for this file)
-        os.chmod(path, filemode & ~stat.S_IEXEC)
-        os.chown(path, stat_parent_path.st_uid, stat_parent_path.st_gid)
+
+        # Change permissions to be world-readable, owner-writable (GH #1795)
+        old_umask = os.umask(0o022)
+
+        try:
+            with open(path, "w") as validation_file:
+                validation_file.write(validation.encode())
+        finally:
+            os.umask(old_umask)
 
         return response
 

letsencrypt/plugins/webroot_test.py

@@ -1,9 +1,10 @@
 """Tests for letsencrypt.plugins.webroot."""
+import errno
 import os
 import shutil
+import stat
 import tempfile
 import unittest
-import stat
 
 import mock
 
@@ -35,7 +36,6 @@ def setUp(self):
         self.config = mock.MagicMock(webroot_path=self.path,
                                      webroot_map={"thing.com": self.path})
         self.auth = Authenticator(self.config, "webroot")
-        self.auth.prepare()
 
     def tearDown(self):
         shutil.rmtree(self.path)
@@ -48,7 +48,7 @@ def test_more_info(self):
     def test_add_parser_arguments(self):
         add = mock.MagicMock()
         self.auth.add_parser_arguments(add)
-        self.assertEqual(0, add.call_count) # became 0 when we moved the args to cli.py!
+        self.assertEqual(0, add.call_count)  # args moved to cli.py!
 
     def test_prepare_bad_root(self):
         self.config.webroot_path = os.path.join(self.path, "null")
@@ -70,24 +70,41 @@ def test_prepare_reraises_other_errors(self):
         self.assertRaises(errors.PluginError, self.auth.prepare)
         os.chmod(self.path, 0o700)
 
+    @mock.patch("letsencrypt.plugins.webroot.os.chown")
+    def test_failed_chown_eacces(self, mock_chown):
+        mock_chown.side_effect = OSError(errno.EACCES, "msg")
+        self.auth.prepare()  # exception caught and logged
+
+    @mock.patch("letsencrypt.plugins.webroot.os.chown")
+    def test_failed_chown_not_eacces(self, mock_chown):
+        mock_chown.side_effect = OSError()
+        self.assertRaises(errors.PluginError, self.auth.prepare)
+
     def test_prepare_permissions(self):
+        self.auth.prepare()
 
         # Remove exec bit from permission check, so that it
         # matches the file
         self.auth.perform([self.achall])
-        parent_permissions = (stat.S_IMODE(os.stat(self.path).st_mode) &
-                              ~stat.S_IEXEC)
+        path_permissions = stat.S_IMODE(os.stat(self.validation_path).st_mode)
+        self.assertEqual(path_permissions, 0o644)
 
-        actual_permissions = stat.S_IMODE(os.stat(self.validation_path).st_mode)
+        # Check permissions of the directories
+
+        for dirpath, dirnames, _ in os.walk(self.path):
+            for directory in dirnames:
+                full_path = os.path.join(dirpath, directory)
+                dir_permissions = stat.S_IMODE(os.stat(full_path).st_mode)
+                self.assertEqual(dir_permissions, 0o755)
 
-        self.assertEqual(parent_permissions, actual_permissions)
         parent_gid = os.stat(self.path).st_gid
         parent_uid = os.stat(self.path).st_uid
 
         self.assertEqual(os.stat(self.validation_path).st_gid, parent_gid)
         self.assertEqual(os.stat(self.validation_path).st_uid, parent_uid)
 
     def test_perform_cleanup(self):
+        self.auth.prepare()
         responses = self.auth.perform([self.achall])
         self.assertEqual(1, len(responses))
         self.assertTrue(os.path.exists(self.validation_path))