122 أسطر
4.6 KiB
Python
122 أسطر
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Ghaymah Lightweight SIEM Engine
|
|
Author: AbdulRhman Ewais
|
|
Description: Consolidated Log Analyzer for Auth, Nginx, and API services.
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
import json
|
|
from collections import defaultdict
|
|
|
|
# مسارات السجلات الافتراضية المربوطة بالـ Volume
|
|
LOG_SOURCES = {
|
|
"auth": "/var/log/siem_logs/auth.log",
|
|
"web": "/var/log/siem_logs/nginx_access.log",
|
|
"api": "/var/log/siem_logs/ghaymah_api.log"
|
|
}
|
|
|
|
# أنماط الكشف (Detection Signatures)
|
|
BRUTE_FORCE_THRESHOLD = 5
|
|
XSS_PATTERN = re.compile(r'(<script>|javascript:|onerror=|onload=)', re.IGNORECASE)
|
|
SQLI_PATTERN = re.compile(r'(UNION\s+SELECT|SELECT\s+.*\s+FROM|OR\s+1\s*=\s*1|--|/\*|\*/)', re.IGNORECASE)
|
|
|
|
def parse_auth_logs(filepath):
|
|
"""تحليل سجلات المصادقة لكشف هجمات الـ SSH Brute Force"""
|
|
alerts = []
|
|
failed_attempts = defaultdict(int)
|
|
|
|
if not os.path.exists(filepath):
|
|
return alerts
|
|
|
|
# النمط: Failed password for invalid user admin from 192.168.1.50 port 54322 ssh2
|
|
pattern = re.compile(r'Failed password for .* from (?P<ip>\d+\.\d+\.\d+\.\d+)')
|
|
|
|
with open(filepath, 'r', encoding='utf-8') as f:
|
|
for line in f:
|
|
match = pattern.search(line)
|
|
if match:
|
|
ip = match.group('ip')
|
|
failed_attempts[ip] += 1
|
|
if failed_attempts[ip] == BRUTE_FORCE_THRESHOLD:
|
|
alerts.append({
|
|
"source": "Auth Logs (SSH)",
|
|
"ip": ip,
|
|
"type": "SSH Brute Force Attempt",
|
|
"severity": "High",
|
|
"details": f"Detected {failed_attempts[ip]} failed SSH logins."
|
|
})
|
|
return alerts
|
|
|
|
def parse_web_logs(filepath):
|
|
"""تحليل سجلات Nginx لكشف محاولات الاختراق (SQLi / XSS)"""
|
|
alerts = []
|
|
if not os.path.exists(filepath):
|
|
return alerts
|
|
|
|
# النمط القياسي لـ Nginx Log
|
|
pattern = re.compile(r'(?P<ip>\d+\.\d+\.\d+\.\d+) - - \[.*\] "(?P<method>\w+) (?P<path>\S+) HTTP/.*" (?P<status>\d{3})')
|
|
|
|
with open(filepath, 'r', encoding='utf-8') as f:
|
|
for line in f:
|
|
match = pattern.search(line)
|
|
if match:
|
|
data = match.groupdict()
|
|
path_decoded = data['path']
|
|
|
|
if SQLI_PATTERN.search(path_decoded):
|
|
alerts.append({
|
|
"source": "Web Logs (Nginx)",
|
|
"ip": data['ip'],
|
|
"type": "SQL Injection Attempt",
|
|
"severity": "Critical",
|
|
"details": f"Suspicious pattern in URL: {path_decoded}"
|
|
})
|
|
elif XSS_PATTERN.search(path_decoded):
|
|
alerts.append({
|
|
"source": "Web Logs (Nginx)",
|
|
"ip": data['ip'],
|
|
"type": "Cross-Site Scripting (XSS)",
|
|
"severity": "High",
|
|
"details": f"Payload detected in path: {path_decoded}"
|
|
})
|
|
return alerts
|
|
|
|
def parse_api_logs(filepath):
|
|
"""تحليل سجلات تطبيق غيمة لكشف إساءة استخدام الـ Token أو الـ Rate Limit"""
|
|
alerts = []
|
|
if not os.path.exists(filepath):
|
|
return alerts
|
|
|
|
with open(filepath, 'r', encoding='utf-8') as f:
|
|
for line in f:
|
|
try:
|
|
log_data = json.loads(line)
|
|
if log_data.get("status") == 429: # Too Many Requests
|
|
alerts.append({
|
|
"source": "Ghaymah API",
|
|
"ip": log_data.get("client_ip", "Unknown"),
|
|
"type": "API Rate Limit Exceeded",
|
|
"severity": "Medium",
|
|
"details": f"Client hammered Endpoint: {log_data.get('endpoint')}"
|
|
})
|
|
except json.JSONDecodeError:
|
|
continue
|
|
return alerts
|
|
|
|
def main():
|
|
all_alerts = []
|
|
all_alerts.extend(parse_auth_logs(LOG_SOURCES["auth"]))
|
|
all_alerts.extend(parse_web_logs(LOG_SOURCES["web"]))
|
|
all_alerts.extend(parse_api_logs(LOG_SOURCES["api"]))
|
|
|
|
# تصدير النتائج لملف JSON لتقرأه الـ Dashboard
|
|
output_path = "/var/log/siem_logs/siem_alerts.json"
|
|
with open(output_path, 'w', encoding='utf-8') as out_f:
|
|
json.dump(all_alerts, out_f, indent=4)
|
|
|
|
print(f"[+] SIEM Analysis complete. {len(all_alerts)} alerts written to {output_path}")
|
|
|
|
if __name__ == "__main__":
|
|
main() |