IDS/IPS - أنظمة كشف ومنع التسلل — أمن الشبكات

IDS/IPS - كشف ومنع التسلل

IDS - Intrusion Detection System

نظام يكتشف الهجمات ويُنبّه.

IPS - Intrusion Prevention System

نظام يكتشف ويمنع الهجمات تلقائياً.

أنواع IDS/IPS

Network-based (NIDS/NIPS)

- يراقب حركة الشبكة

- يُوضع على نقاط استراتيجية

- أمثلة: Snort, Suricata

Host-based (HIDS/HIPS)

- يراقب جهاز واحد

- يفحص الملفات والـ logs

- أمثلة: OSSEC, Wazuh

طرق الكشف

Signature-based

- يبحث عن أنماط هجمات معروفة

- سريع ودقيق ✅

- لا يكتشف الجديد ❌

Anomaly-based

- يتعلم السلوك الطبيعي

- يكتشف الانحرافات

- يكتشف الجديد ✅

- False positives أكثر ❌

Behavior-based

- يحلل سلوك البرامج

- يكتشف الأنماط المشبوهة

أمثلة على القواعد (Snort)

# كشف محاولة SQL Injection
alert tcp any any -> any 80 (content:"' OR '1'='1"; msg:"SQL Injection Attempt";)

# كشف فحص المنافذ
alert tcp any any -> any any (flags:S; threshold:type both, track by_src, count 20, seconds 10; msg:"Port Scan Detected";)

Placement

أين نضع IDS/IPS؟

- بعد الـ Firewall

- قبل الخوادم المهمة

- على نقاط الدخول

# محاكاة IDS بسيط
import re
from datetime import datetime
from collections import defaultdict

class SimpleIDS:
    """
    نظام كشف تسلل بسيط للتعليم
    """
    
    def __init__(self):
        self.alerts = []
        self.connection_count = defaultdict(int)
        self.failed_logins = defaultdict(int)
        
        # قواعد Signature-based
        self.signatures = {
            "sql_injection": [
                r"'\s*OR\s*'1'\s*=\s*'1",
                r"UNION\s+SELECT",
                r"DROP\s+TABLE",
                r"--\s*$",
            ],
            "xss": [
                r"<script[^>]*>",
                r"javascript:",
                r"onerror\s*=",
                r"onload\s*=",
            ],
            "command_injection": [
                r";\s*cat\s+/etc/passwd",
                r"\|\s*nc\s+",
                r";\s*rm\s+-rf",
                r"$\(.*\)",
            ],
            "path_traversal": [
                r"\.\./\.\./",
                r"etc/passwd",
                r"windows/system32",
            ],
        }
        
        # عتبات Anomaly detection
        self.thresholds = {
            "connections_per_minute": 100,
            "failed_logins": 5,
            "request_size": 10000,
        }
    
    def log_alert(self, severity, alert_type, description, source_ip, details=None):
        """تسجيل تنبيه"""
        alert = {
            "timestamp": datetime.now().isoformat(),
            "severity": severity,
            "type": alert_type,
            "description": description,
            "source_ip": source_ip,
            "details": details or {}
        }
        self.alerts.append(alert)
        
        emoji = {"HIGH": "🔴", "MEDIUM": "🟡", "LOW": "🟢"}
        print(f"[{emoji.get(severity, '⚪')} {severity}] {alert_type}: {description}")
        print(f"    Source: {source_ip}")
    
    def check_signatures(self, request_data, source_ip):
        """فحص Signature-based"""
        for attack_type, patterns in self.signatures.items():
            for pattern in patterns:
                if re.search(pattern, request_data, re.IGNORECASE):
                    self.log_alert(
                        severity="HIGH",
                        alert_type=attack_type.upper(),
                        description=f"تم اكتشاف {attack_type}",
                        source_ip=source_ip,
                        details={"pattern": pattern, "data": request_data[:100]}
                    )
                    return True
        return False
    
    def check_anomalies(self, source_ip, request_size=0, is_failed_login=False):
        """فحص Anomaly-based"""
        alerts = []
        
        # فحص عدد الاتصالات
        self.connection_count[source_ip] += 1
        if self.connection_count[source_ip] > self.thresholds["connections_per_minute"]:
            self.log_alert(
                severity="MEDIUM",
                alert_type="RATE_LIMIT",
                description=f"عدد اتصالات غير طبيعي: {self.connection_count[source_ip]}",
                source_ip=source_ip
            )
            alerts.append("rate_limit")
        
        # فحص محاولات تسجيل الدخول الفاشلة
        if is_failed_login:
            self.failed_logins[source_ip] += 1
            if self.failed_logins[source_ip] >= self.thresholds["failed_logins"]:
                self.log_alert(
                    severity="HIGH",
                    alert_type="BRUTE_FORCE",
                    description=f"محاولات تسجيل دخول فاشلة: {self.failed_logins[source_ip]}",
                    source_ip=source_ip
                )
                alerts.append("brute_force")
        
        # فحص حجم الطلب
        if request_size > self.thresholds["request_size"]:
            self.log_alert(
                severity="MEDIUM",
                alert_type="LARGE_REQUEST",
                description=f"طلب كبير جداً: {request_size} bytes",
                source_ip=source_ip
            )
            alerts.append("large_request")
        
        return alerts
    
    def analyze_request(self, request, source_ip):
        """تحليل طلب HTTP"""
        print(f"\n📥 تحليل طلب من {source_ip}...")
        
        # فحص Signatures
        if self.check_signatures(request, source_ip):
            return {"blocked": True, "reason": "signature_match"}
        
        # فحص Anomalies
        anomalies = self.check_anomalies(source_ip, len(request))
        if anomalies:
            return {"blocked": False, "warnings": anomalies}
        
        print("   ✅ الطلب يبدو آمناً")
        return {"blocked": False}
    
    def get_summary(self):
        """ملخص التنبيهات"""
        print("\n" + "=" * 50)
        print("📊 ملخص التنبيهات")
        print("=" * 50)
        
        severity_count = {"HIGH": 0, "MEDIUM": 0, "LOW": 0}
        for alert in self.alerts:
            severity_count[alert["severity"]] += 1
        
        print(f"  🔴 عالية: {severity_count['HIGH']}")
        print(f"  🟡 متوسطة: {severity_count['MEDIUM']}")
        print(f"  🟢 منخفضة: {severity_count['LOW']}")
        print(f"  📋 الإجمالي: {len(self.alerts)}")


# ===== اختبار =====
ids = SimpleIDS()

# طلبات للاختبار
test_requests = [
    ("192.168.1.100", "GET /index.html HTTP/1.1"),
    ("10.0.0.50", "GET /search?q=' OR '1'='1 HTTP/1.1"),
    ("172.16.0.25", "POST /login HTTP/1.1\n<script>alert('xss')</script>"),
    ("192.168.1.100", "GET /files?path=../../../../etc/passwd HTTP/1.1"),
    ("10.0.0.50", "GET /api/data HTTP/1.1"),
]

for ip, request in test_requests:
    ids.analyze_request(request, ip)

# محاكاة Brute Force
print("\n📥 محاكاة Brute Force Attack...")
for i in range(6):
    ids.check_anomalies("attacker_ip", is_failed_login=True)

ids.get_summary()