Fixes bug1073931 "Corrupted databases stop OpenLP from starting"

Checks if the database session is available before trying to use it.
Use a sha256 hash to verify downloaded files. See also: https://code.launchpad.net/~phill-ridout/openlp/sha256

bzr-revno: 2500
This commit is contained in:
Philip Ridout 2015-02-17 05:47:19 +00:00 committed by Tim Bentley
commit 16758d1b80
5 changed files with 117 additions and 36 deletions

View File

@ -60,6 +60,35 @@ def init_db(url, auto_flush=True, auto_commit=False, base=None):
return session, metadata
def get_db_path(plugin_name, db_file_name=None):
"""
Create a path to a database from the plugin name and database name
:param plugin_name: Name of plugin
:param db_file_name: File name of database
:return: The path to the database as type str
"""
if db_file_name is None:
return 'sqlite:///%s/%s.sqlite' % (AppLocation.get_section_data_path(plugin_name), plugin_name)
else:
return 'sqlite:///%s/%s' % (AppLocation.get_section_data_path(plugin_name), db_file_name)
def handle_db_error(plugin_name, db_file_name):
"""
Log and report to the user that a database cannot be loaded
:param plugin_name: Name of plugin
:param db_file_name: File name of database
:return: None
"""
db_path = get_db_path(plugin_name, db_file_name)
log.exception('Error loading database: %s', db_path)
critical_error_message_box(translate('OpenLP.Manager', 'Database Error'),
translate('OpenLP.Manager', 'OpenLP cannot load your database.\n\nDatabase: %s')
% db_path)
def init_url(plugin_name, db_file_name=None):
"""
Return the database URL.
@ -69,13 +98,9 @@ def init_url(plugin_name, db_file_name=None):
"""
settings = Settings()
settings.beginGroup(plugin_name)
db_url = ''
db_type = settings.value('db type')
if db_type == 'sqlite':
if db_file_name is None:
db_url = 'sqlite:///%s/%s.sqlite' % (AppLocation.get_section_data_path(plugin_name), plugin_name)
else:
db_url = 'sqlite:///%s/%s' % (AppLocation.get_section_data_path(plugin_name), db_file_name)
db_url = get_db_path(plugin_name, db_file_name)
else:
db_url = '%s://%s:%s@%s/%s' % (db_type, urlquote(settings.value('db username')),
urlquote(settings.value('db password')),
@ -212,7 +237,7 @@ class Manager(object):
try:
db_ver, up_ver = upgrade_db(self.db_url, upgrade_mod)
except (SQLAlchemyError, DBAPIError):
log.exception('Error loading database: %s', self.db_url)
handle_db_error(plugin_name, db_file_name)
return
if db_ver > up_ver:
critical_error_message_box(
@ -225,10 +250,7 @@ class Manager(object):
try:
self.session = init_schema(self.db_url)
except (SQLAlchemyError, DBAPIError):
log.exception('Error loading database: %s', self.db_url)
critical_error_message_box(translate('OpenLP.Manager', 'Database Error'),
translate('OpenLP.Manager', 'OpenLP cannot load your database.\n\nDatabase: %s')
% self.db_url)
handle_db_error(plugin_name, db_file_name)
def save_object(self, object_instance, commit=True):
"""

View File

@ -22,6 +22,7 @@
"""
This module contains the first time wizard.
"""
import hashlib
import logging
import os
import time
@ -47,10 +48,10 @@ class ThemeScreenshotWorker(QtCore.QObject):
"""
This thread downloads a theme's screenshot
"""
screenshot_downloaded = QtCore.pyqtSignal(str, str)
screenshot_downloaded = QtCore.pyqtSignal(str, str, str)
finished = QtCore.pyqtSignal()
def __init__(self, themes_url, title, filename, screenshot):
def __init__(self, themes_url, title, filename, sha256, screenshot):
"""
Set up the worker object
"""
@ -58,6 +59,7 @@ class ThemeScreenshotWorker(QtCore.QObject):
self.themes_url = themes_url
self.title = title
self.filename = filename
self.sha256 = sha256
self.screenshot = screenshot
super(ThemeScreenshotWorker, self).__init__()
@ -71,7 +73,7 @@ class ThemeScreenshotWorker(QtCore.QObject):
urllib.request.urlretrieve('%s%s' % (self.themes_url, self.screenshot),
os.path.join(gettempdir(), 'openlp', self.screenshot))
# Signal that the screenshot has been downloaded
self.screenshot_downloaded.emit(self.title, self.filename)
self.screenshot_downloaded.emit(self.title, self.filename, self.sha256)
except:
log.exception('Unable to download screenshot')
finally:
@ -221,8 +223,9 @@ class FirstTimeForm(QtGui.QWizard, UiFirstTimeWizard, RegistryProperties):
self.application.process_events()
title = self.config.get('songs_%s' % song, 'title')
filename = self.config.get('songs_%s' % song, 'filename')
sha256 = self.config.get('songs_%s' % song, 'sha256', fallback='')
item = QtGui.QListWidgetItem(title, self.songs_list_widget)
item.setData(QtCore.Qt.UserRole, filename)
item.setData(QtCore.Qt.UserRole, (filename, sha256))
item.setCheckState(QtCore.Qt.Unchecked)
item.setFlags(item.flags() | QtCore.Qt.ItemIsUserCheckable)
bible_languages = self.config.get('bibles', 'languages')
@ -237,8 +240,9 @@ class FirstTimeForm(QtGui.QWizard, UiFirstTimeWizard, RegistryProperties):
self.application.process_events()
title = self.config.get('bible_%s' % bible, 'title')
filename = self.config.get('bible_%s' % bible, 'filename')
sha256 = self.config.get('bible_%s' % bible, 'sha256', fallback='')
item = QtGui.QTreeWidgetItem(lang_item, [title])
item.setData(0, QtCore.Qt.UserRole, filename)
item.setData(0, QtCore.Qt.UserRole, (filename, sha256))
item.setCheckState(0, QtCore.Qt.Unchecked)
item.setFlags(item.flags() | QtCore.Qt.ItemIsUserCheckable)
self.bibles_tree_widget.expandAll()
@ -249,8 +253,9 @@ class FirstTimeForm(QtGui.QWizard, UiFirstTimeWizard, RegistryProperties):
self.application.process_events()
title = self.config.get('theme_%s' % theme, 'title')
filename = self.config.get('theme_%s' % theme, 'filename')
sha256 = self.config.get('theme_%s' % theme, 'sha256', fallback='')
screenshot = self.config.get('theme_%s' % theme, 'screenshot')
worker = ThemeScreenshotWorker(self.themes_url, title, filename, screenshot)
worker = ThemeScreenshotWorker(self.themes_url, title, filename, sha256, screenshot)
self.theme_screenshot_workers.append(worker)
worker.screenshot_downloaded.connect(self.on_screenshot_downloaded)
thread = QtCore.QThread(self)
@ -356,7 +361,7 @@ class FirstTimeForm(QtGui.QWizard, UiFirstTimeWizard, RegistryProperties):
time.sleep(0.1)
self.application.set_normal_cursor()
def on_screenshot_downloaded(self, title, filename):
def on_screenshot_downloaded(self, title, filename, sha256):
"""
Add an item to the list when a theme has been downloaded
@ -364,7 +369,7 @@ class FirstTimeForm(QtGui.QWizard, UiFirstTimeWizard, RegistryProperties):
:param filename: The filename of the theme
"""
item = QtGui.QListWidgetItem(title, self.themes_list_widget)
item.setData(QtCore.Qt.UserRole, filename)
item.setData(QtCore.Qt.UserRole, (filename, sha256))
item.setCheckState(QtCore.Qt.Unchecked)
item.setFlags(item.flags() | QtCore.Qt.ItemIsUserCheckable)
@ -385,7 +390,7 @@ class FirstTimeForm(QtGui.QWizard, UiFirstTimeWizard, RegistryProperties):
self.was_cancelled = True
self.close()
def url_get_file(self, url, f_path):
def url_get_file(self, url, f_path, sha256=None):
""""
Download a file given a URL. The file is retrieved in chunks, giving the ability to cancel the download at any
point. Returns False on download error.
@ -400,16 +405,24 @@ class FirstTimeForm(QtGui.QWizard, UiFirstTimeWizard, RegistryProperties):
try:
url_file = urllib.request.urlopen(url, timeout=CONNECTION_TIMEOUT)
filename = open(f_path, "wb")
if sha256:
hasher = hashlib.sha256()
# Download until finished or canceled.
while not self.was_cancelled:
data = url_file.read(block_size)
if not data:
break
filename.write(data)
if sha256:
hasher.update(data)
block_count += 1
self._download_progress(block_count, block_size)
filename.close()
except ConnectionError:
if sha256 and hasher.hexdigest() != sha256:
log.error('sha256 sums did not match for file: {}'.format(f_path))
os.remove(f_path)
return False
except urllib.error.URLError:
trace_error_handler(log)
filename.close()
os.remove(f_path)
@ -449,7 +462,7 @@ class FirstTimeForm(QtGui.QWizard, UiFirstTimeWizard, RegistryProperties):
site = urllib.request.urlopen(url, timeout=CONNECTION_TIMEOUT)
meta = site.info()
return int(meta.get("Content-Length"))
except ConnectionException:
except urllib.error.URLError:
if retries > CONNECTION_RETRIES:
raise
else:
@ -491,7 +504,7 @@ class FirstTimeForm(QtGui.QWizard, UiFirstTimeWizard, RegistryProperties):
self.application.process_events()
item = self.songs_list_widget.item(i)
if item.checkState() == QtCore.Qt.Checked:
filename = item.data(QtCore.Qt.UserRole)
filename, sha256 = item.data(QtCore.Qt.UserRole)
size = self._get_file_size('%s%s' % (self.songs_url, filename))
self.max_progress += size
# Loop through the Bibles list and increase for each selected item
@ -500,7 +513,7 @@ class FirstTimeForm(QtGui.QWizard, UiFirstTimeWizard, RegistryProperties):
self.application.process_events()
item = iterator.value()
if item.parent() and item.checkState(0) == QtCore.Qt.Checked:
filename = item.data(0, QtCore.Qt.UserRole)
filename, sha256 = item.data(0, QtCore.Qt.UserRole)
size = self._get_file_size('%s%s' % (self.bibles_url, filename))
self.max_progress += size
iterator += 1
@ -509,10 +522,10 @@ class FirstTimeForm(QtGui.QWizard, UiFirstTimeWizard, RegistryProperties):
self.application.process_events()
item = self.themes_list_widget.item(i)
if item.checkState() == QtCore.Qt.Checked:
filename = item.data(QtCore.Qt.UserRole)
filename, sha256 = item.data(QtCore.Qt.UserRole)
size = self._get_file_size('%s%s' % (self.themes_url, filename))
self.max_progress += size
except ConnectionError:
except urllib.error.URLError:
trace_error_handler(log)
critical_error_message_box(translate('OpenLP.FirstTimeWizard', 'Download Error'),
translate('OpenLP.FirstTimeWizard', 'There was a connection problem during '
@ -608,31 +621,33 @@ class FirstTimeForm(QtGui.QWizard, UiFirstTimeWizard, RegistryProperties):
for i in range(self.songs_list_widget.count()):
item = self.songs_list_widget.item(i)
if item.checkState() == QtCore.Qt.Checked:
filename = item.data(QtCore.Qt.UserRole)
filename, sha256 = item.data(QtCore.Qt.UserRole)
self._increment_progress_bar(self.downloading % filename, 0)
self.previous_size = 0
destination = os.path.join(songs_destination, str(filename))
if not self.url_get_file('%s%s' % (self.songs_url, filename), destination):
if not self.url_get_file('%s%s' % (self.songs_url, filename), destination, sha256):
return False
# Download Bibles
bibles_iterator = QtGui.QTreeWidgetItemIterator(self.bibles_tree_widget)
while bibles_iterator.value():
item = bibles_iterator.value()
if item.parent() and item.checkState(0) == QtCore.Qt.Checked:
bible = item.data(0, QtCore.Qt.UserRole)
bible, sha256 = item.data(0, QtCore.Qt.UserRole)
self._increment_progress_bar(self.downloading % bible, 0)
self.previous_size = 0
if not self.url_get_file('%s%s' % (self.bibles_url, bible), os.path.join(bibles_destination, bible)):
if not self.url_get_file('%s%s' % (self.bibles_url, bible), os.path.join(bibles_destination, bible),
sha256):
return False
bibles_iterator += 1
# Download themes
for i in range(self.themes_list_widget.count()):
item = self.themes_list_widget.item(i)
if item.checkState() == QtCore.Qt.Checked:
theme = item.data(QtCore.Qt.UserRole)
theme, sha256 = item.data(QtCore.Qt.UserRole)
self._increment_progress_bar(self.downloading % theme, 0)
self.previous_size = 0
if not self.url_get_file('%s%s' % (self.themes_url, theme), os.path.join(themes_destination, theme)):
if not self.url_get_file('%s%s' % (self.themes_url, theme), os.path.join(themes_destination, theme),
sha256):
return False
return True

View File

@ -131,6 +131,7 @@ class BibleDB(QtCore.QObject, Manager, RegistryProperties):
log.info('BibleDB loaded')
QtCore.QObject.__init__(self)
self.bible_plugin = parent
self.session = None
if 'path' not in kwargs:
raise KeyError('Missing keyword argument "path".')
if 'name' not in kwargs and 'file' not in kwargs:
@ -144,8 +145,8 @@ class BibleDB(QtCore.QObject, Manager, RegistryProperties):
if 'file' in kwargs:
self.file = kwargs['file']
Manager.__init__(self, 'bibles', init_schema, self.file, upgrade)
if 'file' in kwargs:
self.get_name()
if self.session and 'file' in kwargs:
self.get_name()
if 'path' in kwargs:
self.path = kwargs['path']
self.wizard = None

View File

@ -121,6 +121,8 @@ class BibleManager(RegistryProperties):
self.old_bible_databases = []
for filename in files:
bible = BibleDB(self.parent, path=self.path, file=filename)
if not bible.session:
continue
name = bible.get_name()
# Remove corrupted files.
if name is None:

View File

@ -125,7 +125,7 @@ class TestInit(TestCase, TestMixin):
# WHEN: Calling parse_options
results = parse_options(options)
# THEN: A tuple should be returned with the parsed options and left over args
# THEN: A tuple should be returned with the parsed options and left over options
self.assertEqual(results, (Values({'no_error_form': True, 'dev_version': True, 'portable': True,
'style': 'style', 'loglevel': 'debug'}), ['extra', 'qt', 'args']))
@ -136,10 +136,51 @@ class TestInit(TestCase, TestMixin):
# GIVEN: A list of valid options
options = ['openlp.py', '-e', '-l', 'debug', '-pd', '-s', 'style', 'extra', 'qt', 'args']
# WHEN: Passing in the options through sys.argv and calling parse_args with None
# WHEN: Passing in the options through sys.argv and calling parse_options with None
with patch.object(sys, 'argv', options):
results = parse_options(None)
# THEN: parse_args should return a tuple of valid options and of left over options that OpenLP does not use
# THEN: parse_options should return a tuple of valid options and of left over options that OpenLP does not use
self.assertEqual(results, (Values({'no_error_form': True, 'dev_version': True, 'portable': True,
'style': 'style', 'loglevel': 'debug'}), ['extra', 'qt', 'args']))
def test_parse_options_valid_long_options(self):
"""
Test that parse_options parses valid long options correctly
"""
# GIVEN: A list of valid options
options = ['--no-error-form', 'extra', '--log-level', 'debug', 'qt', '--portable', '--dev-version', 'args',
'--style=style']
# WHEN: Calling parse_options
results = parse_options(options)
# THEN: parse_options should return a tuple of valid options and of left over options that OpenLP does not use
self.assertEqual(results, (Values({'no_error_form': True, 'dev_version': True, 'portable': True,
'style': 'style', 'loglevel': 'debug'}), ['extra', 'qt', 'args']))
def test_parse_options_help_option(self):
"""
Test that parse_options raises an SystemExit exception when called with invalid options
"""
# GIVEN: The --help option
options = ['--help']
# WHEN: Calling parse_options
# THEN: parse_options should raise an SystemExit exception with exception code 0
with self.assertRaises(SystemExit) as raised_exception:
parse_options(options)
self.assertEqual(raised_exception.exception.code, 0)
def test_parse_options_invalid_option(self):
"""
Test that parse_options raises an SystemExit exception when called with invalid options
"""
# GIVEN: A list including invalid options
options = ['-t']
# WHEN: Calling parse_options
# THEN: parse_options should raise an SystemExit exception with exception code 2
with self.assertRaises(SystemExit) as raised_exception:
parse_options(options)
self.assertEqual(raised_exception.exception.code, 2)