import weasyprint import os import json import hashlib import re import sys from weasyprint import HTML from pathlib import Path from datetime import datetime from utils import get_config class Report(object): def __init__(self, capture_directory): self.capture_directory = capture_directory self.alerts = self.read_json(os.path.join( capture_directory, "assets/alerts.json")) self.whitelist = self.read_json(os.path.join( capture_directory, "assets/whitelist.json")) self.conns = self.read_json(os.path.join( capture_directory, "assets/conns.json")) self.device = self.read_json(os.path.join( capture_directory, "assets/device.json")) self.capinfos = self.read_json(os.path.join( capture_directory, "assets/capinfos.json")) try: with open(os.path.join(self.capture_directory, "capture.pcap"), "rb") as f: self.capture_sha1 = hashlib.sha1(f.read()).hexdigest() except: self.capture_sha1 = "N/A" self.userlang = get_config(("frontend", "user_lang")) # Load template language if not re.match("^[a-z]{2,3}$", self.userlang): self.userlang = "en" with open(os.path.join(os.path.dirname(os.path.realpath(sys.argv[0])), "locales/{}.json".format(self.userlang))) as f: self.template = json.load(f)["report"] def read_json(self, json_path): """ Read and convert a JSON file. :return: array or dict. """ with open(json_path, "r") as json_file: return json.load(json_file) def generate_report(self): """ Generate the full report in PDF :return: nothing """ content = self.generate_page_header() content += self.generate_header() content += self.generate_warning() content += self.generate_alerts() content += self.generate_suspect_conns_block() content += self.generate_uncat_conns_block() content += self.generate_whitelist_block() htmldoc = HTML(string=content, base_url="").write_pdf() Path(os.path.join(self.capture_directory, "report.pdf")).write_bytes(htmldoc) def generate_warning(self): """ Generate the warning message. :return: str """ if len(self.alerts["high"]): msg = "
" msg += self.template["high_msg"].format( self.nb_translate(len(self.alerts["high"]))) msg += "
" return msg elif len(self.alerts["moderate"]): msg = "
" msg += self.template["moderate_msg"].format( self.nb_translate(len(self.alerts["moderate"]))) msg += "
" return msg elif len(self.alerts["low"]): msg = "
" msg += self.template["low_msg"].format( self.nb_translate(len(self.alerts["low"]))) msg += "
" return msg else: msg = "
" msg += self.template["none_msg"] msg += "
" return msg def nb_translate(self, nb): """ Translate a number in a string. :return: str """ a = self.template["numbers"] return a[nb-1] if nb <= 9 else str(nb) def generate_suspect_conns_block(self): """ Generate the table of the network non-whitelisted communications. :return: string """ if not len([c for c in self.conns if c["alert_tiggered"] == True]): return "" title = "

{}

".format(self.template["suspect_title"]) table = "" table += " " table += " " table += " ".format(self.template["protocol"]) table += " ".format(self.template["domain"]) table += " ".format(self.template["dst_ip"]) table += " ".format(self.template["dst_port"]) table += " " table += " " table += "" for rec in self.conns: if rec["alert_tiggered"] == True: table += "" table += "".format(rec["proto"].upper()) table += "".format(rec["resolution"] if rec["resolution"] != rec["ip_dst"] else "--") table += "".format(rec["ip_dst"]) table += "".format(rec["port_dst"]) table += "" table += "
{}{}{}{}
{}{}{}{}
" return title + table def generate_uncat_conns_block(self): """ Generate the table of the network non-whitelisted communications. :return: string """ if not len([c for c in self.conns if c["alert_tiggered"] == False]): return "" title = "

{}

".format(self.template["uncat_title"]) table = "" table += " " table += " " table += " ".format(self.template["protocol"]) table += " ".format(self.template["domain"]) table += " ".format(self.template["dst_ip"]) table += " ".format(self.template["dst_port"]) table += " " table += " " table += "" for rec in self.conns: if rec["alert_tiggered"] == False: table += "" table += "".format(rec["proto"].upper()) table += "".format(rec["resolution"] if rec["resolution"] != rec["ip_dst"] else "--") table += "".format(rec["ip_dst"]) table += "".format(rec["port_dst"]) table += "" table += "
{}{}{}{}
{}{}{}{}
" return title + table def generate_whitelist_block(self): """ Generate the table of the whitelisted communications. :return: string """ if not len(self.whitelist): return "" title = "

{}

".format(self.template["whitelist_title"]) table = "" table += " " table += " " table += " ".format(self.template["protocol"]) table += " ".format(self.template["domain"]) table += " ".format(self.template["dst_ip"]) table += " ".format(self.template["dst_port"]) table += " " table += " " table += "" for rec in sorted(self.whitelist, key=lambda k: k['resolution']): table += "" table += "".format(rec["proto"].upper()) table += "".format(rec["resolution"] if rec["resolution"] != rec["ip_dst"] else "--") table += "".format(rec["ip_dst"]) table += "".format(rec["port_dst"]) table += "" table += "
{}{}{}{}
{}{}{}{}
" return title + table def generate_header(self): """ Generate the report header with context data. :return: string """ header = "
" header += "
" header += "


{}: {}
".format(self.template["device_name"], self.device["name"]) header += "{}: {}
".format(self.template["device_mac"], self.device["mac_address"]) header += "{} {}
".format(self.template["report_generated_on"], datetime.now().strftime("%d/%m/%Y - %H:%M:%S")) header += "{}: {}s
".format(self.template["capture_duration"], self.capinfos["Capture duration"]) header += "{}: {}
".format(self.template["packets_number"], self.capinfos["Number of packets"]) header += "{}: {}
".format( self.template["capture_sha1"], self.capture_sha1) header += "

" header += "
" return header def generate_alerts(self): """ Generate the alerts. :return: string """ alerts = "" return alerts def generate_page_footer(self): """ Generate the html footer. :return: string """ return "" def generate_page_header(self): """ Generate the html header. :return: string """ return """ """.replace("REPORT_HEADER", "{} {}".format(self.template["report_for_the_capture"], self.capture_sha1)).replace("REPORT_FOOTER", self.template["report_footer"])