0001-add-alternative-storage-system-39517.patch

     # -*- coding: utf-8 -*-
     import os
     import mock
     import pytest
     from webtest import Upload
     from wcs.qommon.ident.password_accounts import PasswordAccount
     from wcs.formdef import FormDef
     from wcs.categories import Category
     from wcs import fields
     from utilities import get_app, login, create_temporary_pub, clean_temporary_pub
     def pytest_generate_tests(metafunc):
         if 'pub' in metafunc.fixturenames:
             metafunc.parametrize('pub', ['pickle', 'sql', 'pickle-templates', 'pickle-lazy'], indirect=True)
     @pytest.fixture
     def pub(request, emails):
         pub = create_temporary_pub(
                 sql_mode=bool('sql' in request.param),
                 templates_mode=bool('templates' in request.param),
                 lazy_mode=bool('lazy' in request.param),
+                )
         pub.cfg['identification'] = {'methods': ['password']}
         pub.cfg['language'] = {'language': 'en'}
         pub.write_cfg()
         open(os.path.join(pub.app_dir, 'site-options.cfg'), 'w').write('''
     [storage-remote]
     label = test out storage
     class = wcs.qommon.upload_storage.RemoteOpaqueUploadStorage
     ws = https://crypto.example.net/ws/
     [api-secrets]
     crypto.example.net = 1234
     [wscall-secrets]
     crypto.example.net = 1234
     ''')
         return pub
     def teardown_module(module):
         clean_temporary_pub()
     def create_formdef():
         FormDef.wipe()
         formdef = FormDef()
         formdef.name = 'test'
         formdef.fields = [
                 fields.FileField(id='0', label='file', varname='file'),
                 fields.FileField(id='1', label='remote file', varname='remote_file')
+                ]
         formdef.store()
         return formdef
     def create_user_and_admin(pub):
         pub.user_class.wipe()
         PasswordAccount.wipe()
         user = pub.user_class()
         user.email = 'foo@localhost'
         user.store()
         account = PasswordAccount(id='foo')
         account.set_password('foo')
         account.user_id = user.id
         account.store()
         admin = pub.user_class()
         admin.email = 'admin@localhost'
         admin.is_admin = True
         admin.store()
         account = PasswordAccount(id='admin')
         account.set_password('admin')
         account.user_id = admin.id
         account.store()
         return user, admin
     @mock.patch('wcs.wscalls.call_webservice')
     def test_form_file_field_upload_storage(wscall, pub):
         create_user_and_admin(pub)
         formdef = create_formdef()
         formdef.data_class().wipe()
         assert formdef.fields[0].storage == formdef.fields[1].storage == 'default'
         assert 'remote' in pub.get_site_storages()
         formdef.fields[1].storage = 'remote'
         formdef.store()
         assert formdef.fields[0].storage == 'default'
         assert formdef.fields[1].storage == 'remote'
         wscall.return_value = None, 200, '{"err": 0, "data": {"redirect_url": "https://crypto.example.net/"}}'
         image_content = open(os.path.join(os.path.dirname(__file__), 'image-with-gps-data.jpeg'), 'rb').read()
         upload_0 = Upload('file.jpg', image_content, 'image/jpeg')
         upload_1 = Upload('remote.jpg', image_content, 'image/jpeg')
         resp = get_app(pub).get('/test/')
         resp.forms[0]['f0$file'] = upload_0
         resp.forms[0]['f1$file'] = upload_1
         resp = resp.forms[0].submit('submit')
         assert 'Check values then click submit.' in resp.text
         resp = resp.forms[0].submit('submit')
         assert resp.status_int == 302
         resp = resp.follow()
         assert 'The form has been recorded' in resp.text
         assert resp.text.count('thumbnail=1') == 1  # thumbnail only for first file
         resp = resp.click('remote.jpg')
         assert resp.location.startswith('https://crypto.example.net/')
         assert '&signature=' in resp.location
         admin_app = login(get_app(pub), username='admin', password='admin')
         resp = admin_app.get('/api/forms/test/1/', status=200)
         assert resp.json['fields']['file']['content'].startswith('/9j/4AAQSkZJRg')
         assert resp.json['fields']['remote_file']['content'] == ''
         assert resp.json['fields']['remote_file']['storage_attrs']['redirect_url'] == 'https://crypto.example.net/'

         max_file_size = None
         automatic_image_resize = False
         allow_portfolio_picking = False
         storage = 'default'
         widget_class = FileWithPreviewWidget
         extra_attributes = [
-...
                 'max_file_size',
                 'allow_portfolio_picking',
                 'automatic_image_resize',
                 'storage',
+                ]
         def __init__(self, *args, **kwargs):
-...
                         title=_('Allow user to pick a file from a portfolio'),
                         value=self.allow_portfolio_picking,
                         advanced=(self.allow_portfolio_picking is FileField.allow_portfolio_picking))
             storages = get_publisher().get_site_storages()
             if storages:
                 storage_options = [('default', '---', {})]
                 storage_options += [(key, value['label'], key) for key, value in storages.items()]
                 form.add(SingleSelectWidget, 'storage', title=_('File storage system'),
                         value=self.storage, options=storage_options,
                         advanced=bool(not self.storage or self.storage == 'default'))
         def get_admin_attributes(self):
             return WidgetField.get_admin_attributes(self) + [
                     'document_type', 'max_file_size', 'allow_portfolio_picking',
                     'automatic_image_resize']
                     'automatic_image_resize', 'storage']
         @classmethod
         def convert_value_from_anything(cls, value):
-...
             t = TemplateIO(html=True)
             t += htmltext('<div class="file-field">')
             t += htmltext('<a download="%s" href="[download]?f=%s">') % (value.base_filename, self.id)
             if include_image_thumbnail and can_thumbnail(value.content_type):
             if include_image_thumbnail and can_thumbnail(value.content_type) and not hasattr(value, 'storage_attrs'):
                 t += htmltext('<img alt="" src="[download]?f=%s&thumbnail=1"/>') % self.id
             t += htmltext('<span>%s</span>') % value
             t += htmltext('</a></div>')
-...
             return [str(value) if value else '']
         def get_json_value(self, value):
             return {
             out = {
                 'field_id': self.id,
                 'filename': value.base_filename,
                 'content_type': value.content_type or 'application/octet-stream',
                 'content': force_text(base64.b64encode(value.get_content())),
+            }
             if hasattr(value, 'storage_attrs'):
                 out['storage_attrs'] = value.storage_attrs
             return out
         def from_json_value(self, value):
             if value and 'filename' in value and 'content' in value:

     from quixote.util import randbytes
     from wcs import data_sources
     from wcs.api_utils import get_user_from_api_query_string, is_url_signed
     from wcs.api_utils import get_user_from_api_query_string, is_url_signed, sign_url_auto_orig
     from wcs.fields import WidgetField, FileField
     from wcs.workflows import EditableWorkflowStatusItem
-...
             if component and component != file.base_filename:
                 raise errors.TraversalError()
             if hasattr(file, 'storage_attrs'):  # remote storage
                 if self.thumbnails:
                     raise errors.TraversalError()
                 if not file.storage_attrs.get('redirect_url'):
                     raise errors.TraversalError()
                 redirect_url = sign_url_auto_orig(file.storage_attrs['redirect_url'])
                 return redirect(redirect_url)
             response = get_response()
             if file.content_type:
                 response.set_content_type(file.content_type)
-...
                     continue
                 if field.type == 'file':
                     # add back file to session
                     tempfile = session.add_tempfile(form_data[field.id])
                     tempfile = session.add_tempfile(form_data[field.id], storage=field.storage)
                     form_data[field.id].token = tempfile['token']
             form_data['is_recalled_draft'] = True
             form_data['draft_formdata_id'] = filled.id
-...
             if not hasattr(file, 'content_type'):
                 raise errors.TraversalError()
             if hasattr(file, 'storage_attrs'):  # remote storage
                 if get_request().form.get('thumbnail') == '1':
                     raise errors.TraversalError()
                 if not file.storage_attrs.get('redirect_url'):
                     raise errors.TraversalError()
                 redirect_url = sign_url_auto_orig(file.storage_attrs['redirect_url'])
                 return redirect(redirect_url)
             file_url = 'files/%s/' % fn
             if get_request().form.get('thumbnail') == '1':
                 file_url += 'thumbnail/'

     import quixote.form.widget
     from quixote import get_publisher, get_request, get_response, get_session
     from quixote.http_request import Upload
     from quixote.form import *
     from quixote.html import htmltext, htmltag, htmlescape, TemplateIO
     from quixote.http_request import Upload
     from quixote.util import randbytes
     from django.utils.encoding import force_bytes, force_text
     from django.utils import six
     from django.utils.six.moves.html_parser import HTMLParser
     from django.utils.six import StringIO
     from django.conf import settings
     from django.utils.safestring import mark_safe
-...
     from .misc import strftime, C_, HAS_PDFTOPPM, json_loads
     from .publisher import get_cfg
     from .template_utils import render_block_to_string
     from .upload_storage import PicklableUpload
     QuixoteForm = Form
-...
         max_file_size_bytes = None # will be filled automatically
         def __init__(self, name, value=None, **kwargs):
             self.storage = kwargs.pop('storage', None)
             CompositeWidget.__init__(self, name, value, **kwargs)
             self.value = value
             self.readonly = kwargs.get('readonly')
-...
             self.add(HiddenWidget, 'token')
             if not self.readonly:
                 attrs = {'data-url': get_publisher().get_root_url() + 'tmp-upload'}
                 if self.storage:
                     attrs['data-url'] += '?storage=%s' % self.storage
                 self.file_type = kwargs.pop('file_type', None)
                 if self.file_type:
                     attrs['accept'] = ','.join(self.file_type)
-...
                     # oops, it has a token but it's not in the session; this is
                     # probably because it was restored from a draft file created
                     # from an expired session.
                     self.value.token = get_session().add_tempfile(self.value).get('token')
                     self.value.token = get_session().add_tempfile(self.value, storage=self.storage).get('token')
                     self.get_widget('token').set_value(self.value.token)
         def add_media(self):
-...
             if self.get('token'):
                 token = self.get('token')
             elif self.get('file'):
                 token = get_session().add_tempfile(self.get('file'))['token']
                 token = get_session().add_tempfile(self.get('file'), storage=self.storage)['token']
                 request.form[self.get_widget('token').get_name()] = token
             else:
                 token = None
-...
                 # there's no file, the other checks are irrelevant.
                 return
             if self.storage and self.storage != self.storage:
                 self.error = _('unknown storage system (system error)')
             # Don't trust the browser supplied MIME type, update the Upload object
             # with a MIME type created with magic (or based on the extension if the
             # module is missing).
-...
             # This also helps people uploading PDF files that were downloaded from
             # sites setting a wrong MIME type (like application/force-download) for
             # various reasons.
             if magic:
             if magic and self.value.fp:
                 if hasattr(magic, 'MagicException'):
                     mime = magic.Magic(mime=True)
                     filetype = mime.from_file(self.value.fp.name)
-...
                     filetype = magic_object.file(self.value.fp.name).split(';')[0]
                     magic_object.close()
             else:
                 filetype, encoding = mimetypes.guess_type(self.value.base_filename)
                 filetype = getattr(self.value, 'storage_attrs', {}).get('content_type')
                 if not filetype:
                     filetype, encoding = mimetypes.guess_type(self.value.base_filename)
             if not filetype:
                 filetype = 'application/octet-stream'
-...
                 self.error = _('forbidden file type')
     class PicklableUpload(Upload):
         def __getstate__(self):
             odict = self.__dict__.copy()
             if 'fp' in odict:
                 del odict['fp']
             basedir = os.path.join(get_publisher().app_dir, 'uploads')
             if not os.path.exists(basedir):
                 os.mkdir(basedir)
             if 'qfilename' in odict:
                 filepath = os.path.join(basedir, self.qfilename)
             else:
                 self.qfilename = misc.file_digest(self.fp)
                 filepath = os.path.join(basedir, self.qfilename)
             if 'fp' in self.__dict__ and not self.fp.closed:
                 self.fp.seek(0)
                 atomic_write(filepath, self.fp, async_op=False)
             odict['qfilename'] = self.qfilename
             return odict
         def get_file_pointer(self):
             if 'fp' in self.__dict__ and self.__dict__.get('fp') is not None:
                 return self.__dict__.get('fp')
             elif hasattr(self, 'qfilename'):
                 basedir = os.path.join(get_publisher().app_dir, 'uploads')
                 self.fp = open(os.path.join(basedir, self.qfilename), 'rb')
                 return self.fp
             return None
         def __setstate__(self, dict):
             self.__dict__.update(dict)
             if hasattr(self, 'data'):
                 # backward compatibility with older w.c.s. version
                 self.fp = StringIO(self.data)
                 del self.data
         def get_file(self):
             # quack like UploadedFile
             return self.get_file_pointer()
         def get_filename(self):
             if not hasattr(self, 'qfilename'):
                 raise AttributeError('filename')
             basedir = os.path.join(get_publisher().app_dir, 'uploads')
             return os.path.join(basedir, self.qfilename)
         def get_content(self):
             if hasattr(self, 'qfilename'):
                 filename = os.path.join(get_publisher().app_dir, 'uploads', self.qfilename)
                 return open(filename, 'rb').read()
             return None
         def get_base64_content(self):
             content = self.get_content()
             if content:
                 return base64.encodestring(content)
             return None
     class EmailWidget(StringWidget):
         HTML_TYPE = 'email'

                 return obj.decode('ascii')
             if hasattr(obj, 'base_filename'):
                 return {
                 out = {
                     'filename': obj.base_filename,
                     'content_type': obj.content_type or 'application/octet-stream',
                     'content': base64.b64encode(obj.get_content()),
+                 }
+                }
                 if hasattr(obj, 'storage_attrs'):  # remote storage
                     out['storage_attrs'] = obj.storage_attrs
                 return out
             # Let the base class default method raise the TypeError
             return json.JSONEncoder.default(self, obj)

             except ConfigParser.NoOptionError:
                 return None
         def get_site_storages(self):
             if self.site_options is None:
                 self.load_site_options()
             storages = {}
             for section, definition in self.site_options.items():
                 if section.startswith('storage-') and 'label' in definition and 'class' in definition:
                     storage_id = section[8:]
                     storages[storage_id] = dict(definition)
                     storages[storage_id]['id'] = storage_id
             return storages
         def set_config(self, request = None):
             self.reload_cfg()
             self.site_options = None # reset at the beginning of a request

     from . import misc
     from .storage import StorableObject
     from .publisher import get_cfg
     from .upload_storage import get_storage_object
     from quixote.publish import get_session, get_session_manager, get_request
-...
         def get_signer(self):
             return Signer(settings.SECRET_KEY + self.id)
         def add_tempfile(self, upload):
             from wcs.qommon.form import PicklableUpload
             token = randbytes(8)
             upload.__class__ = PicklableUpload
             upload.time = time.time()
             upload.token = token
             # and saves it
         def add_tempfile(self, upload, storage=None):
             dirname = os.path.join(get_publisher().app_dir, 'tempfiles')
             if not os.path.exists(dirname):
                 os.mkdir(dirname)
             filename = os.path.join(dirname, token)
             fd = open(filename, 'wb')
             upload.get_file_pointer().seek(0)
             fd.write(upload.get_file_pointer().read())
             size = fd.tell()
             fd.close()
             token = randbytes(8)
             upload.time = time.time()
             upload.token = token
             upload.storage = storage
             get_storage_object(upload.storage).save_tempfile(upload)
             self.has_uploads = True
             if not self.id:
-...
                 'base_filename': upload.base_filename,
                 'content_type': upload.content_type,
                 'charset': upload.charset,
                 'size': size,
                 'size': getattr(upload, 'size', None),
                 'session': self.id,
                 'token': signer.sign(token),
                 'unsigned_token': token,
                 'storage': upload.storage,
                 'storage-attrs': getattr(upload, 'storage_attrs', None),
+            }
             filename = os.path.join(get_publisher().app_dir, 'tempfiles', upload.token)
             with open(filename + '.json', 'w') as fd:
                 json.dump(data, fd, indent=2)
-...
             if not temp:
                 return temp
             from wcs.qommon.form import PicklableUpload
             value = PicklableUpload(temp['orig_filename'],
                             temp['content_type'], temp['charset'])
             dirname = os.path.join(get_publisher().app_dir, 'tempfiles')
             filename = os.path.join(dirname, temp['unsigned_token'])
             value.token = token
             value.fp = open(filename, 'rb')
             return value
             return get_storage_object(temp.get('storage')).get_tempfile(temp)
         def add_extra_variables(self, **kwargs):
             if not self.extra_variables:

     # w.c.s. - web application for online forms
     # Copyright (C) 2005-2020  Entr'ouvert
+    #
     # This program is free software; you can redistribute it and/or modify
     # it under the terms of the GNU General Public License as published by
     # the Free Software Foundation; either version 2 of the License, or
     # (at your option) any later version.
+    #
     # This program is distributed in the hope that it will be useful,
     # but WITHOUT ANY WARRANTY; without even the implied warranty of
     # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
     # GNU General Public License for more details.
+    #
     # You should have received a copy of the GNU General Public License
     # along with this program; if not, see <http://www.gnu.org/licenses/>.
     import base64
     import os
     from quixote import get_publisher
     from quixote.http_request import Upload
     from django.utils.module_loading import import_string
     from django.utils.six import StringIO
     from .errors import ConnectionError
     from .misc import json_loads, file_digest
     from .storage import atomic_write
     class PicklableUpload(Upload):
         def __getstate__(self):
             odict = self.__dict__.copy()
             if 'fp' in odict:
                 del odict['fp']
             get_storage_object(getattr(self, 'storage', None)).save(self)
             odict['qfilename'] = getattr(self, 'qfilename', None)
             return odict
         def get_file_pointer(self):
             if 'fp' in self.__dict__ and self.__dict__.get('fp') is not None:
                 return self.__dict__.get('fp')
             elif hasattr(self, 'qfilename'):
                 basedir = os.path.join(get_publisher().app_dir, 'uploads')
                 self.fp = open(os.path.join(basedir, self.qfilename), 'rb')
                 return self.fp
             return None
         def __setstate__(self, dict):
             self.__dict__.update(dict)
             if hasattr(self, 'data'):
                 # backward compatibility with older w.c.s. version
                 self.fp = StringIO(self.data)
                 del self.data
         def get_file(self):
             # quack like UploadedFile
             return self.get_file_pointer()
         def get_filename(self):
             if not hasattr(self, 'qfilename'):
                 raise AttributeError('filename')
             basedir = os.path.join(get_publisher().app_dir, 'uploads')
             return os.path.join(basedir, self.qfilename)
         def get_content(self):
             if hasattr(self, 'storage_attrs'):  # remote storage
                 return b''
             if hasattr(self, 'qfilename'):
                 filename = os.path.join(get_publisher().app_dir, 'uploads', self.qfilename)
                 return open(filename, 'rb').read()
             return None
         def get_base64_content(self):
             content = self.get_content()
             if content:
                 return base64.encodestring(content)
             return b''
     class UploadStorageError(Exception):
         pass
     class UploadStorage(object):
         def save_tempfile(self, upload):
             upload.__class__ = PicklableUpload
             dirname = os.path.join(get_publisher().app_dir, 'tempfiles')
             filename = os.path.join(dirname, upload.token)
             fd = open(filename, 'wb')
             upload.get_file_pointer().seek(0)
             fd.write(upload.get_file_pointer().read())
             upload.size = fd.tell()
             fd.close()
         def get_tempfile(self, temp_data):
             value = PicklableUpload(
                     temp_data['orig_filename'],
                     temp_data['content_type'],
                     temp_data['charset'])
             value.storage = temp_data.get('storage')
             dirname = os.path.join(get_publisher().app_dir, 'tempfiles')
             filename = os.path.join(dirname, temp_data['unsigned_token'])
             value.token = temp_data['token']
             value.fp = open(filename, 'rb')
             return value
         def save(self, upload):
             basedir = os.path.join(get_publisher().app_dir, 'uploads')
             if not os.path.exists(basedir):
                 os.mkdir(basedir)
             if getattr(upload, 'qfilename', None):
                 filepath = os.path.join(basedir, upload.qfilename)
             else:
                 upload.qfilename = file_digest(upload.fp)
                 filepath = os.path.join(basedir, upload.qfilename)
             if getattr(upload, 'fp', None) and not upload.fp.closed:
                 upload.fp.seek(0)
                 atomic_write(filepath, upload.fp, async_op=False)
     class RemoteOpaqueUploadStorage(object):
         def __init__(self, ws, **kwargs):
             self.ws = ws
         def save_tempfile(self, upload):
             if getattr(upload, 'storage_attrs', None):
                 # upload is already a remote PicklableUpload, it does not
                 # have content. We are certainly restoring a draft.
                 return
             upload.__class__ = PicklableUpload
             dirname = os.path.join(get_publisher().app_dir, 'tempfiles')
             filename = os.path.join(dirname, upload.token)
             fd = open(filename, 'wb')
             upload.get_file_pointer().seek(0)
             base64content = base64.b64encode(upload.get_file_pointer().read())
             fd.close()
             post_data = {
                     'file': {
                         'filename': upload.base_filename,
                         'content_type': upload.content_type or 'application/octet-stream',
                         'content': base64content,
+                        }
+                    }
             try:
                 from wcs.wscalls import call_webservice
                 response, status, data = call_webservice(self.ws, method='POST', post_data=post_data)
             except ConnectionError as e:
                 raise UploadStorageError('remote storage connection error (%r)' % e)
             if status not in (200, 201):
                 raise UploadStorageError('remote storage returns status %s' % status)
             try:
                 ws_result = json_loads(data)
             except (ValueError, TypeError) as e:
                 raise UploadStorageError('remote storage returns invalid JSON')
             if not isinstance(ws_result, dict):
                 raise UploadStorageError('remote storage returns non-dict JSON')
             if ws_result.get('err') != 0:
                 raise UploadStorageError('remote storage returns err = %s' % ws_result.get('err'))
             ws_result_data = ws_result.get('data', {})
             if not ws_result_data.get('redirect_url'):
                 raise UploadStorageError('remote storage returns data.redirect_url= %s' %
                         ws_result_data.get('redirect_url'))
             upload.storage_attrs = ws_result_data
         def get_tempfile(self, temp_data):
             value = PicklableUpload(
                     temp_data['orig_filename'],
                     temp_data['content_type'],
                     temp_data['charset'])
             value.storage = temp_data.get('storage')
             value.storage_attrs = temp_data['storage-attrs']
             value.token = temp_data['token']
             value.fp = None
             return value
         def save(self, upload):
             pass
     def get_storage_object(storage):
         if not storage or storage == 'default':
             return UploadStorage()
         storage_cfg = get_publisher().get_site_storages().get(storage)
         if not storage_cfg:
             raise UploadStorageError('unknown storage %s' % storage)
         try:
             storage_class = import_string(storage_cfg['class'])
         except ImportError:
             raise UploadStorageError('failed to import storage class %s' % storage_class)
         return storage_class(**storage_cfg)

         def tmp_upload(self):
             results = []
             storage = get_request().form.get('storage')
             for k, v in get_request().form.items():
                 if hasattr(v, 'fp'):
                     tempfile = get_session().add_tempfile(v)
                     tempfile = get_session().add_tempfile(v, storage=storage)
                     results.append({'name': tempfile.get('base_filename'),
                                     'type': tempfile.get('content_type'),
                                     'size': tempfile.get('size'),
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.