45 أسطر
1.8 KiB
Python
45 أسطر
1.8 KiB
Python
import json
|
|
|
|
logs_data = [
|
|
'{"source": "Nginx", "timestamp": "2026-07-27T01:00:01", "ip": "192.168.1.10", "status": 200, "message": "GET /index.html"}',
|
|
'{"source": "Auth", "timestamp": "2026-07-27T01:00:05", "ip": "45.33.32.156", "status": 401, "message": "Failed login for admin"}',
|
|
'{"source": "Auth", "timestamp": "2026-07-27T01:00:06", "ip": "45.33.32.156", "status": 401, "message": "Failed login for admin"}',
|
|
'{"source": "Auth", "timestamp": "2026-07-27T01:00:07", "ip": "45.33.32.156", "status": 401, "message": "Failed login for admin"}',
|
|
'{"source": "WAF", "timestamp": "2026-07-27T01:00:10", "ip": "103.15.28.1", "status": 403, "message": "SQL Injection pattern detected: UNION SELECT"}'
|
|
]
|
|
|
|
failed_attempts = {}
|
|
alerts = []
|
|
|
|
def analyze_logs():
|
|
for entry in logs_data:
|
|
log = json.loads(entry)
|
|
ip = log.get("ip")
|
|
msg = log.get("message", "")
|
|
status = log.get("status")
|
|
|
|
if status == 401:
|
|
failed_attempts[ip] = failed_attempts.get(ip, 0) + 1
|
|
if failed_attempts[ip] >= 3:
|
|
alerts.append({
|
|
"severity": "HIGH",
|
|
"type": "Brute Force Detected",
|
|
"ip": ip,
|
|
"details": f"Multiple failed logins ({failed_attempts[ip]} times)"
|
|
})
|
|
|
|
if "SQL" in msg or "UNION SELECT" in msg:
|
|
alerts.append({
|
|
"severity": "CRITICAL",
|
|
"type": "Web Attack (SQLi)",
|
|
"ip": ip,
|
|
"details": msg
|
|
})
|
|
|
|
with open("q4-siem/alerts.json", "w", encoding="utf-8") as f:
|
|
json.dump(alerts, f, indent=2)
|
|
print("[✓] SIEM Analysis Complete. Alerts generated in q4-siem/alerts.json")
|
|
|
|
if __name__ == "__main__":
|
|
analyze_logs()
|