StatsDatabase.py 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458
  1. import os.path
  2. import random as rnd
  3. import typing
  4. import sqlite3
  5. import sys
  6. # TODO: double check this import
  7. # does it complain because libpcapreader is not a .py?
  8. import ID2TLib.libpcapreader as pr
  9. import Core.QueryParser as qp
  10. import pyparsing as pp
  11. def dict_gen(curs: sqlite3.Cursor):
  12. """
  13. Generates a dictionary of a sqlite3.Cursor object by fetching the query's results.
  14. Taken from Python Essential Reference by David Beazley.
  15. """
  16. field_names = [d[0] for d in curs.description]
  17. while True:
  18. rows = curs.fetchmany()
  19. if not rows:
  20. return
  21. for row in rows:
  22. yield dict(zip(field_names, row))
  23. class QueryExecutionException(Exception):
  24. pass
  25. class StatsDatabase:
  26. def __init__(self, db_path: str):
  27. """
  28. Creates a new StatsDatabase.
  29. :param db_path: The path to the database file
  30. """
  31. self.query_parser = qp.QueryParser()
  32. self.existing_db = os.path.exists(db_path)
  33. self.database = sqlite3.connect(db_path)
  34. self.cursor = self.database.cursor()
  35. self.current_interval_statistics_tables = []
  36. # If DB not existing, create a new DB scheme
  37. if self.existing_db:
  38. if self.get_db_outdated():
  39. print('Statistics database outdated. Recreating database at: ', db_path)
  40. else:
  41. print('Located statistics database at: ', db_path)
  42. else:
  43. print('Statistics database not found. Creating new database at: ', db_path)
  44. def get_file_info(self):
  45. """
  46. Retrieves general file statistics from the database. This includes:
  47. - packetCount : Number of packets in the PCAP file
  48. - captureDuration : Duration of the packet capture in seconds
  49. - timestampFirstPacket : Timestamp of the first captured packet
  50. - timestampLastPacket : Timestamp of the last captured packet
  51. - avgPacketRate : Average packet rate
  52. - avgPacketSize : Average packet size
  53. - avgPacketsSentPerHost : Average number of packets sent per host
  54. - avgBandwidthIn : Average incoming bandwidth
  55. - avgBandwidthOut : Average outgoing bandwidth
  56. :return: a dictionary of keys (see above) and their respective values
  57. """
  58. return [r for r in dict_gen(
  59. self.cursor.execute('SELECT * FROM file_statistics'))][0]
  60. def get_db_exists(self):
  61. """
  62. :return: True if the database was already existent, otherwise False
  63. """
  64. return self.existing_db
  65. def get_db_outdated(self):
  66. """
  67. Retrieves the database version from the database and compares it to the version
  68. it should have to check whether the database is outdated and needs to be recreated.
  69. :return: True if the versions match, otherwise False
  70. """
  71. self.cursor.execute('PRAGMA user_version;')
  72. return self.cursor.fetchall()[0][0] != pr.pcap_processor.get_db_version()
  73. @staticmethod
  74. def _get_selector_keywords():
  75. """
  76. :return: a list of selector keywords
  77. """
  78. return ['most_used', 'least_used', 'avg', 'all']
  79. @staticmethod
  80. def _get_parametrized_selector_keywords():
  81. """
  82. :return: a list of parameterizable selector keywords
  83. """
  84. return ['ipaddress', 'macaddress']
  85. @staticmethod
  86. def _get_extractor_keywords():
  87. """
  88. :return: a list of extractor keywords
  89. """
  90. return ['random', 'first', 'last']
  91. def get_all_named_query_keywords(self):
  92. """
  93. :return: a list of all named query keywords, used to identify named queries
  94. """
  95. return (
  96. self._get_selector_keywords() + self._get_parametrized_selector_keywords() + self._get_extractor_keywords())
  97. @staticmethod
  98. def get_all_sql_query_keywords():
  99. """
  100. :return: a list of all supported SQL keywords, used to identify SQL queries
  101. """
  102. return ["select", "insert"]
  103. def process_user_defined_query(self, query_string: str, query_parameters: tuple = None):
  104. """
  105. Takes as input a SQL query query_string and optional a tuple of parameters which are marked by '?' in the query
  106. and later substituted.
  107. :param query_string: The query to execute
  108. :param query_parameters: The tuple of parameters to inject into the query
  109. :return: the results of the query
  110. """
  111. if query_parameters is not None:
  112. self.cursor.execute(query_string, query_parameters)
  113. else:
  114. self.cursor.execute(query_string)
  115. self.database.commit()
  116. return self.cursor.fetchall()
  117. def get_field_types(self, *table_names):
  118. """
  119. Creates a dictionary whose keys are the fields of the given table(s) and whose values are the appropriate field
  120. types, like TEXT for strings and REAL for float numbers.
  121. :param table_names: The name of table(s)
  122. :return: a dictionary of {field_name : field_type} for fields of all tables
  123. """
  124. dic = {}
  125. for table in table_names:
  126. self.cursor.execute("PRAGMA table_info('%s')" % table)
  127. results = self.cursor.fetchall()
  128. for field in results:
  129. dic[field[1].lower()] = field[2]
  130. return dic
  131. def get_current_interval_statistics_table(self):
  132. """
  133. :return: the current interval statistics table used for internal calculations
  134. """
  135. if len(self.current_interval_statistics_tables) > 0:
  136. return self.current_interval_statistics_tables[0]
  137. else:
  138. return ""
  139. def get_all_current_interval_statistics_tables(self):
  140. """
  141. :return: the list of all current interval statistics tables
  142. """
  143. if len(self.current_interval_statistics_tables) == 0:
  144. return [self.process_db_query("SELECT name FROM interval_tables WHERE is_default=1")]
  145. return self.current_interval_statistics_tables
  146. def set_current_interval_statistics_tables(self, current_intervals: list):
  147. """
  148. Sets the current interval statistics table, which should be used for internal calculations.
  149. :param current_intervals: a list of current intervals in seconds, first of which should be used for internal
  150. calculations
  151. """
  152. for current_interval in current_intervals:
  153. if current_interval == 0.0:
  154. table_name = self.process_db_query("SELECT name FROM interval_tables WHERE is_default=1")
  155. print("No user specified interval found. Using default interval: " +
  156. str(float(table_name[len("interval_statistics_"):])/1000000) + "s")
  157. else:
  158. self.current_interval_statistics_tables.append("interval_statistics_" +
  159. str(int(current_interval*1000000)))
  160. if current_interval == current_intervals[0]:
  161. print("User specified interval(s) found. Using first interval length given for internal "
  162. "calculations: " + str(current_interval) + "s")
  163. def named_query_parameterized(self, keyword: str, param_op_val: list):
  164. """
  165. Executes a parameterizable named query.
  166. :param keyword: The query to be executed, like ipaddress or macadress
  167. :param param_op_val: A list consisting of triples with (parameter, operator, value)
  168. :return: the results of the executed query
  169. """
  170. named_queries = {
  171. "ipaddress": "SELECT DISTINCT ip_statistics.ipAddress from ip_statistics INNER JOIN ip_mac, ip_ttl, "
  172. "ip_ports, ip_protocols ON ip_statistics.ipAddress=ip_mac.ipAddress AND "
  173. "ip_statistics.ipAddress=ip_ttl.ipAddress AND ip_statistics.ipAddress=ip_ports.ipAddress "
  174. "AND ip_statistics.ipAddress=ip_protocols.ipAddress WHERE ",
  175. "macaddress": "SELECT DISTINCT macAddress from ip_mac WHERE "}
  176. query = named_queries.get(keyword)
  177. field_types = self.get_field_types('ip_mac', 'ip_ttl', 'ip_ports', 'ip_protocols', 'ip_statistics', 'ip_mac')
  178. conditions = []
  179. for key, op, value in param_op_val:
  180. # Check whether the value is not a simple value, but another query (or list)
  181. if isinstance(value, pp.ParseResults):
  182. if value[0] == "list":
  183. # We have a list, cut the token off and use the remaining elements
  184. value = value[1:]
  185. # Lists can only be used with "in"
  186. if op is not "in":
  187. raise QueryExecutionException("List values require the usage of the 'in' operator!")
  188. else:
  189. # If we have another query instead of a direct value, execute and replace it
  190. rvalue = self._execute_query_list(value)
  191. # Do we have a comparison operator with a multiple-result query?
  192. if op is not "in" and value[0] in ['most_used', 'least_used', 'all', 'ipaddress_param',
  193. 'macaddress_param']:
  194. raise QueryExecutionException("The extractor '" + value[0] +
  195. "' may return more than one result!")
  196. # Make value contain a simple list with the results of the query
  197. value = map(lambda x: str(x[0]), rvalue)
  198. else:
  199. # Make sure value is a list now to simplify handling
  200. value = [value]
  201. # this makes sure that TEXT fields are queried by strings,
  202. # e.g. ipAddress=192.168.178.1 --is-converted-to--> ipAddress='192.168.178.1'
  203. if field_types.get(key) == 'TEXT':
  204. def ensure_string(x):
  205. if not str(x).startswith("'") and not str(x).startswith('"'):
  206. return "'" + x + "'"
  207. else:
  208. return x
  209. value = map(ensure_string, value)
  210. # If we have more than one value, join them together, separated by commas
  211. value = ",".join(map(str, value))
  212. # this replacement is required to remove ambiguity in SQL query
  213. if key == 'ipAddress':
  214. key = 'ip_mac.ipAddress'
  215. conditions.append(key + " " + op + " (" + str(value) + ")")
  216. where_clause = " AND ".join(conditions)
  217. query += where_clause
  218. self.cursor.execute(query)
  219. return self.cursor.fetchall()
  220. named_queries = {
  221. "most_used.ipaddress": "SELECT ipAddress FROM ip_statistics WHERE (pktsSent+pktsReceived) == "
  222. "(SELECT MAX(pktsSent+pktsReceived) from ip_statistics) ORDER BY ipAddress ASC",
  223. "most_used.macaddress": "SELECT macAddress FROM (SELECT macAddress, COUNT(*) as occ from ip_mac GROUP BY "
  224. "macAddress) WHERE occ=(SELECT COUNT(*) as occ from ip_mac GROUP BY macAddress "
  225. "ORDER BY occ DESC LIMIT 1) ORDER BY macAddress ASC",
  226. "most_used.portnumber": "SELECT portNumber FROM ip_ports GROUP BY portNumber HAVING COUNT(portNumber)="
  227. "(SELECT MAX(cntPort) from (SELECT portNumber, COUNT(portNumber) as cntPort FROM "
  228. "ip_ports GROUP BY portNumber)) ORDER BY portNumber ASC",
  229. "most_used.protocolname": "SELECT protocolName FROM ip_protocols GROUP BY protocolName HAVING "
  230. "COUNT(protocolCount)=(SELECT COUNT(protocolCount) as cnt FROM ip_protocols "
  231. "GROUP BY protocolName ORDER BY cnt DESC LIMIT 1) ORDER BY protocolName ASC",
  232. "most_used.ttlvalue": "SELECT ttlValue FROM (SELECT ttlValue, SUM(ttlCount) as occ FROM ip_ttl GROUP BY "
  233. "ttlValue) WHERE occ=(SELECT SUM(ttlCount) as occ FROM ip_ttl GROUP BY ttlValue "
  234. "ORDER BY occ DESC LIMIT 1) ORDER BY ttlValue ASC",
  235. "most_used.mssvalue": "SELECT mssValue FROM (SELECT mssValue, SUM(mssCount) as occ FROM tcp_mss GROUP BY "
  236. "mssValue) WHERE occ=(SELECT SUM(mssCount) as occ FROM tcp_mss GROUP BY mssValue "
  237. "ORDER BY occ DESC LIMIT 1) ORDER BY mssValue ASC",
  238. "most_used.winsize": "SELECT winSize FROM (SELECT winSize, SUM(winCount) as occ FROM tcp_win GROUP BY "
  239. "winSize) WHERE occ=(SELECT SUM(winCount) as occ FROM tcp_win GROUP BY winSize ORDER "
  240. "BY occ DESC LIMIT 1) ORDER BY winSize ASC",
  241. "most_used.ipclass": "SELECT ipClass FROM (SELECT ipClass, COUNT(*) as occ from ip_statistics GROUP BY "
  242. "ipClass ORDER BY occ DESC) WHERE occ=(SELECT COUNT(*) as occ from ip_statistics "
  243. "GROUP BY ipClass ORDER BY occ DESC LIMIT 1) ORDER BY ipClass ASC",
  244. "least_used.ipaddress": "SELECT ipAddress FROM ip_statistics WHERE (pktsSent+pktsReceived) == (SELECT "
  245. "MIN(pktsSent+pktsReceived) from ip_statistics) ORDER BY ipAddress ASC",
  246. "least_used.macaddress": "SELECT macAddress FROM (SELECT macAddress, COUNT(*) as occ from ip_mac GROUP "
  247. "BY macAddress) WHERE occ=(SELECT COUNT(*) as occ from ip_mac GROUP BY macAddress "
  248. "ORDER BY occ ASC LIMIT 1) ORDER BY macAddress ASC",
  249. "least_used.portnumber": "SELECT portNumber FROM ip_ports GROUP BY portNumber HAVING COUNT(portNumber)="
  250. "(SELECT MIN(cntPort) from (SELECT portNumber, COUNT(portNumber) as cntPort FROM "
  251. "ip_ports GROUP BY portNumber)) ORDER BY portNumber ASC",
  252. "least_used.protocolname": "SELECT protocolName FROM ip_protocols GROUP BY protocolName HAVING "
  253. "COUNT(protocolCount)=(SELECT COUNT(protocolCount) as cnt FROM ip_protocols "
  254. "GROUP BY protocolName ORDER BY cnt ASC LIMIT 1) ORDER BY protocolName ASC",
  255. "least_used.ttlvalue": "SELECT ttlValue FROM (SELECT ttlValue, SUM(ttlCount) as occ FROM ip_ttl GROUP BY "
  256. "ttlValue) WHERE occ=(SELECT SUM(ttlCount) as occ FROM ip_ttl GROUP BY ttlValue "
  257. "ORDER BY occ ASC LIMIT 1) ORDER BY ttlValue ASC",
  258. "least_used.mssvalue": "SELECT mssValue FROM (SELECT mssValue, SUM(mssCount) as occ FROM tcp_mss GROUP BY "
  259. "mssValue) WHERE occ=(SELECT SUM(mssCount) as occ FROM tcp_mss GROUP BY mssValue "
  260. "ORDER BY occ ASC LIMIT 1) ORDER BY mssValue ASC",
  261. "least_used.winsize": "SELECT winSize FROM (SELECT winSize, SUM(winCount) as occ FROM tcp_win GROUP BY "
  262. "winSize) WHERE occ=(SELECT SUM(winCount) as occ FROM tcp_win GROUP BY winSize "
  263. "ORDER BY occ ASC LIMIT 1) ORDER BY winSize ASC",
  264. "least_used.ipclass": "SELECT ipClass FROM (SELECT ipClass, COUNT(*) as occ from ip_statistics GROUP BY "
  265. "ipClass ORDER BY occ DESC) WHERE occ=(SELECT COUNT(*) as occ from ip_statistics "
  266. "GROUP BY ipClass ORDER BY occ ASC LIMIT 1) ORDER BY ipClass ASC",
  267. "avg.pktsreceived": "SELECT avg(pktsReceived) from ip_statistics",
  268. "avg.pktssent": "SELECT avg(pktsSent) from ip_statistics",
  269. "avg.kbytesreceived": "SELECT avg(kbytesReceived) from ip_statistics",
  270. "avg.kbytessent": "SELECT avg(kbytesSent) from ip_statistics",
  271. "avg.ttlvalue": "SELECT avg(ttlValue) from ip_ttl",
  272. "avg.mss": "SELECT avg(mssValue) from tcp_mss",
  273. "all.ipaddress": "SELECT ipAddress from ip_statistics ORDER BY ipAddress ASC",
  274. "all.ttlvalue": "SELECT DISTINCT ttlValue from ip_ttl ORDER BY ttlValue ASC",
  275. "all.mss": "SELECT DISTINCT mssValue from tcp_mss ORDER BY mssValue ASC",
  276. "all.macaddress": "SELECT DISTINCT macAddress from ip_mac ORDER BY macAddress ASC",
  277. "all.portnumber": "SELECT DISTINCT portNumber from ip_ports ORDER BY portNumber ASC",
  278. "all.protocolname": "SELECT DISTINCT protocolName from ip_protocols ORDER BY protocolName ASC",
  279. "all.winsize": "SELECT DISTINCT winSize FROM tcp_win ORDER BY winSize ASC",
  280. "all.ipclass": "SELECT DISTINCT ipClass FROM ip_statistics ORDER BY ipClass ASC"}
  281. def _execute_query_list(self, query_list):
  282. """
  283. Recursively executes a list of named queries. They are of the following form:
  284. ['macaddress_param', [['ipaddress', 'in', ['most_used', 'ipaddress']]]]
  285. :param query_list: The query statement list obtained from the query parser
  286. :return: The result of the query (either a single result or a list).
  287. """
  288. if query_list[0] == "random":
  289. return [rnd.choice(self._execute_query_list(query_list[1:]))]
  290. elif query_list[0] == "first":
  291. return [self._execute_query_list(query_list[1:])[0]]
  292. elif query_list[0] == "last":
  293. return [self._execute_query_list(query_list[1:])[-1]]
  294. elif query_list[0] == "macaddress_param":
  295. return self.named_query_parameterized("macaddress", query_list[1])
  296. elif query_list[0] == "ipaddress_param":
  297. return self.named_query_parameterized("ipaddress", query_list[1])
  298. else:
  299. query = self.named_queries.get(query_list[0] + "." + query_list[1])
  300. if query is None:
  301. raise QueryExecutionException("The requested query '" + query_list[0] + "(" + query_list[1] +
  302. ")' was not found in the internal query list!")
  303. self.cursor.execute(str(query))
  304. # TODO: fetch query on demand
  305. last_result = self.cursor.fetchall()
  306. return last_result
  307. def process_db_query(self, query_string_in: str, print_results=False, sql_query_parameters: tuple = None):
  308. """
  309. Processes a database query. This can either be a standard SQL query or a named query (predefined query).
  310. :param query_string_in: The string containing the query
  311. :param print_results: Indicated whether the results should be printed to terminal (True) or not (False)
  312. :param sql_query_parameters: Parameters for the SQL query (optional)
  313. :return: the results of the query
  314. """
  315. named_query_keywords = self.get_all_named_query_keywords()
  316. # Clean query_string
  317. query_string = query_string_in.lower().lstrip()
  318. # query_string is a user-defined SQL query
  319. result = None
  320. if sql_query_parameters is not None or query_string.startswith("select") or query_string.startswith("insert"):
  321. result = self.process_user_defined_query(query_string, sql_query_parameters)
  322. # query string is a named query -> parse it and pass it to statisticsDB
  323. elif any(k in query_string for k in named_query_keywords) and all(k in query_string for k in ['(', ')']):
  324. if query_string[-1] != ";":
  325. query_string += ";"
  326. query_list = self.query_parser.parse_query(query_string)
  327. result = self._execute_query_list(query_list)
  328. else:
  329. sys.stderr.write(
  330. "Query invalid. Only named queries and SQL SELECT/INSERT allowed. Please check the query's syntax!\n")
  331. return
  332. # If result is tuple/list with single element, extract value from list
  333. requires_extraction = (isinstance(result, list) or isinstance(result, tuple)) and len(result) == 1 and \
  334. (not isinstance(result[0], tuple) or len(result[0]) == 1)
  335. while requires_extraction:
  336. if isinstance(result, list) or isinstance(result, tuple):
  337. result = result[0]
  338. else:
  339. requires_extraction = False
  340. # If tuple of tuples or list of tuples, each consisting of single element is returned,
  341. # then convert it into list of values, because the returned column is clearly specified by the given query
  342. if (isinstance(result, tuple) or isinstance(result, list)) and all(len(val) == 1 for val in result):
  343. result = [c for c in result for c in c]
  344. # Print results if option print_results is True
  345. if print_results:
  346. if isinstance(result, list) and len(result) == 1:
  347. result = result[0]
  348. print("Query returned 1 record:\n")
  349. for i in range(0, len(result)):
  350. print(str(self.cursor.description[i][0]) + ": " + str(result[i]))
  351. else:
  352. self._print_query_results(query_string_in, result if isinstance(result, list) else [result])
  353. return result
  354. def process_interval_statistics_query(self, query_string_in: str, table_param: str=""):
  355. """
  356. :param query_string_in: a query to be executed over the current internal interval statistics table
  357. :param table_param: a name of a specific interval statistics table
  358. :return: the result of the query
  359. """
  360. if table_param != "":
  361. table_name = table_param
  362. elif self.get_current_interval_statistics_table() != "":
  363. table_name = self.get_current_interval_statistics_table()
  364. else:
  365. table_name = self.process_db_query("SELECT name FROM interval_tables WHERE is_default=1")
  366. return self.process_user_defined_query(query_string_in % table_name)
  367. def _print_query_results(self, query_string_in: str, result: typing.List[typing.Union[str, float, int]]) -> None:
  368. """
  369. Prints the results of a query.
  370. Based on http://stackoverflow.com/a/20383011/3017719.
  371. :param query_string_in: The query the results belong to
  372. :param result: The list of query results
  373. """
  374. # Print number of results according to type of result
  375. if len(result) == 1:
  376. print("Query returned 1 record:\n")
  377. else:
  378. print("Query returned " + str(len(result)) + " records:\n")
  379. # Print query results
  380. if query_string_in.lstrip().upper().startswith(
  381. "SELECT") and result is not None and self.cursor.description is not None:
  382. widths = []
  383. columns = []
  384. tavnit = '|'
  385. separator = '+'
  386. for index, cd in enumerate(self.cursor.description):
  387. max_col_length = 0
  388. if len(result) > 0:
  389. max_col_length = max(list(map(lambda x:
  390. len(str(x[index] if len(self.cursor.description) > 1 else x)),
  391. result)))
  392. widths.append(max(len(cd[0]), max_col_length))
  393. columns.append(cd[0])
  394. for w in widths:
  395. tavnit += " %-" + "%ss |" % (w,)
  396. separator += '-' * w + '--+'
  397. print(separator)
  398. print(tavnit % tuple(columns))
  399. print(separator)
  400. for row in result:
  401. print(tavnit % row)
  402. print(separator)
  403. else:
  404. print(result)