rjmcmahon
/
fiwi_monitor
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: rjmcmahon:d3a4937d0a43c68618423ab34084d5a3675ce9ac
Branches Tags
rjmcmahon:main
..
compare: rjmcmahon:5132a667e9bc08526c1b494f1b41f5392b8693fb
Branches Tags
rjmcmahon:main

No commits in common. "d3a4937d0a43c68618423ab34084d5a3675ce9ac" and "5132a667e9bc08526c1b494f1b41f5392b8693fb" have entirely different histories.
d3a4937d0a ... 5132a667e9

2 changed files with 222 additions and 586 deletions
Show Stats Expand all files Collapse all files

586
test_monitor.py
View File

@ -1,586 +0,0 @@
#!/usr/bin/env python3
"""
Monitor mode WiFi packet capture and analysis using scapy.
Replaces the bash script with a pure Python solution.
Usage:
# Live capture from interface:
sudo python3 test_monitor.py [interface] [channel] [duration_seconds] [--keep-pcap] [--full-packet]
sudo python3 test_monitor.py wlan0 36 10
sudo python3 test_monitor.py wlan0 36 10 --keep-pcap
sudo python3 test_monitor.py wlan0 36 10 --full-packet # Capture entire packets (default: header only)
# Read from pcap file:
python3 test_monitor.py <pcap_file>
python3 test_monitor.py /tmp/capture.pcap
"""
import sys
import os
import subprocess
import signal
import time
import tempfile
from datetime import datetime
from collections import Counter, defaultdict
from scapy.all import *
from scapy.layers.dot11 import Dot11, Dot11QoS, Dot11Beacon, Dot11ProbeReq, Dot11ProbeResp
from scapy.layers.dot11 import Dot11AssoReq, Dot11AssoResp, Dot11Auth, Dot11Deauth, Dot11Disas
# Configuration
WIFI_INTERFACE = os.environ.get("WIFI_INTERFACE", "wlan0")
KEEP_PCAP = False
FULL_PACKET = False
# Default snaplen: 256 bytes (enough for 802.11 header ~30-40 bytes + some payload)
# 802.11 header can be up to ~40 bytes, plus radiotap header ~20-30 bytes
# 256 bytes gives us header + ~200 bytes of payload, which is usually enough for analysis
DEFAULT_SNAPLEN = 256
def parse_args():
"""Parse command line arguments.
Returns:
tuple: (mode, interface_or_pcap, channel, duration)
mode: 'live' or 'pcap'
interface_or_pcap: interface name (for live) or pcap file path (for pcap)
channel: channel number (for live mode only)
duration: capture duration in seconds (for live mode only)
"""
global WIFI_INTERFACE, KEEP_PCAP, FULL_PACKET
args = sys.argv[1:]
if len(args) == 0:
print("Usage:")
print(" Live capture: sudo python3 test_monitor.py [interface] [channel] [duration] [--keep-pcap] [--full-packet]")
print(" Read pcap: python3 test_monitor.py <pcap_file>")
sys.exit(1)
# Check for --keep-pcap flag
if "--keep-pcap" in args or "-k" in args:
KEEP_PCAP = True
args = [a for a in args if a not in ["--keep-pcap", "-k"]]
# Check for --full-packet flag
if "--full-packet" in args or "-f" in args:
FULL_PACKET = True
args = [a for a in args if a not in ["--full-packet", "-f"]]
# Check if first argument is a pcap file (ends with .pcap or .cap)
if len(args) > 0 and (args[0].endswith('.pcap') or args[0].endswith('.cap')):
pcap_file = args[0]
if not os.path.isfile(pcap_file):
print(f"Error: PCAP file not found: {pcap_file}")
sys.exit(1)
return ('pcap', pcap_file, None, None)
# Otherwise, parse as live capture arguments
if len(args) > 0:
# Check if first arg is an interface name
if args[0].startswith("wl") and len(args[0]) <= 6:
WIFI_INTERFACE = args[0]
CHANNEL = int(args[1]) if len(args) > 1 else 36
DURATION = int(args[2]) if len(args) > 2 else 10
else:
# First arg is channel (using default interface)
CHANNEL = int(args[0]) if len(args) > 0 else 36
DURATION = int(args[1]) if len(args) > 1 else 10
else:
CHANNEL = 36
DURATION = 10
return ('live', WIFI_INTERFACE, CHANNEL, DURATION)
def setup_monitor_mode(interface, channel):
"""Set WiFi interface to monitor mode."""
print(f"=== Setting up monitor mode on {interface} ===")
# Try to unmanage interface from NetworkManager
try:
result = subprocess.run(["which", "nmcli"], capture_output=True, text=True)
if result.returncode == 0:
print("Unmanaging interface from NetworkManager...")
subprocess.run(
["nmcli", "device", "set", interface, "managed", "no"],
capture_output=True,
stderr=subprocess.DEVNULL
)
except Exception:
pass
# Unblock WiFi if blocked by rfkill
try:
subprocess.run(["rfkill", "unblock", "wifi"], check=False)
except Exception:
pass
# Check current mode
try:
result = subprocess.run(
["iw", "dev", interface, "info"],
capture_output=True,
text=True,
check=True
)
if "type monitor" in result.stdout:
print(f"Already in monitor mode")
else:
print(f"Setting {interface} to monitor mode...")
# Bring down interface
subprocess.run(
["ip", "link", "set", interface, "down"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
)
time.sleep(0.5)
# Set monitor mode
subprocess.run(
["iw", "dev", interface, "set", "type", "monitor"],
check=True,
capture_output=True,
text=True
)
time.sleep(0.5)
# Bring up interface
subprocess.run(
["ip", "link", "set", interface, "up"],
capture_output=True,
text=True,
check=True
)
print("Monitor mode activated")
except subprocess.CalledProcessError as e:
print(f"Error setting monitor mode: {e}")
if hasattr(e, 'stderr') and e.stderr:
print(f"Error details: {e.stderr}")
sys.exit(1)
# Set channel
try:
subprocess.run(
["iw", "dev", interface, "set", "channel", str(channel)],
check=True,
capture_output=True
)
print(f"Channel set to {channel}")
except subprocess.CalledProcessError as e:
print(f"Error setting channel: {e}")
print("Continuing anyway - channel may not be set correctly")
# Verify monitor mode
print("\nVerifying monitor mode...")
try:
result = subprocess.run(
["iw", "dev", interface, "info"],
capture_output=True,
text=True,
check=True
)
for line in result.stdout.splitlines():
if "type" in line or "channel" in line:
print(f"\t{line.strip()}")
except Exception:
pass
def get_frame_type_name(pkt):
"""Get human-readable frame type name."""
if not pkt.haslayer(Dot11):
return "Unknown"
dot11 = pkt[Dot11]
fc = dot11.FCfield
if dot11.type == 0: # Management
if dot11.subtype == 8:
return "Beacon"
elif dot11.subtype == 4:
return "Probe Request"
elif dot11.subtype == 5:
return "Probe Response"
elif dot11.subtype == 11:
return "Authentication"
elif dot11.subtype == 12:
return "Deauthentication"
elif dot11.subtype == 0:
return "Association Request"
elif dot11.subtype == 1:
return "Association Response"
elif dot11.subtype == 10:
return "Disassociation"
else:
return f"Management ({dot11.subtype})"
elif dot11.type == 1: # Control
return f"Control ({dot11.subtype})"
elif dot11.type == 2: # Data
if dot11.subtype == 8:
return "QoS Data"
elif dot11.subtype == 0:
return "Data"
else:
return f"Data ({dot11.subtype})"
else:
return f"Unknown ({dot11.type}/{dot11.subtype})"
def get_ra_ta(pkt):
"""Extract RA and TA from 802.11 frame."""
if not pkt.haslayer(Dot11):
return None, None
dot11 = pkt[Dot11]
# RA is typically addr1, TA is typically addr2
# But this depends on frame type and direction
ra = dot11.addr1 if hasattr(dot11, 'addr1') else None
ta = dot11.addr2 if hasattr(dot11, 'addr2') else None
# Format MAC addresses
if ra:
ra = ra.lower() if isinstance(ra, str) else ':'.join(f'{b:02x}' for b in ra)
if ta:
ta = ta.lower() if isinstance(ta, str) else ':'.join(f'{b:02x}' for b in ta)
return ra, ta
def get_phy_info(pkt):
"""Extract PHY rate and MCS from packet (if available in radiotap)."""
phy_rate = None
mcs = None
# Check for radiotap layer
if pkt.haslayer(RadioTap):
radiotap = pkt[RadioTap]
# Try to get rate (in 0.5 Mbps units)
if hasattr(radiotap, 'Rate'):
rate_val = radiotap.Rate
if rate_val:
phy_rate = rate_val * 0.5 # Convert to Mbps
# Try to get MCS
if hasattr(radiotap, 'MCS'):
mcs_data = radiotap.MCS
if mcs_data and hasattr(mcs_data, 'index'):
mcs = mcs_data.index
return phy_rate, mcs
def analyze_packets(packets, duration):
"""Analyze captured packets and generate statistics."""
print("\n=== Capture Statistics ===")
total_count = len(packets)
print(f"Total packets captured: {total_count}")
if total_count == 0:
print("\n(No packets captured)")
print("\n=== Summary ===")
print("✗ No packets captured. Check:")
print(f" 1. Is there WiFi traffic on the channel?")
print(f" 2. Is the interface actually in monitor mode?")
print(f" 3. Try a different channel or longer duration")
return
# Count PLCP headers (radiotap present)
plcp_count = sum(1 for pkt in packets if pkt.haslayer(RadioTap))
print(f"PLCP headers: {plcp_count}")
if total_count > 0:
rate = total_count / duration
print(f"Packet rate: {rate:.1f} packets/second")
print()
# Sample packets
print("Sample packets (first 10):")
for i, pkt in enumerate(packets[:10]):
ra, ta = get_ra_ta(pkt)
ra_str = ra if ra else "N/A"
ta_str = ta if ta else "N/A"
frame_type = get_frame_type_name(pkt)
# Check if encrypted
encrypted = "encrypted" if (pkt.haslayer(Dot11) and pkt[Dot11].FCfield & 0x40) else "unencrypted"
# Check retry
retry = " [retry]" if (pkt.haslayer(Dot11) and pkt[Dot11].FCfield & 0x08) else ""
# Duration
duration_val = "N/A"
if pkt.haslayer(Dot11):
duration_val = pkt[Dot11].Duration if hasattr(pkt[Dot11], 'Duration') else "N/A"
# PLCP
plcp = "yes" if pkt.haslayer(RadioTap) else "no"
print(f" Frame {i+1}: RA={ra_str}, TA={ta_str}, type={frame_type}, {encrypted}, dur={duration_val}, PLCP={plcp}{retry}")
print()
# Unique RA/TA pairs
print("Unique RA/TA pairs (with counts):")
ra_ta_pairs = Counter()
for pkt in packets:
ra, ta = get_ra_ta(pkt)
if ra or ta:
ra_str = ra if ra else "N/A"
ta_str = ta if ta else "N/A"
pair = f"{ra_str} -> {ta_str}"
ra_ta_pairs[pair] += 1
if ra_ta_pairs:
for pair, count in ra_ta_pairs.most_common():
print(f" {pair}: {count} frame(s)")
else:
print(" (no valid RA/TA pairs found)")
print()
# PHY rate and MCS histograms per RA/TA pair
print("PHY Rate and MCS Histograms per RA/TA pair:")
rate_histograms = defaultdict(Counter)
mcs_histograms = defaultdict(Counter)
for pkt in packets:
ra, ta = get_ra_ta(pkt)
if not (ra or ta):
continue
# Only process data frames (type 2)
if not pkt.haslayer(Dot11) or pkt[Dot11].type != 2:
continue
ra_str = ra if ra else "N/A"
ta_str = ta if ta else "N/A"
pair = f"{ra_str} -> {ta_str}"
phy_rate, mcs = get_phy_info(pkt)
if phy_rate:
rate_histograms[pair][phy_rate] += 1
if mcs is not None:
mcs_histograms[pair][mcs] += 1
for pair in sorted(set(list(rate_histograms.keys()) + list(mcs_histograms.keys()))):
print(f"\n {pair}:")
# PHY Rate histogram
print(" PHY Rate (Mbps):")
if pair in rate_histograms:
for rate in sorted(rate_histograms[pair].keys()):
print(f" {rate} Mbps: {rate_histograms[pair][rate]} frame(s)")
else:
print(" (no PHY rate data)")
# MCS histogram
print(" MCS Index:")
if pair in mcs_histograms:
for mcs_val in sorted(mcs_histograms[pair].keys()):
print(f" MCS {mcs_val}: {mcs_histograms[pair][mcs_val]} frame(s)")
else:
print(" (no MCS data)")
print()
# Frame type breakdown
print("Frame type breakdown:")
frame_types = Counter()
for pkt in packets:
if pkt.haslayer(Dot11):
dot11 = pkt[Dot11]
if dot11.type == 0:
frame_types["Management"] += 1
elif dot11.type == 1:
frame_types["Control"] += 1
elif dot11.type == 2:
frame_types["Data"] += 1
else:
frame_types["Unknown"] += 1
for frame_type, count in frame_types.most_common():
print(f" {frame_type}: {count} frame(s)")
print()
# Data frame analysis (QoS Data frames, subtype 8)
print("Data frame analysis (iperf typically uses QoS Data frames, subtype 8):")
qos_data_frames = [pkt for pkt in packets if pkt.haslayer(Dot11) and pkt[Dot11].type == 2 and pkt[Dot11].subtype == 8]
qos_count = len(qos_data_frames)
print(f" QoS Data frames (type 2, subtype 8): {qos_count}")
encrypted_count = sum(1 for pkt in qos_data_frames if pkt.haslayer(Dot11) and pkt[Dot11].FCfield & 0x40)
unencrypted_count = qos_count - encrypted_count
print(f" Encrypted: {encrypted_count}")
print(f" Unencrypted: {unencrypted_count}")
if qos_count > 0:
print(" Sample QoS Data frames (likely iperf traffic):")
for i, pkt in enumerate(qos_data_frames[:5]):
ra, ta = get_ra_ta(pkt)
ra_str = ra if ra else "N/A"
ta_str = ta if ta else "N/A"
encrypted = "encrypted" if (pkt.haslayer(Dot11) and pkt[Dot11].FCfield & 0x40) else "unencrypted"
retry = " [retry]" if (pkt.haslayer(Dot11) and pkt[Dot11].FCfield & 0x08) else ""
duration_val = pkt[Dot11].Duration if pkt.haslayer(Dot11) and hasattr(pkt[Dot11], 'Duration') else "N/A"
print(f" Frame {i+1}: RA={ra_str}, TA={ta_str}, {encrypted}, dur={duration_val}{retry}")
print()
# Frames involving server MAC (80:84:89:93:c4:b6)
server_mac = "80:84:89:93:c4:b6"
print(f"Frames involving server MAC ({server_mac}):")
server_frames = []
for pkt in packets:
ra, ta = get_ra_ta(pkt)
if (ra and ra.lower() == server_mac.lower()) or (ta and ta.lower() == server_mac.lower()):
server_frames.append(pkt)
server_count = len(server_frames)
print(f" Total frames with server MAC: {server_count}")
if server_count > 0:
server_frame_types = Counter()
for pkt in server_frames:
if pkt.haslayer(Dot11):
dot11 = pkt[Dot11]
if dot11.type == 0:
server_frame_types["Management"] += 1
elif dot11.type == 1:
server_frame_types["Control"] += 1
elif dot11.type == 2:
server_frame_types["Data"] += 1
print(" Frame type breakdown:")
for frame_type, count in server_frame_types.most_common():
print(f" {frame_type}: {count} frame(s)")
print(" Sample frames:")
for i, pkt in enumerate(server_frames[:5]):
ra, ta = get_ra_ta(pkt)
ra_str = ra if ra else "N/A"
ta_str = ta if ta else "N/A"
frame_type = get_frame_type_name(pkt)
encrypted = "encrypted" if (pkt.haslayer(Dot11) and pkt[Dot11].FCfield & 0x40) else "unencrypted"
retry = " [retry]" if (pkt.haslayer(Dot11) and pkt[Dot11].FCfield & 0x08) else ""
duration_val = pkt[Dot11].Duration if pkt.haslayer(Dot11) and hasattr(pkt[Dot11], 'Duration') else "N/A"
print(f" Frame {i+1}: RA={ra_str}, TA={ta_str}, type={frame_type}, {encrypted}, dur={duration_val}{retry}")
print()
# Summary
print("=== Summary ===")
if total_count > 0:
print(f"✓ Monitor mode is working! Captured {total_count} packet(s)")
if plcp_count > 0:
print(f"✓ PLCP headers detected: {plcp_count} packet(s) with radiotap information")
else:
print("⚠ No PLCP headers detected (may be using DLT_IEEE802_11 instead of DLT_IEEE802_11_RADIO)")
def packet_handler(pkt, packets_list):
"""Callback function to handle captured packets."""
packets_list.append(pkt)
def main():
"""Main function."""
mode, interface_or_pcap, channel, duration = parse_args()
if mode == 'pcap':
# Read from pcap file
pcap_file = interface_or_pcap
print("=== Reading packets from PCAP file ===")
print(f"PCAP file: {pcap_file}")
print()
try:
packets = rdpcap(pcap_file)
print(f"Loaded {len(packets)} packets from {pcap_file}")
# Get file size
file_size = os.path.getsize(pcap_file)
print(f"File size: {file_size} bytes")
print()
# Analyze packets (duration doesn't matter for pcap mode)
analyze_packets(packets, 1.0) # Use 1.0 as placeholder duration
except Exception as e:
print(f"Error reading PCAP file: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
else:
# Live capture from interface
interface = interface_or_pcap
print("=== Testing Monitor Mode with scapy ===")
print(f"Interface: {interface}")
print(f"Channel: {channel}")
print(f"Duration: {duration} seconds")
print()
# Setup monitor mode
setup_monitor_mode(interface, channel)
# Test capture (1 second)
print("\nChecking Data Link Type (1 second test capture)...")
print("(This may take up to 2 seconds if no packets are present)")
test_packets = []
snaplen = 0 if FULL_PACKET else DEFAULT_SNAPLEN
try:
sniff(iface=interface, prn=lambda pkt: test_packets.append(pkt), timeout=1, store=False, snaplen=snaplen)
except Exception as e:
print(f"Error during test capture: {e}")
test_count = len(test_packets)
test_plcp = sum(1 for pkt in test_packets if pkt.haslayer(RadioTap))
if test_count > 0:
print("Sample packets:")
for i, pkt in enumerate(test_packets[:5]):
ra, ta = get_ra_ta(pkt)
ra_str = ra if ra else "N/A"
ta_str = ta if ta else "N/A"
plcp = "yes" if pkt.haslayer(RadioTap) else "no"
print(f" Frame {i+1 if i > 0 else test_count}: RA={ra_str}, TA={ta_str}, PLCP={plcp}")
print(f"\nTest capture results:")
print(f" Packets captured: {test_count}")
print(f" PLCP headers: {test_plcp}")
if test_plcp == 0 and test_count > 0:
print(" Note: Packets captured but no radiotap headers (may be using DLT_IEEE802_11 instead of DLT_IEEE802_11_RADIO)")
# Main capture
print(f"\n=== Starting scapy capture ({duration} seconds) ===")
print("Press Ctrl+C to stop early\n")
# Set snaplen based on FULL_PACKET flag
snaplen = 0 if FULL_PACKET else DEFAULT_SNAPLEN
if FULL_PACKET:
print("Capturing full packets...")
else:
print(f"Capturing packets (snaplen={snaplen} bytes, header + some payload)...")
packets = []
pcap_file = None
if KEEP_PCAP:
pcap_file = tempfile.NamedTemporaryFile(delete=False, suffix='.pcap', prefix='scapy_capture_')
pcap_path = pcap_file.name
pcap_file.close()
print(f"Capturing to file: {pcap_path}")
try:
if KEEP_PCAP:
sniff(iface=interface, prn=lambda pkt: packets.append(pkt), timeout=duration, store=True, snaplen=snaplen)
wrpcap(pcap_path, packets)
print(f"Pcap file size: {os.path.getsize(pcap_path)} bytes")
print(f"Keeping pcap file: {pcap_path}")
print(f" (Use: python3 test_monitor.py {pcap_path} to analyze)")
else:
sniff(iface=interface, prn=lambda pkt: packets.append(pkt), timeout=duration, store=True, snaplen=snaplen)
except KeyboardInterrupt:
print("\nCapture interrupted by user")
except Exception as e:
print(f"\nError during capture: {e}")
import traceback
traceback.print_exc()
# Analyze packets
analyze_packets(packets, duration)
if __name__ == "__main__":
main()

222
test_monitor_tshark.sh Executable file
View File

@ -0,0 +1,222 @@
#!/bin/bash
# Test script to verify monitor mode works with tshark
# Usage: ./test_monitor_tshark.sh [interface] [channel] [duration_seconds] [--keep-pcap]
# Or: KEEP_PCAP=1 sudo -E ./test_monitor_tshark.sh [interface] [channel] [duration_seconds]
set -e
# Parse arguments
KEEP_PCAP_FLAG=""
ARGS=()
for arg in "$@"; do
if [ "$arg" = "--keep-pcap" ] || [ "$arg" = "-k" ]; then
KEEP_PCAP_FLAG="1"
else
ARGS+=("$arg")
fi
done
INTERFACE="${ARGS[0]:-wlan0}"
CHANNEL="${ARGS[1]:-36}"
DURATION="${ARGS[2]:-10}" # Default 10 seconds, minimum 1 second
# Check for KEEP_PCAP environment variable or flag
if [ -n "$KEEP_PCAP_FLAG" ]; then
KEEP_PCAP="1"
elif [ -z "${KEEP_PCAP:-}" ]; then
KEEP_PCAP=""
fi
# Ensure minimum 1 second
if [ "$DURATION" -lt 1 ]; then
DURATION=1
fi
echo "=== Testing Monitor Mode with tshark ==="
echo "Interface: $INTERFACE"
echo "Channel: $CHANNEL"
echo "Duration: $DURATION seconds"
echo ""
# Check if running as root
if [ "$EUID" -ne 0 ]; then
echo "Please run as root (use sudo)"
exit 1
fi
# Check if tshark is installed
if ! command -v tshark &> /dev/null; then
echo "tshark is not installed. Installing..."
if command -v apt-get &> /dev/null; then
sudo apt-get update
sudo apt-get install -y tshark
elif command -v dnf &> /dev/null; then
sudo dnf install -y wireshark-cli
else
echo "Please install tshark manually"
exit 1
fi
fi
# Unmanage from NetworkManager
if command -v nmcli &> /dev/null; then
echo "Unmanaging interface from NetworkManager..."
nmcli device set "$INTERFACE" managed no 2>/dev/null || true
fi
# Unblock WiFi
rfkill unblock wifi 2>/dev/null || true
# Bring down interface
echo "Bringing down interface..."
ip link set "$INTERFACE" down 2>/dev/null || true
sleep 0.5
# Set monitor mode
echo "Setting monitor mode..."
if ! iw dev "$INTERFACE" set type monitor; then
echo "Error: Failed to set monitor mode"
exit 1
fi
sleep 0.5
# Bring up interface
echo "Bringing up interface..."
ip link set "$INTERFACE" up || echo "Warning: Failed to bring interface up"
sleep 0.5
# Set channel
echo "Setting channel to $CHANNEL..."
iw dev "$INTERFACE" set channel "$CHANNEL" || echo "Warning: Failed to set channel"
# Verify monitor mode
echo ""
echo "Verifying monitor mode..."
iw dev "$INTERFACE" info | grep -E "(type|channel)" || echo "Could not verify"
# Check DLT with tshark (capture for 1 second)
echo ""
echo "Checking Data Link Type (1 second test capture)..."
echo "(This may take up to 2 seconds if no packets are present)"
# Use timeout with -c to limit packets and avoid hanging
# Capture both stdout and stderr
# Fields: frame.number, wlan.ra (Receiver Address), wlan.ta (Transmitter Address), radiotap.present
# Note: wlan.ra and wlan.ta may not be available for all frame types, but are correct for monitor mode
TEST_OUTPUT=$(timeout 2 tshark -i "$INTERFACE" -c 100 -T fields \
-e frame.number \
-e wlan.ra \
-e wlan.ta \
-e radiotap.present \
2>&1 || true)
TEST_EXIT_CODE=${PIPESTATUS[0]}
# Show any warnings/errors from tshark (but not packet data)
echo "$TEST_OUTPUT" | grep -E "(Running as|tshark:|Warning|Error|Capturing)" || true
# Count packets (lines starting with a number, excluding error messages)
PACKET_LINES=$(echo "$TEST_OUTPUT" | grep -E '^[0-9]+' || true)
PACKET_COUNT=$(echo "$PACKET_LINES" | wc -l || echo "0")
# Count lines with radiotap.present (4th field) set to 1
PLCP_COUNT=$(echo "$PACKET_LINES" | awk -F'\t' 'NF >= 4 && $1 != "" && $4 != "" && $4 != "0" && $4 != "-" {count++} END {print count+0}' || echo "0")
# Show sample output with better formatting
if [ "$PACKET_COUNT" -gt 0 ]; then
echo "Sample packets:"
echo "$PACKET_LINES" | head -5 | awk -F'\t' '{
ra = ($2 != "" && $2 != "-") ? $2 : "N/A"
ta = ($3 != "" && $3 != "-") ? $3 : "N/A"
radiotap = ($4 == "1" || $4 == "1.0") ? "yes" : "no"
printf " Frame %s: RA=%s, TA=%s, PLCP=%s\n", $1, ra, ta, radiotap
}'
fi
echo ""
echo "Test capture results:"
echo " Packets captured: $PACKET_COUNT"
echo " PLCP headers: $PLCP_COUNT"
if [ "$PLCP_COUNT" -eq 0 ] && [ "$PACKET_COUNT" -gt 0 ]; then
echo " Note: Packets captured but no radiotap headers (may be using DLT_IEEE802_11 instead of DLT_IEEE802_11_RADIO)"
fi
echo ""
echo "=== Starting tshark capture ($DURATION seconds) ==="
echo "Press Ctrl+C to stop early"
echo ""
# Capture for specified duration and count packets
echo "Capturing packets for $DURATION seconds..."
# Use a temporary pcap file to avoid field extraction errors during capture
# Capture to file first, then parse it - this prevents tshark from exiting early
TEMP_PCAP=$(mktemp /tmp/tshark_capture_XXXXXX.pcap)
echo "Capturing to temporary file: $TEMP_PCAP"
set +e # Temporarily disable exit on error
# Capture to pcap file - this won't error on missing fields
# Use -b filesize:100000 to rotate files if needed, but we'll only use the first
timeout "$DURATION" tshark -q -i "$INTERFACE" -n -w "$TEMP_PCAP" 2>/dev/null
CAPTURE_EXIT_CODE=$?
set -e # Re-enable exit on error
# Force sync to ensure file is written
sync
# Exit code 124 means timeout occurred (expected), 0 means command completed normally
if [ "$CAPTURE_EXIT_CODE" -ne 0 ] && [ "$CAPTURE_EXIT_CODE" -ne 124 ]; then
echo "Warning: tshark capture exited with code $CAPTURE_EXIT_CODE"
fi
# Check if pcap file exists and get its size
if [ -f "$TEMP_PCAP" ]; then
PCAP_SIZE=$(stat -c%s "$TEMP_PCAP" 2>/dev/null || stat -f%z "$TEMP_PCAP" 2>/dev/null || echo "0")
echo "Pcap file size: $PCAP_SIZE bytes"
# Count packets in raw pcap file using capinfos or tshark
if command -v capinfos &> /dev/null; then
RAW_PACKET_COUNT=$(capinfos -c "$TEMP_PCAP" 2>/dev/null | grep "^Number of packets:" | awk '{print $4}' || echo "0")
echo "Raw packets in pcap file: $RAW_PACKET_COUNT"
else
# Fallback: use tshark to count packets
RAW_PACKET_COUNT=$(tshark -q -r "$TEMP_PCAP" -n -T fields -e frame.number 2>/dev/null | tail -1 || echo "0")
if [ "$RAW_PACKET_COUNT" != "0" ] && [ -n "$RAW_PACKET_COUNT" ]; then
echo "Raw packets in pcap file: $RAW_PACKET_COUNT"
else
echo "Raw packets in pcap file: (could not determine)"
fi
fi
else
echo "ERROR: Pcap file was not created: $TEMP_PCAP"
exit 1
fi
# Now parse the pcap file using Python script (more robust than bash parsing)
# The Python script handles missing fields gracefully and provides detailed statistics
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
PARSE_SCRIPT="${SCRIPT_DIR}/parse_tshark_pcap.py"
# Check if Python script exists
if [ ! -f "$PARSE_SCRIPT" ]; then
echo "Error: Python parser script not found: $PARSE_SCRIPT"
echo " Falling back to basic tshark output"
exit 1
fi
# Clean up temp file (unless KEEP_PCAP is set)
if [ -z "$KEEP_PCAP" ]; then
echo "Cleaning up temporary pcap file: $TEMP_PCAP"
rm -f "$TEMP_PCAP"
else
echo "Keeping temporary pcap file: $TEMP_PCAP"
echo " (Use: tshark -r $TEMP_PCAP to analyze)"
fi
# Force output flush
sync
# Parse pcap file using Python script (more robust than bash/awk parsing)
echo ""
python3 "$PARSE_SCRIPT" "$TEMP_PCAP" "$DURATION" "$RAW_PACKET_COUNT"