"""
Sugaroid Backend is a SQL Database handler which stores
incoming messages into an sqlite3 database, which is later used
for analytics and usage history. With this, it will be possible
to chart out the most used command / adapters and accordingly
move them above / increase their preference of processing.
"""
import sqlite3
import logging
logger = logging.getLogger("sugaroid")
CREATE_TABLE = """CREATE TABLE IF NOT EXISTS sugaroid_hist (
statement varchar(500),
in_response_to varchar(500),
time FLOAT,
processing_time FLOAT
);
"""
[docs]class PossibleSQLInjectionPanicError(ValueError):
"""
Raises PossibleSQLInjectionPanicError in case
of possible SQL Injection.
SQL Injection is an attempt to change the data
by altering data within the string by attempting
to manipulate the database entry by multiple
semicolons for example
"""
pass
[docs]def convert_data_escaped_string(data: tuple):
"""
Converts data from tuple form to a string statement
to a SQL string statement
data: tuple
return: a SQL formatted string
rtype: str
"""
_processed_data = list()
for i in data:
if isinstance(i, str):
b = i.replace(";", ",")
# append the data with enclosing double quotes
if ";" in b:
# the data should be preprocessed to remove
# semicolons in case if its necessary
# if the caller did not escape the semicolon
# we should raise a PossibleSQLInjection panic
raise PossibleSQLInjectionPanicError(
"An attempt to inject SQL issues was found"
)
if len(i) > 50:
_processed_data.append('"{}"'.format(b))
else:
_processed_data.append('"{}"'.format("LONG_BLOB_TEXT"))
elif isinstance(i, int) or isinstance(i, float):
# append the data as raw string
_processed_data.append("{}".format(i))
elif i is None:
# append to the line
_processed_data.append("NULL")
else:
logger.warn(
"Unknown data type encountered {} for {}. "
"Make sure that data type matches "
"`int`, `str`, `float`, `None`".format(type(i), i)
)
_processed_data.append('"{}"'.format(i))
# debug: Assert that all the elements are of string type
# which otherwise may crash with a type error
assert all([isinstance(x, str) for x in _processed_data])
return ", ".join(_processed_data)
[docs]class SqlDatabaseManagement:
"""
Sugaroid stores some data for analytics and research
in an ``sqlite3`` database
"""
def __init__(self, path_to_db, table="sugaroid_hist"):
"""
Initialized the Sql Database Management object
and then creates the table ``sugaroid_hist`` in the table
if it does not exist.
"""
self._path_to_db = path_to_db
self.database_instance = sqlite3.connect(self._path_to_db)
self._cnx = self.database_instance.cursor()
self._table = table
self._execute(CREATE_TABLE)
@property
def table(self):
"""
Return the table name on target
"""
return self._table
@property
def database_path(self) -> str:
"""
Return the path to the database
:return: path to the database
:rtype: str
"""
return self._path_to_db
def _execute(self, command: str):
"""
Protected, Private method to execute an sql command
:param command: a valid SQL statement
:type command: str
"""
self._cnx.execute(command)
def _add(self, data: tuple):
"""
Adds data to the SQLite3 database into the table
:param data: data to be inserted
:type data: tuple
"""
self._cnx.execute(
"INSERT INTO {tablename} VALUES({values})".format(
tablename=self.table, values=convert_data_escaped_string(data)
)
)
[docs] def append(self, statement, in_reponse_to, time, processing_time):
self._add((statement, in_reponse_to, time, processing_time))
[docs] def close(self):
"""
Closes the connection to the mysql database
"""
self.database_instance.commit()
self.database_instance.close()