Analisando Comunicação de Comando e Controle (C2)

Quando Usar

• A engenharia reversa de uma amostra de malware revelou comunicação de rede que precisa de análise de protocolo
• Construindo assinaturas de detecção em nível de rede para um framework C2 específico (Cobalt Strike, Metasploit, Sliver)
• Mapeando infraestrutura C2, incluindo servidores primários, domínios de fallback e "dead drops" (locais de troca de dados)
• Analisando tráfego C2 criptografado ou codificado para entender o conjunto de comandos e o formato dos dados
• Atribuindo malware a um ator de ameaça com base em padrões de infraestrutura C2 e ferramentas

Não use para detecção geral de anomalias de rede; isso é especificamente para entender protocolos C2 conhecidos ou suspeitos a partir da análise de malware.

Pré-requisitos

• Captura PCAP de tráfego de rede de malware (de sandbox, "network tap" ou captura de pacotes completa)
• Wireshark/tshark para análise em nível de pacote
• Ferramentas de engenharia reversa (Ghidra, dnSpy) para entender o código C2 no binário do malware
• Python 3.8+ com scapy, dpkt e requests para análise e reprodução de protocolo
• Bancos de dados de inteligência de ameaças para correlação de infraestrutura C2 (VirusTotal, Shodan, Censys)
• Bancos de dados de impressão digital JA3/JA3S para identificação de C2 baseada em TLS

Fluxo de Trabalho

Passo 1: Identificar o Canal C2

Determine o protocolo e o transporte usados para a comunicação C2:

C2 Communication Channels:
━━━━━━━━━━━━━━━━━━━━━━━━━
HTTP/HTTPS:     Most common; uses standard web traffic to blend in
                Indicators: Regular POST/GET requests, specific URI patterns, custom headers

DNS:            Tunneling data through DNS queries and responses
                Indicators: High-volume TXT queries, long subdomain names, high entropy

Custom TCP/UDP: Proprietary binary protocol on non-standard port
                Indicators: Non-HTTP traffic on high ports, unknown protocol

ICMP:           Data encoded in ICMP echo/reply payloads
                Indicators: ICMP packets with large or non-standard payloads

WebSocket:      Persistent bidirectional connection for real-time C2
                Indicators: WebSocket upgrade followed by binary frames

Cloud Services: Using legitimate APIs (Telegram, Discord, Slack, GitHub)
                Indicators: API calls to cloud services from unexpected processes

Email:          SMTP/IMAP for C2 commands and data exfiltration
                Indicators: Automated email operations from non-email processes

Passo 2: Analisar Padrão de Beacon

Caracterize o padrão de comunicação periódico:

from scapy.all import rdpcap, IP, TCP
from collections import defaultdict
import statistics
import json

packets = rdpcap("c2_traffic.pcap")

# Group TCP SYN packets by destination
connections = defaultdict(list)
for pkt in packets:
    if IP in pkt and TCP in pkt and (pkt[TCP].flags & 0x02):
        key = f"{pkt[IP].dst}:{pkt[TCP].dport}"
        connections[key].append(float(pkt.time))

# Analyze each destination for beaconing
for dst, times in sorted(connections.items()):
    if len(times) < 3:
        continue

    intervals = [times[i+1] - times[i] for i in range(len(times)-1)]
    avg_interval = statistics.mean(intervals)
    stdev = statistics.stdev(intervals) if len(intervals) > 1 else 0
    jitter_pct = (stdev / avg_interval * 100) if avg_interval > 0 else 0
    duration = times[-1] - times[0]

    beacon_data = {
        "destination": dst,
        "connections": len(times),
        "duration_seconds": round(duration, 1),
        "avg_interval_seconds": round(avg_interval, 1),
        "stdev_seconds": round(stdev, 1),
        "jitter_percent": round(jitter_pct, 1),
        "is_beacon": 5 < avg_interval < 7200 and jitter_pct < 25,
    }

    if beacon_data["is_beacon"]:
        print(f"[!] BEACON DETECTED: {dst}")
        print(f"    Interval: {avg_interval:.0f}s +/- {stdev:.0f}s ({jitter_pct:.0f}% jitter)")
        print(f"    Sessions: {len(times)} over {duration:.0f}s")

Passo 3: Decodificar Estrutura do Protocolo C2

Faça engenharia reversa do formato da mensagem a partir do tráfego capturado:

# Análise de protocolo C2 baseado em HTTP
import dpkt
import base64

with open("c2_traffic.pcap", "rb") as f:
    pcap = dpkt.pcap.Reader(f)

for ts, buf in pcap:
    eth = dpkt.ethernet.Ethernet(buf)
    if not isinstance(eth.data, dpkt.ip.IP):
        continue
    ip = eth.data
    if not isinstance(ip.data, dpkt.tcp.TCP):
        continue
    tcp = ip.data

    if tcp.dport == 80 or tcp.dport == 443:
        if len(tcp.data) > 0:
            try:

…