540 أسطر
21 KiB
Python
540 أسطر
21 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
siem_analyzer.py
|
|
=================
|
|
A small, dependency-free SIEM (Security Information & Event Management) log
|
|
analyzer.
|
|
|
|
It ingests traffic/activity logs from THREE endpoints:
|
|
|
|
1. Web server access log (endpoint-1 : nginx/Apache "combined" style)
|
|
2. Firewall / router log (endpoint-2 : SRC/DST/PORT/ACTION style)
|
|
3. SSH authentication log (endpoint-3 : syslog "sshd" style)
|
|
|
|
...runs a set of detection rules against each, correlates activity from the
|
|
SAME source IP across DIFFERENT endpoints (the core value-add of a real
|
|
SIEM), and writes the results as JSON for the companion HTML/CSS/JS
|
|
dashboard to render.
|
|
|
|
Usage
|
|
-----
|
|
python3 siem_analyzer.py
|
|
python3 siem_analyzer.py --logs-dir ./logs --out-dir ./dashboard
|
|
|
|
No third-party packages required (standard library only).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sys
|
|
from urllib.parse import unquote
|
|
from collections import Counter, defaultdict
|
|
from dataclasses import dataclass, field, asdict
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from typing import List, Dict, Any, Tuple
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Configuration: detection thresholds & signatures
|
|
# --------------------------------------------------------------------------
|
|
|
|
THRESHOLDS = {
|
|
"brute_force_attempts": 5, # failed logins ...
|
|
"brute_force_window_sec": 300, # ... within this many seconds
|
|
"port_scan_distinct_ports": 8, # distinct dest ports ...
|
|
"port_scan_window_sec": 120, # ... within this many seconds
|
|
"high_rate_requests": 15, # requests ...
|
|
"high_rate_window_sec": 60, # ... within this many seconds
|
|
}
|
|
|
|
# Example threat-intel seed list: IPs already known to be bad regardless of
|
|
# behaviour observed in these logs. In production this would be pulled from
|
|
# a feed (AbuseIPDB, OTX, internal blocklist, etc).
|
|
KNOWN_MALICIOUS_IPS = {
|
|
"192.0.2.77": "Listed in external threat-intel feed (example seed entry)",
|
|
}
|
|
|
|
# (regex, label, severity) -- checked against URL path + query + referer
|
|
SUSPICIOUS_WEB_PATTERNS: List[Tuple[re.Pattern, str, str]] = [
|
|
(re.compile(r"union(\s|%20)+select", re.I), "SQL_INJECTION", "critical"),
|
|
(re.compile(r"'\s*or\s*'?1'?\s*=\s*'?1", re.I), "SQL_INJECTION", "critical"),
|
|
(re.compile(r"sleep\(\d+\)", re.I), "SQL_INJECTION", "critical"),
|
|
(re.compile(r"<script.*?>", re.I), "XSS", "high"),
|
|
(re.compile(r"javascript:", re.I), "XSS", "high"),
|
|
(re.compile(r"\.\./\.\./", re.I), "PATH_TRAVERSAL", "high"),
|
|
(re.compile(r"etc/passwd|etc/shadow", re.I), "PATH_TRAVERSAL", "high"),
|
|
(re.compile(r";\s*(rm|wget|curl|nc)\s", re.I), "COMMAND_INJECTION", "critical"),
|
|
]
|
|
|
|
SCANNER_USER_AGENTS = ["sqlmap", "nikto", "nmap", "masscan", "acunetix", "nessus", "wpscan"]
|
|
|
|
SEVERITY_WEIGHT = {"critical": 10, "high": 5, "medium": 2, "low": 1}
|
|
SEVERITY_ORDER = ["critical", "high", "medium", "low"]
|
|
|
|
ENDPOINT_WEB = "web-server (endpoint-1)"
|
|
ENDPOINT_FW = "firewall (endpoint-2)"
|
|
ENDPOINT_AUTH = "ssh-auth (endpoint-3)"
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Parsing
|
|
# --------------------------------------------------------------------------
|
|
|
|
WEB_LOG_RE = re.compile(
|
|
r'(?P<ip>\S+) \S+ \S+ \[(?P<time>[^\]]+)\] '
|
|
r'"(?P<method>\S+) (?P<path>\S+)(?: \S+)?" '
|
|
r'(?P<status>\d+) (?P<size>\S+) "(?P<referer>[^"]*)" "(?P<ua>[^"]*)"'
|
|
)
|
|
WEB_TIME_FMT = "%d/%b/%Y:%H:%M:%S %z"
|
|
|
|
FW_LOG_RE = re.compile(
|
|
r'(?P<time>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) '
|
|
r'SRC=(?P<src_ip>\S+) SPT=(?P<src_port>\d+) '
|
|
r'DST=(?P<dst_ip>\S+) DPT=(?P<dst_port>\d+) '
|
|
r'PROTO=(?P<proto>\S+) ACTION=(?P<action>\S+)'
|
|
)
|
|
FW_TIME_FMT = "%Y-%m-%d %H:%M:%S"
|
|
|
|
AUTH_LOG_RE = re.compile(
|
|
r'(?P<time>\w{3}\s+\d{1,2} \d{2}:\d{2}:\d{2}) \S+ sshd\[\d+\]: '
|
|
r'(?P<result>Failed password|Accepted password|Accepted publickey) '
|
|
r'for (?:invalid user )?(?P<user>\S+) from (?P<ip>\S+) port (?P<port>\d+)'
|
|
)
|
|
AUTH_TIME_FMT = "%b %d %H:%M:%S"
|
|
AUTH_LOG_YEAR = 2026 # syslog timestamps have no year; assume current year
|
|
|
|
|
|
def parse_web_log(path: Path) -> List[Dict[str, Any]]:
|
|
events = []
|
|
for lineno, line in enumerate(_read_lines(path), start=1):
|
|
m = WEB_LOG_RE.match(line)
|
|
if not m:
|
|
continue
|
|
try:
|
|
ts = datetime.strptime(m.group("time"), WEB_TIME_FMT).replace(tzinfo=None)
|
|
except ValueError:
|
|
continue
|
|
events.append({
|
|
"endpoint": ENDPOINT_WEB,
|
|
"ts": ts,
|
|
"ip": m.group("ip"),
|
|
"method": m.group("method"),
|
|
"path": m.group("path"),
|
|
"status": m.group("status"),
|
|
"referer": m.group("referer"),
|
|
"ua": m.group("ua"),
|
|
"raw": line,
|
|
})
|
|
return events
|
|
|
|
|
|
def parse_firewall_log(path: Path) -> List[Dict[str, Any]]:
|
|
events = []
|
|
for line in _read_lines(path):
|
|
m = FW_LOG_RE.match(line)
|
|
if not m:
|
|
continue
|
|
try:
|
|
ts = datetime.strptime(m.group("time"), FW_TIME_FMT)
|
|
except ValueError:
|
|
continue
|
|
events.append({
|
|
"endpoint": ENDPOINT_FW,
|
|
"ts": ts,
|
|
"ip": m.group("src_ip"),
|
|
"dst_ip": m.group("dst_ip"),
|
|
"dst_port": int(m.group("dst_port")),
|
|
"proto": m.group("proto"),
|
|
"action": m.group("action"),
|
|
"raw": line,
|
|
})
|
|
return events
|
|
|
|
|
|
def parse_auth_log(path: Path) -> List[Dict[str, Any]]:
|
|
events = []
|
|
for line in _read_lines(path):
|
|
m = AUTH_LOG_RE.match(line)
|
|
if not m:
|
|
continue
|
|
try:
|
|
ts = datetime.strptime(f"{AUTH_LOG_YEAR} {m.group('time')}", f"%Y {AUTH_TIME_FMT}")
|
|
except ValueError:
|
|
continue
|
|
events.append({
|
|
"endpoint": ENDPOINT_AUTH,
|
|
"ts": ts,
|
|
"ip": m.group("ip"),
|
|
"user": m.group("user"),
|
|
"result": "failed" if m.group("result") == "Failed password" else "accepted",
|
|
"raw": line,
|
|
})
|
|
return events
|
|
|
|
|
|
def _read_lines(path: Path) -> List[str]:
|
|
if not path.exists():
|
|
print(f"[!] Log file not found, skipping: {path}", file=sys.stderr)
|
|
return []
|
|
with path.open("r", encoding="utf-8", errors="ignore") as f:
|
|
return [ln.rstrip("\n") for ln in f if ln.strip()]
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Alert helper
|
|
# --------------------------------------------------------------------------
|
|
|
|
_alert_counter = 0
|
|
|
|
|
|
def make_alert(ts: datetime, endpoint: str, src_ip: str, alert_type: str,
|
|
severity: str, description: str, evidence: List[str]) -> Dict[str, Any]:
|
|
global _alert_counter
|
|
_alert_counter += 1
|
|
return {
|
|
"id": f"ALT-{_alert_counter:04d}",
|
|
"timestamp": ts.isoformat(),
|
|
"endpoint": endpoint,
|
|
"src_ip": src_ip,
|
|
"alert_type": alert_type,
|
|
"severity": severity,
|
|
"description": description,
|
|
"evidence": evidence[:5], # cap sample evidence lines
|
|
}
|
|
|
|
|
|
def _find_bursts(sorted_ts: List[datetime], window_sec: int, threshold: int) -> List[Tuple[int, int]]:
|
|
"""Greedy sliding window: return non-overlapping (start_idx, end_idx)
|
|
ranges where >= `threshold` events occur within `window_sec` seconds."""
|
|
bursts = []
|
|
n = len(sorted_ts)
|
|
i = 0
|
|
while i < n:
|
|
j = i
|
|
while j < n and (sorted_ts[j] - sorted_ts[i]).total_seconds() <= window_sec:
|
|
j += 1
|
|
count = j - i
|
|
if count >= threshold:
|
|
bursts.append((i, j - 1))
|
|
i = j # skip past this burst to avoid overlapping duplicate alerts
|
|
else:
|
|
i += 1
|
|
return bursts
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Detection rules
|
|
# --------------------------------------------------------------------------
|
|
|
|
def detect_web_attacks(events: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
alerts = []
|
|
for e in events:
|
|
# Decode percent-encoding first -- attack tools URL-encode payloads,
|
|
# and a real WAF/SIEM normalizes input before signature matching.
|
|
haystack = unquote(f"{e['path']} {e['referer']}")
|
|
for pattern, label, severity in SUSPICIOUS_WEB_PATTERNS:
|
|
if pattern.search(haystack):
|
|
alerts.append(make_alert(
|
|
e["ts"], ENDPOINT_WEB, e["ip"], label, severity,
|
|
f"{label.replace('_', ' ').title()} attempt detected in request to {e['path']}",
|
|
[e["raw"]],
|
|
))
|
|
break # one alert per line is enough
|
|
return alerts
|
|
|
|
|
|
def detect_scanner_user_agents(events: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
groups: Dict[Tuple[str, str], List[Dict[str, Any]]] = defaultdict(list)
|
|
for e in events:
|
|
ua_lower = e["ua"].lower()
|
|
for tool in SCANNER_USER_AGENTS:
|
|
if tool in ua_lower:
|
|
groups[(e["ip"], tool)].append(e)
|
|
break
|
|
alerts = []
|
|
for (ip, tool), grp in groups.items():
|
|
if len(grp) >= 3:
|
|
grp.sort(key=lambda x: x["ts"])
|
|
alerts.append(make_alert(
|
|
grp[0]["ts"], ENDPOINT_WEB, ip, "RECON_SCANNER", "medium",
|
|
f"Automated vulnerability scan detected ({tool}): {len(grp)} probe requests "
|
|
f"against paths such as {', '.join(g['path'] for g in grp[:3])}",
|
|
[g["raw"] for g in grp],
|
|
))
|
|
return alerts
|
|
|
|
|
|
def detect_high_request_rate(events: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
by_ip: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
|
|
for e in events:
|
|
by_ip[e["ip"]].append(e)
|
|
alerts = []
|
|
for ip, evs in by_ip.items():
|
|
evs.sort(key=lambda x: x["ts"])
|
|
ts_list = [e["ts"] for e in evs]
|
|
for start, end in _find_bursts(ts_list, THRESHOLDS["high_rate_window_sec"], THRESHOLDS["high_rate_requests"]):
|
|
count = end - start + 1
|
|
severity = "high" if count >= 30 else "medium"
|
|
alerts.append(make_alert(
|
|
evs[start]["ts"], ENDPOINT_WEB, ip, "HIGH_REQUEST_RATE", severity,
|
|
f"{count} requests from a single IP within "
|
|
f"{THRESHOLDS['high_rate_window_sec']}s (possible scraping/DoS/automation)",
|
|
[evs[k]["raw"] for k in range(start, min(start + 5, end + 1))],
|
|
))
|
|
return alerts
|
|
|
|
|
|
def detect_port_scan(events: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
by_ip: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
|
|
for e in events:
|
|
by_ip[e["ip"]].append(e)
|
|
alerts = []
|
|
for ip, evs in by_ip.items():
|
|
evs.sort(key=lambda x: x["ts"])
|
|
n = len(evs)
|
|
i = 0
|
|
while i < n:
|
|
j = i
|
|
ports_seen = set()
|
|
while j < n and (evs[j]["ts"] - evs[i]["ts"]).total_seconds() <= THRESHOLDS["port_scan_window_sec"]:
|
|
ports_seen.add(evs[j]["dst_port"])
|
|
j += 1
|
|
if len(ports_seen) >= THRESHOLDS["port_scan_distinct_ports"]:
|
|
denied = sum(1 for k in range(i, j) if evs[k]["action"] == "DENY")
|
|
allowed = (j - i) - denied
|
|
alerts.append(make_alert(
|
|
evs[i]["ts"], ENDPOINT_FW, ip, "PORT_SCAN", "high",
|
|
f"Port scan detected: {len(ports_seen)} distinct destination ports probed "
|
|
f"against {evs[i]['dst_ip']} within {THRESHOLDS['port_scan_window_sec']}s "
|
|
f"({denied} denied, {allowed} allowed)",
|
|
[evs[k]["raw"] for k in range(i, min(i + 5, j))],
|
|
))
|
|
i = j
|
|
else:
|
|
i += 1
|
|
return alerts
|
|
|
|
|
|
def detect_brute_force(events: List[Dict[str, Any]]) -> Tuple[List[Dict[str, Any]], Dict[str, datetime]]:
|
|
"""Returns (alerts, {ip: last_burst_end_ts}) -- the second value lets us
|
|
check for a successful login shortly after a brute-force burst."""
|
|
by_ip: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
|
|
for e in events:
|
|
by_ip[e["ip"]].append(e)
|
|
|
|
alerts = []
|
|
burst_end_by_ip: Dict[str, datetime] = {}
|
|
for ip, evs in by_ip.items():
|
|
evs.sort(key=lambda x: x["ts"])
|
|
failed = [e for e in evs if e["result"] == "failed"]
|
|
ts_list = [e["ts"] for e in failed]
|
|
for start, end in _find_bursts(ts_list, THRESHOLDS["brute_force_window_sec"], THRESHOLDS["brute_force_attempts"]):
|
|
count = end - start + 1
|
|
users_tried = sorted({failed[k]["user"] for k in range(start, end + 1)})
|
|
alerts.append(make_alert(
|
|
failed[start]["ts"], ENDPOINT_AUTH, ip, "BRUTE_FORCE_SSH", "high",
|
|
f"{count} failed SSH login attempts within {THRESHOLDS['brute_force_window_sec']}s "
|
|
f"trying {len(users_tried)} usernames ({', '.join(users_tried[:6])})",
|
|
[failed[k]["raw"] for k in range(start, min(start + 5, end + 1))],
|
|
))
|
|
burst_end_by_ip[ip] = failed[end]["ts"]
|
|
return alerts, burst_end_by_ip
|
|
|
|
|
|
def detect_compromise_after_brute_force(events: List[Dict[str, Any]],
|
|
burst_end_by_ip: Dict[str, datetime]) -> List[Dict[str, Any]]:
|
|
alerts = []
|
|
by_ip: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
|
|
for e in events:
|
|
by_ip[e["ip"]].append(e)
|
|
for ip, burst_end in burst_end_by_ip.items():
|
|
for e in sorted(by_ip.get(ip, []), key=lambda x: x["ts"]):
|
|
if e["result"] == "accepted" and e["ts"] >= burst_end:
|
|
alerts.append(make_alert(
|
|
e["ts"], ENDPOINT_AUTH, ip, "ACCOUNT_COMPROMISE_SUSPECTED", "critical",
|
|
f"Successful SSH login as '{e['user']}' immediately following a brute-force "
|
|
f"burst from the same IP -- account may be compromised",
|
|
[e["raw"]],
|
|
))
|
|
break # only need the first successful login after the burst
|
|
return alerts
|
|
|
|
|
|
def check_blacklist(all_events: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
alerts = []
|
|
seen: set = set()
|
|
for e in sorted(all_events, key=lambda x: x["ts"]):
|
|
ip = e["ip"]
|
|
if ip in KNOWN_MALICIOUS_IPS and (ip, e["endpoint"]) not in seen:
|
|
seen.add((ip, e["endpoint"]))
|
|
alerts.append(make_alert(
|
|
e["ts"], e["endpoint"], ip, "BLACKLISTED_IP_ACTIVITY", "critical",
|
|
f"Traffic from known-malicious IP {ip}: {KNOWN_MALICIOUS_IPS[ip]}",
|
|
[e["raw"]],
|
|
))
|
|
return alerts
|
|
|
|
|
|
def correlate_cross_endpoint(alerts: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
by_ip: Dict[str, set] = defaultdict(set)
|
|
earliest_ts: Dict[str, datetime] = {}
|
|
for a in alerts:
|
|
by_ip[a["src_ip"]].add(a["endpoint"])
|
|
ts = datetime.fromisoformat(a["timestamp"])
|
|
if a["src_ip"] not in earliest_ts or ts < earliest_ts[a["src_ip"]]:
|
|
earliest_ts[a["src_ip"]] = ts
|
|
|
|
correlated = []
|
|
for ip, endpoints in by_ip.items():
|
|
if len(endpoints) >= 2:
|
|
correlated.append(make_alert(
|
|
earliest_ts[ip], "correlation-engine", ip, "MULTI_VECTOR_ATTACK", "critical",
|
|
f"IP {ip} triggered alerts across {len(endpoints)} different endpoints "
|
|
f"({', '.join(sorted(endpoints))}) -- consistent with a coordinated, "
|
|
f"multi-stage attack (recon -> exploitation -> access)",
|
|
[],
|
|
))
|
|
return correlated
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Aggregation
|
|
# --------------------------------------------------------------------------
|
|
|
|
def build_ip_summary(alerts: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
by_ip: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
|
|
for a in alerts:
|
|
by_ip[a["src_ip"]].append(a)
|
|
|
|
summary = []
|
|
for ip, a_list in by_ip.items():
|
|
sev_counts = Counter(a["severity"] for a in a_list)
|
|
score = sum(SEVERITY_WEIGHT[a["severity"]] for a in a_list)
|
|
endpoints = sorted({a["endpoint"] for a in a_list if a["endpoint"] != "correlation-engine"})
|
|
alert_types = sorted({a["alert_type"] for a in a_list})
|
|
timestamps = [datetime.fromisoformat(a["timestamp"]) for a in a_list]
|
|
summary.append({
|
|
"ip": ip,
|
|
"threat_score": score,
|
|
"severity_counts": {s: sev_counts.get(s, 0) for s in SEVERITY_ORDER},
|
|
"total_alerts": len(a_list),
|
|
"alert_types": alert_types,
|
|
"endpoints_involved": endpoints,
|
|
"first_seen": min(timestamps).isoformat(),
|
|
"last_seen": max(timestamps).isoformat(),
|
|
"known_threat_intel": KNOWN_MALICIOUS_IPS.get(ip),
|
|
})
|
|
summary.sort(key=lambda x: x["threat_score"], reverse=True)
|
|
return summary
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Main pipeline
|
|
# --------------------------------------------------------------------------
|
|
|
|
def run(logs_dir: Path, out_dir: Path) -> Dict[str, Any]:
|
|
web_path = logs_dir / "endpoint1_web_access.log"
|
|
fw_path = logs_dir / "endpoint2_firewall.log"
|
|
auth_path = logs_dir / "endpoint3_auth.log"
|
|
|
|
web_events = parse_web_log(web_path)
|
|
fw_events = parse_firewall_log(fw_path)
|
|
auth_events = parse_auth_log(auth_path)
|
|
all_events = web_events + fw_events + auth_events
|
|
|
|
alerts: List[Dict[str, Any]] = []
|
|
alerts += detect_web_attacks(web_events)
|
|
alerts += detect_scanner_user_agents(web_events)
|
|
alerts += detect_high_request_rate(web_events)
|
|
alerts += detect_port_scan(fw_events)
|
|
|
|
bf_alerts, burst_ends = detect_brute_force(auth_events)
|
|
alerts += bf_alerts
|
|
alerts += detect_compromise_after_brute_force(auth_events, burst_ends)
|
|
|
|
alerts += check_blacklist(all_events)
|
|
|
|
# correlation must run AFTER all per-endpoint alerts exist
|
|
alerts += correlate_cross_endpoint(alerts)
|
|
|
|
alerts.sort(key=lambda a: a["timestamp"], reverse=True)
|
|
malicious_ips = build_ip_summary(alerts)
|
|
|
|
sev_counts = Counter(a["severity"] for a in alerts)
|
|
report = {
|
|
"generated_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
|
|
"meta": {
|
|
"endpoints": [ENDPOINT_WEB, ENDPOINT_FW, ENDPOINT_AUTH],
|
|
"lines_parsed": {
|
|
ENDPOINT_WEB: len(web_events),
|
|
ENDPOINT_FW: len(fw_events),
|
|
ENDPOINT_AUTH: len(auth_events),
|
|
},
|
|
},
|
|
"stats": {
|
|
"total_alerts": len(alerts),
|
|
"by_severity": {s: sev_counts.get(s, 0) for s in SEVERITY_ORDER},
|
|
"unique_malicious_ips": len(malicious_ips),
|
|
"total_events_parsed": len(all_events),
|
|
},
|
|
"alerts": alerts,
|
|
"malicious_ips": malicious_ips,
|
|
}
|
|
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
json_path = out_dir / "siem_report.json"
|
|
js_path = out_dir / "data.js"
|
|
json_text = json.dumps(report, indent=2)
|
|
json_path.write_text(json_text, encoding="utf-8")
|
|
|
|
# Alert evidence is raw log data and may legitimately contain the literal
|
|
# text "</script>" (e.g. a captured XSS payload). Escape the closing tag
|
|
# sequence so this JSON can never be mistaken for the end of a <script>
|
|
# block if it's ever inlined into an HTML document.
|
|
js_safe_text = json_text.replace("</script", "<\\/script")
|
|
js_path.write_text(f"// Auto-generated by siem_analyzer.py -- do not edit by hand\n"
|
|
f"const SIEM_DATA = {js_safe_text};\n", encoding="utf-8")
|
|
|
|
_print_console_summary(report)
|
|
print(f"\n[+] Full report written to: {json_path}")
|
|
print(f"[+] Dashboard data written to: {js_path}")
|
|
return report
|
|
|
|
|
|
def _print_console_summary(report: Dict[str, Any]) -> None:
|
|
stats = report["stats"]
|
|
print("=" * 70)
|
|
print(" SIEM ANALYSIS SUMMARY")
|
|
print("=" * 70)
|
|
print(f" Events parsed : {stats['total_events_parsed']}")
|
|
print(f" Total alerts : {stats['total_alerts']}")
|
|
for sev in SEVERITY_ORDER:
|
|
print(f" - {sev:<9}: {stats['by_severity'][sev]}")
|
|
print(f" Malicious IPs : {stats['unique_malicious_ips']}")
|
|
print("-" * 70)
|
|
print(" TOP MALICIOUS IPs")
|
|
print("-" * 70)
|
|
for ip_info in report["malicious_ips"][:10]:
|
|
print(f" {ip_info['ip']:<16} score={ip_info['threat_score']:<4} "
|
|
f"alerts={ip_info['total_alerts']:<3} "
|
|
f"endpoints={len(ip_info['endpoints_involved'])} "
|
|
f"types={','.join(ip_info['alert_types'])}")
|
|
print("=" * 70)
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
p = argparse.ArgumentParser(description="Analyze SIEM logs from 3 endpoints and emit dashboard JSON.")
|
|
p.add_argument("--logs-dir", type=Path, default=Path(__file__).parent / "logs",
|
|
help="Directory containing endpoint1_web_access.log, endpoint2_firewall.log, endpoint3_auth.log")
|
|
p.add_argument("--out-dir", type=Path, default=Path(__file__).parent / "dashboard",
|
|
help="Directory to write siem_report.json and data.js into")
|
|
return p.parse_args()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
args = parse_args()
|
|
run(args.logs_dir, args.out_dir)
|