second push

q4-siem/analyzer.py

@@ -0,0 +1,129 @@
+"""
+Ghaymah SIEM - Log Correlation Analyzer
+-----------------------------------------
+Correlates events across 3 log sources (Web/Nginx, SSH/Auth, Database) to
+detect a single attacker IP behind multiple attack patterns.
+
+Usage:
+    python3 analyzer.py                                  -> runs on built-in demo data
+    python3 analyzer.py web.log auth.log db.log          -> runs on real log files
+
+When real file paths are given, only NEW lines since the last run are
+processed (tracked via analyzer_state.json), so this script is safe to run
+every minute from a cron job without re-scoring the same events twice.
+"""
+import json
+import re
+import os
+import sys
+from collections import defaultdict
+
+STATE_FILE = "analyzer_state.json"
+
+# --- Demo data (used only when no real log paths are provided) ---
+DEMO_WEB_LOGS = [
+    "192.168.1.50 - POST /login HTTP/1.1 401 Unauthorized",
+    "192.168.1.50 - POST /login HTTP/1.1 401 Unauthorized",
+    "192.168.1.50 - POST /login HTTP/1.1 401 Unauthorized",
+    "10.0.0.5 - GET /index.html HTTP/1.1 200 OK",
+]
+DEMO_SSH_LOGS = [
+    "Failed password for root from 192.168.1.50 port 22 ssh2",
+    "Accepted password for admin from 10.0.0.2 port 22 ssh2",
+]
+DEMO_DB_LOGS = [
+    "Query executed by 192.168.1.50: SELECT * FROM users WHERE id = '1' OR '1'='1'",
+    "Query executed by 10.0.0.2: SELECT name FROM products WHERE id = 5",
+]
+
+
+def load_state():
+    if os.path.exists(STATE_FILE):
+        with open(STATE_FILE) as f:
+            return json.load(f)
+    return {}
+
+
+def save_state(state):
+    with open(STATE_FILE, "w") as f:
+        json.dump(state, f)
+
+
+def read_new_lines(path, state):
+    """Read only the lines appended to `path` since the last recorded offset."""
+    if not path or not os.path.exists(path):
+        return []
+    last_offset = state.get(path, 0)
+    with open(path, "r", errors="ignore") as f:
+        f.seek(last_offset)
+        new_lines = f.readlines()
+        state[path] = f.tell()
+    return [line.strip() for line in new_lines if line.strip()]
+
+
+def flag(bucket, reason, weight):
+    """Add score/reason to an IP bucket, without duplicating the same reason twice."""
+    bucket["score"] += weight
+    if reason not in bucket["reasons"]:
+        bucket["reasons"].append(reason)
+
+
+def analyze_logs(web_path=None, ssh_path=None, db_path=None):
+    using_real_files = any([web_path, ssh_path, db_path])
+    state = load_state() if using_real_files else {}
+
+    web_logs = read_new_lines(web_path, state) if web_path else DEMO_WEB_LOGS
+    ssh_logs = read_new_lines(ssh_path, state) if ssh_path else DEMO_SSH_LOGS
+    db_logs = read_new_lines(db_path, state) if db_path else DEMO_DB_LOGS
+
+    suspicious_ips = defaultdict(lambda: {"score": 0, "reasons": []})
+
+    # Web logs -> Brute Force detection
+    for log in web_logs:
+        if "401 Unauthorized" in log:
+            ip = log.split()[0]
+            flag(suspicious_ips[ip], "Web Brute Force Attempt", 10)
+
+    # SSH logs -> server intrusion attempts
+    for log in ssh_logs:
+        if "Failed password" in log:
+            match = re.search(r"from (\d+\.\d+\.\d+\.\d+)", log)
+            if match:
+                flag(suspicious_ips[match.group(1)], "SSH Failed Login", 20)
+
+    # DB logs -> SQL Injection patterns
+    for log in db_logs:
+        if "OR '1'='1'" in log or "DROP TABLE" in log.upper():
+            match = re.search(r"by (\d+\.\d+\.\d+\.\d+):", log)
+            if match:
+                flag(suspicious_ips[match.group(1)], "SQL Injection Attempt", 50)
+
+    # Build alerts with severity tiers instead of a flat "Critical" for everything
+    alerts = []
+    for ip, data in suspicious_ips.items():
+        if data["score"] >= 30:
+            severity = "Critical" if data["score"] >= 50 else "Warning"
+            alerts.append({
+                "ip": ip,
+                "threat_score": data["score"],
+                "events": data["reasons"],
+                "status": severity,
+            })
+
+    alerts.sort(key=lambda a: a["threat_score"], reverse=True)
+
+    with open("alerts.json", "w") as f:
+        json.dump(alerts, f, indent=4)
+
+    if using_real_files:
+        save_state(state)
+
+    print(f"Analysis complete. {len(alerts)} alert(s) saved to alerts.json")
+
+
+if __name__ == "__main__":
+    args = sys.argv[1:]
+    web_p = args[0] if len(args) > 0 else None
+    ssh_p = args[1] if len(args) > 1 else None
+    db_p = args[2] if len(args) > 2 else None
+    analyze_logs(web_p, ssh_p, db_p)