303 أسطر
11 KiB
Python
303 أسطر
11 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
siem.py — Simplified SIEM for Ghaymah Storage
|
|
Collects logs from 3 sources, analyzes for suspicious patterns,
|
|
and exposes alerts as JSON for the dashboard.
|
|
"""
|
|
|
|
import re
|
|
import json
|
|
import os
|
|
import time
|
|
import random
|
|
import logging
|
|
from datetime import datetime, timedelta
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
# ─── Configuration ─────────────────────────────────────────────
|
|
LOG_SOURCES = {
|
|
"auth": "/mnt/logs/auth.log",
|
|
"nginx": "/mnt/logs/nginx/access.log",
|
|
"app": "/mnt/logs/app/application.log",
|
|
}
|
|
OUTPUT_FILE = "/var/www/siem/alerts.json"
|
|
ALERT_THRESHOLD_FAIL = 5 # Failed logins before alert
|
|
ALERT_THRESHOLD_RATE = 100 # HTTP requests/min before alert
|
|
SCAN_INTERVAL_SEC = 30 # How often to re-scan logs
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(message)s"
|
|
)
|
|
log = logging.getLogger("siem")
|
|
|
|
# ─── Patterns ──────────────────────────────────────────────────
|
|
PATTERNS = {
|
|
"brute_force": re.compile(
|
|
r"Failed password for .+ from (?P<ip>\d+\.\d+\.\d+\.\d+)"
|
|
),
|
|
"invalid_user": re.compile(
|
|
r"Invalid user .+ from (?P<ip>\d+\.\d+\.\d+\.\d+)"
|
|
),
|
|
"nginx_request": re.compile(
|
|
r'(?P<ip>\d+\.\d+\.\d+\.\d+) .+ "(?P<method>\w+) (?P<path>\S+) HTTP.+" (?P<status>\d{3})'
|
|
),
|
|
"app_error": re.compile(
|
|
r"\[(?P<level>ERROR|CRITICAL)\] (?P<msg>.+)"
|
|
),
|
|
"sql_injection": re.compile(
|
|
r"\b(select|union|insert|drop|update|delete|benchmark|sleep)\b\s*[\(%'\"()]",
|
|
re.IGNORECASE
|
|
),
|
|
}
|
|
|
|
# ─── State ─────────────────────────────────────────────────────
|
|
failed_logins: dict = defaultdict(int) # ip → count
|
|
request_counts: dict = defaultdict(int) # ip → count/window
|
|
sql_attempts: list = []
|
|
alerts: list = []
|
|
|
|
|
|
def generate_sample_logs():
|
|
"""
|
|
Generate sample log lines (used when real log files don't exist).
|
|
This allows the SIEM to demonstrate functionality without root access.
|
|
"""
|
|
attacker_ip = "185.220.100.42"
|
|
legit_ip = "192.168.1.10"
|
|
internal_ip = "10.0.0.5"
|
|
sample_lines = {
|
|
"auth": [
|
|
f"Jul 26 19:01:00 server sshd[1234]: Failed password for root from {attacker_ip} port 54321 ssh2",
|
|
f"Jul 26 19:01:02 server sshd[1234]: Failed password for admin from {attacker_ip} port 54321 ssh2",
|
|
f"Jul 26 19:01:04 server sshd[1234]: Failed password for ubuntu from {attacker_ip} port 54321 ssh2",
|
|
f"Jul 26 19:01:06 server sshd[1234]: Failed password for user from {attacker_ip} port 54321 ssh2",
|
|
f"Jul 26 19:01:08 server sshd[1234]: Failed password for deploy from {attacker_ip} port 54321 ssh2",
|
|
f"Jul 26 19:01:10 server sshd[1234]: Failed password for git from {attacker_ip} port 54321 ssh2",
|
|
f"Jul 26 19:01:12 server sshd[1234]: Invalid user hacker from {attacker_ip} port 54321",
|
|
f"Jul 26 19:05:00 server sshd[5678]: Accepted password for deploy from {legit_ip} port 22122 ssh2",
|
|
],
|
|
"nginx": [
|
|
f'185.220.100.42 - - [26/Jul/2026:19:01:00 +0300] "POST /api/v1/auth/login HTTP/1.1" 401 120',
|
|
f'185.220.100.42 - - [26/Jul/2026:19:01:01 +0300] "POST /api/v1/auth/login HTTP/1.1" 401 120',
|
|
f'185.220.100.42 - - [26/Jul/2026:19:01:02 +0300] "POST /api/v1/auth/login HTTP/1.1" 401 120',
|
|
f'185.220.100.42 - - [26/Jul/2026:19:01:03 +0300] "GET /api/v1/users?id=1 UNION SELECT * FROM users-- HTTP/1.1" 400 80',
|
|
f'192.168.1.10 - - [26/Jul/2026:19:02:00 +0300] "GET /dashboard HTTP/1.1" 200 4096',
|
|
f'10.0.0.5 - - [26/Jul/2026:19:03:00 +0300] "GET /health HTTP/1.1" 200 32',
|
|
],
|
|
"app": [
|
|
"[ERROR] Database connection timeout after 30s — retrying",
|
|
"[CRITICAL] Unhandled exception in /api/v1/users/export: PermissionError",
|
|
"[ERROR] JWT verification failed for token eyJ0eXAi... from IP 185.220.100.42",
|
|
"[INFO] User admin@ghaymah.systems logged in from 192.168.1.10",
|
|
"[ERROR] SQL query failed: syntax error near 'UNION'",
|
|
],
|
|
}
|
|
return sample_lines
|
|
|
|
|
|
def read_log_source(name: str, path: str) -> list[str]:
|
|
"""Read lines from a log file, or fall back to sample data."""
|
|
if os.path.exists(path):
|
|
try:
|
|
with open(path, "r", errors="replace") as f:
|
|
lines = f.readlines()
|
|
log.info(f"[{name}] Read {len(lines)} lines from {path}")
|
|
return lines
|
|
except PermissionError:
|
|
log.warning(f"[{name}] Permission denied reading {path} — using sample data")
|
|
else:
|
|
log.warning(f"[{name}] File not found: {path} — using sample data")
|
|
|
|
# Return sample data
|
|
return generate_sample_logs().get(name, [])
|
|
|
|
|
|
def analyze_auth_log(lines: list[str]):
|
|
"""Detect brute force SSH attempts."""
|
|
for line in lines:
|
|
m = PATTERNS["brute_force"].search(line)
|
|
if m:
|
|
ip = m.group("ip")
|
|
failed_logins[ip] += 1
|
|
|
|
m = PATTERNS["invalid_user"].search(line)
|
|
if m:
|
|
ip = m.group("ip")
|
|
failed_logins[ip] += 1
|
|
|
|
for ip, count in failed_logins.items():
|
|
if count >= ALERT_THRESHOLD_FAIL:
|
|
create_alert(
|
|
source="auth.log",
|
|
alert_type="brute_force_ssh",
|
|
severity="CRITICAL",
|
|
ip=ip,
|
|
detail=f"{count} failed SSH login attempts detected",
|
|
)
|
|
|
|
|
|
def analyze_nginx_log(lines: list[str]):
|
|
"""Detect HTTP flooding and SQL injection attempts."""
|
|
ip_requests = defaultdict(list)
|
|
|
|
for line in lines:
|
|
m = PATTERNS["nginx_request"].search(line)
|
|
if m:
|
|
ip = m.group("ip")
|
|
path = m.group("path")
|
|
status = m.group("status")
|
|
ip_requests[ip].append((path, status))
|
|
|
|
# Check for SQL injection in path
|
|
if PATTERNS["sql_injection"].search(path):
|
|
create_alert(
|
|
source="nginx/access.log",
|
|
alert_type="sql_injection_attempt",
|
|
severity="HIGH",
|
|
ip=ip,
|
|
detail=f"Possible SQL injection detected in request path: {path[:120]}",
|
|
)
|
|
|
|
# Check high request rate
|
|
for ip, reqs in ip_requests.items():
|
|
failed = [r for r in reqs if r[1].startswith(("4", "5"))]
|
|
if len(reqs) >= ALERT_THRESHOLD_RATE:
|
|
create_alert(
|
|
source="nginx/access.log",
|
|
alert_type="http_flood",
|
|
severity="HIGH",
|
|
ip=ip,
|
|
detail=f"{len(reqs)} requests detected in log window ({len(failed)} errors)",
|
|
)
|
|
elif len(failed) >= ALERT_THRESHOLD_FAIL:
|
|
create_alert(
|
|
source="nginx/access.log",
|
|
alert_type="repeated_http_errors",
|
|
severity="MEDIUM",
|
|
ip=ip,
|
|
detail=f"{len(failed)} HTTP error responses to IP in log window",
|
|
)
|
|
|
|
|
|
def analyze_app_log(lines: list[str]):
|
|
"""Detect application-level errors and anomalies."""
|
|
for line in lines:
|
|
m = PATTERNS["app_error"].search(line)
|
|
if m:
|
|
level = m.group("level")
|
|
msg = m.group("msg")
|
|
|
|
# Extract IP if present
|
|
ip_match = re.search(r"\d+\.\d+\.\d+\.\d+", msg)
|
|
ip = ip_match.group(0) if ip_match else "unknown"
|
|
|
|
if "JWT" in msg or "token" in msg.lower():
|
|
severity = "HIGH"
|
|
elif level == "CRITICAL":
|
|
severity = "CRITICAL"
|
|
else:
|
|
severity = "MEDIUM"
|
|
create_alert(
|
|
source="app/application.log",
|
|
alert_type="application_error",
|
|
severity=severity,
|
|
ip=ip,
|
|
detail=msg[:200],
|
|
)
|
|
|
|
|
|
def create_alert(source: str, alert_type: str, severity: str, ip: str, detail: str):
|
|
"""Create a deduplicated alert entry."""
|
|
# Deduplicate: only merge if same type + IP + first 60 chars of detail match.
|
|
# This prevents a low-severity alert from absorbing a different high-severity one.
|
|
for existing in alerts:
|
|
if (existing["type"] == alert_type
|
|
and existing["ip"] == ip
|
|
and existing["detail"][:60] == detail[:60]):
|
|
existing["count"] += 1
|
|
existing["last_seen"] = datetime.now().isoformat()
|
|
return
|
|
|
|
alert_entry = {
|
|
"id": len(alerts) + 1,
|
|
"timestamp": datetime.now().isoformat(),
|
|
"last_seen": datetime.now().isoformat(),
|
|
"source": source,
|
|
"type": alert_type,
|
|
"severity": severity,
|
|
"ip": ip,
|
|
"detail": detail,
|
|
"count": 1,
|
|
"status": "open",
|
|
}
|
|
alerts.append(alert_entry)
|
|
log.warning(f"[ALERT] [{severity}] {alert_type} — IP: {ip} — {detail[:80]}")
|
|
|
|
|
|
def run_scan():
|
|
"""Main scan loop — reads all 3 log sources and runs analysis.
|
|
|
|
NOTE: This implementation is stateless — alert history resets on each
|
|
scan cycle. This is acceptable for demo/project use. Production deployments
|
|
should persist alert history in a database or SIEM backend (e.g.,
|
|
Elasticsearch/OpenSearch) to retain trends and support correlation rules.
|
|
"""
|
|
global alerts, failed_logins
|
|
alerts = []
|
|
failed_logins = defaultdict(int)
|
|
|
|
log.info("Starting SIEM scan cycle...")
|
|
|
|
# Source 1: Auth logs
|
|
auth_lines = read_log_source("auth", LOG_SOURCES["auth"])
|
|
analyze_auth_log(auth_lines)
|
|
|
|
# Source 2: Nginx access logs
|
|
nginx_lines = read_log_source("nginx", LOG_SOURCES["nginx"])
|
|
analyze_nginx_log(nginx_lines)
|
|
|
|
# Source 3: Application logs
|
|
app_lines = read_log_source("app", LOG_SOURCES["app"])
|
|
analyze_app_log(app_lines)
|
|
|
|
# Sort by severity
|
|
severity_order = {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "LOW": 3}
|
|
alerts.sort(key=lambda a: severity_order.get(a["severity"], 9))
|
|
|
|
output = {
|
|
"generated_at": datetime.now().isoformat(),
|
|
"total_alerts": len(alerts),
|
|
"critical_count": sum(1 for a in alerts if a["severity"] == "CRITICAL"),
|
|
"high_count": sum(1 for a in alerts if a["severity"] == "HIGH"),
|
|
"medium_count": sum(1 for a in alerts if a["severity"] == "MEDIUM"),
|
|
"alerts": alerts,
|
|
}
|
|
|
|
with open(OUTPUT_FILE, "w") as f:
|
|
json.dump(output, f, indent=2)
|
|
|
|
log.info(f"Scan complete. {len(alerts)} alerts written to {OUTPUT_FILE}")
|
|
return output
|
|
|
|
|
|
if __name__ == "__main__":
|
|
log.info("=== Ghaymah SIEM Starting ===")
|
|
log.info(f"Monitoring: {', '.join(LOG_SOURCES.values())}")
|
|
log.info(f"Output: {OUTPUT_FILE}")
|
|
|
|
while True:
|
|
try:
|
|
results = run_scan()
|
|
print(f"\n[{datetime.now().strftime('%H:%M:%S')}] Alerts: {results['total_alerts']} "
|
|
f"(CRITICAL: {results['critical_count']}, HIGH: {results['high_count']}, "
|
|
f"MEDIUM: {results['medium_count']})")
|
|
except KeyboardInterrupt:
|
|
log.info("SIEM stopped by user.")
|
|
break
|
|
except Exception as e:
|
|
log.error(f"Scan error: {e}")
|
|
|
|
time.sleep(SCAN_INTERVAL_SEC)
|