Moh-Khair/ghaymah-exam-Mohamed-Ashraf-Helmi-SecOps
0
0
نسخ 0
الشفرة المشكلات طلبات السحب Actions Packages المشاريع الإصدارات الويكي النشاط

Upload files to "q4-SIEM-build"

تصفح المصدر
هذا الالتزام موجود في:
Moh-Khair
2026-07-29 01:07:36 +00:00
الأصل 4d09f3b967
التزام 466da0ba68
3 ملفات معدلة مع 788 إضافات و0 حذوفات
تنزيل ملف التصحيح تنزيل ملف الاختلاف توسيع جميع الملفات طي جميع الملفات

154
q4-SIEM-build/PUBLISH_TO_GHAYMAH.md Normal file
عرض الملف

@@ -0,0 +1,154 @@
# Publishing this SIEM Dashboard to Ghaymah Storage Block
## Read this first: what I could and couldn't verify
I looked for Ghaymah Systems' own documentation before writing this. I
found:
- Their docs portal: **https://docs.ghaymah.cloud/**
- Their deployment portal: **https://deploy.ghaymah.systems/**
Both are JavaScript-rendered apps, so I could confirm they exist but
couldn't pull the actual step-by-step page content for the "Storage Block"
feature specifically — I don't have exact screenshots/menu names to give
you with certainty, and I don't want to hand you confidently-wrong click
paths for something that's going to be graded.
What follows is the **standard workflow that essentially every cloud
"storage block"/bucket/object-storage service uses** (AWS S3, DigitalOcean
Spaces, Linode/Akamai Object Storage, etc.), written so you can map it onto
Ghaymah's actual console. Before your exam submission, open
**docs.ghaymah.cloud** and search for "storage" to confirm exact menu
names, size limits, and whether static-site hosting is a toggle you need to
enable — the mechanics below are almost certainly right, the exact button
labels might differ slightly.
There are two different things a "storage block" can mean on a cloud
platform — check which one Ghaymah's product actually is before you start,
since the workflow is different:
| If "Storage Block" is... | It behaves like... | Use it to... |
|---|---|---|
| **Object storage** (a bucket you upload files into, with a public URL) | AWS S3 / DigitalOcean Spaces | Host this static dashboard directly — this is the common case, and the one the rest of this guide assumes |
| **Block storage** (a raw virtual disk volume, like AWS EBS) | An attached hard drive | You'd still need a compute instance/VM running a web server (nginx, `python3 -m http.server`, etc.) — the volume just gives that VM extra disk space. See the [alternate path](#alternate-path-if-storage-block-means-a-disk-volume) below. |
---
## Path A: publishing to object/bucket-style storage (most likely case)
### 1. Regenerate fresh data before publishing
Run the analyzer one last time so `siem_report.json` and `data.js` are
current:
```bash
cd siem_project
python3 siem_analyzer.py
```
### 2. Collect exactly what needs to go live
Only the `dashboard/` folder needs to be published — the Python script and
raw `logs/` are your local tooling, not part of the served site:
```
dashboard/
├── index.html
├── style.css
├── script.js
├── siem_report.json
└── data.js
```
(You can optionally publish `logs/` and `siem_analyzer.py` alongside it too
if the exam wants the whole repo visible — just make sure `index.html` ends
up at the bucket's root so it's served as the default page.)
### 3. Create the storage block / bucket
In the Ghaymah console (or `deploy.ghaymah.systems`):
1. Log in and find **Storage** in the main navigation.
2. Create a new storage block — give it a unique name (e.g.
`siem-dashboard-<yourname>`), and pick a region close to you/your
grader.
3. If there's a **public access / static website hosting** toggle, enable
it — object storage buckets are private by default on most platforms,
and a private bucket will just return "Access Denied" in a browser.
4. If it asks for an **index document**, set it to `index.html`.
### 4. Upload the files
Most storage-block consoles support drag-and-drop upload in the browser —
upload the 5 files from `dashboard/` (keep them at the bucket's root, not
inside a subfolder, so relative paths like `href="style.css"` still
resolve).
If Ghaymah instead exposes an **S3-compatible API** (common for
smaller/regional clouds — check their docs for an endpoint URL, access
key, and secret key), you can upload with the standard AWS CLI pointed at
their endpoint instead of the console:
```bash
aws s3 cp ./dashboard s3://siem-dashboard-<yourname> \
--recursive \
--endpoint-url https://<ghaymah-storage-endpoint>
```
(Only use this if their docs confirm an S3-compatible endpoint and give
you access/secret keys — don't guess the endpoint URL.)
### 5. Get the public URL and verify
The console should show a public URL after upload (something like
`https://<bucket-name>.<region>.ghaymah.systems/` or a
platform-generated domain). Open it and confirm:
- The stat cards, radar, and alert table all render (not a blank page —
that usually means `style.css`/`script.js`/`data.js` didn't upload to
the same folder as `index.html`).
- The alert table search/filter and row-expand still work.
### 6. (Optional) Custom domain
If Ghaymah supports attaching a custom domain/CNAME to a storage block,
that's normally: add a CNAME record at your DNS provider pointing your
subdomain at the bucket's platform domain, then add that domain in the
bucket's settings and wait for it to verify (and, if offered, request/attach
an SSL certificate).
---
## Alternate path: if "Storage Block" means a disk volume
If Ghaymah's "Storage Block" is closer to AWS EBS (a virtual disk you
attach to a VM rather than something with its own public URL), the flow is
instead:
1. Provision a small compute instance (VM) on Ghaymah.
2. Create and attach a storage block/volume to it, then mount it
(`mkfs`, `mount` — same as any Linux block device).
3. Copy `dashboard/` onto that mounted volume (`scp`, `git clone`, etc.).
4. Serve it from the VM:
```bash
cd /mnt/<your-volume>/dashboard
python3 -m http.server 80
# or install nginx and point its document root at this folder
```
5. Open the VM's public IP (or attach a domain to it) to verify.
---
## Either way — a pre-submission checklist
- [ ] Ran `python3 siem_analyzer.py` so `siem_report.json`/`data.js` reflect
the current logs
- [ ] `index.html` is reachable at the bucket/site root, not nested in a
subfolder
- [ ] `style.css`, `script.js`, `data.js`, `siem_report.json` are in the
**same folder** as `index.html`
- [ ] Public URL loads with no blank/broken page and no browser console
errors
- [ ] Confirmed the actual click-path against **docs.ghaymah.cloud**, since
the steps above are the generic pattern, not a verified Ghaymah
screenshot-by-screenshot guide

95
q4-SIEM-build/README.md Normal file
عرض الملف

@@ -0,0 +1,95 @@
# Simple SIEM — Log Analyzer + Threat Dashboard
A small SIEM built for a 3-endpoint environment: a **web server**, a
**firewall/router**, and an **SSH authentication** log source. A Python
script parses and correlates suspicious activity across all three; a static
HTML/CSS/JS dashboard visualizes the resulting alerts and malicious IPs.
```
siem_project/
├── logs/
│ ├── endpoint1_web_access.log # sample nginx-style access log
│ ├── endpoint2_firewall.log # sample SRC/DST/PORT/ACTION firewall log
│ └── endpoint3_auth.log # sample sshd auth log
├── siem_analyzer.py # Part 1 — the analyzer
├── dashboard/
│ ├── index.html # Part 2 — the dashboard
│ ├── style.css
│ ├── script.js
│ ├── siem_report.json # generated by siem_analyzer.py
│ └── data.js # generated by siem_analyzer.py (offline fallback)
├── README.md # this file
└── PUBLISH_TO_GHAYMAH.md # Part 3 — publishing guide
```
## Part 1 — `siem_analyzer.py`
Pure standard library, no `pip install` needed.
```bash
python3 siem_analyzer.py
# or point it at different files:
python3 siem_analyzer.py --logs-dir ./logs --out-dir ./dashboard
```
It parses all three log formats, runs the detection rules below, prints a
console summary, and writes `siem_report.json` + `data.js` into `dashboard/`.
### Detection rules
| Rule | Source | Logic | Severity |
|---|---|---|---|
| `SQL_INJECTION` / `XSS` / `PATH_TRAVERSAL` / `COMMAND_INJECTION` | web | Regex signatures against the (URL-decoded) request path | critical/high |
| `RECON_SCANNER` | web | ≥3 requests with a known scanner user-agent (sqlmap, nikto, nmap…) | medium |
| `HIGH_REQUEST_RATE` | web | ≥15 requests from one IP within 60s | medium/high |
| `PORT_SCAN` | firewall | ≥8 distinct destination ports from one IP within 120s | high |
| `BRUTE_FORCE_SSH` | auth | ≥5 failed logins from one IP within 300s | high |
| `ACCOUNT_COMPROMISE_SUSPECTED` | auth | A successful login from an IP right after it triggered a brute-force alert | critical |
| `BLACKLISTED_IP_ACTIVITY` | any | IP matches a seeded threat-intel list | critical |
| `MULTI_VECTOR_ATTACK` | correlation | Same source IP triggered alerts on ≥2 different endpoints | critical |
That last rule is the actual point of a SIEM: no single log tells the whole
story, but seeing the *same* IP port-scan the firewall, brute-force SSH,
*and* throw SQLi at the web app is what turns three noisy logs into one
clear "this IP is attacking us" signal. All thresholds live in the
`THRESHOLDS` dict at the top of the script if you want to tune them.
## Part 2 — the dashboard
Open `dashboard/index.html` directly in a browser — it works out of the box
because it falls back to the embedded `data.js` snapshot. For the "live"
experience (auto re-fetching `siem_report.json` when you click Refresh),
serve the folder instead:
```bash
cd dashboard
python3 -m http.server 8000
# open http://localhost:8000
```
What's on it:
- **Stat cards** — total alerts and the severity breakdown.
- **Threat Radar** — the top malicious IPs plotted by score (closer to
center = more dangerous); a critical IP gets a pulsing ring.
- **Severity Mix** — a proportional bar of critical/high/medium/low.
- **Top Malicious IPs** — ranked list with a relative-score bar.
- **Alerts table** — searchable, filterable by severity/endpoint; click a
row to expand the raw log line(s) behind that alert.
Re-running `siem_analyzer.py` regenerates both output files — refresh the
page (or click the in-app Refresh button if you're on a local server) to
see updated results.
### A note on why the table doesn't use `innerHTML`
Alert descriptions and evidence are literally attacker-supplied log text —
some of the sample data contains a real `<script>` payload. `script.js`
builds every row with `createElement`/`textContent`, never string-built
HTML, so the console can't be XSS'd by the very payloads it's reporting on.
## Regenerating with your own logs
Swap the three files in `logs/` for real exports (keep the same filenames,
or pass `--logs-dir`) and adjust the regexes in `siem_analyzer.py` if your
log format differs from the nginx / iptables-style / sshd formats assumed
here.

539
q4-SIEM-build/siem_analyzer.py Normal file
عرض الملف

@@ -0,0 +1,539 @@
#!/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)