fix(hooks): make compat for WSL filesystem better by explicity only changing permissions if they are lower at creation time.

vcsserver/hook_utils/__init__.py

@@ -1,208 +1,220 b''
 # RhodeCode VCSServer provides access to different vcs backends via network.
 # Copyright (C) 2014-2023 RhodeCode GmbH
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation; either version 3 of the License, or
 # (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU General Public License
 # along with this program; if not, write to the Free Software Foundation,
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
 
 import re
 import os
 import sys
 import datetime
 import logging
 import pkg_resources
 
 import vcsserver
 from vcsserver.str_utils import safe_bytes
 
 log = logging.getLogger(__name__)
 
+HOOKS_DIR_MODE = 0o755
+HOOKS_FILE_MODE = 0o755
+
+
+def set_permissions_if_needed(path_to_check, perms: oct):
+    # Get current permissions
+    current_permissions = os.stat(path_to_check).st_mode & 0o777  # Extract permission bits
+
+    # Check if current permissions are lower than required
+    if current_permissions < int(perms):
+        # Change the permissions if they are lower than required
+        os.chmod(path_to_check, perms)
+
 
 def get_git_hooks_path(repo_path, bare):
     hooks_path = os.path.join(repo_path, 'hooks')
     if not bare:
         hooks_path = os.path.join(repo_path, '.git', 'hooks')
 
     return hooks_path
 
 
 def install_git_hooks(repo_path, bare, executable=None, force_create=False):
     """
     Creates a RhodeCode hook inside a git repository
 
     :param repo_path: path to repository
     :param bare: defines if repository is considered a bare git repo
     :param executable: binary executable to put in the hooks
     :param force_create: Creates even if the same name hook exists
     """
     executable = executable or sys.executable
     hooks_path = get_git_hooks_path(repo_path, bare)
 
     # we always call it to ensure dir exists and it has a proper mode
     if not os.path.exists(hooks_path):
         # If it doesn't exist, create a new directory with the specified mode
-        os.makedirs(hooks_path, mode=0o777, exist_ok=True)
-    else:
-        # If it exists, change the directory's mode to the specified mode
-        os.chmod(hooks_path, mode=0o777)
+        os.makedirs(hooks_path, mode=HOOKS_DIR_MODE, exist_ok=True)
+    # If it exists, change the directory's mode to the specified mode
+    set_permissions_if_needed(hooks_path, perms=HOOKS_DIR_MODE)
 
     tmpl_post = pkg_resources.resource_string(
         'vcsserver', '/'.join(
             ('hook_utils', 'hook_templates', 'git_post_receive.py.tmpl')))
     tmpl_pre = pkg_resources.resource_string(
         'vcsserver', '/'.join(
             ('hook_utils', 'hook_templates', 'git_pre_receive.py.tmpl')))
 
     path = ''  # not used for now
     timestamp = datetime.datetime.utcnow().isoformat()
 
     for h_type, template in [('pre', tmpl_pre), ('post', tmpl_post)]:
         log.debug('Installing git hook in repo %s', repo_path)
         _hook_file = os.path.join(hooks_path, f'{h_type}-receive')
         _rhodecode_hook = check_rhodecode_hook(_hook_file)
 
         if _rhodecode_hook or force_create:
             log.debug('writing git %s hook file at %s !', h_type, _hook_file)
             try:
                 with open(_hook_file, 'wb') as f:
                     template = template.replace(b'_TMPL_', safe_bytes(vcsserver.__version__))
                     template = template.replace(b'_DATE_', safe_bytes(timestamp))
                     template = template.replace(b'_ENV_', safe_bytes(executable))
                     template = template.replace(b'_PATH_', safe_bytes(path))
                     f.write(template)
-                os.chmod(_hook_file, 0o755)
+                set_permissions_if_needed(_hook_file, perms=HOOKS_FILE_MODE)
             except OSError:
                 log.exception('error writing hook file %s', _hook_file)
         else:
             log.debug('skipping writing hook file')
 
     return True
 
 
 def get_svn_hooks_path(repo_path):
     hooks_path = os.path.join(repo_path, 'hooks')
 
     return hooks_path
 
 
 def install_svn_hooks(repo_path, executable=None, force_create=False):
     """
     Creates RhodeCode hooks inside a svn repository
 
     :param repo_path: path to repository
     :param executable: binary executable to put in the hooks
     :param force_create: Create even if same name hook exists
     """
     executable = executable or sys.executable
     hooks_path = get_svn_hooks_path(repo_path)
     if not os.path.isdir(hooks_path):
         os.makedirs(hooks_path, mode=0o777, exist_ok=True)
 
     tmpl_post = pkg_resources.resource_string(
         'vcsserver', '/'.join(
             ('hook_utils', 'hook_templates', 'svn_post_commit_hook.py.tmpl')))
     tmpl_pre = pkg_resources.resource_string(
         'vcsserver', '/'.join(
             ('hook_utils', 'hook_templates', 'svn_pre_commit_hook.py.tmpl')))
 
     path = ''  # not used for now
     timestamp = datetime.datetime.utcnow().isoformat()
 
     for h_type, template in [('pre', tmpl_pre), ('post', tmpl_post)]:
         log.debug('Installing svn hook in repo %s', repo_path)
         _hook_file = os.path.join(hooks_path, f'{h_type}-commit')
         _rhodecode_hook = check_rhodecode_hook(_hook_file)
 
         if _rhodecode_hook or force_create:
             log.debug('writing svn %s hook file at %s !', h_type, _hook_file)
 
             try:
                 with open(_hook_file, 'wb') as f:
                     template = template.replace(b'_TMPL_', safe_bytes(vcsserver.__version__))
                     template = template.replace(b'_DATE_', safe_bytes(timestamp))
                     template = template.replace(b'_ENV_', safe_bytes(executable))
                     template = template.replace(b'_PATH_', safe_bytes(path))
 
                     f.write(template)
                 os.chmod(_hook_file, 0o755)
             except OSError:
                 log.exception('error writing hook file %s', _hook_file)
         else:
             log.debug('skipping writing hook file')
 
     return True
 
 
 def get_version_from_hook(hook_path):
     version = b''
     hook_content = read_hook_content(hook_path)
     matches = re.search(rb'RC_HOOK_VER\s*=\s*(.*)', hook_content)
     if matches:
         try:
             version = matches.groups()[0]
             log.debug('got version %s from hooks.', version)
         except Exception:
             log.exception("Exception while reading the hook version.")
     return version.replace(b"'", b"")
 
 
 def check_rhodecode_hook(hook_path):
     """
     Check if the hook was created by RhodeCode
     """
     if not os.path.exists(hook_path):
         return True
 
     log.debug('hook exists, checking if it is from RhodeCode')
 
     version = get_version_from_hook(hook_path)
     if version:
         return True
 
     return False
 
 
 def read_hook_content(hook_path) -> bytes:
     content = b''
     if os.path.isfile(hook_path):
         with open(hook_path, 'rb') as f:
             content = f.read()
     return content
 
 
 def get_git_pre_hook_version(repo_path, bare):
     hooks_path = get_git_hooks_path(repo_path, bare)
     _hook_file = os.path.join(hooks_path, 'pre-receive')
     version = get_version_from_hook(_hook_file)
     return version
 
 
 def get_git_post_hook_version(repo_path, bare):
     hooks_path = get_git_hooks_path(repo_path, bare)
     _hook_file = os.path.join(hooks_path, 'post-receive')
     version = get_version_from_hook(_hook_file)
     return version
 
 
 def get_svn_pre_hook_version(repo_path):
     hooks_path = get_svn_hooks_path(repo_path)
     _hook_file = os.path.join(hooks_path, 'pre-commit')
     version = get_version_from_hook(_hook_file)
     return version
 
 
 def get_svn_post_hook_version(repo_path):
     hooks_path = get_svn_hooks_path(repo_path)
     _hook_file = os.path.join(hooks_path, 'post-commit')
     version = get_version_from_hook(_hook_file)
     return version

vcsserver/tests/test_install_hooks.py

@@ -1,206 +1,289 b''
 # RhodeCode VCSServer provides access to different vcs backends via network.
 # Copyright (C) 2014-2023 RhodeCode GmbH
 #
 # This program is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation; either version 3 of the License, or
 # (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU General Public License
 # along with this program; if not, write to the Free Software Foundation,
 # Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
 
 import os
 import sys
 import stat
 import pytest
 import vcsserver
 import tempfile
 from vcsserver import hook_utils
+from vcsserver.hook_utils import set_permissions_if_needed, HOOKS_DIR_MODE, HOOKS_FILE_MODE
 from vcsserver.tests.fixture import no_newline_id_generator
-from vcsserver.str_utils import safe_bytes, safe_str
+from vcsserver.str_utils import safe_bytes
 from vcsserver.utils import AttributeDict
 
 
 class TestCheckRhodecodeHook:
 
     def test_returns_false_when_hook_file_is_wrong_found(self, tmpdir):
         hook = os.path.join(str(tmpdir), 'fake_hook_file.py')
         with open(hook, 'wb') as f:
             f.write(b'dummy test')
             result = hook_utils.check_rhodecode_hook(hook)
             assert result is False
 
     def test_returns_true_when_no_hook_file_found(self, tmpdir):
         hook = os.path.join(str(tmpdir), 'fake_hook_file_not_existing.py')
         result = hook_utils.check_rhodecode_hook(hook)
         assert result
 
     @pytest.mark.parametrize("file_content, expected_result", [
         ("RC_HOOK_VER = '3.3.3'\n", True),
         ("RC_HOOK = '3.3.3'\n", False),
     ], ids=no_newline_id_generator)
     def test_signatures(self, file_content, expected_result, tmpdir):
         hook = os.path.join(str(tmpdir), 'fake_hook_file_1.py')
         with open(hook, 'wb') as f:
             f.write(safe_bytes(file_content))
 
         result = hook_utils.check_rhodecode_hook(hook)
 
         assert result is expected_result
 
 
 class BaseInstallHooks:
     HOOK_FILES = ()
 
+    def _check_hook_file_dir_mode(self, file_path):
+        dir_path = os.path.dirname(file_path)
+        assert os.path.exists(dir_path), f'dir {file_path} missing'
+        stat_info = os.stat(dir_path)
+
+        file_mode = stat.S_IMODE(stat_info.st_mode)
+        expected_mode = int(HOOKS_DIR_MODE)
+        assert expected_mode == file_mode, f'expected mode: {oct(expected_mode)} got: {oct(file_mode)} for {dir_path}'
+
     def _check_hook_file_mode(self, file_path):
         assert os.path.exists(file_path), f'path {file_path} missing'
         stat_info = os.stat(file_path)
 
         file_mode = stat.S_IMODE(stat_info.st_mode)
-        expected_mode = int('755', 8)
-        assert expected_mode == file_mode
+        expected_mode = int(HOOKS_FILE_MODE)
+        assert expected_mode == file_mode, f'expected mode: {oct(expected_mode)} got: {oct(file_mode)} for {file_path}'
 
     def _check_hook_file_content(self, file_path, executable):
         executable = executable or sys.executable
         with open(file_path, 'rt') as hook_file:
             content = hook_file.read()
 
         expected_env = '#!{}'.format(executable)
         expected_rc_version = "\nRC_HOOK_VER = '{}'\n".format(vcsserver.__version__)
         assert content.strip().startswith(expected_env)
         assert expected_rc_version in content
 
     def _create_fake_hook(self, file_path, content):
         with open(file_path, 'w') as hook_file:
             hook_file.write(content)
 
     def create_dummy_repo(self, repo_type):
         tmpdir = tempfile.mkdtemp()
         repo = AttributeDict()
         if repo_type == 'git':
             repo.path = os.path.join(tmpdir, 'test_git_hooks_installation_repo')
             os.makedirs(repo.path)
             os.makedirs(os.path.join(repo.path, 'hooks'))
             repo.bare = True
 
         elif repo_type == 'svn':
             repo.path = os.path.join(tmpdir, 'test_svn_hooks_installation_repo')
             os.makedirs(repo.path)
             os.makedirs(os.path.join(repo.path, 'hooks'))
 
         return repo
 
     def check_hooks(self, repo_path, repo_bare=True):
         for file_name in self.HOOK_FILES:
             if repo_bare:
                 file_path = os.path.join(repo_path, 'hooks', file_name)
             else:
                 file_path = os.path.join(repo_path, '.git', 'hooks', file_name)
+
+            self._check_hook_file_dir_mode(file_path)
             self._check_hook_file_mode(file_path)
             self._check_hook_file_content(file_path, sys.executable)
 
 
 class TestInstallGitHooks(BaseInstallHooks):
     HOOK_FILES = ('pre-receive', 'post-receive')
 
     def test_hooks_are_installed(self):
         repo = self.create_dummy_repo('git')
         result = hook_utils.install_git_hooks(repo.path, repo.bare)
         assert result
         self.check_hooks(repo.path, repo.bare)
 
     def test_hooks_are_replaced(self):
         repo = self.create_dummy_repo('git')
         hooks_path = os.path.join(repo.path, 'hooks')
         for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]:
             self._create_fake_hook(
                 file_path, content="RC_HOOK_VER = 'abcde'\n")
 
         result = hook_utils.install_git_hooks(repo.path, repo.bare)
         assert result
         self.check_hooks(repo.path, repo.bare)
 
     def test_non_rc_hooks_are_not_replaced(self):
         repo = self.create_dummy_repo('git')
         hooks_path = os.path.join(repo.path, 'hooks')
         non_rc_content = 'echo "non rc hook"\n'
         for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]:
             self._create_fake_hook(
                 file_path, content=non_rc_content)
 
         result = hook_utils.install_git_hooks(repo.path, repo.bare)
         assert result
 
         for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]:
             with open(file_path, 'rt') as hook_file:
                 content = hook_file.read()
             assert content == non_rc_content
 
     def test_non_rc_hooks_are_replaced_with_force_flag(self):
         repo = self.create_dummy_repo('git')
         hooks_path = os.path.join(repo.path, 'hooks')
         non_rc_content = 'echo "non rc hook"\n'
         for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]:
             self._create_fake_hook(
                 file_path, content=non_rc_content)
 
         result = hook_utils.install_git_hooks(
             repo.path, repo.bare, force_create=True)
         assert result
         self.check_hooks(repo.path, repo.bare)
 
 
 class TestInstallSvnHooks(BaseInstallHooks):
     HOOK_FILES = ('pre-commit', 'post-commit')
 
     def test_hooks_are_installed(self):
         repo = self.create_dummy_repo('svn')
         result = hook_utils.install_svn_hooks(repo.path)
         assert result
         self.check_hooks(repo.path)
 
     def test_hooks_are_replaced(self):
         repo = self.create_dummy_repo('svn')
         hooks_path = os.path.join(repo.path, 'hooks')
         for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]:
             self._create_fake_hook(
                 file_path, content="RC_HOOK_VER = 'abcde'\n")
 
         result = hook_utils.install_svn_hooks(repo.path)
         assert result
         self.check_hooks(repo.path)
 
     def test_non_rc_hooks_are_not_replaced(self):
         repo = self.create_dummy_repo('svn')
         hooks_path = os.path.join(repo.path, 'hooks')
         non_rc_content = 'echo "non rc hook"\n'
         for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]:
             self._create_fake_hook(
                 file_path, content=non_rc_content)
 
         result = hook_utils.install_svn_hooks(repo.path)
         assert result
 
         for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]:
             with open(file_path, 'rt') as hook_file:
                 content = hook_file.read()
             assert content == non_rc_content
 
     def test_non_rc_hooks_are_replaced_with_force_flag(self):
         repo = self.create_dummy_repo('svn')
         hooks_path = os.path.join(repo.path, 'hooks')
         non_rc_content = 'echo "non rc hook"\n'
         for file_path in [os.path.join(hooks_path, f) for f in self.HOOK_FILES]:
             self._create_fake_hook(
                 file_path, content=non_rc_content)
 
         result = hook_utils.install_svn_hooks(
             repo.path, force_create=True)
         assert result
         self.check_hooks(repo.path, )
+
+
+def create_test_file(filename):
+    """Utility function to create a test file."""
+    with open(filename, 'w') as f:
+        f.write("Test file")
+
+
+def remove_test_file(filename):
+    """Utility function to remove a test file."""
+    if os.path.exists(filename):
+        os.remove(filename)
+
+
[e-mail]
+def test_file():
+    filename = 'test_file.txt'
+    create_test_file(filename)
+    yield filename
+    remove_test_file(filename)
+
+
+def test_increase_permissions(test_file):
+    # Set initial lower permissions
+    initial_perms = 0o644
+    os.chmod(test_file, initial_perms)
+
+    # Set higher permissions
+    new_perms = 0o666
+    set_permissions_if_needed(test_file, new_perms)
+
+    # Check if permissions were updated
+    assert (os.stat(test_file).st_mode & 0o777) == new_perms
+
+
+def test_no_permission_change_needed(test_file):
+    # Set initial permissions
+    initial_perms = 0o666
+    os.chmod(test_file, initial_perms)
+
+    # Attempt to set the same permissions
+    set_permissions_if_needed(test_file, initial_perms)
+
+    # Check if permissions were unchanged
+    assert (os.stat(test_file).st_mode & 0o777) == initial_perms
+
+
+def test_no_permission_reduction(test_file):
+    # Set initial higher permissions
+    initial_perms = 0o666
+    os.chmod(test_file, initial_perms)
+
+    # Attempt to set lower permissions
+    lower_perms = 0o644
+    set_permissions_if_needed(test_file, lower_perms)
+
+    # Check if permissions were not reduced
+    assert (os.stat(test_file).st_mode & 0o777) == initial_perms
+
+
+def test_no_permission_reduction_when_on_777(test_file):
+    # Set initial higher permissions
+    initial_perms = 0o777
+    os.chmod(test_file, initial_perms)
+
+    # Attempt to set lower permissions
+    lower_perms = 0o755
+    set_permissions_if_needed(test_file, lower_perms)
+
+    # Check if permissions were not reduced
+    assert (os.stat(test_file).st_mode & 0o777) == initial_perms