SOC PCAP Traffic Analysis & Beacon Detection (Python)

1

Environment Setup & Python Packet Libraries

Install scapy, pyshark, and pandas to process packet captures natively in Python.

Terminal - Setup Dependencies

pip install scapy pyshark pandas numpy matplotlib

PYTHON • test_imports.py

import scapy.all as scapy
import pandas as pd
import pyshark

print(f"Scapy Version: {scapy.VERSION}")
print(f"Pandas Version: {pd.__version__}")

2

Parsing PCAP Dumps & Extracting IP Layer Metadata

Write a Scapy parser function to read PCAP packet dumps and extract source IPs, destination IPs, ports, and timestamp offsets.

PYTHON • pcap_parser.py

from scapy.all import rdpcap, IP, TCP, UDP
import pandas as pd

def parse_pcap(file_path: str) -> pd.DataFrame:
    packets = rdpcap(file_path)
    records = []

    for pkt in packets:
        if pkt.haslayer(IP):
            record = {
                "timestamp": float(pkt.time),
                "src_ip": pkt[IP].src,
                "dst_ip": pkt[IP].dst,
                "protocol": pkt[IP].proto,
                "length": len(pkt),
                "src_port": pkt[TCP].sport if pkt.haslayer(TCP) else (pkt[UDP].sport if pkt.haslayer(UDP) else 0),
                "dst_port": pkt[TCP].dport if pkt.haslayer(TCP) else (pkt[UDP].dport if pkt.haslayer(UDP) else 0),
            }
            records.append(record)

    return pd.DataFrame(records)

3

Extracting TLS Server Name Indication (SNI) & DNS Exfiltration

Inspect TLS Handshake Extensions for Server Name Indication (SNI) hostnames and extract DNS query domain strings to detect subdomains used for C2 tunneling.

PYTHON • dns_tls_analyzer.py

from scapy.all import rdpcap, DNS, DNSQR
from typing import List, Dict

def extract_dns_activity(pcap_path: str) -> List[Dict]:
    packets = rdpcap(pcap_path)
    dns_records = []

    for pkt in packets:
        if pkt.haslayer(DNS) and pkt.haslayer(DNSQR):
            query_name = pkt[DNSQR].qname.decode('utf-8', errors='ignore')
            dns_records.append({
                "time": float(pkt.time),
                "query": query_name.rstrip('.'),
                "length": len(query_name)
            })

    return dns_records

4

Statistical C2 Beaconing Jitter Analysis in Python

Malware beacons communicate back to Command & Control (C2) servers on fixed interval timers with slight randomized jitter. Calculate variance and standard deviation of inter-arrival times.

PYTHON • beacon_detector.py

import numpy as np
import pandas as pd

def analyze_beaconing(df: pd.DataFrame, ip_pair: tuple) -> dict:
    src_ip, dst_ip = ip_pair
    sub_df = df[(df["src_ip"] == src_ip) & (df["dst_ip"] == dst_ip)].sort_values("timestamp")

    if len(sub_df) < 5:
        return {"is_beacon": False, "reason": "Insufficient packet count"}

    timestamps = sub_df["timestamp"].values
    time_deltas = np.diff(timestamps)

    mean_delta = np.mean(time_deltas)
    std_delta = np.std(time_deltas)
    score = std_delta / (mean_delta + 1e-5)

    is_beacon = score < 0.2
    return {
        "is_beacon": is_beacon,
        "mean_interval_sec": round(mean_delta, 2),
        "score": round(score, 4)
    }

5

Automated Snort / Suricata Signature Generator in Python

Convert detected suspicious IP endpoints and DNS domains into production Snort alert signatures for automated SIEM deployment.

PYTHON • snort_exporter.py

def generate_snort_rule(c2_ip: str, c2_port: int = 443, sid: int = 1000888) -> str:
    rule = (
        f'alert tcp $HOME_NET any -> {c2_ip} {c2_port} ('
        f'msg:"ECHO-SOC Malicious C2 Beacon IP Detected"; '
        f'flow:to_server,established; '
        f'sid:{sid}; rev:1;)'
    )
    return rule

6

Complete End-to-End Threat Hunting Pipeline Script

Run this complete Python script to load mock packet flows, detect beaconing pairs, and output a SOC Incident Summary.

PYTHON • run_soc_threat_hunter.py

import pandas as pd
from beacon_detector import analyze_beaconing
from snort_exporter import generate_snort_rule

def execute_threat_hunt():
    print("=== STARTING SOC PCAP THREAT HUNTING PIPELINE ===")
    packets = [
        {"timestamp": 100.0, "src_ip": "10.0.0.15", "dst_ip": "198.51.100.4"},
        {"timestamp": 130.1, "src_ip": "10.0.0.15", "dst_ip": "198.51.100.4"},
        {"timestamp": 160.2, "src_ip": "10.0.0.15", "dst_ip": "198.51.100.4"},
        {"timestamp": 190.0, "src_ip": "10.0.0.15", "dst_ip": "198.51.100.4"},
        {"timestamp": 220.1, "src_ip": "10.0.0.15", "dst_ip": "198.51.100.4"},
    ]
    df = pd.DataFrame(packets)

    pair = ("10.0.0.15", "198.51.100.4")
    res = analyze_beaconing(df, pair)

    print(f"\n[+] Analyzing Flow {pair[0]} -> {pair[1]}:")
    print(f"    * Is C2 Beacon: {res['is_beacon']}")
    print(f"    * Variance Score: {res['score']}")

if __name__ == "__main__":
    execute_threat_hunt()

Expected Terminal Output

=== STARTING SOC PCAP THREAT HUNTING PIPELINE ===
[+] Analyzing Flow 10.0.0.15 -> 198.51.100.4:
* Is C2 Beacon: True
* Variance Score: 0.0028
=== THREAT HUNTING PIPELINE COMPLETE ===