441 أسطر
18 KiB
Python
441 أسطر
18 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
SIEM مبسط - نظام جمع وتحليل السجلات
|
|
يجمع ويحلل logs من 3 مصادر ويكتشف الأنماط المشبوهة
|
|
|
|
المصادر:
|
|
1. Nginx Access Logs
|
|
2. Auth/SSH Logs
|
|
3. Application Logs
|
|
"""
|
|
|
|
import re
|
|
import json
|
|
import os
|
|
from datetime import datetime, timedelta
|
|
from collections import defaultdict
|
|
from typing import Dict, List, Tuple
|
|
import hashlib
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# إعدادات المصادر
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
LOG_SOURCES = {
|
|
'nginx': '/var/log/nginx/access.log',
|
|
'auth': '/var/log/auth.log',
|
|
'app': '/var/log/app/application.log'
|
|
}
|
|
|
|
# للتجربة - ملفات محلية
|
|
SAMPLE_LOG_SOURCES = {
|
|
'nginx': 'sample_logs/nginx_access.log',
|
|
'auth': 'sample_logs/auth.log',
|
|
'app': 'sample_logs/app.log'
|
|
}
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# أنماط الكشف عن التهديدات
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
THREAT_PATTERNS = {
|
|
'brute_force': {
|
|
'description': 'محاولات تسجيل دخول فاشلة متكررة',
|
|
'threshold': 5,
|
|
'window_minutes': 5,
|
|
'severity': 'critical'
|
|
},
|
|
'sql_injection': {
|
|
'description': 'محاولة SQL Injection',
|
|
'patterns': [
|
|
r"(\%27)|(\')|(\-\-)|(\%23)|(#)",
|
|
r"((\%3D)|(=))[^\n]*((\%27)|(\')|(\-\-)|(\%3B)|(;))",
|
|
r"\w*((\%27)|(\'))((\%6F)|o|(\%4F))((\%72)|r|(\%52))",
|
|
r"union.*select",
|
|
r"select.*from",
|
|
r"insert.*into",
|
|
r"drop.*table"
|
|
],
|
|
'severity': 'critical'
|
|
},
|
|
'xss_attempt': {
|
|
'description': 'محاولة XSS',
|
|
'patterns': [
|
|
r"<script[^>]*>",
|
|
r"javascript:",
|
|
r"on\w+\s*=",
|
|
r"<iframe",
|
|
r"<object"
|
|
],
|
|
'severity': 'high'
|
|
},
|
|
'path_traversal': {
|
|
'description': 'محاولة Path Traversal',
|
|
'patterns': [
|
|
r"\.\./",
|
|
r"\.\.\\",
|
|
r"%2e%2e%2f",
|
|
r"%2e%2e/"
|
|
],
|
|
'severity': 'high'
|
|
},
|
|
'suspicious_user_agent': {
|
|
'description': 'User-Agent مشبوه',
|
|
'patterns': [
|
|
r"nikto",
|
|
r"sqlmap",
|
|
r"nmap",
|
|
r"masscan",
|
|
r"dirbuster",
|
|
r"gobuster",
|
|
r"wfuzz"
|
|
],
|
|
'severity': 'medium'
|
|
},
|
|
'ssh_brute_force': {
|
|
'description': 'محاولات SSH فاشلة',
|
|
'threshold': 3,
|
|
'window_minutes': 5,
|
|
'severity': 'critical'
|
|
}
|
|
}
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# فئة تحليل السجلات
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
class LogAnalyzer:
|
|
def __init__(self, use_sample=True):
|
|
self.sources = SAMPLE_LOG_SOURCES if use_sample else LOG_SOURCES
|
|
self.alerts = []
|
|
self.suspicious_ips = defaultdict(lambda: {
|
|
'count': 0,
|
|
'first_seen': None,
|
|
'last_seen': None,
|
|
'threats': []
|
|
})
|
|
self.stats = {
|
|
'total_lines': 0,
|
|
'threats_detected': 0,
|
|
'sources_analyzed': 0
|
|
}
|
|
|
|
def parse_nginx_log(self, line: str) -> Dict:
|
|
"""تحليل سطر من Nginx access log"""
|
|
pattern = r'(\d+\.\d+\.\d+\.\d+) - - \[([^\]]+)\] "(\w+) ([^"]+)" (\d+) (\d+) "([^"]*)" "([^"]*)"'
|
|
match = re.match(pattern, line)
|
|
if match:
|
|
return {
|
|
'ip': match.group(1),
|
|
'timestamp': match.group(2),
|
|
'method': match.group(3),
|
|
'path': match.group(4),
|
|
'status': int(match.group(5)),
|
|
'size': int(match.group(6)),
|
|
'referer': match.group(7),
|
|
'user_agent': match.group(8),
|
|
'source': 'nginx'
|
|
}
|
|
return None
|
|
|
|
def parse_auth_log(self, line: str) -> Dict:
|
|
"""تحليل سطر من Auth log"""
|
|
# Failed password
|
|
failed_pattern = r'(\w+\s+\d+\s+[\d:]+).*Failed password for (?:invalid user )?(\w+) from (\d+\.\d+\.\d+\.\d+)'
|
|
match = re.search(failed_pattern, line)
|
|
if match:
|
|
return {
|
|
'timestamp': match.group(1),
|
|
'username': match.group(2),
|
|
'ip': match.group(3),
|
|
'event': 'failed_login',
|
|
'source': 'auth'
|
|
}
|
|
|
|
# Accepted password
|
|
success_pattern = r'(\w+\s+\d+\s+[\d:]+).*Accepted password for (\w+) from (\d+\.\d+\.\d+\.\d+)'
|
|
match = re.search(success_pattern, line)
|
|
if match:
|
|
return {
|
|
'timestamp': match.group(1),
|
|
'username': match.group(2),
|
|
'ip': match.group(3),
|
|
'event': 'successful_login',
|
|
'source': 'auth'
|
|
}
|
|
return None
|
|
|
|
def parse_app_log(self, line: str) -> Dict:
|
|
"""تحليل سطر من Application log"""
|
|
pattern = r'\[(\d{4}-\d{2}-\d{2} [\d:]+)\] \[(\w+)\] \[(\d+\.\d+\.\d+\.\d+)\] (.+)'
|
|
match = re.match(pattern, line)
|
|
if match:
|
|
return {
|
|
'timestamp': match.group(1),
|
|
'level': match.group(2),
|
|
'ip': match.group(3),
|
|
'message': match.group(4),
|
|
'source': 'app'
|
|
}
|
|
return None
|
|
|
|
def detect_pattern_threat(self, log_entry: Dict, threat_type: str) -> bool:
|
|
"""كشف التهديدات بناءً على الأنماط"""
|
|
threat = THREAT_PATTERNS.get(threat_type)
|
|
if not threat or 'patterns' not in threat:
|
|
return False
|
|
|
|
# البحث في الحقول المناسبة
|
|
search_fields = ['path', 'message', 'user_agent', 'referer']
|
|
for field in search_fields:
|
|
if field in log_entry:
|
|
for pattern in threat['patterns']:
|
|
if re.search(pattern, str(log_entry[field]), re.IGNORECASE):
|
|
return True
|
|
return False
|
|
|
|
def analyze_log_entry(self, entry: Dict):
|
|
"""تحليل سجل واحد للكشف عن التهديدات"""
|
|
if not entry:
|
|
return
|
|
|
|
ip = entry.get('ip', 'unknown')
|
|
|
|
# تحديث معلومات IP
|
|
self.suspicious_ips[ip]['count'] += 1
|
|
if not self.suspicious_ips[ip]['first_seen']:
|
|
self.suspicious_ips[ip]['first_seen'] = entry.get('timestamp')
|
|
self.suspicious_ips[ip]['last_seen'] = entry.get('timestamp')
|
|
|
|
# كشف SQL Injection
|
|
if self.detect_pattern_threat(entry, 'sql_injection'):
|
|
self.add_alert('sql_injection', entry)
|
|
|
|
# كشف XSS
|
|
if self.detect_pattern_threat(entry, 'xss_attempt'):
|
|
self.add_alert('xss_attempt', entry)
|
|
|
|
# كشف Path Traversal
|
|
if self.detect_pattern_threat(entry, 'path_traversal'):
|
|
self.add_alert('path_traversal', entry)
|
|
|
|
# كشف User-Agent مشبوه
|
|
if self.detect_pattern_threat(entry, 'suspicious_user_agent'):
|
|
self.add_alert('suspicious_user_agent', entry)
|
|
|
|
# كشف محاولات تسجيل دخول فاشلة
|
|
if entry.get('event') == 'failed_login':
|
|
self.suspicious_ips[ip]['threats'].append('failed_login')
|
|
if len([t for t in self.suspicious_ips[ip]['threats'] if t == 'failed_login']) >= 5:
|
|
self.add_alert('brute_force', entry)
|
|
|
|
# كشف أكواد HTTP مشبوهة
|
|
status = entry.get('status')
|
|
if status:
|
|
if status == 401 or status == 403:
|
|
self.suspicious_ips[ip]['threats'].append('auth_failure')
|
|
elif status >= 500:
|
|
self.suspicious_ips[ip]['threats'].append('server_error')
|
|
|
|
def add_alert(self, threat_type: str, entry: Dict):
|
|
"""إضافة تنبيه جديد"""
|
|
threat_info = THREAT_PATTERNS.get(threat_type, {})
|
|
alert = {
|
|
'id': hashlib.md5(f"{threat_type}{entry.get('ip')}{datetime.now()}".encode()).hexdigest()[:8],
|
|
'timestamp': datetime.now().isoformat(),
|
|
'threat_type': threat_type,
|
|
'description': threat_info.get('description', threat_type),
|
|
'severity': threat_info.get('severity', 'medium'),
|
|
'source_ip': entry.get('ip', 'unknown'),
|
|
'details': entry,
|
|
'status': 'new'
|
|
}
|
|
self.alerts.append(alert)
|
|
self.stats['threats_detected'] += 1
|
|
|
|
def analyze_file(self, source_name: str, filepath: str):
|
|
"""تحليل ملف سجلات"""
|
|
if not os.path.exists(filepath):
|
|
print(f"⚠️ الملف غير موجود: {filepath}")
|
|
return
|
|
|
|
parser = {
|
|
'nginx': self.parse_nginx_log,
|
|
'auth': self.parse_auth_log,
|
|
'app': self.parse_app_log
|
|
}.get(source_name)
|
|
|
|
if not parser:
|
|
print(f"⚠️ لا يوجد محلل لـ: {source_name}")
|
|
return
|
|
|
|
print(f"📂 تحليل {source_name}: {filepath}")
|
|
|
|
with open(filepath, 'r', encoding='utf-8', errors='ignore') as f:
|
|
for line in f:
|
|
self.stats['total_lines'] += 1
|
|
entry = parser(line.strip())
|
|
if entry:
|
|
self.analyze_log_entry(entry)
|
|
|
|
self.stats['sources_analyzed'] += 1
|
|
|
|
def analyze_all(self):
|
|
"""تحليل جميع المصادر"""
|
|
print("=" * 60)
|
|
print("🔍 بدء تحليل السجلات - SIEM مبسط")
|
|
print("=" * 60)
|
|
|
|
for source_name, filepath in self.sources.items():
|
|
self.analyze_file(source_name, filepath)
|
|
|
|
print("\n" + "=" * 60)
|
|
print("📊 نتائج التحليل")
|
|
print("=" * 60)
|
|
|
|
def get_suspicious_ips(self, min_threats: int = 3) -> List[Dict]:
|
|
"""الحصول على قائمة IPs المشبوهة"""
|
|
suspicious = []
|
|
for ip, data in self.suspicious_ips.items():
|
|
threat_count = len(data['threats'])
|
|
if threat_count >= min_threats:
|
|
suspicious.append({
|
|
'ip': ip,
|
|
'request_count': data['count'],
|
|
'threat_count': threat_count,
|
|
'threats': list(set(data['threats'])),
|
|
'first_seen': data['first_seen'],
|
|
'last_seen': data['last_seen']
|
|
})
|
|
return sorted(suspicious, key=lambda x: x['threat_count'], reverse=True)
|
|
|
|
def get_alerts_by_severity(self) -> Dict[str, List]:
|
|
"""تجميع التنبيهات حسب الخطورة"""
|
|
by_severity = defaultdict(list)
|
|
for alert in self.alerts:
|
|
by_severity[alert['severity']].append(alert)
|
|
return dict(by_severity)
|
|
|
|
def generate_report(self) -> Dict:
|
|
"""إنشاء تقرير شامل"""
|
|
return {
|
|
'generated_at': datetime.now().isoformat(),
|
|
'statistics': self.stats,
|
|
'alerts': self.alerts,
|
|
'alerts_by_severity': self.get_alerts_by_severity(),
|
|
'suspicious_ips': self.get_suspicious_ips(),
|
|
'summary': {
|
|
'total_alerts': len(self.alerts),
|
|
'critical': len([a for a in self.alerts if a['severity'] == 'critical']),
|
|
'high': len([a for a in self.alerts if a['severity'] == 'high']),
|
|
'medium': len([a for a in self.alerts if a['severity'] == 'medium']),
|
|
'suspicious_ips_count': len(self.get_suspicious_ips())
|
|
}
|
|
}
|
|
|
|
def print_summary(self):
|
|
"""طباعة ملخص النتائج"""
|
|
report = self.generate_report()
|
|
|
|
print(f"\n📈 الإحصائيات:")
|
|
print(f" - إجمالي الأسطر المحللة: {report['statistics']['total_lines']}")
|
|
print(f" - المصادر المحللة: {report['statistics']['sources_analyzed']}")
|
|
print(f" - التهديدات المكتشفة: {report['statistics']['threats_detected']}")
|
|
|
|
print(f"\n🚨 التنبيهات حسب الخطورة:")
|
|
print(f" - حرجة (Critical): {report['summary']['critical']}")
|
|
print(f" - عالية (High): {report['summary']['high']}")
|
|
print(f" - متوسطة (Medium): {report['summary']['medium']}")
|
|
|
|
print(f"\n🔴 عناوين IP المشبوهة ({report['summary']['suspicious_ips_count']}):")
|
|
for ip_data in report['suspicious_ips'][:10]:
|
|
print(f" - {ip_data['ip']}: {ip_data['threat_count']} تهديدات, {ip_data['request_count']} طلبات")
|
|
|
|
if self.alerts:
|
|
print(f"\n⚠️ آخر 5 تنبيهات:")
|
|
for alert in self.alerts[-5:]:
|
|
print(f" [{alert['severity'].upper()}] {alert['description']} من {alert['source_ip']}")
|
|
|
|
def export_json(self, filepath: str):
|
|
"""تصدير التقرير كـ JSON"""
|
|
report = self.generate_report()
|
|
with open(filepath, 'w', encoding='utf-8') as f:
|
|
json.dump(report, f, ensure_ascii=False, indent=2)
|
|
print(f"\n💾 تم حفظ التقرير: {filepath}")
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# إنشاء ملفات سجلات تجريبية
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
def create_sample_logs():
|
|
"""إنشاء ملفات سجلات تجريبية للاختبار"""
|
|
os.makedirs('sample_logs', exist_ok=True)
|
|
|
|
# Nginx sample logs
|
|
nginx_logs = """192.168.1.100 - - [26/Jul/2026:14:05:01 +0000] "POST /api/login HTTP/1.1" 401 45 "-" "Mozilla/5.0"
|
|
192.168.1.100 - - [26/Jul/2026:14:05:02 +0000] "POST /api/login HTTP/1.1" 401 45 "-" "Mozilla/5.0"
|
|
192.168.1.100 - - [26/Jul/2026:14:05:03 +0000] "POST /api/login HTTP/1.1" 401 45 "-" "Mozilla/5.0"
|
|
192.168.1.100 - - [26/Jul/2026:14:05:04 +0000] "POST /api/login HTTP/1.1" 401 45 "-" "Mozilla/5.0"
|
|
192.168.1.100 - - [26/Jul/2026:14:05:05 +0000] "POST /api/login HTTP/1.1" 401 45 "-" "Mozilla/5.0"
|
|
192.168.1.100 - - [26/Jul/2026:14:05:06 +0000] "POST /api/login HTTP/1.1" 200 512 "-" "Mozilla/5.0"
|
|
10.0.0.50 - - [26/Jul/2026:14:10:01 +0000] "GET /api/users?id=1' OR '1'='1 HTTP/1.1" 400 120 "-" "sqlmap/1.5"
|
|
10.0.0.50 - - [26/Jul/2026:14:10:02 +0000] "GET /api/users?id=1 UNION SELECT * FROM users HTTP/1.1" 400 120 "-" "sqlmap/1.5"
|
|
172.16.0.25 - - [26/Jul/2026:14:15:01 +0000] "GET /page?q=<script>alert('xss')</script> HTTP/1.1" 400 80 "-" "Mozilla/5.0"
|
|
172.16.0.25 - - [26/Jul/2026:14:15:02 +0000] "GET /../../etc/passwd HTTP/1.1" 403 50 "-" "Mozilla/5.0"
|
|
8.8.8.8 - - [26/Jul/2026:14:20:01 +0000] "GET / HTTP/1.1" 200 1024 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
|
|
8.8.8.8 - - [26/Jul/2026:14:20:02 +0000] "GET /about HTTP/1.1" 200 2048 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
|
|
"""
|
|
|
|
# Auth sample logs
|
|
auth_logs = """Jul 26 14:00:01 server sshd[1234]: Failed password for invalid user admin from 192.168.1.200 port 22 ssh2
|
|
Jul 26 14:00:02 server sshd[1234]: Failed password for invalid user admin from 192.168.1.200 port 22 ssh2
|
|
Jul 26 14:00:03 server sshd[1234]: Failed password for invalid user root from 192.168.1.200 port 22 ssh2
|
|
Jul 26 14:00:04 server sshd[1234]: Failed password for invalid user test from 192.168.1.200 port 22 ssh2
|
|
Jul 26 14:00:05 server sshd[1234]: Failed password for invalid user user from 192.168.1.200 port 22 ssh2
|
|
Jul 26 14:05:01 server sshd[1235]: Accepted password for developer from 10.0.0.10 port 22 ssh2
|
|
"""
|
|
|
|
# App sample logs
|
|
app_logs = """[2026-07-26 14:00:01] [ERROR] [192.168.1.100] Login failed for user: admin
|
|
[2026-07-26 14:00:02] [ERROR] [192.168.1.100] Login failed for user: admin
|
|
[2026-07-26 14:00:03] [WARN] [10.0.0.50] Suspicious query detected: SELECT * FROM users
|
|
[2026-07-26 14:00:04] [INFO] [8.8.8.8] User logged in successfully
|
|
[2026-07-26 14:00:05] [ERROR] [172.16.0.25] Invalid input: <script>alert(1)</script>
|
|
"""
|
|
|
|
with open('sample_logs/nginx_access.log', 'w') as f:
|
|
f.write(nginx_logs)
|
|
|
|
with open('sample_logs/auth.log', 'w') as f:
|
|
f.write(auth_logs)
|
|
|
|
with open('sample_logs/app.log', 'w') as f:
|
|
f.write(app_logs)
|
|
|
|
print("✅ تم إنشاء ملفات السجلات التجريبية في sample_logs/")
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════
|
|
# التشغيل الرئيسي
|
|
# ═══════════════════════════════════════════════════════════════
|
|
|
|
if __name__ == '__main__':
|
|
import sys
|
|
|
|
# إنشاء ملفات تجريبية إذا لم تكن موجودة
|
|
if not os.path.exists('sample_logs'):
|
|
create_sample_logs()
|
|
|
|
# تحليل السجلات
|
|
analyzer = LogAnalyzer(use_sample=True)
|
|
analyzer.analyze_all()
|
|
analyzer.print_summary()
|
|
|
|
# تصدير التقرير
|
|
analyzer.export_json('siem_report.json')
|
|
|
|
print("\n" + "=" * 60)
|
|
print("✅ اكتمل التحليل - راجع siem_report.json للتفاصيل")
|
|
print("=" * 60)
|