89 lines
3.1 KiB
Python
Executable file
89 lines
3.1 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
import sqlite3
|
|
import csv
|
|
import argparse
|
|
from datetime import datetime
|
|
|
|
def parse_args():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--csv", required=True, help="Input speedtest CSV")
|
|
parser.add_argument("--kismet", required=True, help="Path to .kismet log file")
|
|
parser.add_argument("--output", required=True, help="Output enriched CSV")
|
|
return parser.parse_args()
|
|
|
|
def convert_timestamp_to_epoch(ts_string):
|
|
try:
|
|
return int(datetime.strptime(ts_string.split(".")[0], "%Y-%m-%dT%H:%M:%S").timestamp())
|
|
except Exception as e:
|
|
print(f"[!] Invalid timestamp format: {ts_string}")
|
|
return None
|
|
|
|
def get_rf_metrics(cursor, bssid, channel, timestamp):
|
|
cursor.execute("""
|
|
SELECT COUNT(*) FROM devicelink
|
|
WHERE type = 'Wi-Fi Client' AND mac = ? AND last_time BETWEEN ? AND ?
|
|
""", (bssid.lower(), timestamp - 10, timestamp + 10))
|
|
clients_on_ap = cursor.fetchone()[0]
|
|
|
|
cursor.execute("""
|
|
SELECT COUNT(DISTINCT devicelink.remote_mac)
|
|
FROM devicelink
|
|
JOIN devices ON devicelink.mac = devices.base_mac
|
|
WHERE devicelink.type = 'Wi-Fi Client'
|
|
AND devices.channel = ?
|
|
AND devicelink.last_time BETWEEN ? AND ?
|
|
""", (channel, timestamp - 10, timestamp + 10))
|
|
clients_on_channel = cursor.fetchone()[0]
|
|
|
|
cursor.execute("""
|
|
SELECT COUNT(*) FROM devices
|
|
WHERE type = 'Wi-Fi AP' AND channel = ?
|
|
AND last_time BETWEEN ? AND ?
|
|
""", (channel, timestamp - 10, timestamp + 10))
|
|
aps_on_channel = cursor.fetchone()[0]
|
|
|
|
congestion_score = round(clients_on_channel / aps_on_channel, 2) if aps_on_channel else 0.0
|
|
|
|
return clients_on_ap, clients_on_channel, aps_on_channel, congestion_score
|
|
|
|
def main():
|
|
args = parse_args()
|
|
conn = sqlite3.connect(args.kismet)
|
|
cursor = conn.cursor()
|
|
|
|
with open(args.csv, newline='') as infile, open(args.output, 'w', newline='') as outfile:
|
|
reader = csv.DictReader(infile)
|
|
fieldnames = reader.fieldnames + [
|
|
"ClientsOnAP", "ClientsOnChannel", "APsOnChannel", "CongestionScore"
|
|
]
|
|
writer = csv.DictWriter(outfile, fieldnames=fieldnames)
|
|
writer.writeheader()
|
|
|
|
for row in reader:
|
|
ts = convert_timestamp_to_epoch(row["Timestamp"])
|
|
try:
|
|
bssid = row["BSSID"].strip().lower()
|
|
channel = int(row["Channel"])
|
|
except Exception as e:
|
|
print(f"[!] Failed to extract BSSID/Channel: {e}")
|
|
writer.writerow(row)
|
|
continue
|
|
|
|
if ts is None:
|
|
writer.writerow(row)
|
|
continue
|
|
|
|
clients_ap, clients_chan, aps_chan, congestion = get_rf_metrics(cursor, bssid, channel, ts)
|
|
|
|
row["ClientsOnAP"] = clients_ap
|
|
row["ClientsOnChannel"] = clients_chan
|
|
row["APsOnChannel"] = aps_chan
|
|
row["CongestionScore"] = congestion
|
|
|
|
writer.writerow(row)
|
|
|
|
conn.close()
|
|
print(f"[+] Enrichment complete: {args.output}")
|
|
|
|
if __name__ == "__main__":
|
|
main()
|