Try all chunks when detecting file mimetype

boards/models/attachment/viewers.py

@@ -1,239 +1,240 b''
 import re
 
 from PIL import Image
 
 from django.contrib.staticfiles import finders
 from django.contrib.staticfiles.templatetags.staticfiles import static
 from django.core.files.images import get_image_dimensions
 from django.template.defaultfilters import filesizeformat
 from django.core.urlresolvers import reverse
 from django.utils.translation import ugettext_lazy as _, ungettext_lazy
 
 from boards.utils import get_domain, cached_result
 from boards import settings
 
 
 FILE_STUB_IMAGE = 'images/file.png'
 FILE_STUB_URL = 'url'
 FILE_FILEFORMAT = 'images/fileformats/{}.png'
 
 
 FILE_TYPES_VIDEO = (
     'video/webm',
     'video/mp4',
     'video/mpeg',
     'video/ogv',
 )
 FILE_TYPE_SVG = 'image/svg'
 FILE_TYPES_AUDIO = (
     'audio/ogg',
     'audio/mp3',
     'audio/opus',
     'audio/x-flac',
+    'audio/mpeg',
 )
 FILE_TYPES_IMAGE = (
     'image/jpeg',
     'image/jpg',
     'image/png',
     'image/bmp',
     'image/gif',
 )
 
 PLAIN_FILE_FORMATS = {
     'application/zip': 'archive',
     'application/x-tar': 'archive',
     'gz': 'archive',
     'audio/midi' : 'midi',
 }
 
 URL_PROTOCOLS = {
     'magnet': 'magnet',
 }
 
 CSS_CLASS_IMAGE = 'image'
 CSS_CLASS_THUMB = 'thumb'
 
 ABSTRACT_VIEW = '<div class="image">'\
         '{}'\
         '<div class="image-metadata"><a href="{}" download >{}, {}</a>'\
         ' <a class="file-menu" href="#" data-type="{}" data-search-url="{}" data-filename="{}">🔍 </a></div>'\
         '</div>'
 URL_VIEW = '<div class="image">' \
         '{}' \
         '<div class="image-metadata">{}</div>' \
         '</div>'
 ABSTRACT_FORMAT_VIEW = '<a href="{}">'\
         '<img class="url-image" src="{}" width="{}" height="{}"/>'\
         '</a>'
 VIDEO_FORMAT_VIEW = '<video width="200" height="150" controls src="{}"></video>'
 AUDIO_FORMAT_VIEW = '<audio controls src="{}"></audio>'
 IMAGE_FORMAT_VIEW = '<a class="{}" href="{full}">' \
         '<img class="post-image-preview"' \
         ' src="{}"' \
         ' alt="{}"' \
         ' width="{}"' \
         ' height="{}"' \
         ' data-width="{}"' \
         ' data-height="{}" />' \
         '</a>'
 SVG_FORMAT_VIEW = '<a class="thumb" href="{}">'\
         '<img class="post-image-preview" width="200" height="150" src="{}" />'\
         '</a>'
 URL_FORMAT_VIEW = '<a href="{}">' \
         '<img class="url-image" src="{}" width="{}" height="{}"/>' \
         '</a>'
 
 
 def get_viewers():
     return AbstractViewer.__subclasses__()
 
 
 def get_static_dimensions(filename):
     file_path = finders.find(filename)
     return get_image_dimensions(file_path)
 
 
 # TODO Move this to utils
 def file_exists(filename):
     return finders.find(filename) is not None
 
 
 class AbstractViewer:
     def __init__(self, file, file_type, hash, url):
         self.file = file
         self.file_type = file_type
         self.hash = hash
         self.url = url
 
     @staticmethod
     def supports(file_type):
         return True
 
     def get_view(self):
         search_host = settings.get('External', 'ImageSearchHost')
         if search_host:
             if search_host.endswith('/'):
                 search_host = search_host[:-1]
             search_url = search_host + self.file.url
         else:
             search_url = ''
 
         return ABSTRACT_VIEW.format(self.get_format_view(), self.file.url,
                 self.file_type, filesizeformat(self.file.size),
                 self.file_type, search_url, self.file.name)
 
     def get_format_view(self):
         image_name = PLAIN_FILE_FORMATS.get(self.file_type, self.file_type)
         file_name = FILE_FILEFORMAT.format(image_name)
 
         if file_exists(file_name):
             image = file_name
         else:
             image = FILE_STUB_IMAGE
 
         w, h = get_static_dimensions(image)
 
         return ABSTRACT_FORMAT_VIEW.format(self.file.url, static(image), w, h)
 
 
 class VideoViewer(AbstractViewer):
     @staticmethod
     def supports(file_type):
         return file_type in FILE_TYPES_VIDEO
 
     def get_format_view(self):
         return VIDEO_FORMAT_VIEW.format(self.file.url)
 
 
 class AudioViewer(AbstractViewer):
     @staticmethod
     def supports(file_type):
         return file_type in FILE_TYPES_AUDIO
 
     def get_format_view(self):
         return AUDIO_FORMAT_VIEW.format(self.file.url)
 
 
 class SvgViewer(AbstractViewer):
     @staticmethod
     def supports(file_type):
         return file_type == FILE_TYPE_SVG
 
     def get_format_view(self):
         return SVG_FORMAT_VIEW.format(self.file.url, self.file.url)
 
 
 class ImageViewer(AbstractViewer):
     @staticmethod
     def supports(file_type):
         return file_type in FILE_TYPES_IMAGE
 
     def get_format_view(self):
         metadata = '{}, {}'.format(self.file.name.split('.')[-1],
                                    filesizeformat(self.file.size))
 
         try:
             width, height = get_image_dimensions(self.file.path)
         except Exception:
             # If the image is a decompression bomb, treat it as just a regular
             # file
             return super().get_format_view()
 
         preview_path = self.file.path.replace('.', '.200x150.')
         pre_width, pre_height = get_image_dimensions(preview_path)
 
         split = self.file.url.rsplit('.', 1)
         w, h = 200, 150
         thumb_url = '%s.%sx%s.%s' % (split[0], w, h, split[1])
 
         return IMAGE_FORMAT_VIEW.format(CSS_CLASS_THUMB,
                 thumb_url,
                 self.hash,
                 str(pre_width),
                 str(pre_height), str(width), str(height),
                 full=self.file.url, image_meta=metadata)
 
 
 class UrlViewer(AbstractViewer):
     @staticmethod
     def supports(file_type):
         return file_type is None
 
     def get_view(self):
         return URL_VIEW.format(self.get_format_view(), get_domain(self.url))
 
     def get_format_view(self):
         protocol = self.url.split(':')[0]
 
         domain = get_domain(self.url)
 
         if protocol in URL_PROTOCOLS:
             url_image_name = URL_PROTOCOLS.get(protocol)
         elif domain:
             url_image_name = self._find_image_for_domains(domain) or FILE_STUB_URL
         else:
             url_image_name = FILE_STUB_URL
 
         image_path = 'images/{}.png'.format(url_image_name)
         image = static(image_path)
         w, h = get_static_dimensions(image_path)
 
         return URL_FORMAT_VIEW.format(self.url, image, w, h)
 
     @cached_result()
     def _find_image_for_domains(self, domain):
         """
         Searches for the domain image for every domain level except top.
         E.g. for l3.example.co.uk it will search for l3.example.co.uk, then
         example.co.uk, then co.uk
         """
         levels = domain.split('.')
         while len(levels) > 1:
             domain = '.'.join(levels)
 
             filename = 'images/domains/{}.png'.format(domain)
             if file_exists(filename):
                 return 'domains/' + domain
             else:
                 del levels[0]
 

boards/utils.py

@@ -1,157 +1,157 b''
 """
 This module contains helper functions and helper classes.
 """
 import hashlib
 import uuid
 
 from boards.abstracts.constants import FILE_DIRECTORY
 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.template.defaultfilters import filesizeformat
 from django.utils import timezone
 from django.utils.translation import ugettext_lazy as _
 import magic
 import os
 
 import boards
 from boards.settings import get_bool
 from neboard import settings
 
 CACHE_KEY_DELIMITER = '_'
 
 HTTP_FORWARDED = 'HTTP_X_FORWARDED_FOR'
 META_REMOTE_ADDR = 'REMOTE_ADDR'
 
 SETTING_MESSAGES = 'Messages'
 SETTING_ANON_MODE = 'AnonymousMode'
 
 ANON_IP = '127.0.0.1'
 
 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
 
 
 # TODO Test this carefully
 def cached_result(key_method=None):
     """
     Caches method result in the Django's cache system, persisted by object name,
     object name, model id if object is a Django model, args and kwargs if any.
     """
     def _cached_result(function):
         def inner_func(obj, *args, **kwargs):
             cache_key_params = [obj.__class__.__name__, function.__name__]
 
             cache_key_params += args
             for key, value in kwargs:
                 cache_key_params.append(key + ':' + value)
 
             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)
                 if result is not None:
                     cache.set(cache_key, result)
 
             return result
 
         return inner_func
     return _cached_result
 
 
 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 max_size > 0 and size > max_size:
         raise forms.ValidationError(
             _('File must be less than %s but is %s.')
             % (filesizeformat(max_size), filesizeformat(size)))
 
 
 def get_extension(filename):
     return filename.split(FILE_EXTENSION_DELIMITER)[-1:][0]
 
 
 def get_upload_filename(model_instance, old_filename):
     extension = get_extension(old_filename)
     new_name = '{}.{}'.format(uuid.uuid4(), extension)
 
     return os.path.join(FILE_DIRECTORY, new_name)
 
 
 def get_file_mimetype(file) -> str:
-    file_type = magic.from_buffer(file.chunks().__next__(), mime=True)
+    file_type = magic.from_file(settings.MEDIA_ROOT + file.name, mime=True)
     if file_type is None:
         file_type = 'application/octet-stream'
     elif type(file_type) == bytes:
         file_type = file_type.decode()
     return file_type
 
 
 def get_domain(url: str) -> str:
     """
     Gets domain from an URL with random number of domain levels.
     """
     domain_parts = url.split('/')
     if len(domain_parts) >= 2:
         full_domain = domain_parts[2]
     else:
         full_domain = ''
 
     return full_domain