Skip to content

C2 Protocol Analysis

C2 (command-and-control) protocol analysis is the reverse-engineering of how malware communicates with its operator’s infrastructure: the transport and protocol used, the beacon timing model, the encoding and encryption of tasking and exfiltration, and the profile the implant presents on the wire. Where string extraction (M5) tells you what a sample contains, protocol analysis tells you how it behaves after detonation — the periodic POST to /api/v2/gate with a base64 cookie, the DNS TXT lookups that shuttle commands in subdomain labels, the jitter pattern that makes beaconing look like human traffic. The output of the analysis is threefold: decode capability (read the channel), IOC generation (block and hunt on the infrastructure), and detection logic (signature the pattern of the communication so blocking one domain is not the whole answer).

Every implant must talk to somebody, and the channel is both the attacker’s lifeline and their largest detection surface. Blocking C2 is operationally the highest-value containment action available — acontained implant without a channel is a broken tool, which is why red teams obsess over channel resilience and blue teams over channel detection. Protocol analysis is also how analysts recover tasking: what did the operator actually do during the intrusion — which commands, which credentials exfiltrated, which lateral targets — answers that live in the recorded traffic, not in the binary. For detection engineering, the timing profile (sleep + jitter) and size profile of beacons produce statistical signatures that survive domain rotation and even protocol substitution, making them the most durable network detections available (E17). And for threat intelligence (M15), protocol quirks — a fixed URI path, a distinctive cookie format, a magic value in the first bytes — cluster samples into campaigns faster than any binary feature, because operators reuse protocol templates across builds and even across tool families.

An implant’s communication cycle has four phases: check-in (beacon out), task pickup (receive commands), execution, and result return (exfiltrate). Each phase can ride any transport. The classic model is HTTP(S) polling:

IMPLANT C2 SERVER (e.g. c2.evil-gate.top)
| |
|--- GET /px.gif?id=BASE64 ----->| check-in (random 30-120s + jitter)
|<-- 200, Set-Cookie: TASK=... --| tasking in response/cookie
| |
|--- POST /upload (result) ----->| exfil, then sleep again
| |
timing profile: t0, t0+sleep1, t0+sleep1+sleep2 ...
jitter makes intervals non-uniform but bounded
size profile: check-ins ~fixed small size, results variable

Variants multiply by transport: DNS (commands in TXT answers, data out in subdomain labels — hex-encoded chunks of 30-60 bytes), ICMP (payload in echo data fields), HTTPS with domain fronting, raw TCP with custom binary protocols, and third-party channels (Telegram bots, Pastebin, GitHub gists — implant polls a legit service the perimeter will never block). Defensive analysis decomposes the channel in layers: transport profile (JA3/JA4 TLS fingerprints, cipher order — Cobalt Strike’s default Java TLS stack is instantly distinguishable from Chrome’s [verify current defaults]), protocol semantics (URI paths, header order, cookie names — malleable C2 profiles make these look like anything, but the set is fixed per profile), encoding (base64 → XOR → AES with key derived from a nonce in the packet — peel layer by layer), and timing (mean interval, jitter distribution, size autocorrelation).

To decode a custom protocol: capture traffic from a lab detonation, locate the smallest repeated exchange (the check-in), diff multiple check-ins to find the static envelope (headers, magic bytes) versus the variable payload, then attack the payload — try base64, then single-byte XOR (crib: HTTP inside), then look for AES-CTR patterns (high-entropy body with a 16-byte aligned structure and a nonce field). Session keys derived from a nonce sent in the clear are recoverable; keys fetched from the server’s first response require MITM of the lab environment (hook SSL_read/SSL_write or run mitmproxy as an explicit proxy if the implant honors one).

TechniqueWhat you look atToolingNotes
HTTP beacon statInter-request deltas: mean/σ, size uniformitytshark, zeek, custom Pythonσ/mean < 0.3 with regularity = beacon [verify threshold]
DNS tunnelingSubdomain label entropy, length, NXDOMAIN ratiozeek dns.log, customLabels >30 chars, entropy >3.5 suspicious
JA3/JA4 fingerprintTLS ClientHello hashzeek ja3, suricata tls.logMatch against known-tool fingerprint sets
Malleable profile carvingURI/header/cookie template diffingmultiple check-insThe fixed set identifies the profile
Cookie/param decodingbase64 → XOR → structured fieldsCyberChef, customImplant ID + campaign ID usually cleartext-ish
ICMP payload inspectionEcho data field size/contenttshark -T fields -e data.data>32-byte echo payloads are rare on LANs
Third-party C2 pollingRequests to legit APIs from odd processeszeek http.log + EDR correlationgithub/pastebin/telegram polling from cmd.exe
Long-poll detectionServer-push connections held openflow duration analysisDistinguishes from polling beacons
Beacon replay/C2 emulationRedirect C2 to listener, capture full protocolINetSim, FakeNet, custom fake C2Full decode without the real operator

beacon_hunter.py — extracts HTTP/DNS flows from a PCAP, computes per-host inter-arrival statistics, and scores periodicity to flag beaconing hosts. Dependency: scapy (pip install scapy). Runs on any PCAP from a lab detonation or sensor span port.

#!/usr/bin/env python3
"""beacon_hunter.py — periodicity scoring for HTTP/DNS flows in a PCAP.
Usage: python beacon_hunter.py capture.pcap [--top 20]
Deps: scapy.
"""
import argparse, math, sys
from collections import defaultdict
from scapy.all import rdpcap, IP, TCP, UDP, DNS, DNSQR
def stats(times):
"""Inter-arrival stats for a sorted event list."""
times = sorted(times)
if len(times) < 4:
return None
deltas = [b - a for a, b in zip(times, times[1:])]
n = len(deltas)
mean = sum(deltas) / n
var = sum((d - mean) ** 2 for d in deltas) / n
sd = math.sqrt(var)
return {"count": len(times), "span": times[-1] - times[0],
"mean": mean, "sd": sd,
"cv": sd / mean if mean else 0.0,
"min": min(deltas), "max": max(deltas)}
def beacon_score(st):
"""0-100 score: low CV + enough samples + regular span."""
if st is None:
return 0
s = 0
if st["cv"] < 0.10:
s += 50
elif st["cv"] < 0.30:
s += 30
elif st["cv"] < 0.60:
s += 10
if st["count"] >= 10:
s += 20
elif st["count"] >= 6:
s += 10
if 10 <= st["mean"] <= 3600: # 10s..1h beacon windows
s += 20
if st["span"] >= 300: # observed >= 5 minutes
s += 10
return min(s, 100)
def extract_flows(pcap_path):
"""Collect per-(dst,port) event lists for HTTP/DNS traffic."""
events = defaultdict(list)
meta = defaultdict(lambda: {"uris": set(), "qnames": set(),
"uas": set(), "sizes": []})
pkts = rdpcap(pcap_path)
for pkt in pkts:
if IP not in pkt:
continue
ip = pkt[IP]
if TCP in pkt and pkt[TCP].dport in (80, 8080, 443):
key = (ip.dst, pkt[TCP].dport, "tcp")
events[key].append(float(pkt.time))
load = bytes(pkt[TCP].payload)
if load.startswith((b"GET ", b"POST ")):
try:
head = load.split(b"\r\n")
meta[key]["uris"].add(
head[0].split(b" ")[1].decode(errors="replace"))
for line in head[1:]:
if line.lower().startswith(b"user-agent:"):
meta[key]["uas"].add(
line.split(b":", 1)[1].strip().decode(
errors="replace"))
except IndexError:
pass
meta[key]["sizes"].append(len(load))
elif UDP in pkt and pkt[UDP].dport == 53:
key = (ip.dst, 53, "dns")
events[key].append(float(pkt.time))
if DNS in pkt and pkt[DNS].qr == 0 and DNSQR in pkt:
qn = pkt[DNSQR].qname.decode(errors="replace").rstrip(".")
meta[key]["qnames"].add(qn)
return events, meta
def label_flow(key, meta):
dst, port, proto = key
m = meta[key]
tag = ""
if proto == "tcp" and m["uris"]:
tag = f"uri={sorted(m['uris'])[0][:60]}"
elif proto == "dns" and m["qnames"]:
longest = max(m["qnames"], key=len)
tag = f"q={longest[:60]}"
return f"{dst}:{port}/{proto} {tag}".strip()
def main():
ap = argparse.ArgumentParser()
ap.add_argument("pcap")
ap.add_argument("--top", type=int, default=20)
ap.add_argument("--min-score", type=int, default=60)
args = ap.parse_args()
events, meta = extract_flows(args.pcap)
if not events:
sys.exit("no HTTP/DNS flows found")
scored = []
for key, times in events.items():
st = stats(times)
sc = beacon_score(st)
if sc >= args.min_score:
scored.append((sc, key, st))
scored.sort(reverse=True)
print(f"== beacon candidates (score >= {args.min_score}) ==")
for sc, key, st in scored[:args.top]:
print(f"[{sc:>3}] {label_flow(key, meta)}")
print(f" n={st['count']} span={st['span']:.0f}s "
f"mean={st['mean']:.1f}s sd={st['sd']:.1f} "
f"cv={st['cv']:.2f} range=[{st['min']:.0f},{st['max']:.0f}]")
if not scored:
print("[-] nothing periodic enough; lower --min-score "
"or check for long-poll / third-party channels")
if __name__ == "__main__":
main()
  • Lines 12-24 compute the core statistics: sorted timestamps → inter-arrival deltas → mean, standard deviation, and the coefficient of variation (σ/mean). CV is the single most discriminating beacon feature: fixed-sleep implants land at CV < 0.1, jittered ones at 0.1-0.6, human browsing scatters above 1.0.
  • Lines 27-45 convert statistics into a 0-100 score with weighted bands: CV dominates (up to 50), sample count guards against tiny-flow coincidence (up to 20), plausible beacon window 10 s-1 h (up to 20) and ≥5-minute observation span (up to 10). Every threshold is a tunable documented in the report — calibrate on your own baseline traffic before trusting the defaults.
  • Lines 48-79 parse the PCAP: TCP flows to 80/8080/443 and UDP/53 events are collected per (destination, port, protocol) key. HTTP method lines are minimally parsed for URI and User-Agent (lines 62-72) — enough for labeling, deliberately not a full HTTP parser.
  • Lines 87-101 produce the report: score-sorted candidates with timing stats and a URI/qname label. The [-] fallback message (line 103) names the next analytic moves when nothing scores — long-poll channels and third-party polling are the usual reasons for a silent run.

From the implant’s side, the channel is engineered for resilience: redundant domains baked into config (DGA or hardcoded lists), fallback transports (HTTP → DNS → ICMP), and jittered timing to blend with browser traffic. From the analyst’s side, the workflow starts at detonation: run the sample in the lab with full capture (INetSim or a controlled NAT with a mirrored tap), collect the PCAP, and run beacon analysis on it — this yields the check-in template, timing profile, and initial IOC set. The same analysis then runs against production telemetry: the timing signature hunted across proxy/DNS logs (E17, BT7) finds hosts already infected that you never had a sample for, because the beacon pattern is family-stable even as domains rotate. Decoded tasking flows back into the IR timeline: every command the operator sent is visible in the recovered channel, giving the after-action report actual attacker actions rather than inferred ones. The intelligence cycle completes when the extracted protocol fingerprint (URI path set, cookie format, TLS stack) is registered into the M15 clustering system, where it links this campaign to every other build sharing the profile.

  1. Detonation — lab run of loader sample with full PCAP capture behind INetSim
  2. First pass — beacon_hunter.py flags 185.220.101.4:80 at score 92, CV 0.08
  3. Template — 14 check-ins diffed: fixed /px.gif, cookie cfu=BASE64, UA constant
  4. Peel — cookie is base64(XOR(payload, 0x3D)); inner: id|sleep|cmd-list
  5. Emulate — fake C2 written from the template feeds whoami task to implant
  6. Confirm — implant executes task and POSTs result to fake C2: protocol fully understood
  7. Hunt — timing+size profile queried across proxy logs: 3 additional hosts beaconing
  8. Feed — profile registered in M15 clustering; links 9 prior samples sharing the check-in template

dnstunnel.py — a DNS tunneling detector for zeek-style DNS logs or PCAPs: scores domains by subdomain label length/entropy, NXDOMAIN ratio, and query volume, then separates tunnel traffic from normal lookups. DNS tunneling is the most common “quiet” channel in restricted networks and the one most often missed by HTTP-focused tooling.

  • Score per registered domain, not per query — tunneling is a property of the aggregate (hundreds of unique long labels under one zone), invisible when analyzing single requests.
  • Entropy on the label sequence, not the whole qname — the distinctive signal is long random strings in the leftmost labels; TLD and zone names dilute the metric.
  • MX/TXT query-type weighting — tunnels favor TXT (data-in-answers) and NULL/MX record types; A-record-heavy zones scoring high indicates DGA instead — a different problem with a different response.
  • Pluggable input: accept zeek dns.log TSV (the common sensor format) so the tool runs directly on existing telemetry.
#!/usr/bin/env python3
"""dnstunnel.py — DNS tunneling detector over zeek dns.log or PCAP.
Usage: python dnstunnel.py dns.log [--top 15]
python dnstunnel.py --pcap c.pcap
zeek dns.log TSV expected; PCAP mode needs scapy.
"""
import argparse, math, sys
from collections import defaultdict, Counter
RECORD_TYPES = {1: "A", 16: "TXT", 28: "AAAA", 6: "SOA",
15: "MX", 10: "NULL", 5: "CNAME"}
def shannon(s):
if not s:
return 0.0
c = Counter(s)
n = len(s)
return -sum((v / n) * math.log2(v / n) for v in c.values())
def registered_domain(qname):
"""Crude registrable-domain split: last two labels (+cc 3-label)."""
labels = qname.rstrip(".").split(".")
cc3 = len(labels) >= 3 and labels[-1] in (
"uk", "au", "br", "za", "nz", "in", "jp", "cn")
multi = (len(labels) >= 3 and labels[-2] in ("com", "net", "org",
"gov", "edu") and len(labels[-1]) == 2)
if cc3 or multi:
return ".".join(labels[-3:])
return ".".join(labels[-2:]) if len(labels) >= 2 else qname
def score_domain(queries):
"""queries: list of (qname, qtype). Returns score dict."""
labels_all, qtypes = [], Counter()
unique_sub = set()
for qname, qtype in queries:
labels = qname.rstrip(".").split(".")
# leftmost labels are the tunnel payload zone
payload_labels = labels[:-2] if len(labels) > 2 else labels
for L in payload_labels:
labels_all.append(L)
unique_sub.add(L)
qtypes[RECORD_TYPES.get(qtype, str(qtype))] += 1
if not labels_all:
return None
lens = [len(L) for L in labels_all]
mean_len = sum(lens) / len(lens)
ent = shannon("".join(labels_all))
txt_ratio = (qtypes.get("TXT", 0) + qtypes.get("NULL", 0)
+ qtypes.get("MX", 0)) / sum(qtypes.values())
s = 0
if mean_len > 30:
s += 35
elif mean_len > 18:
s += 15
if ent > 3.5:
s += 25
elif ent > 3.0:
s += 10
if len(unique_sub) > 50:
s += 20
elif len(unique_sub) > 20:
s += 10
if txt_ratio > 0.5:
s += 20
elif txt_ratio > 0.2:
s += 10
return {"score": s, "queries": len(queries),
"unique_labels": len(unique_sub),
"mean_label_len": round(mean_len, 1),
"label_entropy": round(ent, 2),
"txt_mx_null_ratio": round(txt_ratio, 2),
"qtypes": dict(qtypes)}
def parse_zeek(path):
"""Yield (client, qname, qtype) from zeek dns.log TSV."""
fields = None
for line in open(path, encoding="utf-8", errors="replace"):
if line.startswith("#fields"):
fields = line.rstrip("\n").split("\t")[1:]
continue
if line.startswith("#") or not line.strip():
continue
parts = line.rstrip("\n").split("\t")
if not fields:
continue
try:
id_orig_h = parts[fields.index("id.orig_h")]
qname = parts[fields.index("query")]
qtype = int(parts[fields.index("qtype")])
yield id_orig_h, qname, qtype
except (ValueError, IndexError):
continue
def main():
ap = argparse.ArgumentParser()
ap.add_argument("log", nargs="?", help="zeek dns.log")
ap.add_argument("--top", type=int, default=15)
ap.add_argument("--min-score", type=int, default=70)
args = ap.parse_args()
if not args.log:
sys.exit("give me a zeek dns.log (PCAP mode: future work)")
zones = defaultdict(list)
for client, qname, qtype in parse_zeek(args.log):
zones[registered_domain(qname)].append((qname, qtype))
scored = []
for zone, queries in zones.items():
if len(queries) < 5:
continue
sc = score_domain(queries)
if sc and sc["score"] >= args.min_score:
scored.append((sc["score"], zone, sc))
scored.sort(reverse=True)
print(f"== DNS tunnel candidates (score >= {args.min_score}) ==")
for sc, zone, d in scored[:args.top]:
print(f"[{sc:>3}] {zone}")
print(f" q={d['queries']} uniq_labels={d['unique_labels']} "
f"mean_len={d['mean_label_len']} "
f"entropy={d['label_entropy']} "
f"txt_ratio={d['txt_mx_null_ratio']}")
if not scored:
print("[-] no tunnel-shaped zones found "
"(or volume below 5-query floor)")
if __name__ == "__main__":
main()
  • registered_domain() (lines 20-30) is deliberately crude but conservative: multi-label public suffixes get three labels, everything else two. False merges are acceptable — the scoring operates on aggregates and an occasional merged zone only changes denominators.
  • score_domain() (lines 33-65) is the detector brain: mean payload-label length (tunnels carry hex/base32 in 30-60-byte chunks), Shannon entropy over concatenated labels (random payload ≈ 3.7-4.0 bits/char for base32hex), unique-label count (tunnels never repeat), and TXT/NULL/MX ratio (data-return channels). Bands are additive; a real tunnel scores 80-100, a busy CDN zone scores under 40 because CDN labels are short and repetitive even when voluminous.
  • parse_zeek() (lines 72-92) locates the #fields header dynamically instead of hardcoding column positions — zeek’s field order shifts between versions, and hardcoding is the classic reason a sensor-side tool silently returns nothing.
  • The 5-query floor in main() (line 101) suppresses noise: tunneling is volume behavior, and one-off lookups cannot exhibit it.

simulate a DNS tunnel with dnscat2’s reference client (lab only)

Section titled “simulate a DNS tunnel with dnscat2’s reference client (lab only)”

$ git clone https://github.com/iagox86/dnscat2 [verify current repo]

$ ruby dnscat2.rb tunnel.lab.local —no-cache

$ python3 dnscat —dns server=10.0.0.1,port=53,host=tunnel.lab.local

capture the zone’s traffic into zeek dns.log, then:

Section titled “capture the zone’s traffic into zeek dns.log, then:”

$ python3 dnstunnel.py dns.log —min-score 60 [ 95] tunnel.lab.local q=214 uniq_labels=214 mean_len=48.2 entropy=3.81 txt_ratio=0.61

calibrate against normal browsing first — expect near-zero hits

Section titled “calibrate against normal browsing first — expect near-zero hits”

$ python3 dnstunnel.py benign_browsing_dns.log —min-score 60 [-] no tunnel-shaped zones found (or volume below 5-query floor)

Tunnel detection has a built-in precision trap: CDNs, software update services, and IoT telemetry all generate long high-entropy labels at volume. The discriminator set that holds up in production is length + uniqueness + TXT preference together — no single metric survives alone. Your own tooling posture: the detector reads logs offline, spawns nothing, and is invisible on endpoints; deploy it on the sensor host, not the analyst workstation. For the defensive counterpart at scale, register the scoring bands as scheduled hunts (zeek logs into your SIEM daily) rather than static alerts — tunnels appear gradually, one long label at a time, and a daily score trend catches what a single-day threshold misses. Wazuh mapping of the zeek pipeline varies by deployment [verify]; Suricata’s dns keyword plus dns.query; pcre local rules (sid:1000001+) catch the crudest tunnels but statistical detection belongs in the log layer.

  1. Add PCAP mode via scapy (the import scaffolding exists; wire the parser into the same zone aggregation).
  2. Track NXDOMAIN ratio per zone — dnscat-style tools probing for the authoritative server generate distinctive NX streaks.
  3. Emit per-client attribution: which internal host queried the tunnel zone most, from the zeek id.orig_h field.
  4. Add a known-good suffix list (publicsuffix) to fix the crude registered_domain() heuristic.
  5. Time-slice scoring: compute the score per hour instead of per file, so slow tunnels spread across days still surface.
SignalSourceIDNotes
Beacon periodicity (CV<0.3, n≥10)Proxy/zeek http.logstatistical hunt (E17/BT7)Most durable channel signature
Fixed URI template + constant UAProxy logscustom ruleTemplate set identifies C2 profile
JA3/JA4 known-tool fingerprintSuricata tls.log / zeekja3.hash matchingDefault stacks differ from browsers [verify lists]
DNS long-label entropy burstzeek dns.logdaily score hunt (tool above)Pair with TXT-ratio guard
Odd-process API pollingEDR + proxy correlationSysmon EID 3 + EID 22cmd.exe → pastebin = act now
7045 service w/ external hostWindows Security7045 + EID 3 flowReverse-shell persistence pattern
Custom/local Suricata SIDsNDRsid:1000001+ (local rules)Content-match on recovered URI paths
Wazuh decoy-channel alertsHIDS[verify rule IDs per deployment]Map zeek pipeline per environment

Channel evasion is an arms race in profile space. Malleable profiles (Cobalt Strike and imitators) let the operator reshape URIs, headers, cert info, and packing per engagement — defenses that signature one profile’s cosmetics die on the next engagement, which is why the timing and size statistics are the durable layer. Domain fronting and its modern successor encrypted-SNI abuse hide the true C2 host behind a reputable CDN — counters live in JA3+host-pair analysis, not in hostname blocklists. Legit-service channels (Telegram, GitHub, Dropbox) move the problem into trust space: the fix is process-context correlation (which process is polling), not domain reputation. Timing randomization with heavy jitter defeats naive CV thresholds — the counter is longer observation windows and size-autocorrelation, since jitter preserves the size envelope even when it destroys the clock. Key material over the channel (nonce-derived session keys) blocks retrospective decoding of captures — counter: TLS-intercepting lab detonations or hooking crypto APIs during analysis, because production traffic stays opaque. The deepest point: every evasion makes the channel more distinctive in some other dimension — pure HTTP that looks exactly like Chrome costs the operator functional flexibility, and that trade is where detection engineering lives.

⚠️ What caught me: after blocking a C2 domain during an engagement, the beaconing continued within four minutes — same timing envelope, different IP, from a failover list baked in the implant. I had treated the domain as the detection; the operator treated it as disposable. Since then, every C2 block I request ships with the timing/size profile attached as a hunt query, because with channel rotation being one config edit away, the pattern is the only asset that survives their next move.

  1. Detonate a known public beacon (e.g. a Cobalt Strike team server you run yourself in the lab, or an open-source implant) with full PCAP capture; run beacon_hunter.py and confirm the score, then manually verify the CV by plotting timestamps.
  2. Diff ten consecutive check-ins from the capture to carve the check-in template; decode the cookie/parameter encoding step by step in CyberChef until you reach structured fields.
  3. Write a minimal fake C2 (~40 lines of Python http.server) that answers check-ins with a task; execute whoami through the implant — full protocol validation without the real operator.
  4. Generate DNS tunnel traffic with dnscat2 across your lab DNS, capture with zeek, and run dnstunnel.py; then push 30 minutes of normal browsing through the same pipe and confirm the score separation.
  5. Hunt your own baseline: run the beacon scorer against one week of your real proxy logs and tune --min-score until the noise floor is acceptable; document the chosen threshold and its measured FP rate — that number is your credibility in front of the SOC.
  • Technical: Define the beacon timing profile mathematically. Why is coefficient of variation the discriminator, and how does jitter push it?
  • Technical: Walk through peeling an encoded check-in payload: what do you try, in what order, and what cribs guide each step?
  • Scenario: You block a C2 domain and beaconing resumes from a new IP within minutes. What do you do in the next hour, and what does this tell you about the implant’s config?
  • Conceptual: Why are timing and size statistics more durable than URI or header signatures against malleable C2 profiles?
  • Conceptual: How do third-party service channels (Telegram, GitHub) break traditional perimeter detection, and what replaces it?
  • Deep-dive: Design a DNS tunneling detector that survives CDN noise. Which metrics compose, and why does no single metric work?
  • Practical: Your SOC wants beacon detection deployed this week from proxy logs only. What is your minimal viable metric set and its expected FP profile?

M5 (String and Artifact Extraction) · M11 (Botnet Architecture Analysis) · M15 (Threat Intelligence and Campaign Clustering) · E17 (Network Detection) · BT2 (IDS Rules)