#!/usr/bin/env python3 """C2 protocol analyzer — identify and reverse engineer malware C2 communications.""" from __future__ import annotations import argparse import hashlib import json import math import struct import sys from collections import Counter, defaultdict from pathlib import Path def calculate_entropy(data: bytes) -> float: """Calculate Shannon entropy of byte data.""" if not data: return 0.0 freq = Counter(data) length = len(data) return -sum((c / length) * math.log2(c / length) for c in freq.values()) def analyze_beacon_intervals(timestamps: list[float]) -> dict: """Detect beaconing patterns from connection timestamps.""" if len(timestamps) < 3: return {"beaconing_detected": False, "reason": "insufficient_samples"} intervals = [timestamps[i + 1] - timestamps[i] for i in range(len(timestamps) - 1)] avg_interval = sum(intervals) / len(intervals) std_dev = (sum((i - avg_interval) ** 2 for i in intervals) / len(intervals)) ** 0.5 jitter_pct = (std_dev / avg_interval * 100) if avg_interval > 0 else 0 return { "beaconing_detected": std_dev < avg_interval * 0.5, "average_interval_seconds": round(avg_interval, 2), "std_deviation": round(std_dev, 2), "jitter_percentage": round(jitter_pct, 1), "sample_count": len(timestamps), } def detect_dns_tunneling(queries: list[dict]) -> dict: """Analyze DNS queries for tunneling indicators.""" results = { "tunneling_detected": False, "suspicious_domains": [], "indicators": {}, } domain_queries = defaultdict(list) for q in queries: parts = q.get("name", "").split(".") if len(parts) >= 3: base_domain = ".".join(parts[-2:]) subdomain = ".".join(parts[:-2]) domain_queries[base_domain].append(subdomain) for domain, subdomains in domain_queries.items(): avg_len = sum(len(s) for s in subdomains) / max(len(subdomains), 1) unique_ratio = len(set(subdomains)) / max(len(subdomains), 1) avg_entropy = sum(calculate_entropy(s.encode()) for s in subdomains) / max(len(subdomains), 1) if avg_len > 30 or (unique_ratio > 0.9 and len(subdomains) > 10) or avg_entropy > 3.5: results["tunneling_detected"] = True results["suspicious_domains"].append({ "domain": domain, "query_count": len(subdomains), "avg_subdomain_length": round(avg_len, 1), "unique_ratio": round(unique_ratio, 2), "avg_entropy": round(avg_entropy, 2), }) return results def compute_ja3_hash(client_hello: bytes) -> str: """Compute JA3 hash from a TLS ClientHello message. This is a simplified implementation — production use should leverage a proper TLS parser like pyshark or dpkt. """ # Placeholder: actual JA3 requires parsing TLS version, cipher suites, # extensions, elliptic curves, and EC point formats return hashlib.md5(client_hello).hexdigest() def detect_dga(domains: list[str]) -> dict: """Detect domain generation algorithm patterns in DNS queries.""" if not domains: return {"dga_detected": False} nxdomain_domains = [d for d in domains if d] # In practice, filter by NXDOMAIN entropies = [] lengths = [] for domain in nxdomain_domains: sld = domain.split(".")[0] entropies.append(calculate_entropy(sld.encode())) lengths.append(len(sld)) avg_entropy = sum(entropies) / max(len(entropies), 1) avg_length = sum(lengths) / max(len(lengths), 1) tld_counter = Counter(d.split(".")[-1] for d in nxdomain_domains if "." in d) return { "dga_detected": avg_entropy > 3.0 and len(nxdomain_domains) > 20, "avg_sld_entropy": round(avg_entropy, 2), "avg_sld_length": round(avg_length, 1), "unique_tlds": dict(tld_counter.most_common(10)), "sample_count": len(nxdomain_domains), } def analyze_protocol_structure(stream_data: bytes) -> dict: """Analyze raw TCP stream for protocol structure patterns.""" if len(stream_data) < 8: return {"error": "insufficient data"} # Look for magic bytes (repeated patterns at packet boundaries) magic_candidates = [] for width in (2, 4): header = stream_data[:width] count = 0 for i in range(0, len(stream_data) - width, width): if stream_data[i:i + width] == header: count += 1 if count > 3: magic_candidates.append({"bytes": header.hex(), "width": width, "occurrences": count}) # Check for length-prefixed messages length_field = None for offset in (0, 2, 4): if offset + 2 <= len(stream_data): val_be = struct.unpack_from(">H", stream_data, offset)[0] val_le = struct.unpack_from(" None: parser = argparse.ArgumentParser(description="C2 Protocol Analyzer") parser.add_argument("--input", "--pcap", dest="pcap", help="Path to PCAP file for analysis") parser.add_argument( "--mode", choices=["identify", "beacon", "dns-tunnel", "cobalt-strike", "ja3", "dga", "protocol", "emulate"], default="identify", help="Analysis mode", ) parser.add_argument("--stream", type=int, help="TCP stream index for protocol analysis") parser.add_argument("--protocol-spec", help="Protocol spec JSON for emulation mode") parser.add_argument("--listen-port", type=int, help="Port for C2 emulation") parser.add_argument("--output", default="c2_analysis.json", help="Output file path") parser.add_argument("--format", choices=["json", "csv", "markdown"], default="json") args = parser.parse_args() if not args.pcap and args.mode != "emulate": parser.error("--pcap is required for analysis modes") results = { "mode": args.mode, "source": args.pcap, "c2_channels": [], "dns_tunneling": {"detected": False}, "dga": {"detected": False}, "domain_fronting": {"detected": False}, "iocs": {"c2_domains": [], "c2_ips": [], "ja3_hashes": [], "uri_patterns": []}, } # Mode-specific analysis would integrate with pyshark/dpkt here print(f"[*] C2 Protocol Analysis — mode: {args.mode}") print(f"[*] Source: {args.pcap}") if args.mode == "emulate": print(f"[*] Starting C2 emulator on port {args.listen_port}") print("[!] Emulation mode requires manual protocol specification") else: print(f"[*] Analysis complete. Results written to {args.output}") output_path = Path(args.output) output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text(json.dumps(results, indent=2)) if __name__ == "__main__": main()