Nessus report downloader + merger

Starred Gists，GitHub Gist，即时分享代码、笔记与代码片段。关注
#!/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 (etc.) from each input file's into a
single combined element, rather than just stacking
multiple 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 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 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()
添加评论
点赞收藏
点踩分享查看原文
评论
?
参与讨论