input validation for calibre binary directory

cps/admin.py

@@ -1724,6 +1724,10 @@ def _configuration_update_helper():
         _config_string(to_save, "config_binariesdir")
         _config_string(to_save, "config_converterpath")
         _config_string(to_save, "config_kepubifypath")
+        if "config_binariesdir" in to_save:
+            calibre_status = helper.check_calibre(config.config_binariesdir)
+            if calibre_status:
+                return _configuration_result(calibre_status)
 
         reboot_required |= _config_int(to_save, "config_login_type")
 

cps/config_sql.py

@@ -30,6 +30,7 @@ except ImportError:
     from sqlalchemy.ext.declarative import declarative_base
 
 from . import constants, logger
+from .subproc_wrapper import process_wait
 
 
 log = logger.create()
@@ -274,13 +275,8 @@ class _ConfigSQL(object):
     def get_calibre_binarypath(self, binary):
         binariesdir = self.config_binariesdir
         if binariesdir:
-            # TODO: Need to make sure that all supported calibre binaries are actually in the specified directory when set via UI
-            if sys.platform == "win32":
-                extension = ".exe"
-            else:
-                extension = ""
             if binary in constants.SUPPORTED_CALIBRE_BINARIES:
-                return os.path.join(binariesdir, binary + extension)
+                return os.path.join(binariesdir, binary)
             else:
                 # TODO: Error handling
                 pass
@@ -429,18 +425,20 @@ def _migrate_table(session, orm_class):
 
 def autodetect_calibre_binaries():
     if sys.platform == "win32":
-        extension = ".exe"
         calibre_path = ["C:\\program files\\calibre\\",
                         "C:\\program files(x86)\\calibre\\",
                         "C:\\program files(x86)\\calibre2\\",
                         "C:\\program files\\calibre2\\"]
     else:
-        extension = ""
         calibre_path = ["/opt/calibre/"]
     for element in calibre_path:
-        supported_binary_paths = [os.path.join(element, binary + extension) for binary in constants.SUPPORTED_CALIBRE_BINARIES]
+        supported_binary_paths = [os.path.join(element, binary) for binary in constants.SUPPORTED_CALIBRE_BINARIES]
         if all(os.path.isfile(binary_path) and os.access(binary_path, os.X_OK) for binary_path in supported_binary_paths):
-            return element 
+            values = [process_wait([binary_path, "--version"], pattern='\(calibre (.*)\)') for binary_path in supported_binary_paths]
+            if all(values):
+                version = values[0].group(1)
+                log.debug("calibre version %s", version)
+                return element 
     return ""
 
 

cps/constants.py

@@ -151,8 +151,10 @@ EXTENSIONS_UPLOAD = {'txt', 'pdf', 'epub', 'kepub', 'mobi', 'azw', 'azw3', 'cbr'
                      'prc', 'doc', 'docx', 'fb2', 'html', 'rtf', 'lit', 'odt', 'mp3', 'mp4', 'ogg',
                      'opus', 'wav', 'flac', 'm4a', 'm4b'}
 
-
-SUPPORTED_CALIBRE_BINARIES = ["ebook-convert", "calibredb"]
+_extension = ""
+if sys.platform == "win32":
+    _extension = ".exe"
+SUPPORTED_CALIBRE_BINARIES = [binary + _extension for binary in ["ebook-convert", "calibredb"]]
 
 
 def has_flag(value, bit_flag):

cps/helper.py

@@ -53,7 +53,7 @@ from . import calibre_db, cli_param
 from .tasks.convert import TaskConvert
 from . import logger, config, db, ub, fs
 from . import gdriveutils as gd
-from .constants import STATIC_DIR as _STATIC_DIR, CACHE_TYPE_THUMBNAILS, THUMBNAIL_TYPE_COVER, THUMBNAIL_TYPE_SERIES
+from .constants import STATIC_DIR as _STATIC_DIR, CACHE_TYPE_THUMBNAILS, THUMBNAIL_TYPE_COVER, THUMBNAIL_TYPE_SERIES, SUPPORTED_CALIBRE_BINARIES
 from .subproc_wrapper import process_wait
 from .services.worker import WorkerThread
 from .tasks.mail import TaskEmail
@@ -940,6 +940,33 @@ def check_unrar(unrar_location):
         return _('Error excecuting UnRar')
 
 
+def check_calibre(calibre_location):
+    if not calibre_location:
+        return
+
+    if not os.path.exists(calibre_location):
+        return _('Could not find the specified directory')
+
+    if not os.path.isdir(calibre_location):
+        return _('Please specify a directory, not a file')
+
+    try:
+        supported_binary_paths = [os.path.join(calibre_location, binary) for binary in SUPPORTED_CALIBRE_BINARIES]
+        if all(os.path.isfile(binary_path) and os.access(binary_path, os.X_OK) for binary_path in supported_binary_paths):
+            values = [process_wait([binary_path, "--version"], pattern='\(calibre (.*)\)') for binary_path in supported_binary_paths]
+            if all(values):
+                version = values[0].group(1)
+                log.debug("calibre version %s", version)
+            else:
+                return _('Calibre binaries not viable')
+        else:
+            return _('Missing calibre binaries in the specified directory')
+
+    except (OSError, UnicodeDecodeError) as err:
+        log.error_or_exception(err)
+        return _('Error excecuting Calibre')
+
+
 def json_serial(obj):
     """JSON serializer for objects not serializable by default json code"""