add replace_acl method to PosixOperations

lib/vsc/filesystem/posix.py

@@ -22,6 +22,7 @@
 import errno
 import os
 import stat
+import tempfile
 
 from vsc.utils import fancylogger
 from vsc.utils.patterns import Singleton
@@ -558,3 +559,66 @@ def create_stat_directory(self, path, permissions, uid, gid, override_permission
             self.log.debug("Path %s already exists with correct ownership", path)
 
         return created
+
+    def replace_acl(self, path, permissions):
+        """
+        Overwrite ACL permissions on given path
+        """
+        path = self._sanity_check(path)
+        fs = self._what_filesystem(path)
+        if fs is None:
+            self.log.warning(f"ACL replacement requested on  filesystem with unknown support: {path}")
+            return False
+
+        if not isinstance(permissions, (tuple, list,)):
+            raise PosixOperationError("Given ACL permissions are not a list or tuple")
+
+        replace_acl_method = self._replace_acl_posix
+        if fs[0] in ['gpfs', 'nfs4']:
+            replace_acl_method = self._replace_acl_nfs
+
+        return replace_acl_method(path, permissions)
+
+    def _replace_acl_posix(self, path, permissions):
+        """
+        Overwrite ACL permissions in POSIX format on given path
+        """
+        setfacl_exe = 'setfacl'
+
+        with tempfile.NamedTemporaryFile() as acl_file:
+            self.log.debug(f"Input file for {setfacl_exe} created in: {acl_file.name}")
+            acl_file.write('\n'.join(permissions).encode('utf-8'))
+
+            setfacl_cmd = [
+                setfacl_exe,
+                f'--set-file={acl_file.name}',
+                path
+            ]
+
+            try:
+                ec, out = self._execute(setfacl_cmd)
+            except FileNotFoundError as err:
+                raise PosixOperationError(f"Cannot set POSIX ACLs, command {setfacl_exe} not found") from err
+            else:
+                return ec, out
+
+    def _replace_acl_nfs(self, path, permissions):
+        """
+        Overwrite ACL permissions in NFSv4 format on given path
+        """
+        acl_entries = ','.join(permissions)
+
+        setfacl_exe = 'nfs4_setfacl'
+        setfacl_cmd = [
+            'nfs4_setfacl',
+            '-s',
+            f'"{acl_entries}"',
+            path
+        ]
+
+        try:
+            ec, _ = self._execute(setfacl_cmd)
+        except FileNotFoundError as err:
+            raise PosixOperationError(f"Cannot set NFSv4 ACLs, command {setfacl_exe} not found") from err
+        else:
+            return ec

setup.py

@@ -37,7 +37,7 @@
 
 
 PACKAGE = {
-    "version": "2.3.2",
+    "version": "2.4.0",
     "author": [sdw, ag, kh, kw],
     "maintainer": [sdw, ag, kh, kw, wdp],
     "setup_requires": ["vsc-install >= 0.15.2"],

test/posix.py

@@ -240,3 +240,42 @@ def test_create_stat_dir_existing_not_dir(self, mock_chmod, mock_chown, mock_mak
 
         mock_os_stat.assert_called_with(test_path)
         mock_makedirs.assert_not_called
+
+    @mock.patch('vsc.filesystem.posix.PosixOperations._execute')
+    @mock.patch('vsc.filesystem.posix.PosixOperations._what_filesystem')
+    @mock.patch('vsc.filesystem.posix.tempfile.NamedTemporaryFile')
+    def test_replace_acl(self, mock_tempfile, mock_what_filesystem, mock_execute):
+        """
+        Test replacement of ACLs
+        """
+        mock_aclfile = mock.MagicMock()
+        mock_aclfile.name = "/tmp/acl_input_file"
+        mock_context = mock.MagicMock()
+        mock_context.__enter__.return_value = mock_aclfile
+        mock_tempfile.return_value = mock_context
+        mock_execute.return_value = (0, "")
+        test_path = '/tmp/test'
+
+        # POSIX ACLs
+        mock_what_filesystem.return_value = ['posix', '/data', 0, '127.0.0.1@tcp']
+
+        test_acl_posix = "user::rwx"
+        self.assertRaises(PosixOperationError, self.po.replace_acl, test_path, test_acl_posix)
+
+        test_acl_posix = ["user::rwx", "group::r-x", "other::r-x"]
+        self.po.replace_acl(test_path, test_acl_posix)
+        mock_execute.assert_called_with(
+            ['setfacl', '--set-file=/tmp/acl_input_file', '/tmp/test']
+        )
+
+        # NFSv4 ACLs
+        mock_what_filesystem.return_value = ['nfs4', '/data', 0, '127.0.0.1@tcp']
+
+        test_acl_posix = "A:d:OWNER@:rwaDdxtTnNcoy"
+        self.assertRaises(PosixOperationError, self.po.replace_acl, test_path, test_acl_posix)
+
+        test_acl_posix = ["A:d:OWNER@:rwaDdxtTnNcoy", "A:dg:GROUP@:rxtncy", "A:fd:EVERYONE@:tncy"]
+        self.po.replace_acl(test_path, test_acl_posix)
+        mock_execute.assert_called_with(
+            ['nfs4_setfacl', '-s', '"A:d:OWNER@:rwaDdxtTnNcoy,A:dg:GROUP@:rxtncy,A:fd:EVERYONE@:tncy"', '/tmp/test']
+        )