NATS is a high-performance cloud-native messaging system used for microservice communication, event streaming via JetStream, and service mesh. Unlike Kafka (which requires Zookeeper/KRaft), NATS runs as a single binary and is easy to misconfigure. Default configurations allow unauthenticated connections, subject wildcards can be abused to subscribe to topics beyond intended scope, JetStream persists messages that can be replayed by any authorized consumer, and NATS accounts provide multi-tenancy isolation that can be bypassed through operator-level credential exposure. This guide covers systematic NATS security assessment.
# NATS defaults to no authentication — any client can connect and pub/sub
# Test connectivity without credentials
# Test unauthenticated connection (uses Go NATS CLI tool)
nats pub test.subject "probe" --server nats://nats.example.com:4222 2>&1
# If succeeds: no authentication required
# Or with netcat (NATS protocol is plaintext):
echo "CONNECT {}\r\nPUB test.probe 5\r\nhello\r\nPING\r\n" | nc nats.example.com 4222
# Response: PONG = unauthenticated connection succeeded
# Subscribe to all subjects (# wildcard = match everything)
nats sub ">" --server nats://nats.example.com:4222 &
# Then wait for messages published to any subject by any service
# Check NATS server information (publicly exposed)
curl -s http://nats.example.com:8222/varz | python3 -c "
import json,sys
r = json.load(sys.stdin)
print(f'Version: {r.get(\"version\",\"?\")}')
print(f'Auth required: {r.get(\"auth_required\",False)}')
print(f'TLS required: {r.get(\"tls_required\",False)}')
print(f'Max connections: {r.get(\"max_connections\",\"?\")}')
print(f'Connected clients: {r.get(\"connections\",\"?\")}')
print(f'JetStream enabled: {r.get(\"jetstream\",{}).get(\"enabled\",False)}')
"
# List connected clients (monitoring endpoint — no auth by default)
curl -s http://nats.example.com:8222/connz | python3 -c "
import json,sys
r = json.load(sys.stdin)
for conn in r.get('connections',[]):
print(f\"Client: {conn.get('name','?')} IP: {conn.get('ip','?')} subs: {conn.get('num_subscriptions',0)}\")
"# NATS subject authorization restricts which subjects a user can publish/subscribe to
# Test whether wildcard permissions are overly broad
# NATS subject wildcards:
# * = matches single token: orders.* matches orders.create but not orders.v1.create
# > = matches remaining: orders.> matches orders.create AND orders.v1.create.all
# Check a user's permissions (if you have access to server config)
# Overly broad examples to look for:
# publish: [">"] = can publish to ALL subjects
# subscribe: [">"] = can subscribe to ALL subjects (eavesdrop on everything)
# publish: ["admin.>"] = can publish admin commands
# Test: subscribe with wildcard to capture cross-service messages
nats sub ">" --server nats://user:password@nats.example.com:4222 --count 100 2>/dev/null | \
head -100
# Captures ALL messages across all services
# Test: publish to admin/privileged subjects
nats pub "admin.command" '{"action":"purge_all"}' \
--server nats://user:password@nats.example.com:4222 2>&1
# If no -ERR Permissions Violation: user can publish admin commands
# Test: subscribe to other team's subjects
nats sub "team-b.>" --server nats://team-a-user:pass@nats.example.com:4222 2>&1
# Should return -ERR Permissions Violation for 'SUB "team-b.>"'
# If no error: team isolation not enforced# JetStream persists messages in streams — subscribers can replay historical messages
# Test whether stream access controls are properly configured
# List all JetStream streams (requires stream:list permission or admin)
nats stream ls --server nats://nats.example.com:4222 2>/dev/null
# Attempt to subscribe to a JetStream stream as a consumer
nats consumer sub ORDERS new-consumer \
--server nats://user:pass@nats.example.com:4222 \
--deliver all 2>/dev/null | head -20
# --deliver all replays ALL historical messages in the stream
# Test: create a stream on a subject you can publish to
nats stream add test-capture \
--subjects ">" \
--server nats://user:pass@nats.example.com:4222 \
--storage file \
--retention limits \
--max-age 1h 2>/dev/null
# If allowed: captures ALL messages published to any subject for 1 hour
# Check stream metadata
curl -s http://nats.example.com:8222/jsz | python3 -c "
import json,sys
r = json.load(sys.stdin)
for stream in r.get('account_details',[]):
for s in stream.get('stream_detail',[]):
print(f\"Stream: {s.get('name','?')} messages: {s.get('state',{}).get('messages',0)}\")
"no_auth_user in production> wildcard subscribe permission unless explicitly required by a monitoring servicetls_required: true| Security Test | Method | Risk |
|---|---|---|
| Unauthenticated NATS connection | nats pub test "probe" without credentials | Critical |
| Wildcard ">" subscribe permission | nats sub ">" — if succeeds, reads all messages | Critical |
| JetStream historical message replay | nats consumer sub STREAM new --deliver all | High |
| Monitoring endpoint exposes cluster topology | curl :8222/connz — lists all connected clients | High |
| Publish to admin/control subjects | nats pub admin.command payload | High |
| Cross-account subject access without exports | Subscribe to account-B subject from account-A | Medium |
Ironimo tests NATS deployments for unauthenticated connections, subject authorization wildcard abuse, JetStream stream data replay access, account isolation bypass, credentials exposure, monitoring endpoint information disclosure, and TLS misconfiguration.
Start free scan