diff --git a/docs/backends/dropbox.rst b/docs/backends/dropbox.rst new file mode 100644 index 000000000..e884f10cf --- /dev/null +++ b/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 diff --git a/storages/backends/dropbox.py b/storages/backends/dropbox.py index dc1958757..fdeb9a70b 100644 --- a/storages/backends/dropbox.py +++ b/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,39 +32,52 @@ 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: @@ -68,31 +85,27 @@ def listdir(self, 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) diff --git a/tests/test_dropbox.py b/tests/test_dropbox.py index 41be28a13..a29d10468 100644 --- a/tests/test_dropbox.py +++ b/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('.'))