From 9d26a72094865f2327017b7075fd483e78b0bb2b Mon Sep 17 00:00:00 2001 From: dimeko Date: Mon, 20 May 2024 14:29:58 +0300 Subject: [PATCH] Feature: diff between two given xml, without storing to the database --- deltascan/cli/cli_output.py | 15 +- deltascan/cli/cmd.py | 287 +++++++++++++++++++++-------------- deltascan/core/config.py | 1 + deltascan/core/deltascan.py | 136 ++++++++++++++--- deltascan/core/exceptions.py | 3 + deltascan/core/importer.py | 81 +++++++--- deltascan/core/parser.py | 145 ++++++++++++++++++ deltascan/core/scanner.py | 71 +-------- deltascan/core/schemas.py | 1 + deltascan/core/utils.py | 79 +--------- tests/conftest.py | 1 + tests/unit/test_main.py | 3 +- tests/unit/test_parser.py | 31 ++++ tests/unit/test_scanner.py | 7 +- tests/unit/test_utils.py | 28 ---- 15 files changed, 549 insertions(+), 340 deletions(-) create mode 100644 deltascan/core/parser.py create mode 100644 tests/unit/test_parser.py diff --git a/deltascan/cli/cli_output.py b/deltascan/cli/cli_output.py index 941f20a..b726508 100644 --- a/deltascan/cli/cli_output.py +++ b/deltascan/cli/cli_output.py @@ -1,5 +1,4 @@ -from deltascan.core.utils import (diffs_to_output_format, - format_string) +from deltascan.core.utils import (format_string) from deltascan.core.output import Output from deltascan.core.schemas import ReportScanFromDB, ReportDiffs from deltascan.core.exceptions import DScanMethodNotImplemented @@ -9,7 +8,7 @@ from rich.table import Table from rich.panel import Panel from rich.columns import Columns - +from deltascan.core.parser import Parser import datetime @@ -59,7 +58,7 @@ def _validate_data(self, data): articulated_diffs.append( {"date_from": diff["dates"][1], "date_to": diff["dates"][0], - "diffs": diffs_to_output_format(diff), + "diffs": Parser.diffs_to_output_format(diff), "generic": diff["generic"], "uuids": diff["uuids"]}) for d in articulated_diffs: @@ -279,7 +278,7 @@ def display(self): None """ tables = self._display() - panel = Panel.fit(Columns(tables), title=self._display_title, border_style="conceal") + panel = Panel.fit(Columns(tables), title=self._display_title) self.console.print(panel) return self._index_to_uuid_mapping @@ -287,9 +286,9 @@ def display(self): @classmethod def profiles(cls, profiles): _profiles_table = Table(show_header=True) - _profiles_table.add_column("Name", style="bright_yellow", no_wrap=True) - _profiles_table.add_column("Arguments", style="rosy_brown", no_wrap=True) - _profiles_table.add_column("Created at", style="bright_yellow", no_wrap=True) + _profiles_table.add_column("Name", style="bright_yellow", no_wrap=False, width=50) + _profiles_table.add_column("Arguments", style="rosy_brown", no_wrap=False) + _profiles_table.add_column("Created at", style="bright_yellow", no_wrap=False, width=30) for profile in profiles: _profiles_table.add_row(profile["profile_name"], profile["arguments"], str(profile["created_at"])) diff --git a/deltascan/cli/cmd.py b/deltascan/cli/cmd.py index 7f56456..15de1a1 100644 --- a/deltascan/cli/cmd.py +++ b/deltascan/cli/cmd.py @@ -80,77 +80,99 @@ def do_conf(self, v): """conf [key=value] Modify configuration values real-time. Ex. conf output_file=/tmp/output.json""" - if len(v.split("=")) <= 1: + try: + if len(v.split("=")) <= 1: + conf_key = v.split("=")[0] + if conf_key == "output_file" or conf_key == "": + print(f"{'output_file: ' + '':<20} {self._app.output_file}") + if conf_key == "template_file" or conf_key == "": + print(f"{'template_file: ' + '':<20} {self._app.template_file}") + if conf_key == "import_file" or conf_key == "": + print(f"{'import_file: ' + '':<20} {self._app.import_file}") + if conf_key == "diff_files" or conf_key == "": + print(f"{'diff_files: ' + '':<20} {self._app.diff_files}") + if conf_key == "n_scans" or conf_key == "": + print(f"{'n_scans: ' + '':<20} {self._app.n_scans}") + if conf_key == "n_diffs" or conf_key == "": + print(f"{'n_diffs: ' + '':<20} {self._app.n_diffs}") + if conf_key == "fdate" or conf_key == "": + print(f"{'From date [fdate]: ' + '':<20} {self._app.fdate}") + if conf_key == "tdate" or conf_key == "": + print(f"{'To date [tdate]: ' + '':<20} {self._app.tdate}") + if conf_key == "suppress" or conf_key == "": + print(f"{'suppress: ' + '':<20} {self._app.suppress}") + if conf_key == "host" or conf_key == "": + print(f"{'host: ' + '':<20} {self._app.host}") + if conf_key == "profile" or conf_key == "": + print(f"{'profile: ' + '':<20} {self._app.profile}") + return + conf_key = v.split("=")[0] - if conf_key == "output_file" or conf_key == "": - print(f"{'output_file: ' + '':<20} {self._app.output_file}") - if conf_key == "template_file" or conf_key == "": - print(f"{'template_file: ' + '':<20} {self._app.template_file}") - if conf_key == "import_file" or conf_key == "": - print(f"{'import_file: ' + '':<20} {self._app.import_file}") - if conf_key == "n_scans" or conf_key == "": - print(f"{'n_scans: ' + '':<20} {self._app.n_scans}") - if conf_key == "n_diffs" or conf_key == "": - print(f"{'n_diffs: ' + '':<20} {self._app.n_diffs}") - if conf_key == "fdate" or conf_key == "": - print(f"{'From date [fdate]: ' + '':<20} {self._app.fdate}") - if conf_key == "tdate" or conf_key == "": - print(f"{'To date [tdate]: ' + '':<20} {self._app.tdate}") - if conf_key == "suppress" or conf_key == "": - print(f"{'suppress: ' + '':<20} {self._app.suppress}") - if conf_key == "host" or conf_key == "": - print(f"{'host: ' + '':<20} {self._app.host}") - if conf_key == "profile" or conf_key == "": - print(f"{'profile: ' + '':<20} {self._app.profile}") - return - - conf_key = v.split("=")[0] - conf_value = v.split("=")[1] - - def __norm_value(v): - return None if v == "" or v == "None" else v - - if conf_key == "output_file": - self._app.output_file = __norm_value(conf_value) - elif conf_key == "template_file": - self._app.template_file = __norm_value(conf_value) - elif conf_key == "import_file": - self._app.import_file = __norm_value(conf_value) - elif conf_key == "n_scans": - self._app.n_scans = __norm_value(conf_value) - elif conf_key == "n_diffs": - self._app.n_diffs = __norm_value(conf_value) - elif conf_key == "fdate": - self._app.fdate = __norm_value(conf_value) - elif conf_key == "tdate": - self._app.tdate = __norm_value(conf_value) - elif conf_key == "suppress": - self._app.suppress = bool(__norm_value(conf_value)) - elif conf_key == "host": - self._app.host = __norm_value(conf_value) - elif conf_key == "profile": - self._app.profile = __norm_value(conf_value) - else: - print("Invalid configuration value") + conf_value = v.split("=")[1] + + def __norm_value(v): + return None if v == "" or v == "None" else v + + if conf_key == "output_file": + self._app.output_file = __norm_value(conf_value) + elif conf_key == "template_file": + self._app.template_file = __norm_value(conf_value) + elif conf_key == "import_file": + self._app.import_file = __norm_value(conf_value) + elif conf_key == "n_scans": + self._app.n_scans = __norm_value(conf_value) + elif conf_key == "n_diffs": + self._app.n_diffs = __norm_value(conf_value) + elif conf_key == "fdate": + self._app.fdate = __norm_value(conf_value) + elif conf_key == "tdate": + self._app.tdate = __norm_value(conf_value) + elif conf_key == "suppress": + self._app.suppress = bool(__norm_value(conf_value)) + elif conf_key == "host": + self._app.host = __norm_value(conf_value) + elif conf_key == "profile": + self._app.profile = __norm_value(conf_value) + else: + print("Invalid configuration value") + except Exception as e: + print(str(e)) def do_scan(self, v): """scan Add ad-hoc scans: scan 10.10.10.10 PROFILE_NAME""" - if len(v.split(" ")) != 2: - print("Invalid input. Provide a host and a profile: scan ") - return - v1, v2 = v.split(" ") - _r = self._app.add_scan(v1, v2) - if _r is False: - print("Not starting scan. Check your host and profile. Maybe the scan is already in the queue.") - return + try: + if len(v.split(" ")) != 2: + print("Invalid input. Provide a host and a profile: scan ") + return + v1, v2 = v.split(" ") + _r = self._app.add_scan(v1, v2) + if _r is False: + print("Not starting scan. Check your host and profile. Maybe the scan is already in the queue.") + return + except Exception as e: + print(str(e)) def do_view(self, _): """view Execute the view action using the current configuration""" - _r = self._app.view() - output = CliOutput(_r, self._app.suppress) - self.last_index_to_uuid_mapping = output.display() + try: + _r = self._app.view() + output = CliOutput(_r, self._app.suppress) + self.last_index_to_uuid_mapping = output.display() + except Exception as e: + print(str(e)) + + def do_diff_files(self, v): + """diff_files + Execute the difference comparison using the current configuration. + Ex. diff_files file1.xml,file2.xml,file3.xml""" + try: + _r = self._app.files_diff(v) + output = CliOutput(_r) + output.display() + except Exception as e: + print(str(e)) def do_diff(self, v): """diff @@ -159,46 +181,58 @@ def do_diff(self, v): You can also provide a list of indexes from the last view results. Ex. diff 1,2,3,4,5 """ - if len(v.split(",")) > 1 and \ - self.last_index_to_uuid_mapping is not None: - _idxs = v.split(",") - _uuids = [] - for _key in self.last_index_to_uuid_mapping: - if _key in _idxs: - _uuids.append(self.last_index_to_uuid_mapping[_key]) - if len(_uuids) < 2: - print("Provide 2 valid indexes from the view list." - " Re-run view to view the last results.") - return - r = self._app.diffs(uuids=_uuids) - else: - r = self._app.diffs() + try: + if len(v.split(",")) > 1 and \ + self.last_index_to_uuid_mapping is not None: + _idxs = v.split(",") + _uuids = [] + for _key in self.last_index_to_uuid_mapping: + if _key in _idxs: + _uuids.append(self.last_index_to_uuid_mapping[_key]) + if len(_uuids) < 2: + print("Provide 2 valid indexes from the view list." + " Re-run view to view the last results.") + return + r = self._app.diffs(uuids=_uuids) + else: + r = self._app.diffs() - output = CliOutput(r) - output.display() + output = CliOutput(r) + output.display() + except Exception as e: + print(str(e)) def do_report(self, _): """report Generate a report using the current configuration. Ex. report""" - _ = self._app.report_result() - print("File configured", self._app.output_file) + try: + _ = self._app.report_result() + print("File configured", self._app.output_file) + except Exception as e: + print(str(e)) def do_imp(self, v): """imp Import a file using the current configuration. Ex. imp""" # Getting the requested scans from the list of the last scans - if v == "": - v = None + try: + if v == "": + v = None - r = self._app.import_data(v) - output = CliOutput(r) - output.display() + r = self._app.import_data(v) + output = CliOutput(r) + output.display() + except Exception as e: + print(str(e)) def do_profiles(self, _): """profiles List all available profiles""" - r = self._app.list_profiles() - CliOutput.profiles(r) + try: + r = self._app.list_profiles() + CliOutput.profiles(r) + except Exception as e: + print(str(e)) def do_clear(self, _): """clear @@ -208,36 +242,44 @@ def do_clear(self, _): def do_q(self, _): """q or quit Quit interactive shell""" - if self._app.scans_to_wait == 0 and self._app.scans_to_execute == 0: - print("No scans in the queue...") - else: - return True + try: + if self._app.scans_to_wait == 0 and self._app.scans_to_execute == 0: + print("No scans in the queue...") + else: + return True + except Exception as e: + print(str(e)) def do_quit(self, _): """q or quit Quit interactive shell""" - if self._app.scans_to_wait == 0 and self._app.scans_to_execute == 0: - print("No scans in the queue...") - else: - return True + try: + if self._app.scans_to_wait == 0 and self._app.scans_to_execute == 0: + print("No scans in the queue...") + else: + return True + except Exception as e: + print(str(e)) def do_exit(self, _): """exit Exit Deltascan""" - print("Shutting down...") - self._app.cleanup() - while self._app.cleaning_up is False or self._app.is_running is True: - sleep(1) - continue - print("Cancelled all scans. Exiting with grace ...") - os._exit(0) + try: + print("Shutting down...") + self._app.cleanup() + while self._app.cleaning_up is False or self._app.is_running is True: + sleep(1) + continue + print("Cancelled all scans. Exiting with grace ...") + os._exit(0) + except Exception as e: + print(str(e)) def signal_handler(signal, frame): print("Exiting without cleanup :-(") os._exit(1) - def run(): """ Entry point for the command line interface. @@ -248,19 +290,25 @@ def run(): "-a", "--action", help='the command to run', required=False, choices=['scan', 'diff', 'view', 'import']) parser.add_argument("-o", "--output", help='output file', required=False) + parser.add_argument("-d", "--diff-files", + help='comma separated files to find their differences (xml)', + required=False) parser.add_argument( "--single", default=False, action='store_true', - help='export scans as single entries', required=False) + help='if flag exists, it exports scans as single entries', required=False) parser.add_argument( - "--template", help='template file', required=False) + "--template", help='the html template file to generate .html and .pdf reports', + required=False) parser.add_argument( "-i", "--import", dest="import_file", - help='import file', required=False) + help='import file (csv, xml). Csv must be generated by deltascan and XML must be generated by nmap', + required=False) parser.add_argument( - "-p", "--profile", help="select scanning profile", required=False) + "-p", "--profile", help="select scanning profile that exists in config file or already in database", + required=False) parser.add_argument( "-c", "--conf-file", - help="select profile file to load", required=False) + help="path to configuration file", required=False) parser.add_argument( "-v", "--verbose", default=False, action='store_true', help="verbose output", required=False) @@ -268,21 +316,21 @@ def run(): "-s", "--suppress", default=False, action='store_true', help="suppress output", required=False) parser.add_argument( - "--n-scans", help="N scan number", required=False) + "--n-scans", help="limit of scans databse queries. It is applied in scans view as well as scans diff", + required=False) parser.add_argument( "--n-diffs", default=1, - help="N scan differences", required=False) + help="limit of the diff results", required=False) parser.add_argument( - "--from-date", help="Date of oldest scan to compare", required=False) + "--from-date", help="date of oldest scan to compare", required=False) parser.add_argument( - "--to-date", help="Created at date, of the queried scan", - required=False) + "--to-date", help="date of newest scan to compare", required=False) parser.add_argument( "--port-type", default="open,closed,filtered", - help="Type of port status open,filter,closed,all", required=False) + help="Type of port status (open,filter,closed,all)", required=False) parser.add_argument( "-t", "--target", dest="host", - help="select scanning target host", required=False) + help="select target host/subnet to scan", required=False) parser.add_argument( "-it", "--interactive", default=False, action='store_true', help="execute action and go in interactive mode", required=False) @@ -320,6 +368,7 @@ def run(): "single": clargs.single, "template_file": clargs.template, "import_file": clargs.import_file, + "diff_files": clargs.diff_files, "action": clargs.action, "profile": clargs.profile, "conf_file": clargs.conf_file, @@ -391,7 +440,10 @@ def run(): os._exit(0) elif clargs.action == 'diff': - _r = _dscan.diffs() + if clargs.diff_files is not None: + _r = _dscan.files_diff() + else: + _r = _dscan.diffs() output = CliOutput(_r) output.display() elif clargs.action == 'view': @@ -416,12 +468,15 @@ def run(): except DScanException as e: print(f"Error occurred: {str(e)}") + os._exit(1) except KeyboardInterrupt: signal.signal(signal.SIGINT, signal_handler) _dscan.cleanup() print("Cancelling running scans and closing ...") + os._exit(1) except Exception as e: print(f"Unknown error occurred: {str(e)}") + os._exit(1) if __name__ == "__main__": diff --git a/deltascan/core/config.py b/deltascan/core/config.py index f382ffd..0788a25 100644 --- a/deltascan/core/config.py +++ b/deltascan/core/config.py @@ -34,6 +34,7 @@ class Config: single: bool template_file: str import_file: str + diff_files: str action: str profile: str conf_file: str diff --git a/deltascan/core/deltascan.py b/deltascan/core/deltascan.py index d77a08c..8c38e34 100644 --- a/deltascan/core/deltascan.py +++ b/deltascan/core/deltascan.py @@ -2,6 +2,7 @@ import deltascan.core.store as store from deltascan.core.config import ( CONFIG_FILE_PATH, + APP_DATE_FORMAT, Config, ADDED, CHANGED, @@ -14,12 +15,11 @@ from deltascan.core.utils import (datetime_validation, validate_host, check_root_permissions, - validate_port_state_type, - diffs_to_output_format) + validate_port_state_type) from deltascan.core.export import Exporter -from deltascan.core.schemas import (DBScan, ConfigSchema) +from deltascan.core.schemas import (DBScan, ConfigSchema, Scan) from deltascan.core.importer import Importer - +from deltascan.core.parser import Parser from marshmallow import ValidationError from threading import Thread, Event @@ -29,6 +29,7 @@ import json import copy import time +from datetime import datetime from rich.progress import ( BarColumn, @@ -74,6 +75,7 @@ def __init__(self, config, ui_context=None, result=[]): _config["single"], _config["template_file"], _config["import_file"], + _config["diff_files"], _config["action"], _config['profile'], _config['conf_file'], @@ -333,11 +335,7 @@ def diffs(self, uuids=None): to_date=self._config.tdate ) - _split_scans_in_hosts = {} - for _s in scans: - if _s["results"]["host"] not in _split_scans_in_hosts: - _split_scans_in_hosts[_s["results"]["host"]] = [] - _split_scans_in_hosts[_s["results"]["host"]].append(_s) + _split_scans_in_hosts = self.__split_scans_in_hosts([_s for _s in scans]) diffs = self._list_scans_with_diffs([_s for _scans in _split_scans_in_hosts.values() for _s in _scans]) if self._config.output_file is not None: @@ -359,6 +357,98 @@ def diffs(self, uuids=None): else: raise DScanSchemaException("Invalid scan results schema") + @staticmethod + def __split_scans_in_hosts(scans): + """ + Splits a list of scans into a dictionary where the keys are the hosts and the values are lists of scans for each host. + + Args: + scans (list): A list of scans. + + Returns: + dict: A dictionary where the keys are the hosts and the values are lists of scans for each host. + """ + _split_scans_in_hosts = {} + for _s in scans: + if _s["host"] not in _split_scans_in_hosts: + _split_scans_in_hosts[_s["host"]] = [] + _split_scans_in_hosts[_s["host"]].append(_s) + return _split_scans_in_hosts + + def files_diff(self, _diff_files=None): + """ + Compare the scan results from different files and print the differences. + + Args: + _diff_files (str): Comma-separated list of file paths to compare. If not provided or empty, + the file paths will be fetched from the configuration. + + Returns: + None + """ + if _diff_files is not None and _diff_files != "": + _files = _diff_files.split(",") + else: + _files = self._config.diff_files.split(",") + _imported_scans = [] + _importer = None + if _files is None or len(_files) < 2: + raise DScanInputValidationException("At least two files must be provided to compare") + for _f in _files: + if _importer is None: + _importer = Importer(_f, logger=self.logger) + _r = _importer.load_results_from_file() + else: + _importer.filename = _f + _r = _importer.load_results_from_file() + + _host = _r._nmaprun["args"].split(" ")[-1] + _parsed = Parser.extract_port_scan_dict_results(_r) + if "/" in _host: + raise DScanInputValidationException("Subnet is not supported for this operation") + if len(_parsed) > 1: + raise DScanInputValidationException("Only one host per file is supported for this operation") + + _imported_scans.append( + { + "created_at": datetime.fromtimestamp(int( + _r._runstats["finished"]["time"])).strftime( + APP_DATE_FORMAT) if "finished" in _r._runstats else None, + "results": _parsed[0], + "arguments": _r._nmaprun["args"] + } + ) + + _final_diffs = [] + for i, _ in enumerate(_imported_scans, 1): + if i == len(_imported_scans): + break + __diffs = self._diffs_between_dicts( + self._results_to_port_dict(_imported_scans[i-1]["results"]), + self._results_to_port_dict(_imported_scans[i]["results"])) + _final_diffs.append({ + "ids": [0,0], + "uuids": ["",""], + "generic": [ + { + "host": _imported_scans[i-1]["results"]["host"], + "arguments": _imported_scans[i-1]["arguments"], + "profile_name": "" + }, + { + "host": _imported_scans[i]["results"]["host"], + "arguments": _imported_scans[i]["arguments"], + "profile_name": "" + } + ], + "dates": [ + _imported_scans[i-1]["created_at"], + _imported_scans[i]["created_at"]], + "diffs": __diffs, + "result_hashes": ["",""] + }) + return _final_diffs + def _list_scans_with_diffs(self, scans): """ Returns a list of scans with differences between consecutive scans. @@ -400,8 +490,8 @@ def _list_scans_with_diffs(self, scans): str(scans[i-1]["created_at"]), str(scans[i]["created_at"])], "diffs": self._diffs_between_dicts( - self._results_to_port_dict(scans[i-1]), - self._results_to_port_dict(scans[i])), + self._results_to_port_dict(scans[i-1]["results"]), + self._results_to_port_dict(scans[i]["results"])), "result_hashes": [ scans[i-1]["result_hash"], scans[i]["result_hash"]] @@ -429,7 +519,7 @@ def _results_to_port_dict(self, results): DScanResultsSchemaException: If the scan results have an invalid schema. """ try: - DBScan().load(results) + Scan().load(results) except (KeyError, ValidationError) as e: self.logger.error(f"{str(e)}") if self._config.is_interactive is True: @@ -439,13 +529,13 @@ def _results_to_port_dict(self, results): port_dict = copy.deepcopy(results) - port_dict["results"]["new_ports"] = {} - for port in port_dict["results"]["ports"]: - port_dict["results"]["new_ports"][port["portid"]] = port - port_dict["results"]["ports"] = port_dict["results"]["new_ports"] - del port_dict["results"]["new_ports"] + port_dict["new_ports"] = {} + for port in port_dict["ports"]: + port_dict["new_ports"][port["portid"]] = port + port_dict["ports"] = port_dict["new_ports"] + del port_dict["new_ports"] - return port_dict["results"] + return port_dict def _diffs_between_dicts(self, changed_scan, old_scan): """ @@ -621,7 +711,7 @@ def _report_diffs(self, diffs, output_file=None): articulated_diffs.append( {"date_from": diff["dates"][1], "date_to": diff["dates"][0], - "diffs": diffs_to_output_format(diff), + "diffs": Parser.diffs_to_output_format(diff), "generic": diff["generic"], "uuids": diff["uuids"]}) except DScanResultsSchemaException as e: @@ -754,6 +844,14 @@ def import_file(self): @import_file.setter def import_file(self, value): self._config.import_file = value + + @property + def diff_files(self): + return self._config.diff_files + + @diff_files.setter + def diff_files(self, value): + self._config.diff_files = value @property def n_scans(self): diff --git a/deltascan/core/exceptions.py b/deltascan/core/exceptions.py index 6e015bb..1c1a9ec 100644 --- a/deltascan/core/exceptions.py +++ b/deltascan/core/exceptions.py @@ -85,3 +85,6 @@ class DScanRDBMSErrorCreatingEntry(DScanRDBMSException): class DScanMethodNotImplemented(DScanException): pass + +class DScanResultsParsingError(DScanException): + pass \ No newline at end of file diff --git a/deltascan/core/importer.py b/deltascan/core/importer.py index 0255663..52339e3 100644 --- a/deltascan/core/importer.py +++ b/deltascan/core/importer.py @@ -7,7 +7,7 @@ from deltascan.core.utils import ( nmap_arguments_to_list) from libnmap.parser import NmapParser, NmapParserException -from deltascan.core.scanner import Scanner +from deltascan.core.parser import Parser from deltascan.core.config import (APP_DATE_FORMAT, LOG_CONF, XML, CSV) import csv from datetime import datetime @@ -31,18 +31,18 @@ def __init__(self, filename, logger=None): raise DScanImportFileError("File is None") self.logger = logger if logger is not None else logging.basicConfig(**LOG_CONF) - self.filename = filename + self._filename = filename self.store = store.Store() if filename.split(".")[-1] in [CSV, XML]: - self.file_extension = filename.split(".")[-1] - self.filename = filename[:-1*len(self.file_extension)-1] - self.full_name = f"{self.filename}.{self.file_extension}" + self._file_extension = filename.split(".")[-1] + self._filename = filename[:-1*len(self._file_extension)-1] + self._full_name = f"{self._filename}.{self._file_extension}" else: raise DScanImportFileExtensionError("Please specify a valid file extension for the import file.") - if self.file_extension == CSV: + if self._file_extension == CSV: self.import_data = self._import_csv - elif self.file_extension == XML: + elif self._file_extension == XML: self.import_data = self._import_xml else: raise DScanImportFileExtensionError("Please specify a valid file extension for the import file.") @@ -57,7 +57,7 @@ def _import_csv(self): DScanImportDataError: If the CSV data fails to import. """ try: - with open(self.full_name, 'r') as f: + with open(self._full_name, 'r') as f: reader = csv.DictReader(f) # read rows into a dictionary format _csv_data_to_dict = [] for row in reader: # read a row as {column1: value1, column2: value2,...} @@ -96,25 +96,21 @@ def _import_xml(self): DScanImportDataError: If the XML data fails to parse. """ try: - with open(self.full_name, 'r') as file: - self.data = file.read() - - parsed = NmapParser.parse(self.data) - - _imported_scans = Scanner._extract_port_scan_dict_results(parsed) # Lending one method from Scanner :-D - _host = parsed._nmaprun["args"].split(" ")[-1] + _r = self.load_results_from_file(self._full_name) + _parsed = Parser.extract_port_scan_dict_results(_r ) + _host = _r ._nmaprun["args"].split(" ")[-1] _profile_name, _ = \ self._create_or_get_imported_profile( - parsed._nmaprun["args"], parsed._nmaprun["start"]) + _r ._nmaprun["args"], _r ._nmaprun["start"]) _newly_imported_scans = self.store.save_scans( _profile_name, _host, # Subnet - _imported_scans, + _parsed, created_at=datetime.fromtimestamp(int( - parsed._runstats["finished"]["time"])).strftime( - APP_DATE_FORMAT) if "finished" in parsed._runstats else None) + _r ._runstats["finished"]["time"])).strftime( + APP_DATE_FORMAT) if "finished" in _r ._runstats else None) _new_uuids_list = [_s.uuid for _s in list(_newly_imported_scans)] @@ -126,6 +122,27 @@ def _import_xml(self): self.logger.error(f"Failed parsing XML data: {str(e)}") raise DScanImportDataError("Could not import XML file.") + def load_results_from_file(self): + """ + Load results from a file and parse them using NmapParser. + + Args: + filename (str): The path to the file containing the results. + + Returns: + NmapReport: The parsed Nmap report. + + Raises: + FileNotFoundError: If the specified file does not exist. + IOError: If there is an error reading the file. + + """ + data = None + with open(self._full_name, 'r') as file: + data = file.read() + + return NmapParser.parse(data) + def _create_or_get_imported_profile(self, imported_args, new_profile_date=str(datetime.now())): """ Creates a new profile or retrieves an existing profile based on the imported arguments. @@ -174,3 +191,29 @@ def import_data(self): # Add your code here to import the data self.logger.error("Error importing file: 'import_data' not implemented") raise DScanImportError("Something wrong importing file.") + + @property + def full_name(self): + self._full_name = f"{self._filename}.{self._file_extension}" + + @property + def filename(self): + """ + Get the name of the import file. + + Returns: + str: The name of the import file. + """ + return self._filename + + @filename.setter + def filename(self, value): + """ + Set the name of the import file. + + Args: + value (str): The name of the import file. + """ + self._file_extension = value.split(".")[-1] + self._filename = value[:-1*len(self._file_extension)-1] + self._full_name = f"{self._filename}.{self._file_extension}" diff --git a/deltascan/core/parser.py b/deltascan/core/parser.py new file mode 100644 index 0000000..c58a53a --- /dev/null +++ b/deltascan/core/parser.py @@ -0,0 +1,145 @@ +from deltascan.core.exceptions import (DScanResultsParsingError) +from deltascan.core.config import ( + ADDED, + CHANGED, + REMOVED) +import copy +from deltascan.core.schemas import Diffs +from deltascan.core.exceptions import (DScanResultsSchemaException) +from marshmallow import ValidationError + +class Parser: + @classmethod + def diffs_to_output_format(cls, diffs): + """ + Convert the given diffs to a specific output format. + + Args: + diffs (dict): The diffs to be converted. + + Returns: + dict: The converted diffs in the specified output format. + + Raises: + DScanResultsSchemaException: If the diffs have an invalid schema. + """ + try: + Diffs().load(diffs) + except (KeyError, ValidationError) as e: + raise DScanResultsSchemaException(f"Invalid diff results schema: {str(e)}") + + # Here, entity can be many things. In the future an entity, besides port + # can be a service, a host, the osfingerpint. + articulated_diffs = { + ADDED: [], + CHANGED: [], + REMOVED: [], + } + + articulated_diffs[ADDED] = cls._dict_diff_to_list_diff(diffs["diffs"], [], ADDED) + articulated_diffs[CHANGED] = cls._dict_diff_to_list_diff(diffs["diffs"], [], CHANGED) + articulated_diffs[REMOVED] = cls._dict_diff_to_list_diff(diffs["diffs"], [], REMOVED) + + return articulated_diffs + + @classmethod + def _dict_diff_to_list_diff(cls, diff, depth: list, diff_type=CHANGED): + """ + Recursively handles the differences in a dictionary and returns a list of handled differences. + + Args: + diff (dict): The dictionary containing the differences. + depth (list): The list representing the current depth in the + dictionary. + diff_type (str, optional): The type of difference to handle. Defaults to CHANGED. + + Returns: + list: A list of handled differences. + + """ + handled_diff = [] + if (CHANGED in diff or ADDED in diff or REMOVED in diff) and isinstance(diff, dict): + handled_diff.extend(cls._dict_diff_to_list_diff(diff[diff_type], depth, diff_type)) + else: + for k, v in diff.items(): + tmpd = copy.deepcopy(depth) + tmpd.append(k) + + if ("to" in v or "from" in v) and isinstance(v, dict): + tmpd.extend(["from", v["from"], "to", v["to"]]) + handled_diff.append(tmpd) + elif isinstance(v, dict): + handled_diff.extend(cls._dict_diff_to_list_diff(v, tmpd, diff_type)) + else: + tmpd.append(v) + handled_diff.append(tmpd) + return handled_diff + + @classmethod + def extract_port_scan_dict_results(cls, results): + """ + Extracts the port scan results from the provided `results` object and returns a list of dictionaries. + + Args: + results (object): The scan results object. + + Returns: + list: A list of dictionaries containing the extracted scan results. + + Raises: + Exception: If an error occurs during the scan parser. + + """ + try: + scan_results = [] + for host in results.hosts: + scan = {} + scan["host"] = host.address + scan["status"] = host.status + scan["ports"] = [] + for s in host.services: + scan["ports"].append({ + "portid": str(s._portid), + "proto": str(s._protocol), + "state": s._state, + "service": s.service, + "servicefp": "none" if isinstance(s.servicefp, str) and s.servicefp == "" else s.servicefp, + "service_product": "none" if isinstance(s.banner, str) and s.banner == "" else s.banner, + }) + + scan["os"] = [] + try: + for _idx in range(3): + scan["os"].append( + host._extras["os"]["osmatches"][_idx]["osmatch"]["name"]) + except (KeyError, IndexError): + if len(scan["os"]) == 0: + scan["os"] = ["none"] + else: + pass + + scan["hops"] = [] + try: + for _hop in host._extras["trace"]["hops"]: + scan["hops"].append({_k: _hop[_k] for _k in ["ipaddr", "host"]}) + except (KeyError, IndexError): + if len(scan["hops"]) == 0: + scan["hops"] = ["none"] + else: + pass + + try: + scan["osfingerprint"] = host._extras["os"]["osfingerprints"][0]["fingerprint"] + except (KeyError, IndexError): + scan["osfingerprint"] = "none" + + try: + scan["last_boot"] = host._extras["uptime"]["lastboot"] + except KeyError: + scan["last_boot"] = "none" + + scan_results.append(scan) + return scan_results + except Exception as e: + raise DScanResultsParsingError(f"An error occurred with the scan parser: {str(e)}") + diff --git a/deltascan/core/scanner.py b/deltascan/core/scanner.py index 50a5d2a..e5d5a96 100644 --- a/deltascan/core/scanner.py +++ b/deltascan/core/scanner.py @@ -1,5 +1,6 @@ from deltascan.core.nmap.libnmap_wrapper import LibNmapWrapper from deltascan.core.config import LOG_CONF +from deltascan.core.parser import Parser import logging @@ -34,7 +35,7 @@ def scan(cls, target=None, scan_args=None, ui_context=None, logger=None, name=No try: scan_results = LibNmapWrapper.scan(target, scan_args, ui_context, logger=cls.logger, name=name, _cancel_evt=_cancel_evt) - scan_results = cls._extract_port_scan_dict_results(scan_results) + scan_results = Parser.extract_port_scan_dict_results(scan_results) if scan_results is None: raise ValueError("Failed to parse scan results") @@ -43,71 +44,3 @@ def scan(cls, target=None, scan_args=None, ui_context=None, logger=None, name=No except Exception as e: cls.logger.error(f"An error ocurred with nmap: {str(e)}") - - @classmethod - def _extract_port_scan_dict_results(cls, results): - """ - Extracts the port scan results from the provided `results` object and returns a list of dictionaries. - - Args: - results (object): The scan results object. - - Returns: - list: A list of dictionaries containing the extracted scan results. - - Raises: - Exception: If an error occurs during the scan parser. - - """ - try: - scan_results = [] - for host in results.hosts: - scan = {} - scan["host"] = host.address - scan["status"] = host.status - scan["ports"] = [] - for s in host.services: - scan["ports"].append({ - "portid": str(s._portid), - "proto": str(s._protocol), - "state": s._state, - "service": s.service, - "servicefp": "none" if isinstance(s.servicefp, str) and s.servicefp == "" else s.servicefp, - "service_product": "none" if isinstance(s.banner, str) and s.banner == "" else s.banner, - }) - - scan["os"] = [] - try: - for _idx in range(3): - scan["os"].append( - host._extras["os"]["osmatches"][_idx]["osmatch"]["name"]) - except (KeyError, IndexError): - if len(scan["os"]) == 0: - scan["os"] = ["none"] - else: - pass - - scan["hops"] = [] - try: - for _hop in host._extras["trace"]["hops"]: - scan["hops"].append({_k: _hop[_k] for _k in ["ipaddr", "host"]}) - except (KeyError, IndexError): - if len(scan["hops"]) == 0: - scan["hops"] = ["none"] - else: - pass - - try: - scan["osfingerprint"] = host._extras["os"]["osfingerprints"][0]["fingerprint"] - except (KeyError, IndexError): - scan["osfingerprint"] = "none" - - try: - scan["last_boot"] = host._extras["uptime"]["lastboot"] - except KeyError: - scan["last_boot"] = "none" - - scan_results.append(scan) - return scan_results - except Exception as e: - cls.logger.error(f"An error occurred with the scan parser: {str(e)}") diff --git a/deltascan/core/schemas.py b/deltascan/core/schemas.py index 9615829..30bf4ea 100644 --- a/deltascan/core/schemas.py +++ b/deltascan/core/schemas.py @@ -11,6 +11,7 @@ class ConfigSchema(Schema): single = fields.Bool(allow_none=True) template_file = fields.Str(allow_none=True) import_file = fields.Str(allow_none=True) + diff_files = fields.Str(allow_none=True) action = fields.Str(allow_none=True) profile = fields.Str(allow_none=True) conf_file = fields.Str(allow_none=True) diff --git a/deltascan/core/utils.py b/deltascan/core/utils.py index 78d02c1..9275121 100644 --- a/deltascan/core/utils.py +++ b/deltascan/core/utils.py @@ -1,16 +1,9 @@ import hashlib from datetime import datetime -from deltascan.core.config import ( - APP_DATE_FORMAT, - ADDED, - CHANGED, - REMOVED) -from deltascan.core.schemas import Diffs -from deltascan.core.exceptions import DScanResultsSchemaException -from marshmallow import ValidationError +from deltascan.core.config import (APP_DATE_FORMAT) + import re import os -import copy def n_hosts_on_subnet(subnet: str) -> int: @@ -112,73 +105,6 @@ def validate_port_state_type(port_status_type): return False return True - -def diffs_to_output_format(diffs): - """ - Convert the given diffs to a specific output format. - - Args: - diffs (dict): The diffs to be converted. - - Returns: - dict: The converted diffs in the specified output format. - - Raises: - DScanResultsSchemaException: If the diffs have an invalid schema. - """ - try: - Diffs().load(diffs) - except (KeyError, ValidationError) as e: - raise DScanResultsSchemaException(f"Invalid diff results schema: {str(e)}") - - # Here, entity can be many things. In the future an entity, besides port - # can be a service, a host, the osfingerpint. - articulated_diffs = { - ADDED: [], - CHANGED: [], - REMOVED: [], - } - - articulated_diffs[ADDED] = _dict_diff_to_list_diff(diffs["diffs"], [], ADDED) - articulated_diffs[CHANGED] = _dict_diff_to_list_diff(diffs["diffs"], [], CHANGED) - articulated_diffs[REMOVED] = _dict_diff_to_list_diff(diffs["diffs"], [], REMOVED) - - return articulated_diffs - - -def _dict_diff_to_list_diff(diff, depth: list, diff_type=CHANGED): - """ - Recursively handles the differences in a dictionary and returns a list of handled differences. - - Args: - diff (dict): The dictionary containing the differences. - depth (list): The list representing the current depth in the - dictionary. - diff_type (str, optional): The type of difference to handle. Defaults to CHANGED. - - Returns: - list: A list of handled differences. - - """ - handled_diff = [] - if (CHANGED in diff or ADDED in diff or REMOVED in diff) and isinstance(diff, dict): - handled_diff.extend(_dict_diff_to_list_diff(diff[diff_type], depth, diff_type)) - else: - for k, v in diff.items(): - tmpd = copy.deepcopy(depth) - tmpd.append(k) - - if ("to" in v or "from" in v) and isinstance(v, dict): - tmpd.extend(["from", v["from"], "to", v["to"]]) - handled_diff.append(tmpd) - elif isinstance(v, dict): - handled_diff.extend(_dict_diff_to_list_diff(v, tmpd, diff_type)) - else: - tmpd.append(v) - handled_diff.append(tmpd) - return handled_diff - - def format_string(string: str) -> str: """ Formats a string by making the first letter uppercase and replacing underscores with white spaces. @@ -214,3 +140,4 @@ def nmap_arguments_to_list(arguments): _arguments = [_arg for _arg in _arguments.split(" ") if _arg != "" and _arg != " "] return _arguments + diff --git a/tests/conftest.py b/tests/conftest.py index 6792786..7db5de5 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -38,6 +38,7 @@ class Config: single: bool template_file: str import_file: str + diff_files: str action: str profile: str conf_file: str diff --git a/tests/unit/test_main.py b/tests/unit/test_main.py index 4df36d5..49dd671 100644 --- a/tests/unit/test_main.py +++ b/tests/unit/test_main.py @@ -22,6 +22,7 @@ def setUp(self): "single": False, "template_file": None, "import_file": None, + "diff_files": None, "action": "view", "profile": "TEST_V1", "conf_file": CONFIG_FILE, @@ -179,7 +180,7 @@ def test_results_to_port_dict_success(self): _results_to_port_dict_results = SCANS_FROM_DB_TEST_V1_PORTS_KEYS[0] _results_to_port_dict_results["result_hash"] = mock_data_with_real_hash(SCANS_FROM_DB_TEST_V1)[0]["result_hash"] self.assertEqual( - self.dscan._results_to_port_dict(SCANS_FROM_DB_TEST_V1[0]), + self.dscan._results_to_port_dict(SCANS_FROM_DB_TEST_V1[0]["results"]), _results_to_port_dict_results["results"] ) diff --git a/tests/unit/test_parser.py b/tests/unit/test_parser.py new file mode 100644 index 0000000..f98f77c --- /dev/null +++ b/tests/unit/test_parser.py @@ -0,0 +1,31 @@ +import unittest +from deltascan.core.parser import Parser +from .test_data.mock_data import (DIFFS, ARTICULATED_DIFFS) + +class TestParser(unittest.TestCase): + def test_dict_diff_to_list_diff(self): + result = Parser._dict_diff_to_list_diff(DIFFS[0]["diffs"], [], "added") + self.assertEqual(result, ARTICULATED_DIFFS[0]["added"]) + + result = Parser._dict_diff_to_list_diff(DIFFS[0]["diffs"], [], "changed") + self.assertEqual(result, ARTICULATED_DIFFS[0]["changed"]) + + result = Parser._dict_diff_to_list_diff(DIFFS[0]["diffs"], [], "removed") + self.assertEqual(result, ARTICULATED_DIFFS[0]["removed"]) + + result = Parser._dict_diff_to_list_diff(DIFFS[1]["diffs"], [], "added") + self.assertEqual(result, ARTICULATED_DIFFS[1]["added"]) + + result = Parser._dict_diff_to_list_diff(DIFFS[1]["diffs"], [], "changed") + self.assertEqual(result, ARTICULATED_DIFFS[1]["changed"]) + + result = Parser._dict_diff_to_list_diff(DIFFS[1]["diffs"], [], "removed") + self.assertEqual(result, ARTICULATED_DIFFS[1]["removed"]) + + + def test_diffs_to_output_format(self): + results = Parser.diffs_to_output_format(DIFFS[0]) + self.assertEqual(results, ARTICULATED_DIFFS[0]) + + results = Parser.diffs_to_output_format(DIFFS[1]) + self.assertEqual(results, ARTICULATED_DIFFS[1]) \ No newline at end of file diff --git a/tests/unit/test_scanner.py b/tests/unit/test_scanner.py index c5c95ab..df737c3 100644 --- a/tests/unit/test_scanner.py +++ b/tests/unit/test_scanner.py @@ -1,7 +1,6 @@ import unittest from unittest.mock import MagicMock, patch from .test_data.mock_data import (SCAN_NMAP_RESULTS) -import json from deltascan.core.scanner import Scanner from dotmap import DotMap @@ -9,12 +8,12 @@ class TestScanner(unittest.TestCase): - @patch("deltascan.core.scanner.Scanner._extract_port_scan_dict_results", MagicMock()) + @patch("deltascan.core.scanner.Parser.extract_port_scan_dict_results") @patch("deltascan.core.scanner.LibNmapWrapper.scan") - def test_scan_calls(self, mock_nmap): + def test_scan_calls(self, mock_nmap, mock_extract_port_scan_dict_results): self._scanner = Scanner self._scanner.scan("0.0.0.0", "-vv") - self._scanner._extract_port_scan_dict_results.assert_called_once() + mock_extract_port_scan_dict_results.assert_called_once() mock_nmap.assert_called_once() def test_scan(self): diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index 0dc1990..7089d55 100644 --- a/tests/unit/test_utils.py +++ b/tests/unit/test_utils.py @@ -6,8 +6,6 @@ check_root_permissions, find_ports_from_state, validate_port_state_type, - diffs_to_output_format, - _dict_diff_to_list_diff, format_string) from unittest.mock import MagicMock, patch from .test_data.mock_data import (DIFFS, ARTICULATED_DIFFS) @@ -80,32 +78,6 @@ def test_validate_port_state_type(self): r = validate_port_state_type(["wronf"]) self.assertEqual(r, False) - - def test_diffs_to_output_format(self): - results = diffs_to_output_format(DIFFS[0]) - self.assertEqual(results, ARTICULATED_DIFFS[0]) - - results = diffs_to_output_format(DIFFS[1]) - self.assertEqual(results, ARTICULATED_DIFFS[1]) - - def test_dict_diff_to_list_diff(self): - result = _dict_diff_to_list_diff(DIFFS[0]["diffs"], [], "added") - self.assertEqual(result, ARTICULATED_DIFFS[0]["added"]) - - result = _dict_diff_to_list_diff(DIFFS[0]["diffs"], [], "changed") - self.assertEqual(result, ARTICULATED_DIFFS[0]["changed"]) - - result = _dict_diff_to_list_diff(DIFFS[0]["diffs"], [], "removed") - self.assertEqual(result, ARTICULATED_DIFFS[0]["removed"]) - - result = _dict_diff_to_list_diff(DIFFS[1]["diffs"], [], "added") - self.assertEqual(result, ARTICULATED_DIFFS[1]["added"]) - - result = _dict_diff_to_list_diff(DIFFS[1]["diffs"], [], "changed") - self.assertEqual(result, ARTICULATED_DIFFS[1]["changed"]) - - result = _dict_diff_to_list_diff(DIFFS[1]["diffs"], [], "removed") - self.assertEqual(result, ARTICULATED_DIFFS[1]["removed"]) def test_format_string(self): new_string = format_string("This is a test string")