

An ICARAX Tech‑Blog implementation guide
Goal: Provide ready‑to‑run, defensive code that lets you monitor, analyse and alert on suspicious STUN (Session Traversal Utilities for NAT) traffic – the vector abused by a recent Linux backdoor that exploits dozens of STUN‑related flaws.
The examples below do not create or facilitate malware; they are pure detection/mitigation utilities you can drop into a security‑operations stack.
| Item | Why you need it | Minimum version |
|---|---|---|
| Linux host (any recent distro) | To run the sniffers; needs ability to capture UDP packets. | Ubuntu 22.04 / RHEL 9 / Debian 12 |
| Root or CAP_NET_RAW | Raw socket access (required by Scapy & dgram UDP bind on privileged ports <1024 – STUN uses 3478, but many IDs run on non‑privileged ports; still need raw socket for promiscuous mode). | – |
| Python 3.9+ | For the Scapy based detector. | 3.9 |
| Node.js 18+ (LTS) | For the JavaScript/TypeScript detector. | 18 |
| Package managers | pip (Python) and npm (Node). | – |
| Optional – Wireshark/tshark | For manual validation of captured packets. | – |
| Optional – Alert endpoint (Slack webhook, email, SIEM) | To get notified when anomalies are seen. | – |
Note: If you cannot run as root, grant the needed capability:
sudo setcap cap_net_raw,cap_net_admin=eip $(which python3) sudo setcap cap_net_raw,cap_net_admin=eip $(which node)
# 1️⃣ Update system & install build dependencies (needed for libpcap headers)
sudo apt-get update && sudo apt-get install -y \
python3-pip python3-dev libpcap-dev build-essential
# 2️⃣ Install Scapy (latest from PyPI) – includes packet dissection helpers
pip3 install --upgrade scapy
# 3️⃣ (Optional) Install python-dotenv for env var loading
pip3 install python-dotenv
# 1️⃣ Install Node (if not already) – using nvm is recommended
curl -fsSL https://raw.githubusercontent.com/nvm-sh/nvm/v0.39.7/install.sh | bash
source ~/.bashrc
nvm install --lts # installs latest LTS (currently 20.x)
nvm use --lts
# 2️⃣ Create a project folder
mkdir stun-monitor && cd stun-monitor
npm init -y
# 3️⃣ Install helpful dev deps
npm install --save-dev typescript @types/node ts-node
npm install --save dotenv # for .env loading
npm install --save pino # structured logger (optional but recommended)
# 4️⃣ Initialise TS config
npx tsc --init --rootDir src --outDir dist --esModuleInterop --resolveJsonModule --lib es6,dom
Tip: Keep a
.envfile (see Step 4) next to your code; both loaders will read it automatically.
Below are complete, copy‑and‑paste ready scripts that:
Feel free to extend the detection logic (e.g., entropy checks, signature matching).
File: stun_monitor.py
#!/usr/bin/env python3
"""
STUN anomaly detector – Python/Scapy version
Run: sudo python3 stun_monitor.py (or with CAP_NET_RAW)
"""
import os
import json
import time
import socket
from collections import defaultdict
from datetime import datetime, timedelta
from scapy.all import (
sniff,
UDP,
Raw,
conf,
)
# ----------------------------------------------------------------------
# Configuration (can also be overridden by env vars – see Step 4)
# ----------------------------------------------------------------------
IFACE = os.getenv("STUN_IFACE", "any") # network interface to sniff
PORT = int(os.getenv("STUN_PORT", "3478")) # UDP port STUN uses
MAX_PAYLOAD = int(os.getenv("STUN_MAX_PAYLOAD", "500")) # bytes
WEBHOOK_URL = os.getenv("STUN_WEBHOOK", "") # optional alert endpoint
# ----------------------------------------------------------------------
# In‑memory state for simple rate‑limiting / beacon detection
src_msg_counts = defaultdict(list) # src_ip -> [timestamps]
def _post_webhook(payload: dict):
"""Best‑effort POST to a Slack/Discord/SIEM webhook."""
if not WEBHOOK_URL:
return
try:
import urllib.request
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(
WEBHOOK_URL,
data=data,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(req, timeout=5) as resp:
resp.read() # discard
except Exception as e: # pragma: no cover – network errors are non‑fatal
print(f"[WARN] Webhook failed: {e}", flush=True)
def _is_stun_msg_type(msg_type: int) -> bool:
"""STUN method range per RFC 5389: 0x0001–0x010F (0‑31 for class, rest method)."""
return 0x0001 <= msg_type <= 0x010F
def _parse_stun_attributes(data: bytes):
"""
Very lightweight attribute parser.
Returns list of (attr_type, attr_length, attr_value) tuples.
Stops on malformed data.
"""
attrs = []
i = 0
while i + 4 <= len(data):
attr_type = int.from_bytes(data[i:i+2], "big")
attr_len = int.from_bytes(data[i+2:i+4], "big")
i += 4
if i + attr_len > len(data):
break # malformed
attr_val = data[i:i+attr_len]
i += attr_len
attrs.append((attr_type, attr_len, attr_val))
return attrs
def _detect_anomalies(pkt_info: dict) -> list:
"""Return a list of human‑readable reason strings."""
reasons = []
src = pkt_info["src"]
now = datetime.utcnow()
# 1️⃣ Non‑standard STUN method
if not _is_stun_msg_type(pkt_info["msg_type"]):
reasons.append(f"Non‑standard STUN method 0x{pkt_info['msg_type']:04x}")
# 2️⃣ Unknown attribute (type > 0x7FFF indicates experimental/private)
for attr_type, _, _ in pkt_info["attributes"]:
if attr_type > 0x7FFF:
reasons.append(f"Unknown/experimental attribute 0x{attr_type:04x}")
break # one is enough to flag
# 3️⃣ Payload too large
if pkt_info["payload_len"] > MAX_PAYLOAD:
reasons.append(f"Payload size {pkt_info['payload_len']} > {MAX_PAYLOAD} bytes")
# 4️⃣ Simple beacon detection: >5 msgs from same src in 10 s window
src_msg_counts[src].append(now)
cutoff = now - timedelta(seconds=10)
src_msg_counts[src] = [ts for ts in src_msg_counts[src] if ts >= cutoff]
if len(src_msg_counts[src]) > 5:
reasons.append(f"High frequency: {len(src_msg_counts[src])} msgs in 10 s")
# clear to avoid spamming
src_msg_counts[src].clear()
return reasons
def _handle_packet(pkt):
"""Callback passed to scapy.sniff."""
if not pkt.haslayer(UDP) or pkt[UDP].dport != PORT:
return # not our port
if not pkt.haslayer(Raw):
return # no payload
raw = pkt[Raw].load
if len(raw) < 20: # STUN header is 20 bytes
return
# ---- STUN header ----
msg_type = int.from_bytes(raw[0:2], "big") # first 2 bytes
# msg_length = int.from_bytes(raw[2:4], "big") # not needed for detection
magic_cookie = raw[4:8]
transaction_id = raw[8:20]
# Validate magic cookie (0x2112A442) – helps discard non‑STUN traffic
if magic_cookie != b"\x21\x12\xA4\x42":
return
payload = raw[20:] # everything after header
attributes = _parse_stun_attributes(payload)
pkt_info = {
"src": pkt[IP].src if pkt.haslayer(IP) else pkt[IPv6].src if pkt.haslayer(IPv6) else "unknown",
"dst": pkt[IP].dst if pkt.haslayer(IP) else pkt[IPv6].dst if pkt.haslayer(IPv6) else "unknown",
"msg_type": msg_type,
"magic_cookie": magic_cookie.hex(),
"transaction_id": transaction_id.hex(),
"payload_len": len(payload),
"attributes": attributes,
"timestamp": datetime.utcnow().isoformat() + "Z",
}
anomalies = _detect_anomalies(pkt_info)
if anomalies:
log_entry = {
"event": "stun_anomaly",
"info": pkt_info,
"reasons": anomalies,
}
print(json.dumps(log_entry), flush=True)
_post_webhook(log_entry)
else:
# Optional: debug‑level logging of benign STUN traffic
if os.getenv("STUN_DEBUG") == "1":
print(json.dumps({"event": "stun_ok", "info": pkt_info}), flush=True)
def main():
# Use Scapy's built‑in socket; disable IPv6 if not needed
conf.use_pcap = True # faster capture on Linux
print(f"[INFO] Listening on IFACE={IFACE} UDP/{PORT} …", flush=True)
# BPF filter: udp port <PORT>
sniff(
iface=IFACE,
filter=f"udp port {PORT}",
prn=_handle_packet,
store=False,
)
if __name__ == "__main__":
main()
Make it executable
chmod +x stun_monitor.py
Run (with required capability):
# If you gave the binary cap_net_raw:
./stun_monitor.py
# Otherwise:
sudo ./stun_monitor.py
File: src/stunMonitor.ts
#!/usr/bin/env node
/**
* STUN anomaly detector – Node.js/dgram version
* ts-node src/stunMonitor.ts (or compile to JS)
*/
import dgram from "node:dgram";
import net from "node:net";
import { config } from "dotenv";
import { createLogger, format, transports } from "pino";
config(); // loads .env into process.env
// ----------------------------------------------------------------------
// Configuration (env‑overridable)
// ----------------------------------------------------------------------
const IFACE: string = process.env.STUN_IFACE ?? "0.0.0.0"; // bind address
const PORT: number = Number(process.env.STUN_PORT ?? "3478");
const MAX_PAYLOAD: number = Number(process.env.STUN_MAX_PAYLOAD ?? "500");
const WEBHOOK_URL: string = process.env.STUN_WEBHOOK ?? "";
const DEBUG: boolean = process.env.STUN_DEBUG === "1";
// ----------------------------------------------------------------------
const logger = createLogger({
level: "info",
format: format.combine(
format.timestamp(),
format.json()
),
transports: [new transports.Stream({ target: "pino-pretty", options: { colorize: true } })],
});
type StunAttr = { type: number; length: number; value: Buffer };
interface StunPacketInfo {
src: string;
dst: string;
msgType: number; // 0‑0xFFFF
magicCookie: string; // hex
transactionId: string; // hex
payloadLen: number;
attributes: StunAttr[];
timestamp: string; // ISO UTC
}
// Simple in‑memory rate limiter (src IP => timestamps)
const srcMsgMap = new Map<string, number[]>();
const WINDOW_SECONDS = 10;
const MAX_MSGS_IN_WINDOW = 5;
/**
* Parse STUN attributes (very basic, stops on malformed data).
*/
function parseAttributes(buf: Buffer, offset: number): StunAttr[] {
const attrs: StunAttr[] = [];
let i = offset;
while (i + 4 <= buf.length) {
const type = buf.readUInt16BE(i);
const length = buf.readUInt16BE(i + 2);
i += 4;
if (i + length > buf.length) break; // malformed
const value = buf.slice(i, i + length);
i += length;
attrs.push({ type, length, value });
}
return attrs;
}
/**
* Heuristic detection – returns array of reason strings.
*/
function detectAnomalies(pkt: StunPacketInfo): string[] {
const reasons: string[] = [];
// 1️⃣ Non‑standard STUN method (0x0001‑0x010F)
if (!(pkt.msgType >= 0x0001 && pkt.msgType <= 0x010F)) {
reasons.push(`Non‑standard STUN method 0x${pkt.msgType.toString(16).padStart(4, "0")}`);
}
// 2️⃣ Unknown/experimental attribute (>0x7FFF)
for const attr of pkt.attributes {
if (attr.type > 0x7FFF) {
reasons.push(`Unknown/experimental attribute 0x${attr.type.toString(16).padStart(4, "0")}`);
break;
}
}
// 3️⃣ Payload too large
if (pkt.payloadLen > MAX_PAYLOAD) {
reasons.push(`Payload size ${pkt.payloadLen} > ${MAX_PAYLOAD} bytes`);
}
// 4️⃣ Beacon detection
const now = Date.now();
const timestamps = srcMsgMap.get(pkt.src) ?? [];
const recent = timestamps.filter(t => now - t < WINDOW_SECONDS * 1000);
srcMsgMap.set(pkt.src, [...recent, now]);
if (recent.length >= MAX_MSGS_IN_WINDOW) {
reasons.push(`High frequency: ${recent.length + 1} msgs in ${WINDOW_SECONDS}s`);
// clear to avoid repeat alerts
srcMsgMap.set(pkt.src, []);
}
return reasons;
}
/**
* Send alert to webhook (best‑effort).
*/
async function postWebhook(payload: unknown) {
if (!WEBHOOK_URL) return;
try {
const resp = await fetch(WEBHOOK_URL, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(payload),
});
if (!resp.ok) {
logger.warn({ status: resp.status }, "Webhook non‑OK response");
}
} catch (err) {
logger.warn({ err: (err as Error).message }, "Webhook delivery failed");
}
}
/**
* Main UDP listener.
*/
function startListener() {
const socket = dgram.createSocket("udp4");
socket.on("error", (err) => {
logger.error({ err: err.message }, "Socket error");
socket.close();
});
socket.on("listening", () => {
const address = socket.address();
logger.info(
{ address: address.address, port: address.port, family: address.family },
"STUN listener started"
);
});
socket.on("message", (msg, rinfo) => {
// Basic STUN header validation (20 bytes)
if (msg.length < 20) return;
const msgType = msg.readUInt16BE(0);
// Magic cookie check (0x2112A442)
const magicCookie = msg.slice(4, 8);
if (magicCookie.equals(Buffer.from([0x21, 0x12, 0xa4, 0x42])) === false) {
return; // not a STUN packet
}
const transactionId = msg.slice(8, 20).toString("hex");
const payload = msg.slice(20);
const attributes = parseAttributes(payload, 0);
const pktInfo: StunPacketInfo = {
src: rinfo.address,
dst: rinfo.address, // we only have src; dst is local interface
msgType,
magicCookie: magicCookie.toString("hex"),
transactionId,
payloadLen: payload.length,
attributes,
timestamp: new Date().toISOString(),
};
const anomalies = detectAnomalies(pktInfo);
if (anomalies.length > 0) {
const alert = {
event: "stun_anomaly",
info: pktInfo,
reasons: anomalies,
};
logger.info(alert);
void postWebhook(alert);
} else if (DEBUG) {
logger.debug({ event: "stun_ok", info: pktInfo });
}
});
// Bind to the chosen interface/port
socket.bind(PORT, IFACE);
}
// Start
startListener();
Compile & run
# Build (optional)
npm run build # if you added a "build": "tsc" script
# Run directly with ts-node (dev)
npm install -g ts-node # if not already global
ts-node src/stunMonitor.ts
# Or run the compiled JS
node dist/stunMonitor.js
Permissions: Same as Python – you need raw UDP socket access. On most Linux distros binding to port 3478 (>1024) does not require root, but if you choose a privileged port (<1024) or want promiscuous mode on an interface, give the Node binary the capability:
sudo setcap cap_net_raw,cap_net_admin=eip $(which node)
Create a .env file in the project root (same directory as the scripts).
All values are optional; defaults are shown in the code comments.
# Network interface to sniff. Use "any" for all interfaces (Linux) or a specific name like eth0.
STUN_IFACE=eth0
# UDP port STUN uses (default 3478). Change if your environment uses a non‑standard port.
STUN_PORT=3478
# Max allowed STUN payload size (bytes). Anything larger is flagged.
STUN_MAX_PAYLOAD=500
# Optional HTTP(S) webhook URL for alerts (Slack, Discord, Splunk, etc.)
STUN_WEBHOOK=https://hooks.slack.com/services/T00000000/B00000000/XXXXXXXXXXXXXXXXXXXXXXXX
# Set to "1" to get debug logs of every benign STUN packet (useful for tuning).
STUN_DEBUG=0
Loading the env
python-dotenv (from dotenv import load_dotenv; load_dotenv()) – already imported at the top.require('dotenv').config(); (via config()).You can also override any variable directly in the shell:
export STUN_IFACE=eth0 STUN_PORT=3478 STUN_MAX_PAYLOAD=400 STUN_WEBHOOK="https://example.com/alert" STUN_DEBUG=1
| Pattern | Description | Where it appears |
|---|---|---|
| BPF filter in capture | udp port 3478 reduces CPU usage by letting the kernel drop non‑matching packets early. | Python sniff(filter=…), Node (kernel filter not exposed – rely on userspace check; still cheap). |
| Header‑first validation | Check STUN magic cookie (0x2112A442) before parsing attributes – prevents wasting cycles on random UDP traffic. | Both implementations. |
| Attribute whitelist/blacklist | Known STUN attributes (e.g., MESSAGE-INTEGRITY, FINGERPRINT) have ranges 0x0000‑0x7FFF. Anything >0x7FFF is experimental/private – a strong indicator of tunnelling. | if attr.type > 0x7FFF. |
| Rate‑limiting / beacon detection | Sliding window counters per source IP detect periodic beacons used by malware for C2. | srcMsgMap (Node) & src_msg_counts (Python). |
| Structured JSON logging | Emits one JSON object per line – easy to ingest with Loki, Elasticsearch, or Splunk. | logger.info() / print(json.dumps(...)). |
| Best‑effort webhook alerting | Non‑blocking POST; errors are logged but never crash the detector. | _post_webhook / postWebhook. |
| Capability‑based privileges | Instead of running as root, give the binary only CAP_NET_RAW (and optionally CAP_NET_ADMIN). | setcap commands in prerequisites. |
| Graceful shutdown | Node catches SIGINT/SIGTERM to close socket; Python’s sniff stops on KeyboardInterrupt. | Add signal handlers if running as a service. |
| Symptom | Likely cause | Fix |
|---|---|---|
Permission denied when binding to port | Missing CAP_NET_RAW or not root. | Run with sudo or grant capability: sudo setcap cap_net_raw,cap_net_admin=eip $(which python3) (or node). |
| No packets seen, but traffic exists | Sniffing on wrong interface or BPF filter mismatch. | Verify with tcpdump -i <iface> -nn udp port 3478. Adjust STUN_IFACE. |
| High CPU usage | Capturing all traffic without BPF filter or processing every packet in Python without store=False. | Ensure filter="udp port <PORT>" is set; in Python keep store=False. |
| Webhook never receives payload | URL incorrect, outbound firewall blocks, or DNS resolution fails. | Test with curl -v -X POST -H "Content-Type: application/json" -d '{}' $STUN_WEBHOOK. |
ModuleNotFoundError: No module named 'scapy' | Scapy not installed in the current Python environment. | pip3 install scapy (or pip install --user scapy). |
TypeError: Cannot read property 'readUInt16BE' of undefined | Message shorter than expected (malformed UDP). | Guard clauses already exist (if (msg.length < 20) return;). Ensure you’re not capturing fragmented packets; increase OS UDP buffer if needed (sysctl -w net.core.rmem_max=2500000). |
| Detector floods logs with benign STUN | Debug mode left on or thresholds too low. | Set STUN_DEBUG=0 and/or raise STUN_MAX_PAYLOAD or increase beacon window. |
| Service fails to start under systemd | WorkingDir or EnvironmentFile not set. | See Production Checklist below for a sample unit file. |
| ✅ Item | Why it matters | How to implement |
|---|---|---|
| Run as unprivileged user with limited capabilities | Limits damage if the detector is compromised. | Create a dedicated user stunmon:<br>useradd -r -s /usr/sbin/nologin stunmon<br>Grant only CAP_NET_RAW:<br>setcap cap_net_raw,cap_net_admin=eip /usr/bin/python3 (or node). |
| Use a systemd service | Guarantees restart on failure, logs to journal, easy enable/disable. | Example unit file (/etc/systemd/system/stun-monitor.service):<br>ini\n[Unit]\nDescription=STUN anomaly detector\nAfter=network-online.target\nWants=network-online.target\n\n[Service]\nType=simple\nUser=stunmon\nGroup=stunmon\nExecStart=/usr/local/bin/stun_monitor.py # or node dist/stunMonitor.js\nEnvironmentFile=-/etc/stun-monitor.env\nRestart=on-failure\nRestartSec=5\nLimitNOFILE=65535\n\n[Install]\nWantedBy=multi-user.target\n<br>Then: systemctl daemon-reload && systemctl enable --now stun-monitor.service. |
| Log rotation | Prevents disk fill‑up from verbose JSON logs. | Configure journald (SystemMaxUse=100M) or redirect stdout to a file handled by logrotate. |
| Resource limits | Avoid runaway CPU/memory during traffic spikes. | In systemd: CPUQuota=20%, MemoryMax=200M. |
| Network egress control | Only allow outbound alerts to approved endpoints. | Use iptables/nftables or SELinux/AppArmor to restrict the detector user to destination IPs/ports of your webhook. |
| TLS verification for webhook | Prevent MITM of alert payloads. | Ensure webhook URL is https:// and that Node/Python verify certificates (default). |
| Version pinning / SBOM | Guarantees reproducible builds and helps with vulnerability scanning. | Pin exact versions in requirements.txt (scapy==2.5.0) and package-lock.json. Commit lockfiles to repo. |
| Testing in a staging environment | Confirms detection logic does not generate excessive false positives. | Deploy on a mirror of production traffic (e.g., using a TAP or SPAN port) and tune STUN_MAX_PAYLOAD and beacon window. |
| Alert validation | Ensure alerts are actionable and not noisy. | Integrate with SIEM: create a rule that triggers on event:stun_anomaly with severity based on number of reasons. |
| Documentation & Runbooks | Enables SOC analysts to react quickly. | Write a short runbook: “When STUN anomaly alert appears, check source IP, correlate with auth logs, consider isolating host.” |
| Periodic review | Threats evolve; detection signatures may need updates. | Schedule a monthly review of STUN abuse CVE feeds and adjust attribute whitelist or payload thresholds. |
# ---- Python ---------------------------------------------------------
sudo apt-get update && sudo apt-get install -y python3-pip libpcap-dev build-essential
pip3 install --upgrade scapy python-dotenv
# copy stun_monitor.py to /usr/local/bin/
chmod +x /usr/local/bin/stun_monitor.py
# optional: give capability instead of sudo
sudo setcap cap_net_raw,cap_net_admin=eip $(which python3)
# ---- Node -----------------------------------------------------------
# Install Node (via nvm) if missing
curl -fsSL https://raw.githubusercontent.com/nvm-sh/nvm/v0.39.7/install.sh | bash
source ~/.bashrc
nvm install --lts
nvm use --lts
mkdir -p /opt/stun-monitor && cd /opt/stun-monitor
npm init -y
npm install --save dotenv pino
npm install --save-dev typescript @types/node ts-node
# copy src/stunMonitor.ts (and tsconfig) here
npx tsc --init # creates tsconfig.json
npm run build # if you added "build":"tsc" in package.json
# optional capability
sudo setcap cap_net_raw,cap_net_admin=eip $(which node)
# ---- Environment ----------------------------------------------------
cat > /etc/stun-monitor.env <<EOF
STUN_IFACE=eth0
STUN_PORT=3478
STUN_MAX_PAYLOAD=500
STUN_WEBHOOK=https://hooks.slack.com/services/...
STUN_DEBUG=0
EOF
# ---- Systemd service ------------------------------------------------
cat > /etc/systemd/system/stun-monitor.service <<'EOF'
[Unit]
Description=STUN anomaly detector
After=network-online.target
Wants=network-online.target
[Service]
Type=simple
User=stunmon
Group=stunmon
ExecStart=/usr/local/bin/stun_monitor.py # or /opt/stun-monitor/dist/stunMonitor.js
EnvironmentFile=-/etc/stun-monitor.env
Restart=on-failure
RestartSec=5
LimitNOFILE=65535
[Install]
WantedBy=multi-user.target
EOF
sudo useradd -r -s /usr/sbin/nologin stunmon
sudo chown -R stunmon:stunmon /usr/local/bin/stun_monitor.py # adjust path as needed
sudo systemctl daemon-reload
sudo systemctl enable --now stun-monitor.service
You now have a production‑ready STUN anomaly detector running as a limited‑privilege service, logging JSON alerts, and optionally posting to a webhook for SOC integration. Adjust thresholds, add more sophisticated attribute inspection, or plug the output into your SIEM as needed. Happy monitoring! 🚀
Source: Security Week AI
Follow ICARAX for more AI insights and tutorials.
