أنهيت الإمتحان
هذا الالتزام موجود في:
218
q4-siem-log-analysis/siem_analyzer.py
Normal file
218
q4-siem-log-analysis/siem_analyzer.py
Normal file
@@ -0,0 +1,218 @@
|
||||
#!/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()
|
||||
المرجع في مشكلة جديدة
حظر مستخدم