feat(everything): fix doh and increase min TLS to 1.3
This commit is contained in:
@@ -1,405 +1,258 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Fast PCAP Preprocessor for DNS QoS Analysis
|
||||
Loads PCAP into memory first, then uses binary search for matching.
|
||||
Uses LAN IP to determine direction (LAN = sent, non-LAN = received).
|
||||
Fast PCAP Preprocessor for DNS QoS Analysis.
|
||||
|
||||
Loads PCAP into memory, then uses binary search to match packets to query
|
||||
windows. Direction is determined by a configured local IP (the netns veth IP).
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import bisect
|
||||
import csv
|
||||
import shutil
|
||||
import socket
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, NamedTuple
|
||||
import time
|
||||
|
||||
import dpkt
|
||||
from dateutil import parser as date_parser
|
||||
|
||||
|
||||
BANDWIDTH_COLUMNS = [
|
||||
'bytes_sent',
|
||||
'bytes_received',
|
||||
'packets_sent',
|
||||
'packets_received',
|
||||
'total_bytes',
|
||||
"bytes_sent",
|
||||
"bytes_received",
|
||||
"packets_sent",
|
||||
"packets_received",
|
||||
"total_bytes",
|
||||
]
|
||||
|
||||
DEFAULT_LOCAL_IP = "192.168.100.2" # netns veth1 address
|
||||
DEFAULT_PROVIDERS = ["adguard", "cloudflare", "google", "quad9"]
|
||||
|
||||
|
||||
class Packet(NamedTuple):
|
||||
"""Lightweight packet representation."""
|
||||
timestamp: float
|
||||
size: int
|
||||
is_outbound: bool # True if from LAN, False if from internet
|
||||
is_outbound: bool
|
||||
|
||||
|
||||
class QueryWindow:
|
||||
"""Efficient query window representation."""
|
||||
__slots__ = ['index', 'start', 'end', 'sent', 'received', 'pkts_sent', 'pkts_received']
|
||||
|
||||
def __init__(self, index: int, start: float, end: float):
|
||||
self.index = index
|
||||
self.start = start
|
||||
self.end = end
|
||||
self.sent = 0
|
||||
self.received = 0
|
||||
self.pkts_sent = 0
|
||||
self.pkts_received = 0
|
||||
|
||||
|
||||
def is_already_processed(csv_path: Path) -> bool:
|
||||
"""
|
||||
Check if CSV has already been processed.
|
||||
Returns True if bandwidth columns exist AND at least one row has non-zero data.
|
||||
"""
|
||||
def needs_processing(csv_path: Path) -> bool:
|
||||
"""True if file lacks bandwidth columns OR all values are empty/zero."""
|
||||
try:
|
||||
with open(csv_path, 'r', encoding='utf-8') as f:
|
||||
with open(csv_path, "r", encoding="utf-8") as f:
|
||||
reader = csv.DictReader(f)
|
||||
|
||||
# Check if columns exist
|
||||
if not reader.fieldnames:
|
||||
return False
|
||||
|
||||
if not all(col in reader.fieldnames for col in BANDWIDTH_COLUMNS):
|
||||
return False
|
||||
|
||||
# Check if any row has non-zero bandwidth data
|
||||
return True
|
||||
if not all(c in reader.fieldnames for c in BANDWIDTH_COLUMNS):
|
||||
return True
|
||||
for row in reader:
|
||||
for col in BANDWIDTH_COLUMNS:
|
||||
val = row.get(col, '').strip()
|
||||
if val and val != '0':
|
||||
return True
|
||||
|
||||
# All rows have zero/empty values - not truly processed
|
||||
return False
|
||||
|
||||
val = (row.get(col) or "").strip()
|
||||
if val and val != "0":
|
||||
return False # has real data
|
||||
return True # columns exist but all empty/zero
|
||||
except Exception:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def parse_csv_timestamp(ts_str: str) -> float:
|
||||
"""Convert RFC3339Nano timestamp to Unix epoch (seconds)."""
|
||||
dt = date_parser.isoparse(ts_str)
|
||||
return dt.timestamp()
|
||||
return date_parser.isoparse(ts_str).timestamp()
|
||||
|
||||
|
||||
def is_lan_ip(ip_bytes: bytes) -> bool:
|
||||
"""Check if IP is a private/LAN address."""
|
||||
if len(ip_bytes) != 4:
|
||||
return False
|
||||
|
||||
first = ip_bytes[0]
|
||||
second = ip_bytes[1]
|
||||
|
||||
# 10.0.0.0/8
|
||||
if first == 10:
|
||||
return True
|
||||
|
||||
# 172.16.0.0/12
|
||||
if first == 172 and 16 <= second <= 31:
|
||||
return True
|
||||
|
||||
# 192.168.0.0/16
|
||||
if first == 192 and second == 168:
|
||||
return True
|
||||
|
||||
# 127.0.0.0/8 (localhost)
|
||||
if first == 127:
|
||||
return True
|
||||
|
||||
return False
|
||||
def load_pcap(pcap_path: Path, local_ip_bytes: bytes) -> List[Packet]:
|
||||
"""Load PCAP into a list of Packets sorted by timestamp."""
|
||||
print(" Loading PCAP...")
|
||||
t0 = time.time()
|
||||
packets: List[Packet] = []
|
||||
|
||||
with open(pcap_path, "rb") as f:
|
||||
try:
|
||||
reader = dpkt.pcap.Reader(f)
|
||||
except ValueError:
|
||||
f.seek(0)
|
||||
reader = dpkt.pcapng.Reader(f)
|
||||
|
||||
def load_pcap_into_memory(pcap_path: Path) -> List[Packet]:
|
||||
"""Load all packets from PCAP into memory with minimal data."""
|
||||
packets = []
|
||||
|
||||
print(f" Loading PCAP into memory...")
|
||||
start_time = time.time()
|
||||
|
||||
try:
|
||||
with open(pcap_path, 'rb') as f:
|
||||
for ts, buf in reader:
|
||||
try:
|
||||
pcap = dpkt.pcap.Reader(f)
|
||||
except:
|
||||
# Try pcapng format
|
||||
f.seek(0)
|
||||
pcap = dpkt.pcapng.Reader(f)
|
||||
|
||||
for ts, buf in pcap:
|
||||
try:
|
||||
packet_time = float(ts)
|
||||
packet_size = len(buf)
|
||||
|
||||
# Parse to get source IP
|
||||
eth = dpkt.ethernet.Ethernet(buf)
|
||||
|
||||
# Default to outbound if we can't determine
|
||||
is_outbound = True
|
||||
|
||||
if isinstance(eth.data, dpkt.ip.IP):
|
||||
ip = eth.data
|
||||
src_ip = ip.src
|
||||
is_outbound = is_lan_ip(src_ip)
|
||||
|
||||
packets.append(Packet(
|
||||
timestamp=packet_time,
|
||||
size=packet_size,
|
||||
is_outbound=is_outbound
|
||||
))
|
||||
|
||||
except (dpkt.dpkt.NeedData, dpkt.dpkt.UnpackError, AttributeError):
|
||||
eth = dpkt.ethernet.Ethernet(buf)
|
||||
ip = eth.data
|
||||
if not isinstance(ip, dpkt.ip.IP):
|
||||
continue
|
||||
|
||||
except Exception as e:
|
||||
print(f" Error reading PCAP: {e}")
|
||||
return []
|
||||
|
||||
elapsed = time.time() - start_time
|
||||
print(f" Loaded {len(packets):,} packets in {elapsed:.2f}s")
|
||||
|
||||
# Sort by timestamp for binary search
|
||||
if ip.src == local_ip_bytes:
|
||||
is_outbound = True
|
||||
elif ip.dst == local_ip_bytes:
|
||||
is_outbound = False
|
||||
else:
|
||||
continue # not our traffic
|
||||
packets.append(Packet(float(ts), len(buf), is_outbound))
|
||||
except (dpkt.dpkt.NeedData, dpkt.dpkt.UnpackError, AttributeError):
|
||||
continue
|
||||
|
||||
packets.sort(key=lambda p: p.timestamp)
|
||||
|
||||
print(f" Loaded {len(packets):,} packets in {time.time() - t0:.2f}s")
|
||||
return packets
|
||||
|
||||
|
||||
def find_packets_in_window(
|
||||
packets: List[Packet],
|
||||
start_time: float,
|
||||
end_time: float,
|
||||
left_hint: int = 0
|
||||
) -> tuple[List[Packet], int]:
|
||||
"""
|
||||
Binary search to find all packets within time window.
|
||||
Returns (matching_packets, left_index_hint_for_next_search).
|
||||
"""
|
||||
if not packets:
|
||||
return [], 0
|
||||
|
||||
# Binary search for first packet >= start_time
|
||||
left, right = left_hint, len(packets) - 1
|
||||
first_idx = len(packets)
|
||||
|
||||
while left <= right:
|
||||
mid = (left + right) // 2
|
||||
if packets[mid].timestamp >= start_time:
|
||||
first_idx = mid
|
||||
right = mid - 1
|
||||
else:
|
||||
left = mid + 1
|
||||
|
||||
# No packets in range
|
||||
if first_idx >= len(packets) or packets[first_idx].timestamp > end_time:
|
||||
return [], first_idx
|
||||
|
||||
# Collect all packets in window
|
||||
matching = []
|
||||
idx = first_idx
|
||||
while idx < len(packets) and packets[idx].timestamp <= end_time:
|
||||
matching.append(packets[idx])
|
||||
idx += 1
|
||||
|
||||
return matching, first_idx
|
||||
|
||||
|
||||
def load_csv_queries(csv_path: Path) -> List[Dict]:
|
||||
"""Load CSV and create query data structures."""
|
||||
queries = []
|
||||
with open(csv_path, 'r', encoding='utf-8') as f:
|
||||
reader = csv.DictReader(f)
|
||||
for row in reader:
|
||||
with open(csv_path, "r", encoding="utf-8") as f:
|
||||
for row in csv.DictReader(f):
|
||||
try:
|
||||
ts_epoch = parse_csv_timestamp(row['timestamp'])
|
||||
duration_s = float(row['duration_ns']) / 1e9
|
||||
queries.append({
|
||||
'data': row,
|
||||
'start_time': ts_epoch,
|
||||
'end_time': ts_epoch + duration_s,
|
||||
})
|
||||
start = parse_csv_timestamp(row["timestamp"])
|
||||
duration = float(row["duration_ns"]) / 1e9
|
||||
queries.append(
|
||||
{"data": row, "start_time": start, "end_time": start + duration}
|
||||
)
|
||||
except Exception as e:
|
||||
print(f" Warning: Skipping row - {e}")
|
||||
continue
|
||||
print(f" Warning: skipping row - {e}")
|
||||
return queries
|
||||
|
||||
|
||||
def match_packets_to_queries(
|
||||
packets: List[Packet],
|
||||
queries: List[Dict]
|
||||
) -> List[Dict]:
|
||||
"""Match packets to query windows using binary search."""
|
||||
if not queries or not packets:
|
||||
return queries
|
||||
|
||||
print(f" Matching packets to queries...")
|
||||
start_time = time.time()
|
||||
|
||||
# Initialize metrics
|
||||
def match_packets(packets: List[Packet], queries: List[Dict]) -> int:
|
||||
"""Assign bandwidth metrics to each query. Returns total matched packets."""
|
||||
if not packets or not queries:
|
||||
for q in queries:
|
||||
q.update({c: 0 for c in BANDWIDTH_COLUMNS})
|
||||
return 0
|
||||
|
||||
print(" Matching packets to queries...")
|
||||
t0 = time.time()
|
||||
timestamps = [p.timestamp for p in packets]
|
||||
matched = 0
|
||||
|
||||
for q in queries:
|
||||
q['bytes_sent'] = 0
|
||||
q['bytes_received'] = 0
|
||||
q['packets_sent'] = 0
|
||||
q['packets_received'] = 0
|
||||
q['total_bytes'] = 0
|
||||
|
||||
# Sort queries by start time for sequential processing
|
||||
queries_sorted = sorted(enumerate(queries), key=lambda x: x[1]['start_time'])
|
||||
|
||||
matched_packets = 0
|
||||
left_hint = 0 # Optimization: start next search from here
|
||||
|
||||
for original_idx, q in queries_sorted:
|
||||
matching, left_hint = find_packets_in_window(
|
||||
packets,
|
||||
q['start_time'],
|
||||
q['end_time'],
|
||||
left_hint
|
||||
)
|
||||
|
||||
for pkt in matching:
|
||||
matched_packets += 1
|
||||
lo = bisect.bisect_left(timestamps, q["start_time"])
|
||||
hi = bisect.bisect_right(timestamps, q["end_time"])
|
||||
bs = br = ps = pr = 0
|
||||
for pkt in packets[lo:hi]:
|
||||
if pkt.is_outbound:
|
||||
q['bytes_sent'] += pkt.size
|
||||
q['packets_sent'] += 1
|
||||
bs += pkt.size
|
||||
ps += 1
|
||||
else:
|
||||
q['bytes_received'] += pkt.size
|
||||
q['packets_received'] += 1
|
||||
|
||||
q['total_bytes'] = q['bytes_sent'] + q['bytes_received']
|
||||
|
||||
elapsed = time.time() - start_time
|
||||
print(f" Matched {matched_packets:,} packets in {elapsed:.2f}s")
|
||||
|
||||
# Statistics
|
||||
total_sent = sum(q['bytes_sent'] for q in queries)
|
||||
total_recv = sum(q['bytes_received'] for q in queries)
|
||||
queries_with_data = sum(1 for q in queries if q['total_bytes'] > 0)
|
||||
print(f" Total: {total_sent:,} bytes sent, {total_recv:,} bytes received")
|
||||
print(f" Queries with data: {queries_with_data}/{len(queries)}")
|
||||
|
||||
return queries
|
||||
br += pkt.size
|
||||
pr += 1
|
||||
q["bytes_sent"] = bs
|
||||
q["bytes_received"] = br
|
||||
q["packets_sent"] = ps
|
||||
q["packets_received"] = pr
|
||||
q["total_bytes"] = bs + br
|
||||
matched += ps + pr
|
||||
|
||||
print(f" Matched {matched:,} packets in {time.time() - t0:.2f}s")
|
||||
total_sent = sum(q["bytes_sent"] for q in queries)
|
||||
total_recv = sum(q["bytes_received"] for q in queries)
|
||||
with_data = sum(1 for q in queries if q["total_bytes"] > 0)
|
||||
print(f" Total: {total_sent:,} B sent, {total_recv:,} B received")
|
||||
print(f" Queries with data: {with_data}/{len(queries)}")
|
||||
return matched
|
||||
|
||||
|
||||
def write_enriched_csv(
|
||||
csv_path: Path, queries: List[Dict], backup: bool = True
|
||||
):
|
||||
"""Write enriched CSV with bandwidth columns."""
|
||||
def write_csv(csv_path: Path, queries: List[Dict], backup: bool = True):
|
||||
if backup and csv_path.exists():
|
||||
backup_path = csv_path.with_suffix('.csv.bak')
|
||||
if not backup_path.exists(): # Don't overwrite existing backup
|
||||
shutil.copy2(csv_path, backup_path)
|
||||
print(f" Backup: {backup_path.name}")
|
||||
|
||||
# Get fieldnames - filter out any existing bandwidth columns to avoid dupes
|
||||
bak = csv_path.with_suffix(".csv.bak")
|
||||
if not bak.exists():
|
||||
shutil.copy2(csv_path, bak)
|
||||
print(f" Backup: {bak.name}")
|
||||
|
||||
original_fields = [
|
||||
f for f in queries[0]['data'].keys()
|
||||
if f not in BANDWIDTH_COLUMNS
|
||||
f for f in queries[0]["data"].keys() if f not in BANDWIDTH_COLUMNS
|
||||
]
|
||||
fieldnames = original_fields + BANDWIDTH_COLUMNS
|
||||
|
||||
with open(csv_path, 'w', encoding='utf-8', newline='') as f:
|
||||
|
||||
with open(csv_path, "w", encoding="utf-8", newline="") as f:
|
||||
writer = csv.DictWriter(f, fieldnames=fieldnames)
|
||||
writer.writeheader()
|
||||
|
||||
for q in queries:
|
||||
row = {k: v for k, v in q['data'].items() if k not in BANDWIDTH_COLUMNS}
|
||||
for field in BANDWIDTH_COLUMNS:
|
||||
row[field] = q[field]
|
||||
row = {k: q["data"][k] for k in original_fields}
|
||||
for c in BANDWIDTH_COLUMNS:
|
||||
row[c] = q[c]
|
||||
writer.writerow(row)
|
||||
|
||||
print(f" Written: {csv_path.name}")
|
||||
print(f" Written: {csv_path.name}")
|
||||
|
||||
|
||||
def process_provider_directory(provider_path: Path):
|
||||
"""Process all CSV/PCAP pairs in a provider directory."""
|
||||
print(f"\n{'='*60}")
|
||||
print(f"Processing: {provider_path.name.upper()}")
|
||||
print(f"{'='*60}")
|
||||
|
||||
csv_files = sorted(provider_path.glob('*.csv'))
|
||||
processed = 0
|
||||
skipped = 0
|
||||
total_time = 0
|
||||
|
||||
for csv_path in csv_files:
|
||||
# Skip backup files
|
||||
if '.bak' in csv_path.name:
|
||||
def process_provider(provider_path: Path, local_ip_bytes: bytes):
|
||||
print(f"\n{'=' * 60}\nProcessing: {provider_path.name.upper()}\n{'=' * 60}")
|
||||
|
||||
processed = skipped = 0
|
||||
total_time = 0.0
|
||||
|
||||
for csv_path in sorted(provider_path.glob("*.csv")):
|
||||
name = csv_path.name.lower()
|
||||
if ".bak" in name or name.endswith((".cpu.csv", ".mem.csv")):
|
||||
continue
|
||||
|
||||
pcap_path = csv_path.with_suffix('.pcap')
|
||||
|
||||
|
||||
pcap_path = csv_path.with_suffix(".pcap")
|
||||
if not pcap_path.exists():
|
||||
print(f"\n ⚠ Skipping {csv_path.name} - no matching PCAP")
|
||||
print(f"\n ⚠ {csv_path.name}: no matching PCAP")
|
||||
continue
|
||||
|
||||
# Check if already processed
|
||||
if is_already_processed(csv_path):
|
||||
print(f"\n ⏭ Skipping {csv_path.name} - already processed")
|
||||
|
||||
if not needs_processing(csv_path):
|
||||
print(f"\n ⏭ {csv_path.name}: already processed")
|
||||
skipped += 1
|
||||
continue
|
||||
|
||||
|
||||
print(f"\n 📁 {csv_path.name}")
|
||||
file_start = time.time()
|
||||
|
||||
# Load PCAP into memory first
|
||||
packets = load_pcap_into_memory(pcap_path)
|
||||
t0 = time.time()
|
||||
|
||||
packets = load_pcap(pcap_path, local_ip_bytes)
|
||||
if not packets:
|
||||
print(f" ⚠ No packets found in PCAP")
|
||||
print(" ⚠ No usable packets in PCAP")
|
||||
continue
|
||||
|
||||
# Load CSV queries
|
||||
|
||||
queries = load_csv_queries(csv_path)
|
||||
if not queries:
|
||||
print(f" ⚠ No valid queries found")
|
||||
print(" ⚠ No valid queries in CSV")
|
||||
continue
|
||||
|
||||
print(f" Loaded {len(queries):,} queries")
|
||||
|
||||
# Match packets to queries
|
||||
enriched_queries = match_packets_to_queries(packets, queries)
|
||||
|
||||
# Write enriched CSV
|
||||
write_enriched_csv(csv_path, enriched_queries)
|
||||
|
||||
file_time = time.time() - file_start
|
||||
total_time += file_time
|
||||
processed += 1
|
||||
print(f" ✓ Completed in {file_time:.2f}s")
|
||||
|
||||
print(f"\n {'='*58}")
|
||||
print(f" {provider_path.name}: {processed} processed, {skipped} skipped")
|
||||
print(f" Time: {total_time:.2f}s")
|
||||
print(f" {'='*58}")
|
||||
|
||||
match_packets(packets, queries)
|
||||
write_csv(csv_path, queries)
|
||||
|
||||
dt = time.time() - t0
|
||||
total_time += dt
|
||||
processed += 1
|
||||
print(f" ✓ Completed in {dt:.2f}s")
|
||||
|
||||
print(
|
||||
f"\n {provider_path.name}: {processed} processed, "
|
||||
f"{skipped} skipped, {total_time:.2f}s"
|
||||
)
|
||||
|
||||
def main():
|
||||
"""Main preprocessing pipeline."""
|
||||
overall_start = time.time()
|
||||
|
||||
print("\n" + "="*60)
|
||||
print("DNS PCAP PREPROCESSOR - Memory-Optimized Edition")
|
||||
print("="*60)
|
||||
|
||||
results_dir = Path('results')
|
||||
|
||||
if not results_dir.exists():
|
||||
print(f"\n❌ Error: '{results_dir}' directory not found")
|
||||
ap = argparse.ArgumentParser(description=__doc__)
|
||||
ap.add_argument(
|
||||
"--local-ip",
|
||||
default=DEFAULT_LOCAL_IP,
|
||||
help=f"Local (netns veth) IP used to determine direction "
|
||||
f"(default: {DEFAULT_LOCAL_IP})",
|
||||
)
|
||||
ap.add_argument("--results-dir", default="results", type=Path)
|
||||
ap.add_argument("--providers", nargs="+", default=DEFAULT_PROVIDERS)
|
||||
args = ap.parse_args()
|
||||
|
||||
local_ip_bytes = socket.inet_aton(args.local_ip)
|
||||
|
||||
print(f"\n{'=' * 60}\nDNS PCAP PREPROCESSOR\n{'=' * 60}")
|
||||
print(f"Local IP: {args.local_ip}")
|
||||
print(f"Results: {args.results_dir}")
|
||||
|
||||
if not args.results_dir.exists():
|
||||
print(f"\n❌ Directory not found: {args.results_dir}")
|
||||
return
|
||||
|
||||
providers = ['adguard', 'cloudflare', 'google', 'quad9']
|
||||
|
||||
for provider in providers:
|
||||
provider_path = results_dir / provider
|
||||
if provider_path.exists():
|
||||
process_provider_directory(provider_path)
|
||||
|
||||
t0 = time.time()
|
||||
for provider in args.providers:
|
||||
path = args.results_dir / provider
|
||||
if path.exists():
|
||||
process_provider(path, local_ip_bytes)
|
||||
else:
|
||||
print(f"\n⚠ Warning: Provider directory not found: {provider}")
|
||||
|
||||
overall_time = time.time() - overall_start
|
||||
|
||||
print("\n" + "="*60)
|
||||
print(f"✓ PREPROCESSING COMPLETE")
|
||||
print(f" Total time: {overall_time:.2f}s ({overall_time/60:.1f} minutes)")
|
||||
print("="*60 + "\n")
|
||||
print(f"\n⚠ Missing provider directory: {provider}")
|
||||
|
||||
total = time.time() - t0
|
||||
print(f"\n{'=' * 60}")
|
||||
print(f"✓ DONE in {total:.2f}s ({total / 60:.1f} min)")
|
||||
print(f"{'=' * 60}\n")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
Reference in New Issue
Block a user