Last active
September 16, 2026 15:17
-
-
Save gitgotgitgotit/1dda3db240f63b68cfa000dc68df8800 to your computer and use it in GitHub Desktop.
Nessus report downloader + merger
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python3 | |
| ####################################################################################### | |
| # Name: Nessus Report Toolkit (download + merge) | |
| # | |
| # Combines and extends: | |
| # - nessus_report_downloader.py (by Nikhil Raj, nikhilraj149@gmail.com) | |
| # - nessus_merge_vibe.py | |
| # | |
| # Description: Download scan reports from a Nessus server, merge existing .nessus | |
| # files into a single combined report, or do both in one pass | |
| # (download an entire folder, then merge the results). | |
| # | |
| # Usage examples: | |
| # List scans on a server: | |
| # python nessus_toolkit.py -i 127.0.0.1 -u admin -p secret | |
| # | |
| # Download specific scan ids: | |
| # python nessus_toolkit.py -i 127.0.0.1 -u admin -p secret -s 12,15 | |
| # | |
| # Download specific scan ids AND merge them into one file: | |
| # python nessus_toolkit.py -i 127.0.0.1 -u admin -p secret -s 12,15 --merge \ | |
| # -o combined.nessus -r merged_scans | |
| # | |
| # Download every scan in folder(s) 3 and 7, then merge into one file: | |
| # python nessus_toolkit.py -i 127.0.0.1 -u admin -p secret --folder-id 3,7 \ | |
| # --download-folder-then-merge -o combined.nessus -r merged_scans | |
| # | |
| # Merge .nessus files already on disk (no server needed): | |
| # python nessus_toolkit.py --merge-only -d ./reports -o combined.nessus -r merged_scans | |
| # | |
| # Requirements: requests (required), prettytable (optional, nicer scan listing) | |
| ####################################################################################### | |
| import os | |
| import sys | |
| import json | |
| import time | |
| import getpass | |
| import argparse | |
| import xml.etree.ElementTree as ET | |
| from datetime import datetime | |
| import requests | |
| try: | |
| from prettytable import PrettyTable | |
| HAS_PRETTYTABLE = True | |
| except ImportError: | |
| HAS_PRETTYTABLE = False | |
| requests.packages.urllib3.disable_warnings(requests.packages.urllib3.exceptions.InsecureRequestWarning) | |
| SLEEP_TIME = 1.0 | |
| # --------------------------------------------------------------------------------- # | |
| # HTTP helpers | |
| # --------------------------------------------------------------------------------- # | |
| def sendGetRequest(url, headers): | |
| try: | |
| return requests.get(url, verify=False, headers=headers) | |
| except requests.exceptions.ConnectionError: | |
| printMessage("Failed to establish connection", 0) | |
| sys.exit(1) | |
| def sendPostRequest(url, json_data=None, headers=None): | |
| try: | |
| return requests.post(url, verify=False, headers=headers or {}, json=json_data or {}) | |
| except requests.exceptions.ConnectionError: | |
| printMessage("Failed to establish connection", 0) | |
| sys.exit(1) | |
| def sendDeleteRequest(url, json_data=None, headers=None): | |
| try: | |
| return requests.delete(url, verify=False, headers=headers or {}, json=json_data or {}) | |
| except requests.exceptions.ConnectionError: | |
| printMessage("Failed to establish connection", 0) | |
| sys.exit(1) | |
| def printMessage(msg, flag=1): | |
| if flag == 1: | |
| print("[+] " + msg) | |
| elif flag == 0: | |
| print("[-] " + msg) | |
| elif flag == 2: | |
| print("[*] " + msg) | |
| else: | |
| print(msg) | |
| def checkStatus(resp, status_msg, error_msg): | |
| if resp.status_code == 200: | |
| printMessage(status_msg, 1) | |
| return True | |
| printMessage(error_msg, 0) | |
| return False | |
| # --------------------------------------------------------------------------------- # | |
| # Scan listing helpers | |
| # --------------------------------------------------------------------------------- # | |
| def printTable(data, table_headers): | |
| tab = PrettyTable(table_headers) | |
| for row in data: | |
| line = [] | |
| for header in table_headers: | |
| if "date" in header: | |
| line.append(datetime.fromtimestamp(int(row[header])).strftime('%Y-%m-%d %H:%M:%S')) | |
| else: | |
| line.append(str(row[header])) | |
| tab.add_row(line) | |
| print(tab) | |
| def printScanData(scan_data): | |
| folder_info = {folder["id"]: folder["name"] for folder in scan_data["folders"]} | |
| if HAS_PRETTYTABLE: | |
| printTable(scan_data["scans"], ["id", "name", "folder_id", "status", "creation_date", "last_modification_date"]) | |
| else: | |
| print('\t %-10s %-20s %-20s %-40s %-20s %-20s' % ( | |
| "Scan Id", "Folder Name (id)", "Scan status", "Scan Name", "creation_date", "last_modification_date")) | |
| print('\t %-10s %-20s %-20s %-40s %-20s %-20s' % ( | |
| "-------", "---------------", "------------", "-----------------", "-------------------", "--------------------")) | |
| for scan in scan_data["scans"]: | |
| print('\t %-10s %-20s %-20s %-40s %-20s %-20s' % ( | |
| str(scan["id"]), | |
| folder_info[scan["folder_id"]] + ' (' + str(scan["folder_id"]) + ') ', | |
| scan["status"], scan["name"], | |
| datetime.fromtimestamp(int(scan["creation_date"])).strftime('%Y-%m-%d %H:%M:%S'), | |
| datetime.fromtimestamp(int(scan["last_modification_date"])).strftime('%Y-%m-%d %H:%M:%S'))) | |
| print() | |
| def verifyScanId(scan_data, ui_scan_id): | |
| master_ids = [scan["id"] for scan in scan_data["scans"]] | |
| if ui_scan_id == "all": | |
| return master_ids | |
| valid = [] | |
| for scan in ui_scan_id.split(","): | |
| if int(scan) in master_ids: | |
| valid.append(scan) | |
| else: | |
| printMessage("Omitting invalid Scan ID: " + scan, 0) | |
| return valid | |
| def verifyFolderId(scan_data, ui_folder_id): | |
| master_folder_ids = [folder["id"] for folder in scan_data["folders"]] | |
| valid_folder_ids = [] | |
| for folder_id in ui_folder_id.split(","): | |
| if int(folder_id) in master_folder_ids: | |
| valid_folder_ids.append(folder_id) | |
| else: | |
| printMessage("Omitting invalid folder ID: " + folder_id, 0) | |
| scan_ids = [] | |
| for scan in scan_data["scans"]: | |
| for folder_id in valid_folder_ids: | |
| if int(scan["folder_id"]) == int(folder_id): | |
| scan_ids.append(scan["id"]) | |
| return scan_ids | |
| # --------------------------------------------------------------------------------- # | |
| # Download | |
| # --------------------------------------------------------------------------------- # | |
| def getFormatAndChapterList(nessus_format_list, chapter_list, db_pass): | |
| data = [] | |
| chapter_map = { | |
| "0": "vuln_hosts_summary", "1": "vuln_by_host", "2": "vuln_by_plugin", | |
| "3": "compliance_exec", "4": "compliance", "5": "remediations", | |
| } | |
| for nessus_format in nessus_format_list: | |
| if nessus_format == "0": | |
| data.append({'format': 'nessus', 'chapters': ''}) | |
| elif nessus_format in ("1", "2"): | |
| fmt = 'pdf' if nessus_format == "1" else 'html' | |
| for chapter in chapter_list: | |
| if chapter in chapter_map: | |
| data.append({'format': fmt, 'chapters': chapter_map[chapter]}) | |
| elif nessus_format == "3": | |
| data.append({'format': 'csv', 'chapters': ''}) | |
| elif nessus_format == "4": | |
| data.append({'format': 'db', 'chapters': '', 'password': db_pass}) | |
| return data | |
| def downloadNessusReport(base_url, token, scan_id_list, json_user_data, download_dir): | |
| """Downloads one format/chapter combo for every scan id and returns the list of | |
| file paths that were actually written to disk.""" | |
| saved_files = [] | |
| os.makedirs(download_dir, exist_ok=True) | |
| token_header = {'X-Cookie': 'token=' + token['token']} | |
| for scan_id in scan_id_list: | |
| printMessage("Format: {0} | Chapter: {1}".format(json_user_data["format"], json_user_data.get("chapters", ""))) | |
| printMessage("Initiating download request for scan id: " + str(scan_id), 1) | |
| url = base_url + "/scans/{0}/export".format(scan_id) | |
| resp = sendPostRequest(url, json_data=json_user_data, headers=token_header) | |
| file_token = json.loads(resp.text) | |
| url = base_url + "/scans/{0}/export/{1}/status".format(scan_id, file_token["file"]) | |
| resp2 = sendGetRequest(url, headers=token_header) | |
| while json.loads(resp2.text)["status"] == "loading": | |
| printMessage("Report is not ready yet, waiting {0}s".format(SLEEP_TIME), 0) | |
| time.sleep(SLEEP_TIME) | |
| resp2 = sendGetRequest(url, headers=token_header) | |
| if json.loads(resp2.text)["status"] != "ready": | |
| printMessage("Report never became ready for scan id: " + str(scan_id), 0) | |
| continue | |
| url = base_url + "/tokens/{0}/download".format(file_token["token"]) | |
| resp3 = sendGetRequest(url, headers=token_header) | |
| if not checkStatus(resp3, "Started downloading the nessus report", "Unable to download scan: " + str(scan_id)): | |
| continue | |
| filename = resp3.headers["Content-Disposition"].split('"')[1] | |
| filepath = os.path.join(download_dir, filename) | |
| try: | |
| with open(filepath, "w") as fh: | |
| fh.write(resp3.text) | |
| except (IOError, UnicodeEncodeError): | |
| base, ext = os.path.splitext(filename) | |
| chapters = json_user_data.get("chapters", "") | |
| filename2 = f"{base}_{chapters}{ext}" if chapters else filename | |
| filepath = os.path.join(download_dir, filename2) | |
| with open(filepath, "wb") as fh: | |
| fh.write(resp3.content) | |
| printMessage("Report was saved in " + filepath, 1) | |
| printMessage("", 99) | |
| saved_files.append(filepath) | |
| return saved_files | |
| # --------------------------------------------------------------------------------- # | |
| # Merge | |
| # --------------------------------------------------------------------------------- # | |
| def get_nessus_files(directory): | |
| if not os.path.isdir(directory): | |
| return [] | |
| return sorted( | |
| os.path.join(directory, f) | |
| for f in os.listdir(directory) | |
| if f.lower().endswith(".nessus") | |
| ) | |
| def merge_nessus_files(files, output_file, report_name="merged_scans"): | |
| """Merges every <ReportHost> (etc.) from each input file's <Report> into a | |
| single combined <Report name="report_name"> element, rather than just stacking | |
| multiple <Report> tags side by side under the root.""" | |
| if not files: | |
| printMessage("No .nessus files to merge", 0) | |
| return None | |
| combined_root = ET.Element("NessusClientData_v2") | |
| report_elem = ET.SubElement(combined_root, "Report") | |
| report_elem.set("name", report_name) | |
| policy_included = False | |
| for file_path in files: | |
| try: | |
| xml_root = ET.parse(file_path).getroot() | |
| # Keep one <Policy> block (from the first file that has one) so the | |
| # combined file stays structurally valid. | |
| if not policy_included: | |
| policy = xml_root.find("Policy") | |
| if policy is not None: | |
| combined_root.insert(0, policy) | |
| policy_included = True | |
| for report in xml_root.findall(".//Report"): | |
| for child in list(report): | |
| report_elem.append(child) | |
| printMessage("Merged: " + os.path.basename(file_path), 1) | |
| except ET.ParseError as e: | |
| printMessage(f"Skipping invalid file: {file_path} ({e})", 0) | |
| output_dir = os.path.dirname(os.path.abspath(output_file)) | |
| os.makedirs(output_dir, exist_ok=True) | |
| ET.ElementTree(combined_root).write(output_file, encoding="utf-8", xml_declaration=True) | |
| printMessage(f'Combined report (name="{report_name}") written to: {output_file}', 1) | |
| return output_file | |
| # --------------------------------------------------------------------------------- # | |
| # CLI | |
| # --------------------------------------------------------------------------------- # | |
| def build_arg_parser(): | |
| parser = argparse.ArgumentParser( | |
| prog="nessus_toolkit.py", | |
| description="Download Nessus scan reports and/or merge .nessus files into a single combined report.", | |
| formatter_class=argparse.RawDescriptionHelpFormatter, | |
| epilog="""examples: | |
| list scans on a server (prompts for password, hidden input) | |
| %(prog)s -i 127.0.0.1 -u admin | |
| same, with the password passed on the command line instead | |
| %(prog)s -i 127.0.0.1 -u admin -p secret | |
| download specific scan ids | |
| %(prog)s -i 127.0.0.1 -u admin -p secret -s 12,15 | |
| download specific scan ids, saving to a folder, AND merge them into one file | |
| %(prog)s -i 127.0.0.1 -u admin -p secret -s 12,15 -O ./raw -m -o combined.nessus -r merged_scans | |
| download every scan in folder(s), then merge into one file | |
| %(prog)s -i 127.0.0.1 -u admin -p secret -F 3,7 -x -o combined.nessus -r merged_scans | |
| merge .nessus files already on disk (no server needed) | |
| %(prog)s -M -d ./reports -o combined.nessus -r merged_scans | |
| every long option also has a short form (see below); -x is the same as | |
| --download-folder-then-merge / --downloader-folder-then-merge. | |
| """ | |
| ) | |
| conn = parser.add_argument_group("server connection (required unless --merge-only)") | |
| conn.add_argument("-i", "--server", help="IP[:PORT] of nessus server") | |
| conn.add_argument("-u", "--user", help="nessus username") | |
| conn.add_argument("-p", "--passwd", | |
| help="nessus password. If omitted, you'll be prompted for it securely " | |
| "(input is hidden, nothing is echoed to the screen)") | |
| dl = parser.add_argument_group("download options") | |
| dl.add_argument("-s", "--scan-id", help="comma separated scan id(s), or 'all'") | |
| dl.add_argument("-F", "--folder-id", help="comma separated nessus folder id(s)") | |
| dl.add_argument("-f", "--format", default="0", | |
| help="comma separated report format(s): 0-nessus (default), 1-pdf, 2-html, 3-csv, 4-nessus-db") | |
| dl.add_argument("-c", "--chapter", default="1", | |
| help="comma separated chapters: 0-vuln_hosts_summary, 1-vuln_by_host (default), " | |
| "2-vuln_by_plugin, 3-compliance_exec, 4-compliance, 5-remediations") | |
| dl.add_argument("-b", "--db-pass", default="nessus", help="password for nessus-db export (default: nessus)") | |
| dl.add_argument("-O", "--download-dir", default=".", help="directory to save downloaded reports into (default: current directory)") | |
| mg = parser.add_argument_group("merge options") | |
| mg.add_argument("-m", "--merge", action="store_true", | |
| help="merge the reports just downloaded into a single .nessus file") | |
| mg.add_argument("-M", "--merge-only", action="store_true", | |
| help="skip downloading; just merge existing .nessus files found in --merge-dir") | |
| mg.add_argument("-x", "--download-folder-then-merge", "--downloader-folder-then-merge", | |
| dest="download_folder_then_merge", action="store_true", | |
| help="shortcut: download every scan in --folder-id, then merge the results (implies --merge)") | |
| mg.add_argument("-d", "--merge-dir", | |
| help="directory of .nessus files to merge. For --merge-only this is the source directory " | |
| "(default: current directory). When merging right after a download, set this to merge " | |
| "every .nessus file found in that directory instead of just the ones just downloaded " | |
| "(default: whatever --download-dir was)") | |
| mg.add_argument("-o", "--output", default="combined_results.nessus", help="path of the merged output .nessus file") | |
| mg.add_argument("-r", "--report-name", default="merged_scans", | |
| help='name="" attribute to set on the merged <Report> element (default: merged_scans)') | |
| return parser | |
| def nessus_login(base_url, user, passwd): | |
| resp = sendGetRequest(base_url, {}) | |
| if not checkStatus(resp, "Connected to nessus server", "Unable to connect to server at " + base_url): | |
| sys.exit(1) | |
| resp = sendPostRequest(base_url + "/session", {'username': user, 'password': passwd}) | |
| if not checkStatus(resp, "Login successful", "Invalid login credentials"): | |
| sys.exit(1) | |
| return json.loads(resp.text) | |
| def nessus_logout(base_url, token): | |
| resp = sendDeleteRequest(base_url + "/session", headers={'X-Cookie': 'token=' + token['token']}) | |
| checkStatus(resp, "Successfully logged out user session", "Unable to logout the current active session") | |
| def do_merge_only(args): | |
| merge_dir = args.merge_dir or os.getcwd() | |
| files = get_nessus_files(merge_dir) | |
| if not files: | |
| printMessage("No .nessus files found in: " + merge_dir, 0) | |
| sys.exit(1) | |
| merge_nessus_files(files, args.output, args.report_name) | |
| def do_download(args): | |
| if not (args.server and args.user): | |
| printMessage("--server/-i and --user/-u are required unless --merge-only is used", 0) | |
| sys.exit(1) | |
| if not args.passwd: | |
| try: | |
| args.passwd = getpass.getpass("Enter password: ") | |
| except EOFError: | |
| printMessage("No password entered", 0) | |
| sys.exit(1) | |
| if not args.passwd: | |
| printMessage("Password cannot be empty", 0) | |
| sys.exit(1) | |
| if args.download_folder_then_merge and not args.folder_id: | |
| printMessage("--download-folder-then-merge requires --folder-id", 0) | |
| sys.exit(1) | |
| merge_after = args.merge or args.download_folder_then_merge | |
| if ":" in args.server: | |
| ip, port = args.server.split(":", 1) | |
| else: | |
| ip, port = args.server, "8834" | |
| base_url = "https://" + ip + ":" + port | |
| format_arg = args.format | |
| if merge_after and "0" not in format_arg.split(","): | |
| printMessage("Merging requires the 'nessus' (XML) export format; overriding --format to 0", 2) | |
| format_arg = "0" | |
| token = nessus_login(base_url, args.user, args.passwd) | |
| token_header = {'X-Cookie': 'token=' + token['token']} | |
| resp = sendGetRequest(base_url + "/scans", headers=token_header) | |
| if not checkStatus(resp, "Fetching scan reports\n", "Unable to fetch nessus scans"): | |
| nessus_logout(base_url, token) | |
| sys.exit(1) | |
| scan_data = json.loads(resp.text) | |
| if not args.scan_id and not args.folder_id: | |
| printScanData(scan_data) | |
| nessus_logout(base_url, token) | |
| return | |
| if args.scan_id: | |
| scan_id_list = verifyScanId(scan_data, args.scan_id) | |
| else: | |
| scan_id_list = verifyFolderId(scan_data, args.folder_id) | |
| printMessage("Identified " + str(len(scan_id_list)) + " scan(s) for download\n", 2) | |
| downloaded_files = [] | |
| format_specification = getFormatAndChapterList(format_arg.split(","), args.chapter.split(","), args.db_pass) | |
| for report_format in format_specification: | |
| downloaded_files += downloadNessusReport(base_url, token, scan_id_list, report_format, args.download_dir) | |
| nessus_logout(base_url, token) | |
| if merge_after: | |
| if args.merge_dir: | |
| files_to_merge = get_nessus_files(args.merge_dir) | |
| else: | |
| files_to_merge = [f for f in downloaded_files if f.lower().endswith(".nessus")] | |
| merge_nessus_files(files_to_merge, args.output, args.report_name) | |
| def main(): | |
| parser = build_arg_parser() | |
| if len(sys.argv) == 1: | |
| parser.print_help() | |
| sys.exit(1) | |
| args = parser.parse_args() | |
| if args.merge_only: | |
| do_merge_only(args) | |
| else: | |
| do_download(args) | |
| printMessage("✅ All done, here is your mandatory vibe coded emoji.") | |
| if __name__ == '__main__': | |
| main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment