reloader: chmod target dir idempotently

In d57b7bc we introduced a solution to stale watchers, but this
introduced a new observer restart loop caused by the chmod always being
executed.

By using an idempotent chmod, changes are only made when necessary.

Author: tdgroot

nginx_config_reloader/__init__.py

@@ -37,7 +37,7 @@
     UNPRIVILEGED_UID,
     WATCH_IGNORE_FILES,
 )
-from nginx_config_reloader.utils import directory_is_unmounted
+from nginx_config_reloader.utils import apply_chmod, directory_is_unmounted
 
 logger = logging.getLogger(__name__)
 dbus_loop: EventLoop | None = None
@@ -269,19 +269,13 @@ def _apply(self):
 
     def fix_custom_config_dir_permissions(self):
         try:
-            subprocess.check_output(
-                ["chmod", "755", self.dir_to_watch],
-                preexec_fn=as_unprivileged_user,
-            )
+            apply_chmod(self.dir_to_watch, "755", preexec_fn=as_unprivileged_user)
             for root, dirs, _ in os.walk(self.dir_to_watch):
                 for name in dirs:
                     path = os.path.join(root, name)
                     if os.path.islink(path):
                         continue
-                    subprocess.check_output(
-                        ["chmod", "755", path],
-                        preexec_fn=as_unprivileged_user,
-                    )
+                    apply_chmod(path, "755", preexec_fn=as_unprivileged_user)
         except subprocess.CalledProcessError:
             self.logger.info("Failed fixing permissions on watched directory")
 

nginx_config_reloader/utils.py

@@ -1,7 +1,28 @@
 import json
+import os
 import subprocess
 
 
+def apply_chmod(path, mode, preexec_fn=None):
+    if isinstance(mode, int):
+        chmod_mode = oct(mode)[2:]
+        stat_mode = mode
+    else:
+        chmod_mode = str(mode)
+        stat_mode = int(chmod_mode, 8)
+
+    try:
+        if os.stat(path).st_mode & 0o777 == stat_mode:
+            return
+    except OSError:
+        pass
+
+    subprocess.check_call(
+        ["chmod", chmod_mode, path],
+        preexec_fn=preexec_fn,
+    )
+
+
 def directory_is_unmounted(path):
     output = subprocess.check_output(
         ["systemctl", "list-units", "-t", "mount", "--all", "-o", "json"],

tests/test_apply_chmod.py

@@ -0,0 +1,43 @@
+import os
+import tempfile
+
+from nginx_config_reloader import as_unprivileged_user
+from nginx_config_reloader.utils import apply_chmod
+from tests.testcase import TestCase
+
+
+class TestApplyChmod(TestCase):
+    def setUp(self):
+        self.check_call = self.set_up_patch("subprocess.check_call")
+        _, self.path = tempfile.mkstemp()
+
+    def tearDown(self):
+        try:
+            os.unlink(self.path)
+        except OSError:
+            pass
+
+    def test_apply_chmod_changes_mode(self):
+        os.chmod(self.path, 0o600)
+
+        apply_chmod(self.path, "644", preexec_fn=as_unprivileged_user)
+
+        self.check_call.assert_called_once_with(
+            ["chmod", "644", self.path], preexec_fn=as_unprivileged_user
+        )
+
+    def test_apply_chmod_skips_existing_mode(self):
+        os.chmod(self.path, 0o644)
+
+        apply_chmod(self.path, "644", preexec_fn=as_unprivileged_user)
+
+        self.check_call.assert_not_called()
+
+    def test_apply_chmod_accepts_int_mode(self):
+        os.chmod(self.path, 0o600)
+
+        apply_chmod(self.path, 0o644, preexec_fn=as_unprivileged_user)
+
+        self.check_call.assert_called_once_with(
+            ["chmod", "644", self.path], preexec_fn=as_unprivileged_user
+        )

tests/test_fix_custom_config_dir_permissions.py

@@ -8,7 +8,7 @@
 
 class TestFixCustomConfigDirPermissions(TestCase):
     def setUp(self):
-        self.check_output = self.set_up_patch("subprocess.check_output")
+        self.check_call = self.set_up_patch("subprocess.check_call")
         self.temp_dir = tempfile.mkdtemp()
         self.tm = NginxConfigReloader(
             no_magento_config=False,
@@ -17,12 +17,14 @@ def setUp(self):
             magento2_flag=None,
         )
 
-    def test_fix_custom_config_dir_permissions_chmods_all_dirs_to_755(self):
+    def test_fix_custom_config_dir_permissions_chmods_dirs_to_755(self):
+        os.chmod(self.temp_dir, 0o700)
         os.mkdir(self.temp_dir + "/some_dir")
+        os.chmod(self.temp_dir + "/some_dir", 0o700)
 
         self.tm.fix_custom_config_dir_permissions()
 
-        self.check_output.assert_has_calls(
+        self.check_call.assert_has_calls(
             [
                 call(["chmod", "755", self.temp_dir], preexec_fn=as_unprivileged_user),
                 call(
@@ -32,12 +34,22 @@ def test_fix_custom_config_dir_permissions_chmods_all_dirs_to_755(self):
             ]
         )
 
+    def test_fix_custom_config_dir_permissions_skips_dirs_that_are_already_755(self):
+        os.chmod(self.temp_dir, 0o755)
+        os.mkdir(self.temp_dir + "/some_dir")
+        os.chmod(self.temp_dir + "/some_dir", 0o755)
+
+        self.tm.fix_custom_config_dir_permissions()
+
+        self.check_call.assert_not_called()
+
     def test_fix_custom_config_dir_permissions_ignores_symlinks(self):
         other_temp_dir = tempfile.mkdtemp()
         os.symlink(other_temp_dir, self.temp_dir + "/some_pointing_dir")
+        os.chmod(self.temp_dir, 0o700)
 
         self.tm.fix_custom_config_dir_permissions()
 
-        self.check_output.assert_called_once_with(
+        self.check_call.assert_called_once_with(
             ["chmod", "755", self.temp_dir], preexec_fn=as_unprivileged_user
         )

tests/test_symlink_targets.py

@@ -100,6 +100,15 @@ def test_changed_is_true_when_target_inode_is_replaced(self):
 
         self.assertTrue(handler.symlink_targets_changed())
 
+    def test_changed_is_true_when_target_metadata_changes(self):
+        self._symlink("example.com", self.target_a)
+        handler = self._handler()
+        handler.watched_symlink_targets = handler.get_symlink_targets()
+
+        os.chmod(self.target_a, 0o700)
+
+        self.assertTrue(handler.symlink_targets_changed())
+
     def test_changed_is_true_when_a_new_symlink_appears(self):
         handler = self._handler()
         handler.watched_symlink_targets = handler.get_symlink_targets()