45 أسطر
1.7 KiB
Python
45 أسطر
1.7 KiB
Python
import json
|
|
import re
|
|
from datetime import datetime
|
|
|
|
# الأنماط الخبيثة التي نبحث عنها في السجلات
|
|
SUSPICIOUS_PATTERNS = {
|
|
"Brute Force Attempt": re.compile(r"Failed password for .* from (?P<ip>\d+\.\d+\.\d+\.\d+)"),
|
|
"SQL Injection": re.compile(r"UNION SELECT|OR 1=1", re.IGNORECASE),
|
|
"Port Scanning": re.compile(r"BLOCK .* SRC=(?P<ip>\d+\.\d+\.\d+\.\d+)")
|
|
}
|
|
|
|
def analyze_logs():
|
|
alerts = []
|
|
|
|
# محاكاة لبيانات قادمة من 3 مصادر (Authentication, Web Server, Firewall)
|
|
mock_logs = [
|
|
"[AUTH] Failed password for admin from 192.168.1.50",
|
|
"[AUTH] Failed password for root from 192.168.1.50",
|
|
"[WEB] GET /api/data?id=1' OR 1=1-- HTTP/1.1",
|
|
"[FIREWALL] BLOCK TCP SRC=10.0.0.9 DST=192.168.1.10 DPT=22"
|
|
]
|
|
|
|
print("🔍 بدء تحليل السجلات...")
|
|
|
|
for line in mock_logs:
|
|
for attack_type, pattern in SUSPICIOUS_PATTERNS.items():
|
|
match = pattern.search(line)
|
|
if match:
|
|
ip = match.groupdict().get("ip", "Unknown") if "ip" in pattern.groupindex else "Unknown"
|
|
alerts.append({
|
|
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
|
"type": attack_type,
|
|
"source_ip": ip,
|
|
"raw_log": line
|
|
})
|
|
|
|
# حفظ التنبيهات في ملف JSON لتقرأه لوحة التحكم
|
|
with open("alerts.json", "w", encoding="utf-8") as f:
|
|
json.dump(alerts, f, indent=4, ensure_ascii=False)
|
|
|
|
print(f"✅ تم الانتهاء من التحليل. تم العثور على {len(alerts)} تهديدات محتملة.")
|
|
|
|
if __name__ == "__main__":
|
|
analyze_logs()
|