Gateway Share - Add APIs for directly managing NFS Trusted Clients

Mimic the Share ACL functionality

Author: ygalblum

cterasdk/edge/shares.py

@@ -5,7 +5,7 @@
 from ..common import Object
 from ..exception import CTERAException, InputError
 from .base_command import BaseCommand
-from .types import ShareAccessControlEntry, RemoveShareAccessControlEntry
+from .types import NFSv3AccessControlEntry, RemoveNFSv3AccessControlEntry, ShareAccessControlEntry, RemoveShareAccessControlEntry
 
 
 class Shares(BaseCommand):
@@ -269,6 +269,73 @@ def delete(self, name):
             logging.getLogger().error("Share deletion failed.")
             raise CTERAException('Share deletion failed', error)
 
+    def get_trusted_nfs_clients(self, name):
+        """
+        Get the current trusted NFS client entries from an existing share.
+
+        :param str name: The share name
+        """
+        return self._gateway.get('/config/fileservices/share/' + name + '/trustedNFSClients')
+
+    def set_trusted_nfs_clients(self, name, trusted_nfs_clients):
+        """
+        Set a network share's trusted NFS client entries.
+
+        :param str name: The share name
+        :param list[cterasdk.edge.types.NFSv3AccessControlEntry] trusted_nfs_clients: Trusted NFS v3 clients
+
+        .. warning:: this method will override the existing access control entries
+        """
+        Shares._validate_trusted_nfs_clients(trusted_nfs_clients)
+
+        param = [client.to_server_object() for client in (trusted_nfs_clients or [])]
+        self._gateway.put('/config/fileservices/share/' + name + '/trustedNFSClients', param)
+
+    def add_trusted_nfs_clients(self, name, trusted_nfs_clients):
+        """
+        Add one or more trusted NFS client entries to an existing share.
+
+        :param str name: The share name
+        :param list[cterasdk.edge.types.NFSv3AccessControlEntry] trusted_nfs_clients: Trusted NFS v3 clients
+        """
+        Shares._validate_trusted_nfs_clients(trusted_nfs_clients)
+
+        new_trusted_nfs_clients_dict = {
+            trusted_nfs_client.address + '#' + trusted_nfs_client.netmask: trusted_nfs_client.to_server_object()
+            for trusted_nfs_client in trusted_nfs_clients
+        }
+
+        def entry_not_in_new(entry):
+            trusted_nfs_client = NFSv3AccessControlEntry.from_server_object(entry)
+            entry_key = trusted_nfs_client.address + '#' + trusted_nfs_client.netmask
+            return entry_key not in new_trusted_nfs_clients_dict
+
+        param = list(new_trusted_nfs_clients_dict.values()) + list(filter(entry_not_in_new, self.get_trusted_nfs_clients(name)))
+
+        self._gateway.put('/config/fileservices/share/' + name + '/trustedNFSClients', param)
+
+    def remove_trusted_nfs_clients(self, name, trusted_nfs_clients):
+        """
+        Remove one or more trusted NFS client entries from an existing share.
+
+        :param str name: The share name
+        :param list[cterasdk.edge.types.RemoveNFSv3AccessControlEntry] trusted_nfs_clients: Trusted NFS v3 clients
+        """
+        Shares._validate_remove_trusted_nfs_clients(trusted_nfs_clients)
+
+        remove_trusted_nfs_clients_dict = {
+            trusted_nfs_client.address + '#' + trusted_nfs_client.netmask
+            for trusted_nfs_client in trusted_nfs_clients
+        }
+
+        def entry_not_removed(entry):
+            trusted_nfs_client = NFSv3AccessControlEntry.from_server_object(entry)
+            entry_key = trusted_nfs_client.address + '#' + trusted_nfs_client.netmask
+            return entry_key not in remove_trusted_nfs_clients_dict
+
+        param = list(filter(entry_not_removed, self.get_trusted_nfs_clients(name)))
+        self._gateway.put('/config/fileservices/share/' + name + '/trustedNFSClients', param)
+
     def _validate_root_directory(self, name):
         param = Object()
         param.path = '/'
@@ -303,3 +370,31 @@ def _validate_remove_acl(acl):
                     repr(acl_entry),
                     'cterasdk.edge.types.RemoveShareAccessControlEntry'
                 )
+
+    @staticmethod
+    def _validate_trusted_nfs_clients(trusted_nfs_clients):
+        if not isinstance(trusted_nfs_clients, list):
+            raise InputError(
+                'Invalid Trusted NFS Clients list format',
+                repr(trusted_nfs_clients),
+                '[cterasdk.edge.types.NFSv3AccessControlEntry, ...]'
+            )
+        for entry in trusted_nfs_clients:
+            if not isinstance(entry, NFSv3AccessControlEntry):
+                raise InputError('Invalid Trusted NFS Clients entry format', repr(entry), 'cterasdk.edge.types.NFSv3AccessControlEntry')
+
+    @staticmethod
+    def _validate_remove_trusted_nfs_clients(trusted_nfs_clients):
+        if not isinstance(trusted_nfs_clients, list):
+            raise InputError(
+                'Invalid Trusted NFS Clients list format',
+                repr(trusted_nfs_clients),
+                '[cterasdk.edge.types.RemoveNFSv3AccessControlEntry, ...]'
+            )
+        for entry in trusted_nfs_clients:
+            if not isinstance(entry, RemoveNFSv3AccessControlEntry):
+                raise InputError(
+                    'Invalid Trusted NFS Clients entry format',
+                    repr(entry),
+                    'cterasdk.edge.types.RemoveNFSv3AccessControlEntry'
+                )

cterasdk/edge/types.py

@@ -143,6 +143,26 @@ def __str__(self):
         return str(dict(address=self.address, netmask=self.netmask, permission=self.perm))
 
 
+class RemoveNFSv3AccessControlEntry():
+    """
+    Object holding address and netmasak for NFS v3 export access control entry
+    :ivar str address: IP address, hostname or fully qualified domain name of client machine
+    :ivar str netmask: Subnet mask
+    """
+
+    def __init__(self, address, netmask):
+        self._address = address
+        self._netmask = netmask
+
+    @property
+    def address(self):
+        return self._address
+
+    @property
+    def netmask(self):
+        return self._netmask
+
+
 class ShareAccessControlEntry():
     """
     Share access control entry for filer shares

tests/ut/test_edge_shares.py

@@ -3,12 +3,12 @@
 from cterasdk import exception
 from cterasdk.edge import shares
 from cterasdk.edge.enum import Acl, ClientSideCaching, PrincipalType, FileAccessMode
-from cterasdk.edge.types import ShareAccessControlEntry, NFSv3AccessControlEntry
+from cterasdk.edge.types import ShareAccessControlEntry, NFSv3AccessControlEntry, RemoveNFSv3AccessControlEntry
 from cterasdk.common import Object
 from tests.ut import base_edge
 
 
-class TestEdgeShares(base_edge.BaseEdgeTest):
+class TestEdgeShares(base_edge.BaseEdgeTest):  # pylint: disable=too-many-public-methods
 
     def setUp(self):
         super().setUp()
@@ -251,3 +251,66 @@ def test_get_acl(self):
     @staticmethod
     def _get_acl_object():
         return ShareAccessControlEntry(principal_type=PrincipalType.LG, name='Everyone', perm=FileAccessMode.RO)
+
+    def test_get_trusted_nfs_clients(self):
+        share_name = 'share'
+        get_response = self._get_get_trusted_nfs_client_object()
+        self._init_filer(get_response=[get_response.to_server_object()])
+        trusted_nfs_clients = shares.Shares(self._filer).get_trusted_nfs_clients(share_name)
+        self._filer.get.assert_called_once_with('/config/fileservices/share/' + share_name + '/trustedNFSClients')
+        self._assert_equal_objects(NFSv3AccessControlEntry.from_server_object(trusted_nfs_clients[0]), get_response)
+
+    def test_set_trusted_nfs_clients(self):
+        share_name = 'share'
+        new_trusted_nfs_clients = self._get_get_trusted_nfs_client_object()
+        self._init_filer()
+        shares.Shares(self._filer).set_trusted_nfs_clients(share_name, [new_trusted_nfs_clients])
+        self._filer.put.assert_called_once_with('/config/fileservices/share/' + share_name + '/trustedNFSClients', mock.ANY)
+        expected_param = new_trusted_nfs_clients.to_server_object()
+        actual_param = self._filer.put.call_args[0][1][0]
+        self._assert_equal_objects(actual_param, expected_param)
+
+    def test_add_trusted_nfs_clients(self):
+        share_name = 'share'
+        current_trusted_nfs_clients = self._get_get_trusted_nfs_client_object()
+        self._init_filer(get_response=[current_trusted_nfs_clients.to_server_object()])
+
+        new_trusted_nfs_clients = self._get_get_trusted_nfs_client_object(address="192.168.0.0")
+        shares.Shares(self._filer).add_trusted_nfs_clients(share_name, [new_trusted_nfs_clients])
+        self._filer.put.assert_called_once_with('/config/fileservices/share/' + share_name + '/trustedNFSClients', mock.ANY)
+
+        def get_address(elem):
+            return elem.address
+
+        actual_param = self._filer.put.call_args[0][1]
+        actual_param.sort(key=get_address)
+
+        expected_param = [current_trusted_nfs_clients.to_server_object(), new_trusted_nfs_clients.to_server_object()]
+        expected_param.sort(key=get_address)
+
+        self.assertEqual(len(expected_param), len(actual_param))
+
+        for i in range(len(actual_param)):  # pylint: disable=consider-using-enumerate
+            self._assert_equal_objects(actual_param[i], expected_param[i])
+
+    def test_remove_trusted_nfs_clients(self):
+        share_name = 'share'
+        trusted_nfs_client_to_keep = self._get_get_trusted_nfs_client_object(address="192.168.0.0")
+        trusted_nfs_client_to_remove = self._get_get_trusted_nfs_client_object(address="192.168.1.0")
+        self._init_filer(get_response=[trusted_nfs_client_to_keep.to_server_object(), trusted_nfs_client_to_remove.to_server_object()])
+
+        shares.Shares(self._filer).remove_trusted_nfs_clients(
+            share_name,
+            [
+                RemoveNFSv3AccessControlEntry(trusted_nfs_client_to_remove.address, trusted_nfs_client_to_remove.netmask)
+            ]
+        )
+        self._filer.put.assert_called_once_with('/config/fileservices/share/' + share_name + '/trustedNFSClients', mock.ANY)
+
+        expected_param = trusted_nfs_client_to_keep.to_server_object()
+        actual_param = self._filer.put.call_args[0][1][0]
+        self._assert_equal_objects(actual_param, expected_param)
+
+    @staticmethod
+    def _get_get_trusted_nfs_client_object(address=None):
+        return NFSv3AccessControlEntry(address=address or '192.168.68.0', netmask='255.255.255.0', perm=FileAccessMode.RO)