add the main file structure requirements
هذا الالتزام موجود في:
440
q4-siem-system/log_analyzer.py
Normal file
440
q4-siem-system/log_analyzer.py
Normal file
@@ -0,0 +1,440 @@
|
||||
#!/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)
|
||||
المرجع في مشكلة جديدة
حظر مستخدم