"""
Security Log Analyzer
ویژگیها:
1. تحلیل لاگهای مختلف
2. تشخیص الگوهای حمله
3. تولید گزارش امنیتی
4. یکپارچهسازی با SIEM
"""
import re
from collections import Counter
from datetime import datetime, timedelta
import pandas as pd
from typing import List, Dict, Tuple
class LogAnalyzer:
"""تحلیلگر لاگ امنیتی"""
# الگوهای حمله
ATTACK_PATTERNS = {
'SQL Injection': [
r"union.*select",
r"'.*or.*'1'='1",
r"select.*from",
r"insert.*into",
r"drop.*table",
r"' OR '1'='1",
r"--",
r";--",
r"/*",
r"*/",
r"@@version",
r"waitfor delay"
],
'XSS Attack': [
r"<script>",
r"alert\(",
r"onerror=",
r"onload=",
r"javascript:",
r"eval\(",
r"document\.cookie",
r"<iframe",
r"<img src=",
r"onmouseover="
],
'Brute Force': [
r"Failed password",
r"authentication failure",
r"Invalid user",
r"Too many authentication failures",
r"POSSIBLE BREAK-IN ATTEMPT",
r"bad password attempt"
],
'Port Scan': [
r"port scan",
r"SYN flood",
r"connection attempts",
r"multiple connections from",
r"scan detected"
],
'DoS Attack': [
r"Denial of Service",
r"flooding",
r"rate limit exceeded",
r"too many requests",
r"connection limit"
]
}
def __init__(self):
self.logs = []
self.detected_attacks = []
self.stats = Counter()
def load_logs(self, file_path: str):
"""بارگذاری لاگها"""
print(f" بارگذاری لاگها از {file_path}...")
try:
with open(file_path, 'r', encoding='utf-8') as f:
self.logs = f.readlines()
print(f"✅ {len(self.logs)} خط لاگ بارگذاری شد")
except Exception as e:
print(f"❌ خطا در بارگذاری: {e}")
def analyze(self) -> Dict:
"""تحلیل لاگها"""
print(" شروع تحلیل لاگها...")
for i, log in enumerate(self.logs):
log_lower = log.lower()
for attack_type, patterns in self.ATTACK_PATTERNS.items():
for pattern in patterns:
if re.search(pattern, log_lower, re.IGNORECASE):
self.detected_attacks.append({
'line_number': i + 1,
'attack_type': attack_type,
'log_snippet': log[:100] + "...",
'pattern': pattern,
'timestamp': self.extract_timestamp(log)
})
self.stats[attack_type] += 1
break # فقط یک حمله در هر خط
return self.generate_report()
def extract_timestamp(self, log: str) -> str:
"""استخراج تایماستمپ از لاگ"""
# الگوهای مختلف تایماستمپ
patterns = [
r'\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}',
r'\d{2}/\d{2}/\d{4} \d{2}:\d{2}:\d{2}',
r'\[(.*?)\]' # تایماستمپ داخل براکت
]
for pattern in patterns:
match = re.search(pattern, log)
if match:
return match.group()
return "unknown"
def generate_report(self) -> Dict:
"""تولید گزارش"""
print(" تولید گزارش...")
# آمار کلی
total_attacks = sum(self.stats.values())
most_common = self.stats.most_common(3)
# توزیع زمانی
time_distribution = self.analyze_time_distribution()
# IPهای مشکوک
suspicious_ips = self.extract_suspicious_ips()
report = {
'summary': {
'total_logs': len(self.logs),
'total_attacks': total_attacks,
'attack_rate': f"{(total_attacks/len(self.logs)*100):.1f}%" if self.logs else "0%",
'most_common_attacks': most_common,
'analysis_time': datetime.now().isoformat()
},
'details': {
'attack_types': dict(self.stats),
'time_distribution': time_distribution,
'suspicious_ips': suspicious_ips,
'sample_attacks': self.detected_attacks[:10] # ۱۰ نمونه اول
},
'recommendations': self.generate_recommendations()
}
# ذخیره گزارش
self.save_report(report)
return report
def analyze_time_distribution(self) -> Dict:
"""تحلیل توزیع زمانی حملات"""
distribution = Counter()
for attack in self.detected_attacks:
timestamp = attack['timestamp']
if timestamp != "unknown":
# استخراج ساعت
hour_match = re.search(r'(\d{2}):\d{2}:\d{2}', timestamp)
if hour_match:
hour = hour_match.group(1)
distribution[hour] += 1
return dict(distribution)
def extract_suspicious_ips(self) -> List[str]:
"""استخراج IPهای مشکوک"""
ip_pattern = r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}'
ips = []
for attack in self.detected_attacks:
ip_matches = re.findall(ip_pattern, attack['log_snippet'])
ips.extend(ip_matches)
# فقط IPهای تکراری (مشکوک)
ip_counts = Counter(ips)
suspicious = [ip for ip, count in ip_counts.items() if count > 5]
return suspicious[:10] # ۱۰ IP اول
def generate_recommendations(self) -> List[str]:
"""تولید توصیههای امنیتی"""
recommendations = []
if self.stats['SQL Injection'] > 0:
recommendations.append(
"فیلتر کردن ورودیهای کاربر در برنامههای وب"
)
if self.stats['Brute Force'] > 0:
recommendations.extend([
"فعالسازی محدودیت تعداد لاگین ناموفق",
"استفاده از احراز هویت دو عاملی"
])
if self.stats['Port Scan'] > 0:
recommendations.append(
"پیکربندی فایروال برای محدود کردن اسکن پورت"
)
if self.stats['DoS Attack'] > 0:
recommendations.extend([
"استفاده از سرویسهای DDoS Protection",
"پیادهسازی Rate Limiting"
])
if not recommendations:
recommendations.append("وضعیت امنیتی مطلوب است")
return recommendations
def save_report(self, report: Dict):
"""ذخیره گزارش"""
filename = f"security_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
import json
with open(filename, 'w', encoding='utf-8') as f:
json.dump(report, f, indent=2, ensure_ascii=False)
print(f"✅ گزارش در {filename} ذخیره شد")
# ذخیره به صورت CSV برای تحلیل بیشتر
self.save_to_csv(report)
def save_to_csv(self, report: Dict):
"""ذخیره به CSV"""
df_data = []
for attack in self.detected_attacks:
df_data.append({
'line_number': attack['line_number'],
'attack_type': attack['attack_type'],
'pattern': attack['pattern'],
'timestamp': attack['timestamp']
})
if df_data:
df = pd.DataFrame(df_data)
csv_filename = f"attacks_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
df.to_csv(csv_filename, index=False, encoding='utf-8-sig')
print(f" دادهها در {csv_filename} ذخیره شد")
# استفاده
def main():
analyzer = LogAnalyzer()
# بارگذاری لاگهای نمونه
analyzer.load_logs("sample_logs.txt")
# تحلیل
report = analyzer.analyze()
# نمایش نتایج
print("\n" + "="*50)
print("گزارش امنیتی")
print("="*50)
summary = report['summary']
print(f"\n آمار کلی:")
print(f" کل لاگها: {summary['total_logs']}")
print(f" کل حملات: {summary['total_attacks']}")
print(f" نرخ حمله: {summary['attack_rate']}")
print(f"\n شایعترین حملات:")
for attack, count in summary['most_common_attacks']:
print(f" {attack}: {count}")
print(f"\n️ توصیههای امنیتی:")
for i, rec in enumerate(report['recommendations'], 1):
print(f" {i}. {rec}")
# if __name__ == "__main__":
# main()
Buy on AliExpress.com
