130 أسطر
4.3 KiB
Python
130 أسطر
4.3 KiB
Python
"""
|
|
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)
|