131 أسطر
4.9 KiB
Python
131 أسطر
4.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
generate_sample_logs.py
|
|
------------------------
|
|
يولد سجلات تجريبية واقعية (Web / Auth / Firewall) تحتوي على سلوك طبيعي
|
|
+ هجمات مصطنعة، لاختبار محرك التحليل siem_analyzer.py قبل تسليم الامتحان.
|
|
|
|
الاستخدام:
|
|
python3 generate_sample_logs.py
|
|
سينشئ الملفات داخل مجلد logs/:
|
|
logs/access.log (Nginx)
|
|
logs/auth.log (Linux SSH)
|
|
logs/firewall.log (iptables)
|
|
"""
|
|
|
|
import random
|
|
from datetime import datetime, timedelta
|
|
|
|
OUT_DIR = "logs"
|
|
|
|
|
|
def rand_ip(bad=False):
|
|
if bad:
|
|
# نفس مجموعة IPs الخبيثة تتكرر لتصبح واضحة الأنماط
|
|
return random.choice(["203.0.113.45", "198.51.100.23", "45.155.205.7"])
|
|
return f"192.168.1.{random.randint(2, 250)}"
|
|
|
|
|
|
def gen_access_log(out_path, lines=400):
|
|
normal_paths = ["/", "/index.html", "/products", "/login", "/api/users", "/about"]
|
|
sqli_payloads = [
|
|
"/login?id=1' UNION SELECT username,password FROM users--",
|
|
"/products?id=1 OR 1=1",
|
|
"/search?q=SELECT * FROM accounts",
|
|
]
|
|
traversal_payloads = [
|
|
"/download?file=../../../../etc/passwd",
|
|
"/view?page=..%2f..%2f..%2fetc%2fpasswd",
|
|
]
|
|
now = datetime.now()
|
|
rows = []
|
|
for i in range(lines):
|
|
ts = now - timedelta(seconds=random.randint(0, 3600))
|
|
ip = rand_ip()
|
|
req_path = random.choice(normal_paths)
|
|
status = random.choice([200, 200, 200, 304, 404])
|
|
rows.append(f'{ip} - - [{ts.strftime("%d/%b/%Y:%H:%M:%S +0000")}] '
|
|
f'"GET {req_path} HTTP/1.1" {status} {random.randint(200,5000)}')
|
|
|
|
# هجوم SQL Injection من IP واحد
|
|
attacker = "203.0.113.45"
|
|
for payload in sqli_payloads * 3:
|
|
ts = now - timedelta(seconds=random.randint(0, 600))
|
|
rows.append(f'{attacker} - - [{ts.strftime("%d/%b/%Y:%H:%M:%S +0000")}] '
|
|
f'"GET {payload} HTTP/1.1" 500 512')
|
|
|
|
# هجوم Directory Traversal
|
|
attacker2 = "198.51.100.23"
|
|
for payload in traversal_payloads * 3:
|
|
ts = now - timedelta(seconds=random.randint(0, 600))
|
|
rows.append(f'{attacker2} - - [{ts.strftime("%d/%b/%Y:%H:%M:%S +0000")}] '
|
|
f'"GET {payload} HTTP/1.1" 403 300')
|
|
|
|
# هجوم Scanning/Fuzzing: نفس IP يولد عشرات 404
|
|
scanner = "45.155.205.7"
|
|
for _ in range(40):
|
|
ts = now - timedelta(seconds=random.randint(0, 600))
|
|
rows.append(f'{scanner} - - [{ts.strftime("%d/%b/%Y:%H:%M:%S +0000")}] '
|
|
f'"GET /wp-admin/{random.randint(1,999)}.php HTTP/1.1" 404 200')
|
|
|
|
random.shuffle(rows)
|
|
with open(out_path, "w") as f:
|
|
f.write("\n".join(rows) + "\n")
|
|
|
|
|
|
def gen_auth_log(out_path):
|
|
now = datetime.now()
|
|
rows = []
|
|
# محاولات دخول عادية ناجحة
|
|
for _ in range(20):
|
|
ts = now - timedelta(seconds=random.randint(0, 3600))
|
|
ip = rand_ip()
|
|
rows.append(f'{ts.strftime("%b %d %H:%M:%S")} server sshd[1234]: '
|
|
f'Accepted password for admin from {ip} port 55000 ssh2')
|
|
|
|
# هجوم Brute Force: نفس IP، فشل متكرر خلال دقيقة ثم نجاح
|
|
attacker = "198.51.100.99"
|
|
base = now - timedelta(minutes=2)
|
|
for i in range(10):
|
|
ts = base + timedelta(seconds=i * 5)
|
|
rows.append(f'{ts.strftime("%b %d %H:%M:%S")} server sshd[999{i}]: '
|
|
f'Failed password for root from {attacker} port 4444{i} ssh2')
|
|
rows.append(f'{(base + timedelta(seconds=60)).strftime("%b %d %H:%M:%S")} server sshd[9999]: '
|
|
f'Accepted password for root from {attacker} port 44450 ssh2')
|
|
|
|
random.shuffle(rows)
|
|
with open(out_path, "w") as f:
|
|
f.write("\n".join(rows) + "\n")
|
|
|
|
|
|
def gen_firewall_log(out_path):
|
|
now = datetime.now()
|
|
rows = []
|
|
# حركة طبيعية مسموحة
|
|
for _ in range(60):
|
|
ts = now - timedelta(seconds=random.randint(0, 3600))
|
|
ip = rand_ip()
|
|
port = random.choice([80, 443, 22])
|
|
rows.append(f'{ts.strftime("%b %d %H:%M:%S")} kernel: IN=eth0 OUT= '
|
|
f'SRC={ip} DST=10.0.0.5 PROTO=TCP DPT={port} ACTION=ACCEPT')
|
|
|
|
# هجوم Port Scan: IP واحد يحاول عشرات المنافذ المغلقة
|
|
scanner = "45.155.205.7"
|
|
for port in range(1000, 1035):
|
|
ts = now - timedelta(seconds=random.randint(0, 300))
|
|
rows.append(f'{ts.strftime("%b %d %H:%M:%S")} kernel: IN=eth0 OUT= '
|
|
f'SRC={scanner} DST=10.0.0.5 PROTO=TCP DPT={port} ACTION=DROP')
|
|
|
|
random.shuffle(rows)
|
|
with open(out_path, "w") as f:
|
|
f.write("\n".join(rows) + "\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import os
|
|
os.makedirs(OUT_DIR, exist_ok=True)
|
|
gen_access_log(f"{OUT_DIR}/access.log")
|
|
gen_auth_log(f"{OUT_DIR}/auth.log")
|
|
gen_firewall_log(f"{OUT_DIR}/firewall.log")
|
|
print("تم إنشاء سجلات تجريبية داخل مجلد logs/")
|