Chapter 35

ETA Without Decryption

Encrypted Traffic Analysis (ETA) is the set of techniques for classifying and characterizing encrypted network traffic without decrypting it. The insight is that even fully encrypted connections leave statistical fingerprints — packet sizes, timing patterns, inter-arrival times, flow duration, and the TLS metadata from the handshake all correlate with what's inside. Cisco coined the term ETA and built commercial tools around it, but the underlying techniques are openly documented and implementable with open-source tools.

Scenario

85% of your network traffic is TLS. You can't decrypt it all (key management, legal concerns). But you need to detect malware C2. ETA gives you: JA3 fingerprinting from the ClientHello, certificate analysis from the handshake, beaconing detection from connection timing, and statistical analysis of payload sizes — all without decryption. Combined, these techniques let you identify Cobalt Strike, Emotet, and most commercial C2 frameworks with acceptable false-positive rates.

ETA Feature Set

  ETA Features Extractable Without Decryption
  ═══════════════════════════════════════════════════════════════════

  TLS Handshake Features (plaintext, from ClientHello/ServerHello):
  ├── JA3 client fingerprint (cipher suites + extensions + curves)
  ├── JA3S server fingerprint (selected cipher + extensions)
  ├── TLS version (1.0/1.1/1.2/1.3 — old = suspicious)
  ├── SNI presence/absence (no SNI = suspicious for browsers)
  ├── Certificate features (self-signed, age, validity, subject)
  └── ALPN extension (h2, http/1.1, ftp, xmpp — protocol declared)

  Flow-Level Features (connection metadata):
  ├── Connection duration (short = scanning, long = C2 tunnel)
  ├── Bytes per direction (client→server vs server→client ratio)
  ├── Packet count per direction
  ├── Mean/stddev of inter-packet arrival times
  └── Bytes-per-second rate (high = exfil, low+regular = beacon)

  Sequence-of-Packets Features (statistical, per connection):
  ├── Initial data length (length of first client DATA record)
  ├── Sequence of application data record lengths
  │     Different application types have characteristic sequences
  │     YouTube video: large sustained server records
  │     Cobalt Strike: small regular client records (check-in data)
  └── Number of application data records per direction

  Aggregated Features (per flow over time):
  ├── Connection frequency (beaconing CV)
  ├── Consistent destination over time (stable C2)
  └── Diversity of SNIs contacted (high = browser; low = malware)

ETA Analysis Pipeline

Pythoneta-analysis.py

#!/usr/bin/env python3
"""
Encrypted Traffic Analysis (ETA) pipeline.
Extracts features from TLS traffic without decryption.
Uses Zeek logs for efficiency on large captures.
"""
import csv, json, math, sys, subprocess, re
from collections import defaultdict
from pathlib import Path

# ── FEATURE EXTRACTION FROM ZEEK LOGS ─────────────────────────────────
def load_zeek_conn_log(path: str) -> list:
    """Load Zeek conn.log and return list of flow dicts."""
    flows = []
    field_names = []
    with open(path) as f:
        for line in f:
            if line.startswith("#fields"):
                field_names = line.strip().split("\t")[1:]
            elif line.startswith("#"):
                continue
            else:
                values = line.strip().split("\t")
                if field_names:
                    d = dict(zip(field_names, values))
                    if d.get("proto") == "tcp":
                        flows.append(d)
    return flows

def load_zeek_ssl_log(path: str) -> dict:
    """Load Zeek ssl.log → {uid: ssl_fields}"""
    ssl_by_uid = {}
    field_names = []
    with open(path) as f:
        for line in f:
            if line.startswith("#fields"):
                field_names = line.strip().split("\t")[1:]
            elif line.startswith("#"):
                continue
            else:
                values = line.strip().split("\t")
                if field_names:
                    d = dict(zip(field_names, values))
                    uid = d.get("uid", "")
                    if uid:
                        ssl_by_uid[uid] = d
    return ssl_by_uid

def compute_features(flow: dict, ssl: dict) -> dict:
    """Compute ETA features for a flow."""
    try:
        duration = float(flow.get("duration", "0") or "0")
        orig_bytes = int(flow.get("orig_bytes", "0") or "0")
        resp_bytes = int(flow.get("resp_bytes", "0") or "0")
        orig_pkts = int(flow.get("orig_pkts", "0") or "0")
        resp_pkts = int(flow.get("resp_pkts", "0") or "0")
    except ValueError:
        return {}

    # Directionality ratio: what fraction of bytes came from server
    total_bytes = orig_bytes + resp_bytes
    resp_ratio = resp_bytes / total_bytes if total_bytes > 0 else 0.5

    # Bytes per second
    bps = total_bytes / duration if duration > 0 else 0

    # TLS fields
    sni = ssl.get("server_name", "") if ssl else ""
    version = ssl.get("version", "") if ssl else ""
    cipher = ssl.get("cipher", "") if ssl else ""
    subject = ssl.get("subject", "") if ssl else ""
    issuer = ssl.get("issuer", "") if ssl else ""
    ja3 = ssl.get("ja3", "") if ssl else ""
    ja3s = ssl.get("ja3s", "") if ssl else ""
    validation = ssl.get("validation_status", "") if ssl else ""

    # Heuristic flags
    flags = []

    if not sni:
        flags.append("NO_SNI")
    if version in ["TLSv10", "TLSv11", "SSLv3"]:
        flags.append(f"OLD_TLS_{version}")
    if "self signed" in validation.lower():
        flags.append("SELF_SIGNED")
    if duration > 3600 and orig_bytes < 50000:
        flags.append("LONG_LOW_VOLUME")  # Beacon indicator
    if resp_ratio > 0.97 and total_bytes > 50_000_000:
        flags.append("LARGE_DOWNLOAD")
    if orig_bytes > 100_000_000 and resp_ratio < 0.1:
        flags.append("LARGE_UPLOAD")  # Exfil indicator
    if orig_pkts > 100 and duration > 300:
        mean_interval = duration / orig_pkts
        if 1 < mean_interval < 600:
            flags.append(f"REGULAR_INTERVALS_{mean_interval:.0f}s")

    return {
        "uid": flow.get("uid", ""),
        "src": flow.get("id.orig_h", ""),
        "dst": flow.get("id.resp_h", ""),
        "port": flow.get("id.resp_p", ""),
        "duration_s": round(duration, 1),
        "orig_bytes": orig_bytes,
        "resp_bytes": resp_bytes,
        "resp_ratio": round(resp_ratio, 3),
        "bps": round(bps),
        "sni": sni,
        "version": version,
        "ja3": ja3[:16] + "..." if len(ja3) > 16 else ja3,
        "validation": validation,
        "flags": flags,
    }

# Main
if len(sys.argv) < 3:
    print("Usage: eta-analysis.py  ")
    sys.exit(1)

print("Loading Zeek logs...")
flows = load_zeek_conn_log(sys.argv[1])
ssl_map = load_zeek_ssl_log(sys.argv[2])

print(f"Analyzing {len(flows)} TCP flows...")
flagged = []

for flow in flows:
    ssl = ssl_map.get(flow.get("uid", ""))
    features = compute_features(flow, ssl)
    if features and features.get("flags"):
        flagged.append(features)

print(f"\n=== Flagged ETA Sessions ({len(flagged)}) ===")
flagged.sort(key=lambda x: len(x["flags"]), reverse=True)
for f in flagged[:30]:
    print(f"\n  {f['src']}→{f['dst']}:{f['port']} [{f['duration_s']}s]")
    print(f"  SNI: {f['sni'] or '(none)'}  Version: {f['version']}  ja3: {f['ja3']}")
    print(f"  Bytes: {f['orig_bytes']:,}↑ / {f['resp_bytes']:,}↓  ({f['bps']:,} Bps)")
    print(f"  Flags: {f['flags']}")

Mental model: ETA is probabilistic, not deterministic

ETA cannot tell you with certainty that a connection is C2 traffic — it tells you that a connection has statistical properties consistent with C2 traffic. A "NO_SNI + SELF_SIGNED + LONG_LOW_VOLUME" connection is highly suspicious but might be a legacy internal application, an IoT device, or a misconfigured client. ETA output is an investigation priority list, not an alert list. The workflow is: ETA flags connections → analyst investigates flagged connections → uses additional context (endpoint telemetry, known-good asset list, historical baseline) to confirm or dismiss. The power of ETA is volume reduction: out of 100,000 TLS sessions per day, ETA might flag 50 for human review. Without ETA, you'd either review all 100,000 (impossible) or review none (blind). The 50 that ETA flags are the highest-priority sessions to investigate — a much more tractable problem.

Q & A

Q: How do I establish a baseline of normal TLS traffic to compare ETA results against?

Baseline establishment is essential for making ETA results actionable. Without a baseline, every self-signed cert or unusual JA3 is an unknown — you can't tell legitimate unusual from malicious unusual. Building a baseline: (1) Collect 30 days of Zeek ssl.log from a normal period (no known incidents). (2) Build a per-host inventory: for each internal IP, record which external IPs/SNIs it connects to, which JA3 hashes it uses, which certificate issuers it accepts. (3) Store in a database (SQLite or Elasticsearch) keyed by source IP. (4) Flag deviations: a new SNI never seen before from a host, a new JA3 hash from a host that normally uses only one JA3 (Chrome), a new certificate issuer. "New to this host" is a stronger signal than "globally suspicious" — a certificate that's self-signed but has been accepted by the same host for 2 years is probably a known internal service. A self-signed certificate from a new IP appearing for the first time is a real alert. The baseline makes the difference between signal and noise.