Merge remote-tracking branch 'mimetype/python_magic_poc' into Develop

cps/__init__.py

@@ -56,6 +56,7 @@ except ImportError:
 mimetypes.init()
 mimetypes.add_type('application/xhtml+xml', '.xhtml')
 mimetypes.add_type('application/epub+zip', '.epub')
+mimetypes.add_type('application/epub+zip', '.kepub')
 mimetypes.add_type('application/fb2+zip', '.fb2')
 mimetypes.add_type('application/x-mobipocket-ebook', '.mobi')
 mimetypes.add_type('application/x-mobipocket-ebook', '.prc')
@@ -66,6 +67,7 @@ mimetypes.add_type('application/x-cbz', '.cbz')
 mimetypes.add_type('application/x-cbt', '.cbt')
 mimetypes.add_type('application/x-cb7', '.cb7')
 mimetypes.add_type('image/vnd.djv', '.djv')
+mimetypes.add_type('image/vnd.djv', '.djvu')
 mimetypes.add_type('application/mpeg', '.mpeg')
 mimetypes.add_type('application/mpeg', '.mp3')
 mimetypes.add_type('application/mp4', '.m4a')
@@ -73,6 +75,7 @@ mimetypes.add_type('application/mp4', '.m4b')
 mimetypes.add_type('application/ogg', '.ogg')
 mimetypes.add_type('application/ogg', '.oga')
 mimetypes.add_type('text/css', '.css')
+mimetypes.add_type('application/x-ms-reader', '.lit')
 mimetypes.add_type('text/javascript; charset=UTF-8', '.js')
 
 log = logger.create()

cps/admin.py

@@ -1780,7 +1780,7 @@ def _configuration_update_helper():
             to_save["config_upload_formats"] = ','.join(
                 helper.uniq([x.lstrip().rstrip().lower() for x in to_save["config_upload_formats"].split(',')]))
             _config_string(to_save, "config_upload_formats")
-            constants.EXTENSIONS_UPLOAD = config.config_upload_formats.split(',')
+            # constants.EXTENSIONS_UPLOAD = config.config_upload_formats.split(',')
 
         _config_string(to_save, "config_calibre")
         _config_string(to_save, "config_binariesdir")
@@ -1830,6 +1830,7 @@ def _configuration_update_helper():
         reboot_required |= reboot
 
         # security configuration
+        _config_checkbox(to_save, "config_check_extensions")
         _config_checkbox(to_save, "config_password_policy")
         _config_checkbox(to_save, "config_password_number")
         _config_checkbox(to_save, "config_password_lower")

cps/config_sql.py

@@ -169,6 +169,7 @@ class _Settings(_Base):
     config_ratelimiter = Column(Boolean, default=True)
     config_limiter_uri = Column(String, default="")
     config_limiter_options = Column(String, default="")
+    config_check_extensions = Column(Boolean, default=True)
 
     def __repr__(self):
         return self.__class__.__name__
@@ -348,7 +349,7 @@ class ConfigSQL(object):
             db_file = os.path.join(self.config_calibre_dir, 'metadata.db')
             have_metadata_db = os.path.isfile(db_file)
         self.db_configured = have_metadata_db
-        constants.EXTENSIONS_UPLOAD = [x.lstrip().rstrip().lower() for x in self.config_upload_formats.split(',')]
+        # constants.EXTENSIONS_UPLOAD = [x.lstrip().rstrip().lower() for x in self.config_upload_formats.split(',')]
         from . import cli_param
         if os.environ.get('FLASK_DEBUG'):
             logfile = logger.setup(logger.LOG_TO_STDOUT, logger.logging.DEBUG)

cps/editbooks.py

@@ -27,22 +27,7 @@ from shutil import copyfile
 from uuid import uuid4
 from markupsafe import escape, Markup  # dependency of flask
 from functools import wraps
-# from lxml.etree import ParserError
 
-#try:
-#    # at least bleach 6.0 is needed -> incomplatible change from list arguments to set arguments
-#    from bleach import clean_text as clean_html
-#    BLEACH = True
-#except ImportError:
-#    try:
-#        BLEACH = False
-#        from nh3 import clean as clean_html
-#    except ImportError:
-#        try:
-#            BLEACH = False
-#            from lxml.html.clean import clean_html
-#        except ImportError:
-#            clean_html = None
 
 from flask import Blueprint, request, flash, redirect, url_for, abort, Response
 from flask_babel import gettext as _
@@ -62,7 +47,7 @@ from .render_template import render_title_template
 from .usermanagement import login_required_if_no_ano
 from .kobo_sync_status import change_archived_books
 from .redirect import get_redirect_location
-
+from .file_helper import validate_mime_type
 
 editbook = Blueprint('edit-book', __name__)
 log = logger.create()
@@ -738,9 +723,15 @@ def create_book_on_upload(modify_date, meta):
 
 def file_handling_on_upload(requested_file):
     # check if file extension is correct
+    allowed_extensions = config.config_upload_formats.split(',')
+    if requested_file:
+        if config.config_check_extensions:
+            if not validate_mime_type(requested_file, allowed_extensions):
+                flash(_("File type isn't allowed to be uploaded to this server"), category="error")
+                return None, Response(json.dumps({"location": url_for("web.index")}), mimetype='application/json')
     if '.' in requested_file.filename:
         file_ext = requested_file.filename.rsplit('.', 1)[-1].lower()
-        if file_ext not in constants.EXTENSIONS_UPLOAD and '' not in constants.EXTENSIONS_UPLOAD:
+        if file_ext not in allowed_extensions and '' not in allowed_extensions:
             flash(
                 _("File extension '%(ext)s' is not allowed to be uploaded to this server",
                   ext=file_ext), category="error")
@@ -1191,7 +1182,12 @@ def edit_cc_data(book_id, book, to_save, cc):
 def upload_single_file(file_request, book, book_id):
     # Check and handle Uploaded file
     requested_file = file_request.files.get('btn-upload-format', None)
+    allowed_extensions = config.config_upload_formats.split(',')
     if requested_file:
+        if config.config_check_extensions:
+            if not validate_mime_type(requested_file, allowed_extensions):
+                flash(_("File type isn't allowed to be uploaded to this server"), category="error")
+                return False
         # check for empty request
         if requested_file.filename != '':
             if not current_user.role_upload():
@@ -1199,7 +1195,7 @@ def upload_single_file(file_request, book, book_id):
                 return False
             if '.' in requested_file.filename:
                 file_ext = requested_file.filename.rsplit('.', 1)[-1].lower()
-                if file_ext not in constants.EXTENSIONS_UPLOAD and '' not in constants.EXTENSIONS_UPLOAD:
+                if file_ext not in allowed_extensions and '' not in allowed_extensions:
                     flash(_("File extension '%(ext)s' is not allowed to be uploaded to this server", ext=file_ext),
                           category="error")
                     return False
@@ -1216,7 +1212,8 @@ def upload_single_file(file_request, book, book_id):
                 try:
                     os.makedirs(filepath)
                 except OSError:
-                    flash(_("Failed to create path %(path)s (Permission denied).", path=filepath), category="error")
+                    flash(_("Failed to create path %(path)s (Permission denied).", path=filepath),
+                          category="error")
                     return False
             try:
                 requested_file.save(saved_filename)

cps/file_helper.py

@@ -19,6 +19,18 @@
 from tempfile import gettempdir
 import os
 import shutil
+import zipfile
+import mimetypes
+import copy
+from io import BytesIO
+try:
+    import magic
+except ImportError:
+    pass
+
+from . import logger
+
+log = logger.create()
 
 def get_temp_dir():
     tmp_dir = os.path.join(gettempdir(), 'calibre_web')
@@ -30,3 +42,29 @@ def get_temp_dir():
 def del_temp_dir():
     tmp_dir = os.path.join(gettempdir(), 'calibre_web')
     shutil.rmtree(tmp_dir)
+
+
+def validate_mime_type(file_buffer, allowed_extensions):
+    mime = magic.Magic(mime=True)
+    allowed_mimetypes =list()
+    for x in allowed_extensions:
+        try:
+            allowed_mimetypes.append(mimetypes.types_map["." + x])
+        except KeyError as e:
+            log.error("Unkown mimetype for Extension: {}".format(x))
+    tmp_mime_type = mime.from_buffer(file_buffer.read())
+    file_buffer.seek(0)
+    if any(mime_type in tmp_mime_type for mime_type in allowed_mimetypes):
+        return True
+    # Some epubs show up as zip mimetypes
+    elif "zip" in tmp_mime_type:
+        try:
+            with zipfile.ZipFile(BytesIO(file_buffer.read()), 'r') as epub:
+                file_buffer.seek(0)
+                if "mimetype" in epub.namelist():
+                    return True
+        except:
+            file_buffer.seek(0)
+            pass
+    
+    return False

cps/render_template.py

@@ -112,7 +112,7 @@ def render_title_template(*args, **kwargs):
     sidebar, simple = get_sidebar_config(kwargs)
     try:
         return render_template(instance=config.config_calibre_web_title, sidebar=sidebar, simple=simple,
-                               accept=constants.EXTENSIONS_UPLOAD,
+                               accept=config.config_upload_formats.split(','),
                                *args, **kwargs)
     except PermissionError:
         log.error("No permission to access {} file.".format(args[0]))

cps/templates/config_edit.html

@@ -377,6 +377,10 @@
                   <input type="text" class="form-control" id="config_limiter_options" name="config_limiter_options" value="{% if config.config_limiter_options != None %}{{ config.config_limiter_options }}{% endif %}" autocomplete="off">
                </div>
             </div>
+            <div class="form-group">
+                <input type="checkbox" id="config_check_extensions" name="config_check_extensions" {% if config.config_check_extensions %}checked{% endif %}>
+                <label for="config_check_extensions">{{_('Check if file extensions matches file content on upload')}}</label>
+            </div>
             <div class="form-group">
               <label for="config_session">{{_('Session protection')}}</label>
                 <select name="config_session" id="config_session" class="form-control">

cps/uploader.py

@@ -23,7 +23,7 @@ from flask_babel import gettext as _
 from . import logger, comic, isoLanguages
 from .constants import BookMeta
 from .helper import split_authors
-from .file_helper import get_temp_dir
+from .file_helper import get_temp_dir, validate_mime_type
 
 log = logger.create()
 

cps/web.py

@@ -1582,7 +1582,8 @@ def read_book(book_id, book_format):
         return render_title_template('readtxt.html', txtfile=book_id, title=book.title)
     elif book_format.lower() in ["djvu", "djv"]:
         log.debug("Start djvu reader for %d", book_id)
-        return render_title_template('readdjvu.html', djvufile=book_id, title=book.title, extension=book_format.lower())
+        return render_title_template('readdjvu.html', djvufile=book_id, title=book.title,
+                                     extension=book_format.lower())
     else:
         for fileExt in constants.EXTENSIONS_AUDIO:
             if book_format.lower() == fileExt:

requirements.txt

@@ -19,3 +19,4 @@ chardet>=3.0.0,<4.1.0
 advocate>=1.0.0,<1.1.0
 Flask-Limiter>=2.3.0,<3.6.0
 regex>=2022.3.2,<2024.2.25
+python-magic>=0.4.27,<0.5.0