import os import re import math import pwd import grp from datetime import datetime from collections import Counter, defaultdict from html import escape # --------------------------------------------- # LOG FILE PATHS # --------------------------------------------- log_paths = { "LIVE": "/var/log/snort/alerts/alert_fast.txt", "TUESDAY": "/var/log/snort/pcap_results/tuesday/alert_fast.txt", "WEDNESDAY": "/var/log/snort/pcap_results/wednesday/alert_fast.txt", "FRIDAY": "/var/log/snort/pcap_results/friday/alert_fast.txt" } SESSION_ORDER = ["LIVE", "TUESDAY", "WEDNESDAY", "FRIDAY"] # --------------------------------------------- # OUTPUT PATHS # --------------------------------------------- report_dir = "/home/osboxes/nids-reports" detail_dir = os.path.join(report_dir, "details") os.makedirs(report_dir, exist_ok=True) os.makedirs(detail_dir, exist_ok=True) report_file = os.path.join(report_dir, "report_all.html") PAGE_SIZE = 1000 # --------------------------------------------- # SEVERITY MAP # --------------------------------------------- severity_map = { 1000001: "LOW", 1000002: "MEDIUM", 1000003: "MEDIUM", 1000004: "MEDIUM", 1000005: "MEDIUM", 1000006: "MEDIUM", 1000007: "MEDIUM", 1000008: "HIGH", 1000009: "HIGH", 1000010: "HIGH", 1000011: "HIGH", 1000012: "HIGH", 1000013: "MEDIUM", 1000014: "HIGH", 1000015: "MEDIUM", 1000016: "MEDIUM", 1000017: "LOW", 1000018: "LOW", 1000019: "BLOCKED", 1000020: "BLOCKED", 1000021: "BLOCKED", 1000022: "BLOCKED", 1000023: "BLOCKED", 1000024: "BLOCKED", 1000025: "BLOCKED", } # --------------------------------------------- # RULE COVERAGE TABLE # --------------------------------------------- rule_coverage = { # NIDS ALERT RULES 1000001: {"name": "ICMP Ping Detected", "attack": "Ping / Reconnaissance", "protocol": "ICMP", "mode": "NIDS"}, 1000002: {"name": "Possible SYN Scan", "attack": "Port Scanning", "protocol": "TCP", "mode": "NIDS"}, 1000003: {"name": "Suspicious HTTP Admin Probe", "attack": "Web Probe", "protocol": "HTTP", "mode": "NIDS"}, 1000004: {"name": "Possible TCP Portscan (burst)", "attack": "Port Scanning", "protocol": "TCP", "mode": "NIDS"}, 1000005: {"name": "Possible Xmas Scan", "attack": "Port Scanning", "protocol": "TCP", "mode": "NIDS"}, 1000006: {"name": "Possible NULL Scan", "attack": "Port Scanning", "protocol": "TCP", "mode": "NIDS"}, 1000007: {"name": "Possible UDP Scan (burst)", "attack": "Port Scanning", "protocol": "UDP", "mode": "NIDS"}, 1000008: {"name": "Possible SYN Flood (rate)", "attack": "Denial of Service", "protocol": "TCP", "mode": "NIDS"}, 1000009: {"name": "Possible ICMP Flood (rate)", "attack": "Denial of Service", "protocol": "ICMP", "mode": "NIDS"}, 1000010: {"name": "Possible DNS Query Flood (rate)", "attack": "Denial of Service", "protocol": "UDP", "mode": "NIDS"}, 1000011: {"name": "Possible SSH Brute Force", "attack": "Credential / Brute Force", "protocol": "TCP", "mode": "NIDS"}, 1000012: {"name": "Possible RDP Brute Force", "attack": "Credential / Brute Force", "protocol": "TCP", "mode": "NIDS"}, 1000013: {"name": "HTTP Directory Traversal Probe", "attack": "Web Application Attack", "protocol": "HTTP", "mode": "NIDS"}, 1000014: {"name": "Possible SQL Injection Attempt", "attack": "Web Application Attack", "protocol": "HTTP", "mode": "NIDS"}, 1000015: {"name": "HTTP Possible XSS Probe", "attack": "Web Application Attack", "protocol": "HTTP", "mode": "NIDS"}, 1000016: {"name": "HTTP Upload Endpoint Probe", "attack": "Web Application Attack", "protocol": "HTTP", "mode": "NIDS"}, 1000017: {"name": "Telnet Connection Attempt", "attack": "Unauthorised Access", "protocol": "TCP", "mode": "NIDS"}, 1000018: {"name": "SMB Connection Attempt", "attack": "Unauthorised Access", "protocol": "TCP", "mode": "NIDS"}, # IPS DROP RULES 1000019: {"name": "ICMP Flood BLOCKED", "attack": "Denial of Service (Blocked)", "protocol": "ICMP", "mode": "IPS"}, 1000020: {"name": "SSH Brute Force BLOCKED", "attack": "Credential Attack (Blocked)", "protocol": "TCP", "mode": "IPS"}, 1000021: {"name": "SYN Flood BLOCKED", "attack": "Denial of Service (Blocked)", "protocol": "TCP", "mode": "IPS"}, 1000022: {"name": "DNS Flood BLOCKED", "attack": "Denial of Service (Blocked)", "protocol": "UDP", "mode": "IPS"}, 1000023: {"name": "RDP Brute Force BLOCKED", "attack": "Credential Attack (Blocked)", "protocol": "TCP", "mode": "IPS"}, 1000024: {"name": "SQL Injection BLOCKED", "attack": "Web Application (Blocked)", "protocol": "HTTP", "mode": "IPS"}, 1000025: {"name": "XSS Probe BLOCKED", "attack": "Web Application (Blocked)", "protocol": "HTTP", "mode": "IPS"}, } # --------------------------------------------- # ALERT LINE REGEX # --------------------------------------------- alert_re = re.compile( r'^(?P\d{2}/\d{2}-\d{2}:\d{2}:\d{2}\.\d+)' r'\s+(?P\[drop\]\s+)?' r'\[\*\*\]\s+\[(?P\d+):(?P\d+):(?P\d+)\]' r'\s+"(?P.+?)"\s+\[\*\*\].*?' r'\{(?P[A-Z0-9]+)\}\s+' r'(?P\S+)\s+->\s+(?P\S+)' ) # --------------------------------------------- # PERMISSION HELPERS # --------------------------------------------- def fix_permissions(path, user="osboxes", group="osboxes"): try: uid = pwd.getpwnam(user).pw_uid gid = grp.getgrnam(group).gr_gid os.chown(path, uid, gid) except Exception: pass try: os.chmod(path, 0o755 if os.path.isdir(path) else 0o644) except Exception: pass def fix_tree_permissions(root_path, user="osboxes", group="osboxes"): for current_root, dirs, files in os.walk(root_path): fix_permissions(current_root, user, group) for d in dirs: fix_permissions(os.path.join(current_root, d), user, group) for f in files: fix_permissions(os.path.join(current_root, f), user, group) # --------------------------------------------- # PARSING # --------------------------------------------- def parse_endpoint(endpoint): endpoint = endpoint.strip() if endpoint == "-" or not endpoint: return "-", "-" if endpoint.startswith("[") and "]:" in endpoint: host, port = endpoint.rsplit("]:", 1) return host + "]", port if port.isdigit() else "-" if "." in endpoint and endpoint.count(":") == 1: host, port = endpoint.rsplit(":", 1) if port.isdigit(): return host, port if "." in endpoint: return endpoint, "-" return endpoint, "-" def parse_alert_file(path, session): alerts = [] if not os.path.exists(path): print(f"[WARN] {session}: file not found -> {path}", flush=True) return alerts print(f"[INFO] Parsing {session}: {path}", flush=True) with open(path, "r", encoding="utf-8", errors="ignore") as f: for line_num, line in enumerate(f, 1): line = line.strip() if not line: continue m = alert_re.search(line) if not m: continue sid = int(m.group("sid")) action = m.group("action") is_drop = action is not None and "drop" in action.lower() src_ip, src_port = parse_endpoint(m.group("src_raw")) dst_ip, dst_port = parse_endpoint(m.group("dst_raw")) if is_drop or sid in range(1000019, 1000026): sev = "BLOCKED" else: sev = severity_map.get(sid, "UNKNOWN") alerts.append({ "session": session, "timestamp": m.group("timestamp"), "sid": sid, "message": m.group("msg"), "severity": sev, "action": "DROP" if is_drop else "ALERT", "protocol": m.group("proto"), "src_ip": src_ip, "src_port": src_port, "dst_ip": dst_ip, "dst_port": dst_port }) if line_num % 50000 == 0: print(f"[INFO] {session}: read {line_num} lines, matched {len(alerts)} alerts", flush=True) print(f"[INFO] Finished {session}: {len(alerts)} alerts parsed", flush=True) return alerts # --------------------------------------------- # SHARED STYLES # --------------------------------------------- def base_style(): return """ """ def severity_badge(sev): sev_esc = escape(sev) cls = sev.lower() return f'{sev_esc}' def action_badge(action): if action == "DROP": return '⛔ DROP' return 'ALERT' def mode_badge(mode): if mode == "IPS": return 'IPS' return 'NIDS' # --------------------------------------------- # MAIN REPORT # --------------------------------------------- def build_main_report(all_alerts, alerts_by_session, session_totals, severity_totals, sid_totals, src_ip_totals, generated_at): parts = [] blocked_total = severity_totals.get("BLOCKED", 0) drop_alerts = [a for a in all_alerts if a["action"] == "DROP"] parts.append(f""" NIDS and IPS Alert Report {base_style()}

NIDS and IPS Alert Report

Snort 3 running on Ubuntu VM. Covers live traffic and PCAP replay sessions from CICIDS2017.

Report generated: {escape(generated_at)}

""") # Log file status parts.append("""

Log File Status

""") for session in SESSION_ORDER: path = log_paths[session] status = "FOUND" if os.path.exists(path) else "MISSING" count = session_totals.get(session, 0) parts.append( f"" f"" ) parts.append("
SessionPathStatusAlerts Parsed
{escape(session)}{escape(path)}{escape(status)}{count:,}
") # Overall summary cards total_alerts = len(all_alerts) parts.append(f"""

Overall Summary

Total Alerts
{total_alerts:,}
LIVE
{session_totals.get("LIVE", 0):,}
TUESDAY
{session_totals.get("TUESDAY", 0):,}
WEDNESDAY
{session_totals.get("WEDNESDAY", 0):,}
FRIDAY
{session_totals.get("FRIDAY", 0):,}
HIGH Severity
{severity_totals.get("HIGH", 0):,}
MEDIUM Severity
{severity_totals.get("MEDIUM", 0):,}
LOW Severity
{severity_totals.get("LOW", 0):,}
⛔ IPS BLOCKED
{blocked_total:,}
""") # Alerts by severity parts.append("""

Alerts by Severity

""") for sev in ["HIGH", "MEDIUM", "LOW", "BLOCKED", "UNKNOWN"]: count = severity_totals.get(sev, 0) pct = f"{(count / total_alerts * 100):.1f}%" if total_alerts else "0%" parts.append(f"") parts.append("
SeverityCount% of Total
{severity_badge(sev)}{count:,}{pct}
") # IPS Blocked Alerts parts.append("""

⛔ IPS Blocked Alerts

These are alerts from the IPS DROP rules (SIDs 1000019-1000025). Traffic matched by these rules was actively dropped by Snort. Only shows up in LIVE mode when running with --daq afpacket -Q.

""") ips_sids = range(1000019, 1000026) has_ips = False for sid in ips_sids: count = sid_totals.get(sid, 0) if count > 0: has_ips = True info = rule_coverage.get(sid, {}) parts.append( f"" f"" f"" f"" f"" f"" f"" ) if not has_ips: parts.append("") parts.append("
SIDRule NameAttack TypeProtocolCount
{sid}{escape(info.get('name', 'Unknown'))}{escape(info.get('attack', '-'))}{escape(info.get('protocol', '-'))}{count:,}
" "No IPS blocks found. Run Snort in IPS mode (-Q flag) to enable dropping.
") # Full Rule Coverage Table parts.append("""

Full Rule Coverage (NIDS + IPS)

All 25 custom rules: 18 NIDS alert rules and 7 IPS DROP rules.

""") for sid in sorted(rule_coverage.keys()): info = rule_coverage[sid] sev = severity_map.get(sid, "UNKNOWN") count = sid_totals.get(sid, 0) row_cls = "drop-row" if info["mode"] == "IPS" else "" parts.append( f"" f"" f"" f"" f"" f"" f"" f"" f"" ) parts.append("
SIDModeRule NameAttack TypeProtocolSeverityAlerts Triggered
{sid}{mode_badge(info['mode'])}{escape(info['name'])}{escape(info['attack'])}{escape(info['protocol'])}{severity_badge(sev)}{count:,}
") # Alert Frequency by SID parts.append("""

Alert Frequency by SID

""") for sid, count in sid_totals.most_common(): info = rule_coverage.get(sid, {}) name = info.get("name", f"SID {sid}") mode = info.get("mode", "") sev = severity_map.get(sid, "UNKNOWN") row_cls = "drop-row" if mode == "IPS" else "" parts.append( f"" f"" f"" f"" f"" f"" f"" ) parts.append("
SIDModeRule NameSeverityCount
{sid}{mode_badge(mode) if mode else ''}{escape(name)}{severity_badge(sev)}{count:,}
") # Top Source IPs parts.append("""

Top 15 Source IPs

""") for ip, count in src_ip_totals.most_common(15): pct = f"{(count / total_alerts * 100):.1f}%" if total_alerts else "0%" parts.append(f"") parts.append("
Source IPCount% of Total
{escape(ip)}{count:,}{pct}
") # Per-Session Breakdown parts.append("""

Per-Session Breakdown

""") for session in SESSION_ORDER: s_alerts = alerts_by_session[session] s_sev = Counter(a["severity"] for a in s_alerts) parts.append( f"" f"" f"" f"" f"" f"" f"" f"" f"" ) parts.append("
SessionTotalHIGHMEDIUMLOWBLOCKED (IPS)UNKNOWN
{escape(session)}{len(s_alerts):,}{s_sev.get('HIGH', 0):,}{s_sev.get('MEDIUM', 0):,}{s_sev.get('LOW', 0):,}{s_sev.get('BLOCKED', 0):,}{s_sev.get('UNKNOWN', 0):,}
") # Navigation to detail pages parts.append("""

Browse Detailed Alerts

Alert data split into pages of 1000 rows each.

""") for session in SESSION_ORDER: total = len(alerts_by_session[session]) pages = max(1, math.ceil(total / PAGE_SIZE)) first_page = f"details/{session.lower()}_page_1.html" parts.append( f'' f'{escape(session)} ({total:,} alerts, {pages} pages)' ) parts.append("
") parts.append("
") with open(report_file, "w", encoding="utf-8") as f: f.write("".join(parts)) fix_permissions(report_file) print(f"[INFO] Main report written: {report_file}", flush=True) # --------------------------------------------- # SESSION DETAIL PAGES # --------------------------------------------- def build_session_pages(session, alerts): total = len(alerts) total_pages = max(1, math.ceil(total / PAGE_SIZE)) for page_num in range(1, total_pages + 1): start = (page_num - 1) * PAGE_SIZE end = min(start + PAGE_SIZE, total) page_alerts = alerts[start:end] filename = os.path.join(detail_dir, f"{session.lower()}_page_{page_num}.html") parts = [] parts.append(f""" {escape(session)} Alerts - Page {page_num} {base_style()}

{escape(session)} Alert Details

Showing alerts {start + 1:,} to {end:,} of {total:,}. Page {page_num} of {total_pages}.

Navigation

← Main Report """) if page_num > 1: parts.append(f'← Previous') if page_num < total_pages: parts.append(f'Next →') parts.append('
') page_window = 10 start_page = max(1, page_num - page_window // 2) end_page = min(total_pages, start_page + page_window - 1) start_page = max(1, end_page - page_window + 1) for p in range(start_page, end_page + 1): cls = "active" if p == page_num else "" parts.append(f'{p}') parts.append("
") parts.append("""

Alert Table

""") for a in page_alerts: info = rule_coverage.get(a["sid"], {}) rule_name = info.get("name", f"SID {a['sid']}") mode = info.get("mode", "") row_cls = "drop-row" if a["action"] == "DROP" else "" parts.append( f"" f"" f"" f"" f"" f"" f"" f"" f"" f"" f"" f"" f"" f"" f"" ) parts.append("
Action Session Timestamp SID Mode Rule Name Message Severity Protocol Source IP Src Port Dest IP Dst Port
{action_badge(a['action'])}{escape(a['session'])}{escape(a['timestamp'])}{a['sid']}{mode_badge(mode) if mode else ''}{escape(rule_name)}{escape(a['message'])}{severity_badge(a['severity'])}{escape(a['protocol'])}{escape(a['src_ip'])}{escape(a['src_port'])}{escape(a['dst_ip'])}{escape(a['dst_port'])}
") with open(filename, "w", encoding="utf-8") as f: f.write("".join(parts)) fix_permissions(filename) # --------------------------------------------- # MAIN ENTRY POINT # --------------------------------------------- def main(): generated_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"[INFO] Starting report generation at {generated_at}", flush=True) fix_permissions(report_dir) fix_permissions(detail_dir) all_alerts = [] alerts_by_session = defaultdict(list) session_totals = Counter() severity_totals = Counter() sid_totals = Counter() src_ip_totals = Counter() for session in SESSION_ORDER: alerts = parse_alert_file(log_paths[session], session) alerts_by_session[session].extend(alerts) all_alerts.extend(alerts) for a in alerts: session_totals[a["session"]] += 1 severity_totals[a["severity"]] += 1 sid_totals[a["sid"]] += 1 src_ip_totals[a["src_ip"]] += 1 print(f"[INFO] Total parsed alerts : {len(all_alerts):,}", flush=True) print(f"[INFO] IPS blocked alerts : {severity_totals.get('BLOCKED', 0):,}", flush=True) for session in SESSION_ORDER: alerts_by_session[session].sort(key=lambda x: x["timestamp"]) print("[INFO] Writing main HTML report...", flush=True) build_main_report( all_alerts, alerts_by_session, session_totals, severity_totals, sid_totals, src_ip_totals, generated_at ) print("[INFO] Writing paginated detail pages...", flush=True) for session in SESSION_ORDER: print(f"[INFO] Building pages for {session}...", flush=True) build_session_pages(session, alerts_by_session[session]) fix_tree_permissions(report_dir) print("[INFO] Done.", flush=True) print(f"[INFO] Main report : {report_file}", flush=True) print(f"[INFO] Detail pages: {detail_dir}", flush=True) print(f"[INFO] Generated at: {generated_at}", flush=True) if __name__ == "__main__": main()