A short capture is easy to open and surprisingly easy to misread. You find a DNS query, a TCP reset, and an HTTP request, then start telling a story before checking whether those records belong together.
Zeek gives you a second way to inspect the capture: connection, DNS, and HTTP logs with fields you can filter and identifiers you can follow back to related records. I tested that workflow against a small Linux lab with known inputs, including a deliberately unsuccessful DNS lookup and a bounded SYN scan.
The goal is an activity timeline supported by packets. Start with the packet analysis hub if you need capture placement and filter basics; this walkthrough begins with a controlled input and follows it into Zeek's logs.
What goes into the capture
The lab is two network namespaces joined directly by a veth pair:
ple-client · 192.0.2.10/24 ← veth → ple-server · 192.0.2.20/24
UDP 53: synthetic DNS
TCP 8080: plain HTTPNeither namespace has an uplink or default route. The client generates two DNS requests, retrieves /baseline, scans 25 selected TCP ports, then retrieves /recovery. The DNS exchanges and HTTP requests are explicit script inputs; their timing alone would not prove which application used a DNS answer on an unknown host.
Tested with: kernel 7.0.0-38-generic, Python 3.14.4, Nmap 7.98, tcpdump 4.99.6, Zeek 9.0.0. Zeek runs offline in its Docker image with networking disabled; the image digest below pins the tested build. The docker commands assume your user can run Docker. If it can't, prefix them with sudo.
Generate a capture you can explain
On a Linux lab host with Bash, iproute2, Python, Nmap, tcpdump, and Docker, create a fresh directory:
mkdir -p pcap-lab/scripts pcap-lab/evidence
cd pcap-labSave this as scripts/setup.sh. It refuses existing namespace and interface names.
#!/usr/bin/env bash
set -euo pipefail
for ns in ple-client ple-server; do
if ip netns list | awk '{print $1}' | grep -qx "$ns"; then
echo "Refusing existing namespace: $ns" >&2
exit 1
fi
done
for dev in plec0 ples0; do
if ip link show dev "$dev" >/dev/null 2>&1; then
echo "Refusing existing interface: $dev" >&2
exit 1
fi
done
ip netns add ple-client
ip netns add ple-server
ip -n ple-client link set lo up
ip -n ple-server link set lo up
ip link add plec0 type veth peer name ples0
ip link set plec0 netns ple-client
ip link set ples0 netns ple-server
ip -n ple-client link set plec0 name eth0
ip -n ple-server link set ples0 name eth0
ip -n ple-client addr add 192.0.2.10/24 dev eth0
ip -n ple-server addr add 192.0.2.20/24 dev eth0
ip -n ple-client link set eth0 up
ip -n ple-server link set eth0 upSave the following as scripts/services.py. It answers app.test with 192.0.2.20, returns NXDOMAIN for the other lab name, and serves a small HTTP response. This is a synthetic DNS service, not a general resolver.
#!/usr/bin/env python3
import json, socket, struct, threading
from http.server import BaseHTTPRequestHandler, HTTPServer
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
body = b'PingLabz capture lab OK\n'
self.send_response(200)
self.send_header('Content-Length', str(len(body)))
self.end_headers()
self.wfile.write(body)
print(json.dumps({'event': 'http', 'peer': self.client_address, 'path': self.path}), flush=True)
def log_message(self, *args):
pass
def dns():
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
s.bind(('192.0.2.20', 53))
while True:
data, peer = s.recvfrom(4096)
pos = 12
labels = []
while data[pos]:
n = data[pos]
labels.append(data[pos + 1:pos + 1 + n].decode())
pos += n + 1
pos += 1
question = data[12:pos + 4]
name = '.'.join(labels)
good = name == 'app.test'
header = data[:2] + struct.pack('!HHHHH', 33152 if good else 33155, 1, 1 if good else 0, 0, 0)
answer = b'\xc0\x0c' + struct.pack('!HHIH', 1, 1, 60, 4) + socket.inet_aton('192.0.2.20') if good else b''
s.sendto(header + question + answer, peer)
print(json.dumps({'event': 'dns', 'peer': peer, 'query': name, 'rcode': 0 if good else 3}), flush=True)
threading.Thread(target=dns, daemon=True).start()
print('Listening: UDP 53 and HTTP TCP 8080', flush=True)
HTTPServer(('192.0.2.20', 8080), Handler).serve_forever()Save this as scripts/client.py. It checks both DNS outcomes and requires an HTTP 200 response with the expected body.
#!/usr/bin/env python3
import argparse, json, socket, struct
p = argparse.ArgumentParser()
p.add_argument('mode', choices=['baseline', 'recovery'])
a = p.parse_args()
if a.mode == 'baseline':
for name, port, tid in [('app.test', 53001, 4097), ('missing.test', 53002, 4098)]:
q = b''.join((bytes([len(x)]) + x.encode() for x in name.split('.'))) + b'\x00' + struct.pack('!HH', 1, 1)
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
s.bind(('192.0.2.10', port))
s.settimeout(3)
s.sendto(struct.pack('!HHHHHH', tid, 256, 1, 0, 0, 0) + q, ('192.0.2.20', 53))
reply, _ = s.recvfrom(4096)
flags = struct.unpack('!H', reply[2:4])[0]
rcode = flags & 15
assert rcode == (0 if name == 'app.test' else 3)
if name == 'app.test':
assert reply[-4:] == socket.inet_aton('192.0.2.20')
print(json.dumps({'query': name, 'rcode': rcode, 'source_port': port}), flush=True)
with socket.socket() as s:
s.bind(('192.0.2.10', 41001 if a.mode == 'baseline' else 41002))
s.settimeout(3)
s.connect(('192.0.2.20', 8080))
s.sendall(f'GET /{a.mode} HTTP/1.0\r\nHost: app.test\r\nConnection: close\r\n\r\n'.encode())
parts = []
while True:
b = s.recv(4096)
if not b:
break
parts.append(b)
response = b''.join(parts).decode()
assert 'HTTP/1.0 200' in response and 'PingLabz capture lab OK' in response
print(response, flush=True)Finally, save the capture runner as scripts/capture.py. It captures on the server's eth0, records the scan command and output, and saves both the benign baseline and the complete mixed capture.
#!/usr/bin/env python3
import os, pathlib, signal, subprocess, time, json
R = pathlib.Path(__file__).resolve().parent.parent
E = R / 'evidence'
children = []
commands = []
def ns(name, *args):
return ['ip', 'netns', 'exec', 'ple-' + name, *map(str, args)]
def start(args, label):
out = open(E / (label + '.stdout'), 'w')
err = open(E / (label + '.stderr'), 'w')
p = subprocess.Popen(args, cwd=R, stdout=out, stderr=err, start_new_session=True)
children.append((p, out, err))
time.sleep(0.3)
assert p.poll() is None
return p
def stop(p):
if p.poll() is None:
os.killpg(p.pid, signal.SIGINT)
p.wait(timeout=10)
def run(args, label):
p = subprocess.run(args, cwd=R, capture_output=True, text=True, timeout=30)
(E / (label + '.stdout')).write_text(p.stdout)
(E / (label + '.stderr')).write_text(p.stderr)
commands.append({'argv': args, 'returncode': p.returncode, 'label': label})
assert p.returncode == 0, p.stderr
try:
run(['nmap', '--version'], 'nmap-version')
run(['uname', '-r'], 'kernel-version')
run(['date', '-u', '+%Y-%m-%dT%H:%M:%SZ'], 'clock')
start(ns('server', 'python3', '-u', 'scripts/services.py'), 'services')
mixed = start(ns('server', 'tcpdump', '--immediate-mode', '-U', '-n', '-i', 'eth0', '-s', '0', '-w', E / 'mixed.pcap', 'tcp or udp port 53'), 'mixed-capture')
baseline = start(ns('server', 'tcpdump', '--immediate-mode', '-U', '-n', '-i', 'eth0', '-s', '0', '-w', E / 'baseline.pcap', 'tcp or udp port 53'), 'baseline-capture')
run(ns('client', 'python3', 'scripts/client.py', 'baseline'), 'baseline-client')
time.sleep(0.2)
stop(baseline)
run(ns('client', 'nmap', '-n', '-Pn', '-sS', '--reason', '--max-retries', '0', '--scan-delay', '20ms', '-p', '8000-8023,8080', '--source-port', '42000', '192.0.2.20', '-oA', E / 'scan'), 'nmap-scan')
run(ns('client', 'python3', 'scripts/client.py', 'recovery'), 'recovery-client')
time.sleep(0.2)
stop(mixed)
finally:
for p, out, err in children:
if p.poll() is None:
os.killpg(p.pid, signal.SIGTERM)
p.wait(timeout=10)
out.close()
err.close()
(E / 'commands.json').write_text(json.dumps(commands, indent=2))
print('Capture complete; all owned services and capture processes stopped.')Run the setup once, then generate the traffic:
sudo bash scripts/setup.sh
sudo python3 scripts/capture.py
sudo chown -R "$(id -u):$(id -g)" evidenceThe runner scans only 192.0.2.20, ports 8000–8023 and 8080, using SYN probes from source port 42000. It does not perform version detection or scan a subnet. My Nmap result was 24 closed ports and one open port, 8080. The Nmap guide explains scan mechanics.
Check the input before analyzing it
The baseline capture contained 16 packets: two DNS exchanges and one HTTP connection. The mixed capture contained 79 packets: that baseline, the scan, and the recovery request. Both tcpdump logs reported matching captured and filter-received counts with zero kernel drops.
You can inspect those capture logs and count file records:
cat evidence/baseline-capture.stderr evidence/mixed-capture.stderr
tshark -r - -T fields -e frame.number < evidence/mixed.pcap | wc -l
sha256sum evidence/baseline.pcap evidence/mixed.pcapThese counts and frame numbers belong to the saved run; scheduling and Nmap probe order can vary when you generate a fresh capture. The hashes identify the files you analyzed.
Turn the PCAP into Zeek JSON logs
From the lab directory, create separate output directories so another run cannot silently replace the first set of logs:
ZEEK_IMAGE='zeek/zeek@sha256:70733f4e540ba1608e37e00c6a93d009f79734262c9ec2ba09d90aa9abc16de5'
mkdir evidence/zeek-baseline evidence/zeek-mixed
for name in baseline mixed; do
docker run --rm --network none --user "$(id -u):$(id -g)" \
-v "$PWD:/work" -w "/work/evidence/zeek-$name" "$ZEEK_IMAGE" \
zeek -C -r "/work/evidence/$name.pcap" LogAscii::use_json=T \
policy/misc/loaded-scripts policy/protocols/conn/community-id-logging \
CommunityID::seed=0
done-r reads the saved capture, while LogAscii::use_json=T makes each log line a JSON object. loaded_scripts.log records the script set. This invocation loads the Community ID logging policy for the later cross-tool comparison; it does not load an additional scan-detection policy.
The -C option ignores packet checksum validation. These are host-side veth captures, where offload representation can affect checksums. Keep that analysis choice in the evidence manifest. The Zeek quick-start guide documents offline processing and the checksum option.
Start with conn.log, then add protocol detail
Count the records rather than assuming one log line equals one packet:
python3 - <<'EOF'
import collections, json
from pathlib import Path
for name in ['baseline', 'mixed']:
root = Path('evidence') / ('zeek-' + name)
connections = [json.loads(line) for line in (root / 'conn.log').read_text().splitlines()]
print(name, 'connections:', len(connections))
print(collections.Counter(row['conn_state'] for row in connections))
for log in ['dns.log', 'http.log']:
print(log, len((root / log).read_text().splitlines()))
EOFMy baseline had three connection records, two DNS records, and one HTTP record. The mixed capture had 29 connection records, two DNS records, and two HTTP records. Its connection-state counts were 24 REJ, four SF, and one RSTO.
Those labels describe what Zeek inferred from the observed connection behavior. They do not say “25 malicious connections.” In this controlled test, the source-port-42000 records correspond to the 25 Nmap probes. Packet inspection distinguishes the closed-port resets from the open port's SYN/ACK and client reset.
Read the successful and failed DNS exchanges
Filter the actual DNS records:
python3 - <<'EOF'
import json
from pathlib import Path
for line in Path('evidence/zeek-mixed/dns.log').read_text().splitlines():
row = json.loads(line)
print({key: row.get(key) for key in ['uid', 'trans_id', 'query', 'qtype_name', 'rcode_name', 'answers']})
EOFSelected fields from the successful exchange:
{
"uid": "C5l1As1tHzkE8bJNug",
"trans_id": 4097,
"query": "app.test",
"qtype_name": "A",
"rcode_name": "NOERROR",
"answers": [
"192.0.2.20"
]
}The other record identified missing.test, transaction ID 4098, and NXDOMAIN. That is a server response describing a failed lookup, not evidence that the DNS server was unreachable.
Use uid to find the matching UDP connection record. Do not use the DNS transaction ID alone as a global connection key: the same number can appear in unrelated DNS traffic.
Pivot from HTTP back to its connection
For the recovery request, follow the HTTP record's UID into conn.log:
python3 - <<'EOF'
import json
from pathlib import Path
root = Path('evidence/zeek-mixed')
connections = {row['uid']: row for row in map(json.loads, (root / 'conn.log').read_text().splitlines())}
for row in map(json.loads, (root / 'http.log').read_text().splitlines()):
if row.get('uri') == '/recovery':
print('HTTP:', row['uid'], row['method'], row['uri'], row['status_code'])
print('Connection:', connections[row['uid']])
EOFThe saved HTTP record contained:
{
"uid": "CHPWL61Q5CkyYFxi1g",
"id.orig_h": "192.0.2.10",
"id.orig_p": 41002,
"id.resp_h": "192.0.2.20",
"id.resp_p": 8080,
"method": "GET",
"host": "app.test",
"uri": "/recovery",
"status_code": 200
}Its matching connection recorded six packets in each direction, no missed bytes, and a normal completed exchange in this test. The client also checked the response body, and the server logged the /recovery request. Together those observations support application success after the scan.
UIDs may change when you process the same capture again. Discover them from your own log set instead of copying the example UID into an investigation script.
Return to the packets when the logs need context
Inspect the scan probes without losing the distinction between source and destination ports:
tshark -r - -Y 'tcp.srcport == 42000 || tcp.dstport == 42000' \
-T fields -e frame.number -e ip.src -e tcp.srcport \
-e ip.dst -e tcp.dstport -e tcp.flags < evidence/mixed.pcapThe closed probes each had a SYN followed by a server reset. The open-port probe had a SYN/ACK and a client reset, without an HTTP request. By contrast, ports 41001 and 41002 were the script's full HTTP connections.
A missing protocol log does not prove that no activity occurred. Check whether the capture includes the relevant traffic, whether that protocol was visible, and which analyzers and policies were loaded. This lab uses plaintext DNS and HTTP; encrypted application data would change what the protocol logs can reveal.
Add detection evidence to the same input
Zeek gives this capture an activity timeline and connection pivots. Next, use the Suricata EVE lab to test exactly which enabled rule generates an alert. Then follow the Community ID comparison to join the outputs without mistaking record counts or alert labels for additional traffic.