89 lines
3 KiB
Python
Executable file
89 lines
3 KiB
Python
Executable file
import sqlite3
|
|
from pathlib import Path
|
|
|
|
|
|
class GuiStorageDatabaseError(Exception):
|
|
def __init__(self, message, detail=None):
|
|
self.detail = detail
|
|
super().__init__(f"{message} Details: {detail}" if detail else message)
|
|
|
|
|
|
def _connect_readonly(database_path):
|
|
path = Path(database_path).resolve()
|
|
return sqlite3.connect(f"{path.as_uri()}?mode=ro", uri=True)
|
|
|
|
|
|
def validate_storage_database(database_path):
|
|
path = Path(database_path)
|
|
if not path.exists():
|
|
return
|
|
|
|
connection = None
|
|
try:
|
|
connection = _connect_readonly(path)
|
|
connection.execute("PRAGMA schema_version").fetchone()
|
|
result = connection.execute("PRAGMA quick_check").fetchone()
|
|
if result and result[0] != "ok":
|
|
raise GuiStorageDatabaseError(
|
|
"The local storage database failed SQLite integrity checks.",
|
|
str(result[0]),
|
|
)
|
|
table_rows = connection.execute("SELECT name FROM sqlite_master WHERE type = 'table'").fetchall()
|
|
table_names = {row[0] for row in table_rows}
|
|
known_tables = {
|
|
"applications",
|
|
"application_versions",
|
|
"cached_sync",
|
|
"client_versions",
|
|
"database_version",
|
|
"encryptedproxies",
|
|
"locations",
|
|
"operators",
|
|
"subscription_plans",
|
|
}
|
|
if table_names and table_names.isdisjoint(known_tables):
|
|
raise GuiStorageDatabaseError(
|
|
"The local storage database does not look like a HydraVeil storage database.",
|
|
", ".join(sorted(table_names)),
|
|
)
|
|
except GuiStorageDatabaseError:
|
|
raise
|
|
except sqlite3.Error as error:
|
|
raise GuiStorageDatabaseError(
|
|
"The local storage database could not be read. It may be malformed, stale, or incompatible.",
|
|
str(error),
|
|
) from error
|
|
finally:
|
|
if connection is not None:
|
|
connection.close()
|
|
|
|
|
|
def _table_has_rows(database_path, table_name):
|
|
path = Path(database_path)
|
|
if not path.exists():
|
|
return False
|
|
|
|
connection = None
|
|
try:
|
|
connection = _connect_readonly(path)
|
|
return connection.execute(f'SELECT 1 FROM "{table_name}" LIMIT 1').fetchone() is not None
|
|
except sqlite3.OperationalError as error:
|
|
if "no such table" in str(error).lower():
|
|
return False
|
|
raise GuiStorageDatabaseError(
|
|
"The local storage database could not be checked for synced data.",
|
|
str(error),
|
|
) from error
|
|
except sqlite3.Error as error:
|
|
raise GuiStorageDatabaseError(
|
|
"The local storage database could not be checked for synced data.",
|
|
str(error),
|
|
) from error
|
|
finally:
|
|
if connection is not None:
|
|
connection.close()
|
|
|
|
|
|
def has_required_sync_data(database_path):
|
|
required_tables = ("applications", "application_versions", "locations")
|
|
return all(_table_has_rows(database_path, table_name) for table_name in required_tables)
|