Improved DropBox storage

docs/backends/dropbox.rst

@@ -0,0 +1,17 @@
+DropBox
+=======
+
+Settings
+--------
+
+
+``DROPBOX_OAUTH2_TOKEN``
+
+Your DropBox token, if you haven't follow this `guide step`_.
+
+``DROPBOX_ROOT_PATH``
+
+Allow to jail your storage to a defined directory.
+
+
+.. _`guide step`: https://www.dropbox.com/developers/documentation/python#tutorial

storages/backends/dropbox.py

@@ -6,15 +6,19 @@
 #
 # Add below to settings.py:
 # DROPBOX_OAUTH2_TOKEN = 'YourOauthToken'
+# DROPBOX_ROOT_PATH = '/dir/'
 
 from __future__ import absolute_import
 
 from datetime import datetime
+from tempfile import SpooledTemporaryFile
+from shutil import copyfileobj
 
 from django.core.files.base import File
 from django.core.exceptions import ImproperlyConfigured
+from django.utils._os import safe_join
 
-from storages.compat import BytesIO, Storage
+from storages.compat import Storage
 from storages.utils import setting
 
 from dropbox.client import DropboxClient
@@ -28,71 +32,80 @@ class DropBoxStorageException(Exception):
 
 
 class DropBoxFile(File):
-    def __init__(self, name, storage, mode='rb'):
+    def __init__(self, name, storage):
         self.name = name
         self._storage = storage
 
-    def read(self, num_bytes=None):
-        return self._storage._read(self.name, num_bytes=num_bytes)
-
-    def write(self, content):
-        self._storage._save(self.name, content)
+    @property
+    def file(self):
+        if not hasattr(self, '_file'):
+            response = self._storage.client.get_file(self.name)
+            self._file = SpooledTemporaryFile()
+            copyfileobj(response, self._file)
+            self._file.seek(0)
+        return self._file
 
 
 class DropBoxStorage(Storage):
     """DropBox Storage class for Django pluggable storage system."""
 
-    def __init__(self, oauth2_access_token=setting('DROPBOX_OAUTH2_TOKEN')):
+    def __init__(self, oauth2_access_token=None, root_path=None):
+        oauth2_access_token = oauth2_access_token or setting('DROPBOX_OAUTH2_TOKEN')
+        self.root_path = root_path or setting('DROPBOX_ROOT_PATH', '/')
         if oauth2_access_token is None:
             raise ImproperlyConfigured("You must configure a token auth at"
                                        "'settings.DROPBOX_OAUTH2_TOKEN'.")
         self.client = DropboxClient(oauth2_access_token)
 
+    def _full_path(self, name):
+        if name == '/':
+            name = ''
+        return safe_join(self.root_path, name)
+
     def delete(self, name):
-        self.client.file_delete(name)
+        self.client.file_delete(self._full_path(name))
 
     def exists(self, name):
         try:
-            return bool(self.client.metadata(name))
+            return bool(self.client.metadata(self._full_path(name)))
         except ErrorResponse:
             return False
 
     def listdir(self, path):
         directories, files = [], []
-        metadata = self.client.metadata(path)
+        full_path = self._full_path(path)
+        metadata = self.client.metadata(full_path)
         for entry in metadata['contents']:
+            entry['path'] = entry['path'].replace(full_path, '', 1)
+            entry['path'] = entry['path'].replace('/', '', 1)
             if entry['is_dir']:
                 directories.append(entry['path'])
             else:
                 files.append(entry['path'])
         return directories, files
 
     def size(self, name):
-        metadata = self.client.metadata(name)
+        metadata = self.client.metadata(self._full_path(name))
         return metadata['bytes']
 
     def modified_time(self, name):
-        metadata = self.client.metadata(name)
+        metadata = self.client.metadata(self._full_path(name))
         mod_time = datetime.strptime(metadata['modified'], DATE_FORMAT)
         return mod_time
 
     def accessed_time(self, name):
-        metadata = self.client.metadata(name)
+        metadata = self.client.metadata(self._full_path(name))
         acc_time = datetime.strptime(metadata['client_mtime'], DATE_FORMAT)
         return acc_time
 
     def url(self, name):
-        media = self.client.media(name)
+        media = self.client.media(self._full_path(name))
         return media['url']
 
     def _open(self, name, mode='rb'):
-        remote_file = DropBoxFile(name, self)
+        remote_file = DropBoxFile(self._full_path(name), self)
         return remote_file
 
     def _save(self, name, content):
-        self.client.put_file(name, content)
+        self.client.put_file(self._full_path(name), content)
         return name
-
-    def _read(self, name, num_bytes=None):
-        data = self.client.get_file(name)
-        return data.read(num_bytes)

tests/test_dropbox.py

@@ -7,6 +7,8 @@
 
 from django.test import TestCase
 from django.core.files.base import File, ContentFile
+from django.core.exceptions import ImproperlyConfigured, \
+                                   SuspiciousFileOperation
 
 from storages.backends import dropbox
 
@@ -64,7 +66,11 @@ class DropBoxTest(TestCase):
                 re.compile(r'.*'))
     @mock.patch('dropbox.client.DropboxOAuth2Session')
     def setUp(self, *args):
-        self.storage = dropbox.DropBoxStorage('')
+        self.storage = dropbox.DropBoxStorage('foo')
+
+    def test_no_access_token(self, *args):
+        with self.assertRaises(ImproperlyConfigured):
+            dropbox.DropBoxStorage(None)
 
     @mock.patch('dropbox.client.DropboxClient.file_delete',
                 return_value=FILE_FIXTURE)
@@ -89,8 +95,8 @@ def test_listdir(self, *args):
         dirs, files = self.storage.listdir('/')
         self.assertGreater(len(dirs), 0)
         self.assertGreater(len(files), 0)
-        self.assertEqual(dirs[0], '/bar')
-        self.assertEqual(files[0], '/foo.txt')
+        self.assertEqual(dirs[0], 'bar')
+        self.assertEqual(files[0], 'foo.txt')
 
     @mock.patch('dropbox.client.DropboxClient.metadata',
                 return_value=FILE_FIXTURE)
@@ -119,34 +125,55 @@ def test_open(self, *args):
     def test_save(self, *args):
         self.storage._save('foo', b'bar')
 
-    @mock.patch('dropbox.client.DropboxClient.get_file',
-                return_value=ContentFile('bar'))
-    def test_read(self, *args):
-        content = self.storage._read('foo')
-        self.assertEqual(content, 'bar')
-
     @mock.patch('dropbox.client.DropboxClient.media',
                 return_value=FILE_MEDIA_FIXTURE)
     def test_url(self, *args):
         url = self.storage.url('foo')
         self.assertEqual(url, FILE_MEDIA_FIXTURE['url'])
 
+    def test_formats(self, *args):
+        self.storage = dropbox.DropBoxStorage('foo')
+        files = self.storage._full_path('')
+        self.assertEqual(files, self.storage._full_path('/'))
+        self.assertEqual(files, self.storage._full_path('.'))
+        self.assertEqual(files, self.storage._full_path('..'))
+        self.assertEqual(files, self.storage._full_path('../..'))
+
 
 class DropBoxFileTest(TestCase):
     @mock.patch('dropbox.client._OAUTH2_ACCESS_TOKEN_PATTERN',
                 re.compile(r'.*'))
     @mock.patch('dropbox.client.DropboxOAuth2Session')
     def setUp(self, *args):
-        self.storage = dropbox.DropBoxStorage('')
+        self.storage = dropbox.DropBoxStorage('foo')
         self.file = dropbox.DropBoxFile('/foo.txt', self.storage)
 
-    @mock.patch('dropbox.client.DropboxClient.put_file',
-                return_value='foo')
-    def test_write(self, *args):
-        self.storage._save('foo', b'bar')
-
     @mock.patch('dropbox.client.DropboxClient.get_file',
-                return_value=ContentFile('bar'))
+                return_value=ContentFile(b'bar'))
     def test_read(self, *args):
-        content = self.storage._read('foo')
-        self.assertEqual(content, 'bar')
+        file = self.storage._open(b'foo')
+        self.assertEqual(file.read(), b'bar')
+
+
+@mock.patch('dropbox.client._OAUTH2_ACCESS_TOKEN_PATTERN',
+            re.compile(r'.*'))
+@mock.patch('dropbox.client.DropboxOAuth2Session')
+@mock.patch('dropbox.client.DropboxClient.metadata',
+            return_value={'contents': []})
+class DropBoxRootPathTest(TestCase):
+    def test_jailed(self, *args):
+        self.storage = dropbox.DropBoxStorage('foo', '/bar')
+        dirs, files = self.storage.listdir('/')
+        self.assertFalse(dirs)
+        self.assertFalse(files)
+
+    def test_suspicious(self, *args):
+        self.storage = dropbox.DropBoxStorage('foo', '/bar')
+        with self.assertRaises((SuspiciousFileOperation, ValueError)):
+            self.storage._full_path('..')
+
+    def test_formats(self, *args):
+        self.storage = dropbox.DropBoxStorage('foo', '/bar')
+        files = self.storage._full_path('')
+        self.assertEqual(files, self.storage._full_path('/'))
+        self.assertEqual(files, self.storage._full_path('.'))