Merge lp:~brian.curtin/ubuntuone-client/824252-validate-path-export into lp:ubuntuone-client

[Brian Curtin]

- Exported validate_path_from_folder from volume_manager instead of ubuntuone-control-panel's backend (lp:824252)

Fix lp:824252 by moving validate_path_for_folder inside volume_manager.VolumeManager and exposing it via the dbus and IPC interfaces. Corrects an earlier None/False issue that was failing on Linux.

[Roberto Alsina (ralsina)]

+1

review: Approve

[Natalia Bidart (nataliabidart)]

Branch looks good!

Two minor nitpicks:

* Any reason to have this:

from ubuntuone.syncdaemon import config, event_queue, tritcask
from ubuntuone.syncdaemon import volume_manager

instead of:

from ubuntuone.syncdaemon import config, event_queue, tritcask, volume_manager

?

* Can you please add the docstring for this method?

def validate_path_for_folder(self, path):

* Shouldn't we call normpath(path) before checking if the path is inside the user home?

Tests are green in both OSes, yey! (except for bug #929546 which is not relevant to this branch)
Thanks!

review: Needs Fixing

[lp:~brian.curtin/ubuntuone-client/824252-validate-path-export]

1189. By Brian Curtin on 2012-02-09

Address Natalia's review comments.
Rearranged import, added docstring, moved normpath call to earlier.

[Brian Curtin (brian.curtin)]

> Branch looks good!

rev 1189 makes the changes you suggested.

[Natalia Bidart (nataliabidart)]

Looks great!

review: Approve

=== modified file 'tests/platform/test_external_interface.py'
--- tests/platform/test_external_interface.py	2012-01-02 17:53:12 +0000
+++ tests/platform/test_external_interface.py	2012-02-09 23:36:18 +0000
@@ -551,6 +551,16 @@
             in_signature='s', out_signature=None)
 
     @defer.inlineCallbacks
+    def test_validate_path(self):
+        """Test validate_path."""
+        result = False
+        path = 'test'
+        yield self.assert_method_called(self.service.folders,
+                                        'validate_path', result, path)
+        self.assert_remote_method('validate_path',
+            in_signature='s', out_signature='b')
+
+    @defer.inlineCallbacks
     def test_get_folders(self):
         """Test get_folders."""
         result = [{'folder_id': '1'}, {'folder_id': '2'}]

=== modified file 'tests/syncdaemon/test_interaction_interfaces.py'
--- tests/syncdaemon/test_interaction_interfaces.py	2012-01-17 20:00:44 +0000
+++ tests/syncdaemon/test_interaction_interfaces.py	2012-02-09 23:36:18 +0000
@@ -1041,6 +1041,16 @@
 
         self.assertEqual(self._called, ((folder_id,), {}))
 
+    def test_validate_path(self):
+        """Test the validate_path method."""
+        # Use a lambda instead of _set_called since we need to return a tuple.
+        self.patch(self.main.vm, 'validate_path_for_folder',
+                   lambda arg: (True, arg))
+        path = u'this/is/a/test'
+        rv = self.sd_obj.validate_path(path)
+
+        self.assertTrue(rv)
+
     @defer.inlineCallbacks
     def test_get_folders(self):
         """Test the get_folders method."""

=== modified file 'tests/syncdaemon/test_vm.py'
--- tests/syncdaemon/test_vm.py	2012-01-18 20:47:18 +0000
+++ tests/syncdaemon/test_vm.py	2012-02-09 23:36:18 +0000
@@ -44,7 +44,7 @@
     FakeMain,
 )
 from ubuntuone import platform
-from ubuntuone.syncdaemon import config, event_queue, tritcask
+from ubuntuone.syncdaemon import config, event_queue, tritcask, volume_manager
 from ubuntuone.syncdaemon.volume_manager import (
     ACCESS_LEVEL_RO,
     ACCESS_LEVEL_RW,
@@ -2713,6 +2713,112 @@
         self.assertEquals(str(root_volume.node_id), root_node_id)
         self.assertEquals(root_volume.free_bytes, free_bytes)
 
+    def test_validate_UDF_path_inside_home(self):
+        """Test proper validation of path for creating folders."""
+        folder_path = os.path.join(self.home_dir, 'Test Me')
+
+        result, msg = self.vm.validate_path_for_folder(folder_path)
+        self.assertTrue(result)
+        self.assertIs(msg, "", 
+            '%r must be a valid path for creating a folder.' % folder_path)
+
+    def test_validate_UDF_path_if_folder_shares_a_prefix_with_an_udf(self):
+        """Test proper validation of path for creating folders.
+
+        If the user chooses a folder with the same prefix as an UDF, but
+        outside every UDF, the path is valid.
+
+        """
+        tricky_path = self.root_dir
+        assert not tricky_path.endswith(os.path.sep)
+        tricky_path += ' Suffix'
+        assert tricky_path.startswith(self.root_dir)
+
+        result, msg = self.vm.validate_path_for_folder(tricky_path)
+        self.assertTrue(result)
+        self.assertIs(msg, "", 
+            '%r must be a valid path for creating a folder.' % tricky_path)
+
+    def test_validate_UDF_path_not_valid_if_outside_home(self):
+        """A folder outside ~ is not valid."""
+        outside_home = os.path.abspath(os.path.join(self.home_dir, os.path.pardir))
+
+        result, msg = self.vm.validate_path_for_folder(outside_home)
+        self.assertFalse(result)
+        self.assertIsNot(msg, "",
+            '%r must be an invalid path for creating a folder.' % outside_home)
+
+    def test_validate_UDF_not_valid_if_folder_inside_root(self):
+        """A folder inside the root is not valid."""
+        root_path = self.root_dir
+        # create a valid path inside the root
+        inside_root = os.path.abspath(os.path.join(root_path, 'test'))
+
+        result, msg = self.vm.validate_path_for_folder(inside_root)
+        self.assertFalse(result)
+        self.assertIsNot(msg, "",
+            '%r must be an invalid path for creating a folder.' % inside_root)
+
+    @defer.inlineCallbacks
+    def test_validate_UDF_not_valid_if_folder_inside_an_udf(self):
+        """A folder inside an UDF is not valid."""
+        udf = self._create_udf(subscribed=True)
+        yield self.vm.add_udf(udf)
+        # create a valid path inside an existing UDF
+        inside_udf = os.path.abspath(os.path.join(udf.path, 'test'))
+
+        result, msg = self.vm.validate_path_for_folder(inside_udf)
+        self.assertFalse(result)
+        self.assertIsNot(msg, "",
+            '%r must be an invalid path for creating a folder.' % inside_udf)
+
+    @defer.inlineCallbacks
+    def test_validate_UDF_not_valid_if_folder_is_parent_of_an_udf(self):
+        """A folder parent of an UDF is not valid."""
+        udf = self._create_udf(subscribed=True)
+        yield self.vm.add_udf(udf)
+        # create a valid path that is parent from an existing UDF
+        udf_parent = os.path.abspath(os.path.join(self.home_dir, os.path.pardir))
+
+        result, msg = self.vm.validate_path_for_folder(udf_parent)
+        self.assertFalse(result)
+        self.assertIsNot(msg, "",
+            '%r must be an invalid path for creating a folder.' % udf_parent)
+
+    def test_not_valid_if_folder_is_link(self):
+        """A link path is not valid."""
+        self.patch(volume_manager, 'is_link', lambda p: True)
+        path_link = os.path.join(self.home_dir, 'Test Me')
+
+        result, msg = self.vm.validate_path_for_folder(path_link)
+        self.assertFalse(result)
+        self.assertIsNot(msg, "",
+            '%r must be an invalid path for creating a folder.' % path_link)
+
+    @defer.inlineCallbacks
+    def test_no_UDFs_outside_home(self):
+        udf = self._create_udf(subscribed=True)
+        yield self.vm.add_udf(udf)
+        # Use a directory surely outside home
+        # Drive root on Windows, or root on linux
+        outside_path = os.path.splitdrive(udf.path)[0] or "/"
+        # patch FakeAQ
+        def create_udf(path, name, marker):
+            """Fake create_udf"""
+            d = dict(volume_id=uuid.uuid4(), node_id=uuid.uuid4(),
+                     marker=marker)
+            self.main.event_q.push("AQ_CREATE_UDF_OK", **d)
+
+        self.main.action_q.create_udf = create_udf
+        d = defer.Deferred()
+        self._listen_for('VM_UDF_CREATE_ERROR', d.callback)
+        self._listen_for('VM_UDF_CREATED', lambda r: d.errback(Exception(r)))
+        self.vm.create_udf(outside_path)
+        result = yield d
+        self.assertEqual(1, len(list(self.vm.udfs.keys())))
+        self.assertEqual(result,
+                         dict(path=outside_path, error="UDFs must be within home"))
+
     @defer.inlineCallbacks
     def test_no_UDFs_inside_root(self):
         """Test that a UDF can't be created inside the root"""

=== modified file 'ubuntuone/platform/linux/dbus_interface.py'
--- ubuntuone/platform/linux/dbus_interface.py	2012-01-02 20:24:31 +0000
+++ ubuntuone/platform/linux/dbus_interface.py	2012-02-09 23:36:18 +0000
@@ -699,6 +699,12 @@
         """Delete the folder specified by folder_id"""
         return self.service.folders.delete(folder_id)
 
+    @dbus.service.method(DBUS_IFACE_FOLDERS_NAME,
+                         in_signature='s', out_signature='b')
+    def validate_path(self, path):
+        """Return True if the path is valid for a folder"""
+        return self.service.folders.validate_path(path)
+
     @dbus.service.method(DBUS_IFACE_FOLDERS_NAME, out_signature='aa{ss}')
     def get_folders(self):
         """Return the list of folders (a list of dicts)"""

=== modified file 'ubuntuone/platform/tools/__init__.py'
--- ubuntuone/platform/tools/__init__.py	2011-12-19 13:39:46 +0000
+++ ubuntuone/platform/tools/__init__.py	2012-02-09 23:36:18 +0000
@@ -441,6 +441,14 @@
 
     @defer.inlineCallbacks
     @log_call(logger.debug)
+    def validate_path(self, path):
+        """Return True if the path is valid for a folder."""
+        result = yield self.proxy.call_method('folders', 'validate', path)
+        self.log.debug('valid: %r', result)
+        defer.returnValue(result)
+
+    @defer.inlineCallbacks
+    @log_call(logger.debug)
     def get_folders(self):
         """Return the list of folders (a list of dicts)."""
         results = yield self.proxy.call_method('folders', 'get_folders')

=== modified file 'ubuntuone/platform/windows/ipc.py'
--- ubuntuone/platform/windows/ipc.py	2012-01-03 12:40:04 +0000
+++ ubuntuone/platform/windows/ipc.py	2012-02-09 23:36:18 +0000
@@ -773,6 +773,7 @@
     remote_calls = [
         'create',
         'delete',
+        'validate_path',
         'get_folders',
         'subscribe',
         'unsubscribe',
@@ -799,6 +800,10 @@
         """Delete the folder specified by folder_id"""
         return self.service.folders.delete(folder_id)
 
+    def validate_path(self, path):
+        """Return True if the path is valid for a folder."""
+        return self.service.folders.validate_path(path)
+
     def get_folders(self):
         """Return the list of folders (a list of dicts)"""
         return self.service.folders.get_folders()

=== modified file 'ubuntuone/platform/windows/ipc_client.py'
--- ubuntuone/platform/windows/ipc_client.py	2012-01-03 12:40:04 +0000
+++ ubuntuone/platform/windows/ipc_client.py	2012-02-09 23:36:18 +0000
@@ -579,6 +579,10 @@
         """Delete the folder specified by folder_id"""
 
     @remote
+    def validate_path(self, path):
+        """Return True if the path is valid for a folder"""
+
+    @remote
     def get_folders(self):
         """Return the list of folders (a list of dicts)"""
 

=== modified file 'ubuntuone/syncdaemon/interaction_interfaces.py'
--- ubuntuone/syncdaemon/interaction_interfaces.py	2012-01-15 19:31:07 +0000
+++ ubuntuone/syncdaemon/interaction_interfaces.py	2012-02-09 23:36:18 +0000
@@ -674,6 +674,13 @@
         """Delete the folder specified by folder_id."""
         self.vm.delete_volume(folder_id)
 
+    @unicode_to_bytes
+    @log_call(logger.debug)
+    def validate_path(self, path):
+        """Return True if the path is valid for a folder."""
+        # Returns (bool, str), but we only care about the bool from here on.
+        return self.vm.validate_path_for_folder(path)[0]
+
     @log_call(logger.debug)
     def get_folders(self):
         """Return the list of folders (a list of dicts)."""

=== modified file 'ubuntuone/syncdaemon/volume_manager.py'
--- ubuntuone/syncdaemon/volume_manager.py	2011-12-20 15:35:51 +0000
+++ ubuntuone/syncdaemon/volume_manager.py	2012-02-09 23:36:18 +0000
@@ -1178,20 +1178,31 @@
                 return True
         return False
 
-    def create_udf(self, path):
-        """Request the creation of a UDF to AQ."""
-        self.log.debug('create udf: %r', path)
+    def validate_path_for_folder(self, path):
+        """Validate 'folder_path' for folder creation."""
         path = normpath(path)
+        user_home = expand_user('~')
+        # handle folder_path not within '~'
+        if not path.startswith(user_home):
+            return (False, "UDFs must be within home")
+
         # check if path is the realpath, bail out if not
         if is_link(path):
-            self.m.event_q.push('VM_UDF_CREATE_ERROR', path=path,
-                                error="UDFs can not be a symlink")
-            return
+            return (False, "UDFs can not be a symlink")
+
         # check if the path it's ok (outside root and
         # isn't a ancestor or child of another UDF)
         if self._is_nested_udf(path):
-            self.m.event_q.push('VM_UDF_CREATE_ERROR', path=path,
-                                error="UDFs can not be nested")
+            return (False, "UDFs can not be nested")
+
+        return (True, "")
+
+    def create_udf(self, path):
+        """Request the creation of a UDF to AQ."""
+        self.log.debug('create udf: %r', path)
+        valid, msg = self.validate_path_for_folder(path)
+        if not valid:
+            self.m.event_q.push('VM_UDF_CREATE_ERROR', path=path, error=msg)
             return
 
         try: