219 أسطر
8.4 KiB
Python
219 أسطر
8.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
siem_analyzer.py
|
|
-----------------
|
|
نظام SIEM مبسط: يجمع السجلات من 3 مصادر (Web Server / Auth-OS / Firewall)،
|
|
يحللها لاكتشاف أنماط مشبوهة، ويحفظ النتائج في alerts.json ليقرأها الـ Dashboard.
|
|
|
|
الاستخدام:
|
|
python3 siem_analyzer.py --once # تشغيل مرة واحدة
|
|
python3 siem_analyzer.py --watch 30 # تشغيل دوري كل 30 ثانية (مثل خدمة حقيقية)
|
|
|
|
يمكن تعديل مسارات ومحددات (thresholds) الكشف في قسم CONFIG أدناه.
|
|
"""
|
|
|
|
import re
|
|
import json
|
|
import argparse
|
|
import time
|
|
from collections import defaultdict, Counter
|
|
from datetime import datetime
|
|
|
|
# ================== CONFIG ==================
|
|
CONFIG = {
|
|
"web_log_path": "logs/access.log",
|
|
"auth_log_path": "logs/auth.log",
|
|
"firewall_log_path": "logs/firewall.log",
|
|
"output_path": "output/alerts.json",
|
|
"thresholds": {
|
|
"web_404_403_count": 15, # عدد 404/403 من نفس IP يعتبر Scanning
|
|
"brute_force_fails": 5, # عدد محاولات فاشلة من نفس IP قبل التنبيه
|
|
"port_scan_ports": 15, # عدد منافذ مختلفة محظورة من نفس IP
|
|
},
|
|
}
|
|
|
|
SQLI_PATTERN = re.compile(
|
|
r"(\bunion\b.*\bselect\b|\bselect\b.*\bfrom\b|\bor\b\s+1=1|--\s*$|'.*or.*'.*=.*')",
|
|
re.IGNORECASE,
|
|
)
|
|
TRAVERSAL_PATTERN = re.compile(r"(\.\./|%2e%2e%2f|\.\.%2f)", re.IGNORECASE)
|
|
WEB_LOG_LINE = re.compile(
|
|
r'(?P<ip>\d{1,3}(?:\.\d{1,3}){3}).*\[(?P<time>[^\]]+)\]\s+"(?P<method>\w+)\s+'
|
|
r'(?P<path>\S+)\s+HTTP/[\d.]+"\s+(?P<status>\d{3})'
|
|
)
|
|
AUTH_FAILED = re.compile(
|
|
r"(?P<time>\w+\s+\d+\s+[\d:]+).*Failed password.*from (?P<ip>\d{1,3}(?:\.\d{1,3}){3})"
|
|
)
|
|
AUTH_ACCEPTED = re.compile(
|
|
r"(?P<time>\w+\s+\d+\s+[\d:]+).*Accepted password.*from (?P<ip>\d{1,3}(?:\.\d{1,3}){3})"
|
|
)
|
|
FW_LINE = re.compile(
|
|
r"(?P<time>\w+\s+\d+\s+[\d:]+).*SRC=(?P<ip>\d{1,3}(?:\.\d{1,3}){3}).*"
|
|
r"DPT=(?P<port>\d+)\s+ACTION=(?P<action>\w+)"
|
|
)
|
|
|
|
|
|
# ================== ANALYZERS ==================
|
|
|
|
def analyze_web_log(path, thresholds):
|
|
"""يحلل سجلات خادم الويب: SQLi, Directory Traversal, Scanning عبر 404/403 مرتفعة."""
|
|
alerts = []
|
|
error_counts = Counter()
|
|
try:
|
|
with open(path, "r", errors="ignore") as f:
|
|
for line in f:
|
|
m = WEB_LOG_LINE.search(line)
|
|
if not m:
|
|
continue
|
|
ip, path_req, status = m["ip"], m["path"], m["status"]
|
|
|
|
if SQLI_PATTERN.search(path_req):
|
|
alerts.append({
|
|
"source": "web", "type": "SQL Injection",
|
|
"ip": ip, "detail": path_req[:120], "severity": "high",
|
|
})
|
|
if TRAVERSAL_PATTERN.search(path_req):
|
|
alerts.append({
|
|
"source": "web", "type": "Directory Traversal",
|
|
"ip": ip, "detail": path_req[:120], "severity": "high",
|
|
})
|
|
if status in ("404", "403"):
|
|
error_counts[ip] += 1
|
|
except FileNotFoundError:
|
|
print(f"[!] تحذير: ملف السجل غير موجود: {path}")
|
|
return alerts
|
|
|
|
for ip, count in error_counts.items():
|
|
if count >= thresholds["web_404_403_count"]:
|
|
alerts.append({
|
|
"source": "web", "type": "Scanning/Fuzzing",
|
|
"ip": ip, "detail": f"{count} طلب 404/403 من نفس المصدر",
|
|
"severity": "medium",
|
|
})
|
|
return alerts
|
|
|
|
|
|
def analyze_auth_log(path, thresholds):
|
|
"""يحلل سجلات المصادقة: Brute Force على SSH، ونجاح دخول بعد فشل متكرر."""
|
|
alerts = []
|
|
fail_counts = defaultdict(int)
|
|
fail_then_success = defaultdict(int)
|
|
try:
|
|
with open(path, "r", errors="ignore") as f:
|
|
for line in f:
|
|
mf = AUTH_FAILED.search(line)
|
|
if mf:
|
|
fail_counts[mf["ip"]] += 1
|
|
continue
|
|
ms = AUTH_ACCEPTED.search(line)
|
|
if ms and fail_counts[ms["ip"]] >= thresholds["brute_force_fails"]:
|
|
fail_then_success[ms["ip"]] = fail_counts[ms["ip"]]
|
|
except FileNotFoundError:
|
|
print(f"[!] تحذير: ملف السجل غير موجود: {path}")
|
|
return alerts
|
|
|
|
for ip, count in fail_counts.items():
|
|
if count >= thresholds["brute_force_fails"]:
|
|
alerts.append({
|
|
"source": "auth", "type": "Brute Force (SSH)",
|
|
"ip": ip, "detail": f"{count} محاولة فاشلة", "severity": "high",
|
|
})
|
|
for ip, count in fail_then_success.items():
|
|
alerts.append({
|
|
"source": "auth", "type": "Successful login after failures",
|
|
"ip": ip, "detail": f"دخول ناجح بعد {count} محاولة فاشلة",
|
|
"severity": "critical",
|
|
})
|
|
return alerts
|
|
|
|
|
|
def analyze_firewall_log(path, thresholds):
|
|
"""يحلل سجلات الجدار الناري: Port Scanning عبر تعدد المنافذ المحظورة من نفس IP."""
|
|
alerts = []
|
|
blocked_ports = defaultdict(set)
|
|
try:
|
|
with open(path, "r", errors="ignore") as f:
|
|
for line in f:
|
|
m = FW_LINE.search(line)
|
|
if not m:
|
|
continue
|
|
if m["action"] in ("DROP", "REJECT", "BLOCK"):
|
|
blocked_ports[m["ip"]].add(m["port"])
|
|
except FileNotFoundError:
|
|
print(f"[!] تحذير: ملف السجل غير موجود: {path}")
|
|
return alerts
|
|
|
|
for ip, ports in blocked_ports.items():
|
|
if len(ports) >= thresholds["port_scan_ports"]:
|
|
alerts.append({
|
|
"source": "firewall", "type": "Port Scanning",
|
|
"ip": ip, "detail": f"محاولة الاتصال بـ {len(ports)} منفذ مختلف",
|
|
"severity": "high",
|
|
})
|
|
return alerts
|
|
|
|
|
|
# ================== AGGREGATION ==================
|
|
|
|
def run_analysis(cfg):
|
|
alerts = []
|
|
alerts += analyze_web_log(cfg["web_log_path"], cfg["thresholds"])
|
|
alerts += analyze_auth_log(cfg["auth_log_path"], cfg["thresholds"])
|
|
alerts += analyze_firewall_log(cfg["firewall_log_path"], cfg["thresholds"])
|
|
|
|
ip_scores = Counter()
|
|
severity_weight = {"low": 1, "medium": 2, "high": 3, "critical": 5}
|
|
for a in alerts:
|
|
ip_scores[a["ip"]] += severity_weight.get(a["severity"], 1)
|
|
|
|
suspicious_ips = [
|
|
{"ip": ip, "score": score, "alert_count": sum(1 for a in alerts if a["ip"] == ip)}
|
|
for ip, score in ip_scores.most_common()
|
|
]
|
|
|
|
result = {
|
|
"generated_at": datetime.now().isoformat(),
|
|
"stats": {
|
|
"total_alerts": len(alerts),
|
|
"by_source": dict(Counter(a["source"] for a in alerts)),
|
|
"by_severity": dict(Counter(a["severity"] for a in alerts)),
|
|
"unique_suspicious_ips": len(suspicious_ips),
|
|
},
|
|
"alerts": sorted(alerts, key=lambda a: severity_weight.get(a["severity"], 0), reverse=True),
|
|
"suspicious_ips": suspicious_ips,
|
|
}
|
|
return result
|
|
|
|
|
|
def save_result(result, output_path):
|
|
import os
|
|
os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True)
|
|
with open(output_path, "w", encoding="utf-8") as f:
|
|
json.dump(result, f, ensure_ascii=False, indent=2)
|
|
print(f"[+] تم حفظ {result['stats']['total_alerts']} تنبيه في {output_path}")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="نظام SIEM مبسط")
|
|
parser.add_argument("--once", action="store_true", help="تشغيل مرة واحدة فقط")
|
|
parser.add_argument("--watch", type=int, default=0,
|
|
help="تشغيل دوري كل N ثانية (محاكاة خدمة حية)")
|
|
args = parser.parse_args()
|
|
|
|
if args.watch > 0:
|
|
print(f"[*] وضع المراقبة الدورية: كل {args.watch} ثانية. اضغط Ctrl+C للإيقاف.")
|
|
try:
|
|
while True:
|
|
result = run_analysis(CONFIG)
|
|
save_result(result, CONFIG["output_path"])
|
|
time.sleep(args.watch)
|
|
except KeyboardInterrupt:
|
|
print("\n[*] تم إيقاف المراقبة.")
|
|
else:
|
|
result = run_analysis(CONFIG)
|
|
save_result(result, CONFIG["output_path"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|