BGP Abuse Added

Author: prince-7

securetea/__init__.py

@@ -28,6 +28,7 @@
 from .lib.ids.r2l_rules import r2l_engine
 from .lib.ids.r2l_rules import syn_flood
 from .lib.ids.r2l_rules import dns_amp
+from .lib.ids.r2l_rules import bgp_abuse
 from .lib.ids.r2l_rules.wireless import deauth
 from .lib.ids.r2l_rules.wireless import fake_access
 from .lib.ids.r2l_rules.wireless import hidden_node

securetea/lib/ids/__init__.py

@@ -10,6 +10,7 @@
 from .r2l_rules import r2l_engine
 from .r2l_rules import syn_flood
 from .r2l_rules import dns_amp
+from .r2l_rules import bgp_abuse
 from .r2l_rules.wireless import deauth
 from .r2l_rules.wireless import fake_access
 from .r2l_rules.wireless import hidden_node

securetea/lib/ids/r2l_rules/__init__.py

@@ -8,6 +8,7 @@
 from . import r2l_engine
 from . import syn_flood
 from . import dns_amp
+from . import bgp_abuse
 from .wireless import deauth
 from .wireless import fake_access
 from .wireless import hidden_node

securetea/lib/ids/r2l_rules/bgp_abuse.py

@@ -0,0 +1,65 @@
+# -*- coding: utf-8
+u"""BGP Abuse Detection detection module for SecureTea IDS.
+
+Project:
+    ╔═╗┌─┐┌─┐┬ ┬┬─┐┌─┐╔╦╗┌─┐┌─┐
+    ╚═╗├┤ │  │ │├┬┘├┤  ║ ├┤ ├─┤
+    ╚═╝└─┘└─┘└─┘┴└─└─┘ ╩ └─┘┴ ┴
+    Author: Aman Singh <dun930n.m45732@gmail.com> , June 16 2021
+    Version: 1.1
+    Module: SecureTea
+
+"""
+
+import scapy.all as scapy
+import scapy.contrib.bgp as bgp
+from securetea import logger
+
+class BGP_Abuse(object):
+    """BGP Abuse class."""
+
+    def __init__(self, debug=False):
+        """
+        Initialize BGP Abuse class.
+
+        Args:
+            debug (bool): Log on terminal or not
+
+        Raises:
+            None
+
+        Returns:
+            None
+        """
+        # Initialize logger
+        self.logger = logger.SecureTeaLogger(
+                __name__,
+                debug=debug
+        )
+
+    def detect_bgp_abuse(self, pkt):
+        """
+        Detect BGP Abuse Attacks by observing set flags and BGPPathAttributes
+
+        Types of attack detected:-
+        1) Blind Disruption
+
+        Args:
+            pkt (scapy_object): Packet to dissect and observe
+
+        Raises:
+            None
+
+        Returns:
+            None
+        """
+
+        # Blind Disruption Detection
+        if (pkt.haslayer(scapy.IP) 
+            and pkt.haslayer(scapy.TCP)):
+            if('RA' in str(pkt[scapy.TCP].flags)):
+                self.logger.log(
+                            "Possible BGP Abuse,Blind Disruption attack detected.",
+                            logtype="warning"
+                        )
+

securetea/lib/ids/r2l_rules/r2l_engine.py

@@ -19,6 +19,7 @@
 from securetea.lib.ids.r2l_rules.syn_flood import SynFlood
 from securetea.lib.ids.r2l_rules.land_attack import LandAttack
 from securetea.lib.ids.r2l_rules.dns_amp import DNS_Amplification
+from securetea.lib.ids.r2l_rules.bgp_abuse import BGP_Abuse
 from securetea.lib.ids.r2l_rules.wireless.deauth import Deauth
 from securetea.lib.ids.r2l_rules.wireless.fake_access import FakeAccessPoint
 from securetea.lib.ids.r2l_rules.wireless.hidden_node import HiddenNode
@@ -51,6 +52,7 @@ def __init__(self, debug=False, interface=None):
         self.ddos = DDoS(debug=debug)
         self.syn_flood = SynFlood(debug=debug)
         self.dns_amp = DNS_Amplification(debug=debug)
+        self.bgp_abuse = BGP_Abuse(debug=debug)
         # Wireless
         self.deauth = Deauth(debug=debug)
         self.fake_access = FakeAccessPoint(debug=debug)
@@ -80,6 +82,7 @@ def run(self, pkt):
         self.ddos.classify_ddos(pkt)
         self.syn_flood.detect_syn_flood(pkt)
         self.dns_amp.detect_dns_amplification(pkt)
+        self.bgp_abuse.detect_bgp_abuse(pkt)
         # Wireless
         self.deauth.detect_deauth(pkt)
         self.fake_access.detect_fake_ap(pkt)

test/test_bgp_abuse.py

@@ -0,0 +1,48 @@
+# -*- coding: utf-8 -*-
+import unittest
+from securetea.lib.ids.r2l_rules.bgp_abuse import BGP_Abuse
+import scapy.all as scapy
+from securetea.logger import SecureTeaLogger
+
+try:
+    # if python 3.x.x
+    from unittest.mock import patch
+except ImportError:  # python 2.x.x
+    from mock import patch
+
+
+class TestBGP_Abuse(unittest.TestCase):
+    """
+    Test class for SecureTea IDS BGP_Abuse Detection.
+    """
+
+    def setUp(self):
+        """
+        Setup class for BGP_Abuse.
+        """
+        # Create scapy packet (valid attack)
+        self.pkt = scapy.IP(src="10.0.2.15",
+                            dst="200.10.10.1") \
+                   / scapy.TCP(dport=53, sport=179, flags="RA", seq=123, ack=456)
+
+        # Create a scapy packet (invalid attack)
+        self.pkt2 = scapy.IP(src="10.0.2.15",
+                            dst="200.10.10.1") \
+                   / scapy.TCP(dport=53, sport=179, seq=123, ack=456)
+
+        # Create BGP Abuse object
+        self.bgp_abuse_obj = BGP_Abuse()
+
+    @patch.object(SecureTeaLogger, 'log')
+    def test_detect_bgp_abuse(self, mock_log):
+        """
+        Test detect_bgp_abuse.
+        """
+        # Case 1: When condition for bgp abuse is invalid
+        self.bgp_abuse_obj.detect_bgp_abuse(self.pkt2)
+        self.assertFalse(mock_log.called)
+
+        # Case 2: When condition for bgp abuse is valid
+        self.bgp_abuse_obj.detect_bgp_abuse(self.pkt)
+        mock_log.assert_called_with("Possible BGP Abuse,Blind Disruption attack detected.",
+                                    logtype="warning")

test/test_dns_amp.py

@@ -32,7 +32,7 @@ def setUp(self):
                     / scapy.UDP(dport=53) \
                     / scapy.DNS(rd=1, qd=scapy.DNSQR(qname="google.com", qtype="ANY"))
 
-        # Create LandAttack object
+        # Create DNS Amplification object
         self.dns_amp_obj = DNS_Amplification()
 
     @patch.object(SecureTeaLogger, 'log')