Recognize flash mimetype properly. Store extension as mimetype in attachment

object

boards/forms.py

@@ -1,405 +1,406 b''
 import hashlib
 import re
 import time
 import logging
 import pytz
 
 from django import forms
 from django.core.files.uploadedfile import SimpleUploadedFile
 from django.core.exceptions import ObjectDoesNotExist
 from django.forms.util import ErrorList
 from django.utils.translation import ugettext_lazy as _, ungettext_lazy
 
 from boards.mdx_neboard import formatters
 from boards.models.attachment.downloaders import Downloader
 from boards.models.post import TITLE_MAX_LENGTH
 from boards.models import Tag, Post
 from boards.utils import validate_file_size, get_file_mimetype, \
     FILE_EXTENSION_DELIMITER
 from neboard import settings
 import boards.settings as board_settings
 import neboard
 
 REGEX_TAGS = re.compile(r'^[\w\s\d]+$', re.UNICODE)
 
 VETERAN_POSTING_DELAY = 5
 
 ATTRIBUTE_PLACEHOLDER = 'placeholder'
 ATTRIBUTE_ROWS = 'rows'
 
 LAST_POST_TIME = 'last_post_time'
 LAST_LOGIN_TIME = 'last_login_time'
 TEXT_PLACEHOLDER = _('Type message here. Use formatting panel for more advanced usage.')
 TAGS_PLACEHOLDER = _('music images i_dont_like_tags')
 
 LABEL_TITLE = _('Title')
 LABEL_TEXT = _('Text')
 LABEL_TAG = _('Tag')
 LABEL_SEARCH = _('Search')
 
 ERROR_SPEED = 'Please wait %(delay)d second before sending message'
 ERROR_SPEED_PLURAL = 'Please wait %(delay)d seconds before sending message'
 
 TAG_MAX_LENGTH = 20
 
 TEXTAREA_ROWS = 4
 
 TRIPCODE_DELIM = '#'
 
 # TODO Maybe this may be converted into the database table?
 MIMETYPE_EXTENSIONS = {
     'image/jpeg': 'jpeg',
     'image/png': 'png',
     'image/gif': 'gif',
     'video/webm': 'webm',
     'application/pdf': 'pdf',
     'x-diff': 'diff',
     'image/svg+xml': 'svg',
+    'application/x-shockwave-flash': 'swf',
 }
 
 
 def get_timezones():
     timezones = []
     for tz in pytz.common_timezones:
         timezones.append((tz, tz),)
     return timezones
 
 
 class FormatPanel(forms.Textarea):
     """
     Panel for text formatting. Consists of buttons to add different tags to the
     form text area.
     """
 
     def render(self, name, value, attrs=None):
         output = '<div id="mark-panel">'
         for formatter in formatters:
             output += '<span class="mark_btn"' + \
                       ' onClick="addMarkToMsg(\'' + formatter.format_left + \
                       '\', \'' + formatter.format_right + '\')">' + \
                       formatter.preview_left + formatter.name + \
                       formatter.preview_right + '</span>'
 
         output += '</div>'
         output += super(FormatPanel, self).render(name, value, attrs=None)
 
         return output
 
 
 class PlainErrorList(ErrorList):
     def __unicode__(self):
         return self.as_text()
 
     def as_text(self):
         return ''.join(['(!) %s ' % e for e in self])
 
 
 class NeboardForm(forms.Form):
     """
     Form with neboard-specific formatting.
     """
 
     def as_div(self):
         """
         Returns this form rendered as HTML <as_div>s.
         """
 
         return self._html_output(
             # TODO Do not show hidden rows in the list here
             normal_row='<div class="form-row">'
                        '<div class="form-label">'
                        '%(label)s'
                        '</div>'
                        '<div class="form-input">'
                        '%(field)s'
                        '</div>'
                        '</div>'
                        '<div class="form-row">'
                        '%(help_text)s'
                        '</div>',
             error_row='<div class="form-row">'
                       '<div class="form-label"></div>'
                       '<div class="form-errors">%s</div>'
                       '</div>',
             row_ender='</div>',
             help_text_html='%s',
             errors_on_separate_row=True)
 
     def as_json_errors(self):
         errors = []
 
         for name, field in list(self.fields.items()):
             if self[name].errors:
                 errors.append({
                     'field': name,
                     'errors': self[name].errors.as_text(),
                 })
 
         return errors
 
 
 class PostForm(NeboardForm):
 
     title = forms.CharField(max_length=TITLE_MAX_LENGTH, required=False,
                             label=LABEL_TITLE,
                             widget=forms.TextInput(
                                 attrs={ATTRIBUTE_PLACEHOLDER:
                                            'test#tripcode'}))
     text = forms.CharField(
         widget=FormatPanel(attrs={
             ATTRIBUTE_PLACEHOLDER: TEXT_PLACEHOLDER,
             ATTRIBUTE_ROWS: TEXTAREA_ROWS,
         }),
         required=False, label=LABEL_TEXT)
     file = forms.FileField(required=False, label=_('File'),
                              widget=forms.ClearableFileInput(
                                  attrs={'accept': 'file/*'}))
     file_url = forms.CharField(required=False, label=_('File URL'),
                                 widget=forms.TextInput(
                                     attrs={ATTRIBUTE_PLACEHOLDER:
                                                'http://example.com/image.png'}))
 
     # This field is for spam prevention only
     email = forms.CharField(max_length=100, required=False, label=_('e-mail'),
                             widget=forms.TextInput(attrs={
                                 'class': 'form-email'}))
     threads = forms.CharField(required=False, label=_('Additional threads'),
             widget=forms.TextInput(attrs={ATTRIBUTE_PLACEHOLDER:
                 '123 456 789'}))
 
     session = None
     need_to_ban = False
 
     def _update_file_extension(self, file):
         if file:
             mimetype = get_file_mimetype(file)
             extension = MIMETYPE_EXTENSIONS.get(mimetype)
             if extension:
                 filename = file.name.split(FILE_EXTENSION_DELIMITER, 1)[0]
                 new_filename = filename + FILE_EXTENSION_DELIMITER + extension
 
                 file.name = new_filename
             else:
                 logger = logging.getLogger('boards.forms.extension')
 
                 logger.info('Unrecognized file mimetype: {}'.format(mimetype))
 
     def clean_title(self):
         title = self.cleaned_data['title']
         if title:
             if len(title) > TITLE_MAX_LENGTH:
                 raise forms.ValidationError(_('Title must have less than %s '
                                               'characters') %
                                             str(TITLE_MAX_LENGTH))
         return title
 
     def clean_text(self):
         text = self.cleaned_data['text'].strip()
         if text:
             max_length = board_settings.get_int('Forms', 'MaxTextLength')
             if len(text) > max_length:
                 raise forms.ValidationError(_('Text must have less than %s '
                                               'characters') % str(max_length))
         return text
 
     def clean_file(self):
         file = self.cleaned_data['file']
 
         if file:
             validate_file_size(file.size)
         self._update_file_extension(file)
 
         return file
 
     def clean_file_url(self):
         url = self.cleaned_data['file_url']
 
         file = None
         if url:
             file = self._get_file_from_url(url)
 
             if not file:
                 raise forms.ValidationError(_('Invalid URL'))
             else:
                 validate_file_size(file.size)
         self._update_file_extension(file)
 
         return file
 
     def clean_threads(self):
         threads_str = self.cleaned_data['threads']
 
         if len(threads_str) > 0:
             threads_id_list = threads_str.split(' ')
 
             threads = list()
 
             for thread_id in threads_id_list:
                 try:
                     thread = Post.objects.get(id=int(thread_id))
                     if not thread.is_opening() or thread.get_thread().archived:
                         raise ObjectDoesNotExist()
                     threads.append(thread)
                 except (ObjectDoesNotExist, ValueError):
                     raise forms.ValidationError(_('Invalid additional thread list'))
 
             return threads
 
     def clean(self):
         cleaned_data = super(PostForm, self).clean()
 
         if cleaned_data['email']:
             self.need_to_ban = True
             raise forms.ValidationError('A human cannot enter a hidden field')
 
         if not self.errors:
             self._clean_text_file()
 
         if not self.errors and self.session:
             self._validate_posting_speed()
 
         return cleaned_data
 
     def get_file(self):
         """
         Gets file from form or URL.
         """
 
         file = self.cleaned_data['file']
         return file or self.cleaned_data['file_url']
 
     def get_tripcode(self):
         title = self.cleaned_data['title']
         if title is not None and TRIPCODE_DELIM in title:
             code = title.split(TRIPCODE_DELIM, maxsplit=1)[1] + neboard.settings.SECRET_KEY
             tripcode = hashlib.md5(code.encode()).hexdigest()
         else:
             tripcode = ''
         return tripcode
 
     def get_title(self):
         title = self.cleaned_data['title']
         if title is not None and TRIPCODE_DELIM in title:
             return title.split(TRIPCODE_DELIM, maxsplit=1)[0]
         else:
             return title
 
     def _clean_text_file(self):
         text = self.cleaned_data.get('text')
         file = self.get_file()
 
         if (not text) and (not file):
             error_message = _('Either text or file must be entered.')
             self._errors['text'] = self.error_class([error_message])
 
     def _validate_posting_speed(self):
         can_post = True
 
         posting_delay = settings.POSTING_DELAY
 
         if board_settings.get_bool('Forms', 'LimitPostingSpeed'):
             now = time.time()
 
             current_delay = 0
 
             if LAST_POST_TIME not in self.session:
                 self.session[LAST_POST_TIME] = now
 
                 need_delay = True
             else:
                 last_post_time = self.session.get(LAST_POST_TIME)
                 current_delay = int(now - last_post_time)
 
                 need_delay = current_delay < posting_delay
 
             if need_delay:
                 delay = posting_delay - current_delay
                 error_message = ungettext_lazy(ERROR_SPEED, ERROR_SPEED_PLURAL,
                                                delay) % {'delay': delay}
                 self._errors['text'] = self.error_class([error_message])
 
                 can_post = False
 
             if can_post:
                 self.session[LAST_POST_TIME] = now
 
     def _get_file_from_url(self, url: str) -> SimpleUploadedFile:
         """
         Gets an file file from URL.
         """
 
         img_temp = None
 
         try:
             for downloader in Downloader.__subclasses__():
                 if downloader.handles(url):
                     return downloader.download(url)
             # If nobody of the specific downloaders handles this, use generic
             # one
             return Downloader.download(url)
         except forms.ValidationError as e:
             raise e
         except Exception as e:
             # Just return no file
             pass
 
 
 class ThreadForm(PostForm):
 
     tags = forms.CharField(
         widget=forms.TextInput(attrs={ATTRIBUTE_PLACEHOLDER: TAGS_PLACEHOLDER}),
         max_length=100, label=_('Tags'), required=True)
 
     def clean_tags(self):
         tags = self.cleaned_data['tags'].strip()
 
         if not tags or not REGEX_TAGS.match(tags):
             raise forms.ValidationError(
                 _('Inappropriate characters in tags.'))
 
         required_tag_exists = False
         tag_set = set()
         for tag_string in tags.split():
             tag, created = Tag.objects.get_or_create(name=tag_string.strip().lower())
             tag_set.add(tag)
 
             # If this is a new tag, don't check for its parents because nobody
             # added them yet
             if not created:
                 tag_set |= set(tag.get_all_parents())
 
         for tag in tag_set:
             if tag.required:
                 required_tag_exists = True
                 break
 
         if not required_tag_exists:
             raise forms.ValidationError(
                 _('Need at least one section.'))
 
         return tag_set
 
     def clean(self):
         cleaned_data = super(ThreadForm, self).clean()
 
         return cleaned_data
 
 
 class SettingsForm(NeboardForm):
 
     theme = forms.ChoiceField(choices=settings.THEMES, label=_('Theme'))
     image_viewer = forms.ChoiceField(choices=settings.IMAGE_VIEWERS, label=_('Image view mode'))
     username = forms.CharField(label=_('User name'), required=False)
     timezone = forms.ChoiceField(choices=get_timezones(), label=_('Time zone'))
 
     def clean_username(self):
         username = self.cleaned_data['username']
 
         if username and not REGEX_TAGS.match(username):
             raise forms.ValidationError(_('Inappropriate characters.'))
 
         return username
 
 
 class SearchForm(NeboardForm):
     query = forms.CharField(max_length=500, label=LABEL_SEARCH, required=False)

boards/models/attachment/__init__.py

@@ -1,41 +1,41 b''
 from django.db import models
 
 from boards import utils
 from boards.models.attachment.viewers import get_viewers, AbstractViewer
-from boards.utils import get_upload_filename, get_file_mimetype
+from boards.utils import get_upload_filename, get_file_mimetype, get_extension
 
 
 class AttachmentManager(models.Manager):
     def create_with_hash(self, file):
         file_hash = utils.get_file_hash(file)
         existing = self.filter(hash=file_hash)
         if len(existing) > 0:
             attachment = existing[0]
         else:
             # FIXME Use full mimetype here, need to modify viewers too
-            file_type = get_file_mimetype(file).split('/')[-1]
+            file_type = get_extension(file.name)
             attachment = self.create(file=file, mimetype=file_type,
                                      hash=file_hash)
 
         return attachment
 
 
 class Attachment(models.Model):
     objects = AttachmentManager()
 
     file = models.FileField(upload_to=get_upload_filename)
     mimetype = models.CharField(max_length=50)
     hash = models.CharField(max_length=36)
 
     def get_view(self):
         file_viewer = None
         for viewer in get_viewers():
             if viewer.supports(self.mimetype):
                 file_viewer = viewer
                 break
         if file_viewer is None:
             file_viewer = AbstractViewer
 
         return file_viewer(self.file, self.mimetype).get_view()
 
 

boards/utils.py

@@ -1,144 +1,148 b''
 """
 This module contains helper functions and helper classes.
 """
 import hashlib
 from random import random
 import time
 import hmac
 
 from django.core.cache import cache
 from django.db.models import Model
 from django import forms
 from django.utils import timezone
 from django.utils.translation import ugettext_lazy as _
 import magic
 from portage import os
 
 import boards
 from boards.settings import get_bool
 from neboard import settings
 
 CACHE_KEY_DELIMITER = '_'
 PERMISSION_MODERATE = 'moderation'
 
 HTTP_FORWARDED = 'HTTP_X_FORWARDED_FOR'
 META_REMOTE_ADDR = 'REMOTE_ADDR'
 
 SETTING_MESSAGES = 'Messages'
 SETTING_ANON_MODE = 'AnonymousMode'
 
 ANON_IP = '127.0.0.1'
 
 UPLOAD_DIRS ={
     'PostImage': 'images/',
     'Attachment': 'files/',
 }
 FILE_EXTENSION_DELIMITER = '.'
 
 
 def is_anonymous_mode():
     return get_bool(SETTING_MESSAGES, SETTING_ANON_MODE)
 
 
 def get_client_ip(request):
     if is_anonymous_mode():
         ip = ANON_IP
     else:
         x_forwarded_for = request.META.get(HTTP_FORWARDED)
         if x_forwarded_for:
             ip = x_forwarded_for.split(',')[-1].strip()
         else:
             ip = request.META.get(META_REMOTE_ADDR)
     return ip
 
 
 # TODO The output format is not epoch because it includes microseconds
 def datetime_to_epoch(datetime):
     return int(time.mktime(timezone.localtime(
         datetime,timezone.get_current_timezone()).timetuple())
                * 1000000 + datetime.microsecond)
 
 
 def get_websocket_token(user_id='', timestamp=''):
     """
     Create token to validate information provided by new connection.
     """
 
     sign = hmac.new(settings.CENTRIFUGE_PROJECT_SECRET.encode())
     sign.update(settings.CENTRIFUGE_PROJECT_ID.encode())
     sign.update(user_id.encode())
     sign.update(timestamp.encode())
     token = sign.hexdigest()
 
     return token
 
 
 def cached_result(key_method=None):
     """
     Caches method result in the Django's cache system, persisted by object name,
     object name and model id if object is a Django model.
     """
     def _cached_result(function):
         def inner_func(obj, *args, **kwargs):
             # TODO Include method arguments to the cache key
             cache_key_params = [obj.__class__.__name__, function.__name__]
             if isinstance(obj, Model):
                 cache_key_params.append(str(obj.id))
 
             if key_method is not None:
                 cache_key_params += [str(arg) for arg in key_method(obj)]
 
             cache_key = CACHE_KEY_DELIMITER.join(cache_key_params)
 
             persisted_result = cache.get(cache_key)
             if persisted_result is not None:
                 result = persisted_result
             else:
                 result = function(obj, *args, **kwargs)
                 cache.set(cache_key, result)
 
             return result
 
         return inner_func
     return _cached_result
 
 
 def is_moderator(request):
     try:
         moderate = request.user.has_perm(PERMISSION_MODERATE)
     except AttributeError:
         moderate = False
 
     return moderate
 
 
 def get_file_hash(file) -> str:
     md5 = hashlib.md5()
     for chunk in file.chunks():
         md5.update(chunk)
     return md5.hexdigest()
 
 
 def validate_file_size(size: int):
     max_size = boards.settings.get_int('Forms', 'MaxFileSize')
     if size > max_size:
         raise forms.ValidationError(
             _('File must be less than %s bytes')
             % str(max_size))
 
 
+def get_extension(filename):
+    return filename.split(FILE_EXTENSION_DELIMITER)[-1:][0]
+
+
 def get_upload_filename(model_instance, old_filename):
     # TODO Use something other than random number in file name
-    extension = old_filename.split(FILE_EXTENSION_DELIMITER)[-1:][0]
+    extension = get_extension(old_filename)
     new_name = '{}{}.{}'.format(
         str(int(time.mktime(time.gmtime()))),
         str(int(random() * 1000)),
         extension)
 
     directory = UPLOAD_DIRS[type(model_instance).__name__]
 
     return os.path.join(directory, new_name)
 
 
 def get_file_mimetype(file) -> str:
     return magic.from_buffer(file.chunks().__next__(), mime=True).decode()