#!/usr/bin/env python3 """Analyze captured C2 communications from PCAP or log files. Identifies common C2 protocol patterns in HTTP, DNS, and raw TCP/UDP traffic. """ from __future__ import annotations import argparse import json import re import sys from collections import Counter, defaultdict from datetime import datetime from pathlib import Path try: from scapy.all import rdpcap, IP, TCP, UDP, DNS, DNSQR, Raw HAS_SCAPY = True except ImportError: HAS_SCAPY = False def analyze_http_c2(packets) -> list: """Analyze HTTP traffic for C2 indicators.""" findings = [] http_sessions = defaultdict(list) for pkt in packets: if not pkt.haslayer(TCP) or not pkt.haslayer(Raw): continue try: payload = pkt[Raw].load.decode("utf-8", errors="ignore") except Exception: continue dst = pkt[IP].dst if pkt.haslayer(IP) else "unknown" # HTTP request detection http_match = re.match(r"(GET|POST|PUT)\s+(\S+)\s+HTTP", payload) if http_match: method = http_match.group(1) uri = http_match.group(2) http_sessions[dst].append({ "method": method, "uri": uri, "timestamp": float(pkt.time), }) # Check for suspicious patterns suspicious = False reasons = [] # Base64 in URI if re.search(r"[A-Za-z0-9+/]{20,}={0,2}", uri): suspicious = True reasons.append("Base64-like data in URI") # Long random-looking URI if len(uri) > 100: suspicious = True reasons.append(f"Very long URI ({len(uri)} chars)") # Unusual file extensions if re.search(r"\.(php|asp|aspx|jsp)\?[a-z]=[A-Za-z0-9]{32,}", uri): suspicious = True reasons.append("Encoded parameter to script endpoint") if suspicious: findings.append({ "type": "suspicious_http", "severity": "high", "destination": dst, "method": method, "uri": uri[:200], "reasons": reasons, }) # Detect beaconing in HTTP sessions for dst, requests in http_sessions.items(): if len(requests) < 3: continue timestamps = sorted([r["timestamp"] for r in requests]) intervals = [timestamps[i + 1] - timestamps[i] for i in range(len(timestamps) - 1)] if intervals: avg = sum(intervals) / len(intervals) std = (sum((x - avg) ** 2 for x in intervals) / len(intervals)) ** 0.5 jitter = std / avg if avg > 0 else 0 if jitter < 0.3 and avg < 600: # Low jitter, regular intervals under 10min findings.append({ "type": "http_beaconing", "severity": "critical", "destination": dst, "avg_interval_seconds": round(avg, 1), "jitter": round(jitter, 3), "request_count": len(requests), }) return findings def analyze_dns_c2(packets) -> list: """Analyze DNS traffic for C2 indicators.""" findings = [] dns_queries = defaultdict(list) for pkt in packets: if not pkt.haslayer(DNS) or not pkt.haslayer(DNSQR): continue qname = pkt[DNSQR].qname.decode("utf-8", errors="ignore").rstrip(".") dns_queries[qname].append(float(pkt.time)) # Check for DNS tunneling indicators parts = qname.split(".") if len(parts) >= 3: subdomain = parts[0] # Long subdomain (data exfiltration/tunneling) if len(subdomain) > 30: findings.append({ "type": "dns_tunneling_suspect", "severity": "high", "domain": qname, "subdomain_length": len(subdomain), "reason": "Unusually long subdomain (possible DNS tunneling)", }) # High entropy subdomain if len(subdomain) > 10: chars = Counter(subdomain.lower()) entropy = -sum( (c / len(subdomain)) * (c / len(subdomain)) and __import__("math").log2(c / len(subdomain)) * (c / len(subdomain)) for c in chars.values() if c > 0 ) if len(subdomain) > 0 else 0 # High volume queries to single domain (tunneling/DGA) domain_counts = Counter() for qname in dns_queries: parts = qname.split(".") if len(parts) >= 2: base_domain = ".".join(parts[-2:]) domain_counts[base_domain] += 1 for domain, count in domain_counts.most_common(10): if count > 50: findings.append({ "type": "high_volume_dns", "severity": "medium", "domain": domain, "query_count": count, "reason": "High volume DNS queries (possible tunneling or DGA)", }) return findings def analyze_raw_connections(packets) -> list: """Analyze raw TCP/UDP connections for C2 patterns.""" findings = [] connections = defaultdict(lambda: {"packets": 0, "bytes": 0, "timestamps": []}) for pkt in packets: if not pkt.haslayer(IP): continue src = pkt[IP].src dst = pkt[IP].dst if pkt.haslayer(TCP): dport = pkt[TCP].dport proto = "TCP" elif pkt.haslayer(UDP): dport = pkt[UDP].dport proto = "UDP" else: continue key = f"{dst}:{dport}/{proto}" connections[key]["packets"] += 1 connections[key]["bytes"] += len(pkt) connections[key]["timestamps"].append(float(pkt.time)) # Find suspicious high-port connections for key, data in connections.items(): dst_port = int(key.split(":")[1].split("/")[0]) # Unusual ports common_ports = {80, 443, 53, 25, 110, 143, 993, 995, 587, 8080, 8443} if dst_port not in common_ports and data["packets"] > 10: findings.append({ "type": "unusual_port_activity", "severity": "medium", "connection": key, "packets": data["packets"], "bytes": data["bytes"], }) return findings def analyze_pcap(pcap_path) -> dict: """Main PCAP analysis function.""" if not HAS_SCAPY: return { "error": "scapy not installed. Install with: pip install scapy", "findings": [] } try: packets = rdpcap(pcap_path) except Exception as e: return {"error": f"Failed to read PCAP: {e}", "findings": []} findings = [] findings.extend(analyze_http_c2(packets)) findings.extend(analyze_dns_c2(packets)) findings.extend(analyze_raw_connections(packets)) return { "pcap_file": str(pcap_path), "total_packets": len(packets), "findings": findings, } def main() -> None: parser = argparse.ArgumentParser( description="Analyze PCAP files for C2 communication patterns" ) parser.add_argument("--input", "--pcap", required=True, help="PCAP file to analyze") parser.add_argument("--output", "-o", help="Output file (JSON)") parser.add_argument("--format", choices=["json", "text"], default="text") args = parser.parse_args() if not HAS_SCAPY: print("[!] scapy is required. Install: pip install scapy", file=sys.stderr) sys.exit(1) result = analyze_pcap(args.pcap) result["timestamp"] = datetime.now().isoformat() result["tool"] = "c2_protocol_analyzer" if args.format == "json": output = json.dumps(result, indent=2) else: output = f"=== C2 Protocol Analysis: {args.pcap} ===\n" output += f"Packets analyzed: {result.get('total_packets', 0)}\n" output += f"Findings: {len(result.get('findings', []))}\n\n" for f in result.get("findings", []): sev = f.get("severity", "info").upper() output += f"[{sev}] {f['type']}: {json.dumps({k: v for k, v in f.items() if k not in ('type', 'severity')})}\n" if args.output: Path(args.output).write_text(output) print(f"[+] Report saved to {args.output}") else: print(output) if __name__ == "__main__": main()