xmpp-enumeration

name	xmpp-enumeration
description	XMPP/Jabber service enumeration for Openfire, ejabberd, Prosody, and other XMPP servers. Trigger when ports 5222 (client), 5223 (legacy TLS), or 5269 (server-to-server) are found open. Covers authentication testing, user enumeration, MUC room discovery, and server fingerprinting. Do NOT use for AD enumeration or credential spraying — route those to the appropriate skills.
keywords	["xmpp","jabber","openfire","ejabberd","prosody","5222","5223","5269","xep-0077","sasl","anonymous","muc","in-band registration","chat","instant messaging"]
tools	["python3","nmap"]
opsec	low

XMPP/Jabber Enumeration

You are helping a penetration tester enumerate an XMPP/Jabber service. This skill covers service detection, authentication testing, user enumeration, MUC (Multi-User Chat) room discovery, and server fingerprinting. All testing is under explicit written authorization.

Engagement Logging

Check for ./engagement/ directory. If absent, proceed without logging.

When an engagement directory exists:

• Print [xmpp-enumeration] Activated → <target> to the screen on activation.
• Evidence → save significant output to engagement/evidence/ with descriptive filenames (e.g., xmpp-users.txt, xmpp-rooms.txt, xmpp-server-info.txt).

Scope Boundary

This skill covers XMPP service enumeration only. It does NOT cover:

• AD enumeration or Kerberos attacks — route to ad-discovery
• Credential spraying or brute force — route to password-spraying
• Web application testing (even Openfire admin console) — route to web-discovery
• Exploitation of RCE vulnerabilities in XMPP servers — report and return

When enumeration is complete, STOP and return to the orchestrator with discovered users, rooms, server details, and recommendations for next skills.

Stay in methodology. Only use techniques documented in this skill. If you encounter a scenario not covered here, note it and return — do not improvise attacks, write custom exploit code, or apply techniques from other domains. The orchestrator will provide specific guidance or route to a different skill.

State Management

Call get_state_summary() from the state MCP server to read current engagement state. Use it to:

• Skip re-testing targets, parameters, or vulns already confirmed
• Leverage existing credentials or access for this technique
• Understand what's been tried and failed (check Blocked section)

Your return summary must include:

• New targets/hosts discovered (with ports and services)
• New credentials or tokens found
• Access gained or changed (user, privilege level, method)
• Vulnerabilities confirmed (with status and severity)
• Pivot paths identified (what leads where)
• Blocked items (what failed and why, whether retryable)

Prerequisites

• XMPP port open: 5222 (STARTTLS), 5223 (legacy TLS), or 5269 (S2S)
• python3 — for raw XML socket interaction (no external libraries required)
• nmap — for initial service probing (via MCP nmap-server)
• Optional: slixmpp Python library (if installed, simplifies some steps)

Special characters in credentials

Bash history expansion treats ! as a special character (!event), even inside double quotes. Passwords containing !, $, backticks, or other shell metacharacters will be silently mangled when passed as command arguments.

Canonical workaround — write to file, read from file:

# 1. Use the Write tool (not echo/printf) to create a password file
Write("/tmp/claude-1000/cred.txt", "lDaP_1n_th3_cle4r!")

# 2. Read into a variable
PASS=$(cat /tmp/claude-1000/cred.txt)

# 3. Use the variable in commands (double-quote it)
python3 xmpp_enum.py --password "$PASS"

Step 1: Service Detection

Confirm XMPP service and identify the server software.

1a. Nmap Service Probes

Use the nmap MCP to scan XMPP ports:

nmap_scan(target="<IP>", options="-sV -p 5222,5223,5269,5270,5275,5276,7070,7443,9090,9091 -sC")

Key ports:

Port	Service	Notes
5222	XMPP client (STARTTLS)	Primary client connection
5223	XMPP client (legacy TLS)	Direct TLS, older servers
5269	XMPP server-to-server	Federation port
5270	XMPP S2S (TLS)	Secure federation
5275	XMPP component	External component interface
7070	HTTP binding (BOSH)	Web client access
7443	HTTPS binding (BOSH)	Secure web client access
9090	Openfire admin (HTTP)	Admin console — route to web-discovery
9091	Openfire admin (HTTPS)	Admin console — route to web-discovery

1b. TLS Certificate Inspection

Extract hostname and organization from the TLS certificate:

# STARTTLS on 5222
echo | openssl s_client -starttls xmpp -connect <IP>:5222 -servername <domain> 2>/dev/null | openssl x509 -noout -subject -issuer -dates

# Direct TLS on 5223
echo | openssl s_client -connect <IP>:5223 2>/dev/null | openssl x509 -noout -subject -issuer -dates

The certificate CN or SAN fields often reveal the XMPP domain (e.g., chat.corp.local, xmpp.target.local).

1c. Raw XMPP Stream Probe

Send an initial stream header to identify the server and supported features:

#!/usr/bin/env python3
"""XMPP stream probe — identifies server software and SASL mechanisms."""
import socket
import ssl
import sys

TARGET = sys.argv[1]  # IP or hostname
PORT = int(sys.argv[2]) if len(sys.argv) > 2 else 5222
DOMAIN = sys.argv[3] if len(sys.argv) > 3 else TARGET

STREAM_HEADER = f'''<?xml version='1.0'?>
<stream:stream xmlns='jabber:client'
  xmlns:stream='http://etherx.jabber.org/streams'
  to='{DOMAIN}' version='1.0'>'''

def probe(target, port, domain, use_tls=False):
    sock = socket.create_connection((target, port), timeout=10)
    if use_tls:
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        sock = ctx.wrap_socket(sock, server_hostname=domain)

    sock.sendall(STREAM_HEADER.encode())
    data = b""
    while True:
        try:
            chunk = sock.recv(4096)
            if not chunk:
                break
            data += chunk
            # Stop once we have features or stream error
            if b"</stream:features>" in data or b"</stream:error>" in data:
                
         socket.timeout:
            
    sock.close()
     data.decode(errors=)


 use_tls  [, ]:
    :
        label =   use_tls  
        ()
        resp = probe(TARGET, PORT, DOMAIN, use_tls)
        (resp)
        
     Exception  e:
        ()

What to look for in the response:

• <stream:features> block lists supported authentication mechanisms
• SASL mechanisms: PLAIN, SCRAM-SHA-1, ANONYMOUS, EXTERNAL, DIGEST-MD5
• <register xmlns='http://jabber.org/features/iq-register'/> = in-band registration enabled (XEP-0077)
• Server identification in stream header attributes or error messages

Step 2: Authentication Testing

Test what authentication options are available without credentials.

2a. SASL ANONYMOUS

If ANONYMOUS appears in the SASL mechanisms, the server allows anonymous login — this is a significant finding:

#!/usr/bin/env python3
"""Test SASL ANONYMOUS authentication."""
import socket
import ssl
import base64
import sys

TARGET = sys.argv[1]
PORT = int(sys.argv[2]) if len(sys.argv) > 2 else 5222
DOMAIN = sys.argv[3] if len(sys.argv) > 3 else TARGET

STREAM_HEADER = f'''<?xml version='1.0'?>
<stream:stream xmlns='jabber:client'
  xmlns:stream='http://etherx.jabber.org/streams'
  to='{DOMAIN}' version='1.0'>'''

def recv_until(sock, marker, timeout=10):
    sock.settimeout(timeout)
    data = b""
    while marker.encode() not in data:
        try:
            chunk = sock.recv(4096)
            if not chunk:
                break
            data += chunk
        except socket.timeout:
            break
    return data.decode(errors="replace")

sock = socket.create_connection((TARGET, PORT), timeout=10)

# Start stream
sock.sendall(STREAM_HEADER.encode())
features = recv_until(sock, "</stream:features>")
print("[*] Features:", features[:500])

# Check for STARTTLS and upgrade if available
if   features:
    sock.sendall()
    resp = recv_until(sock, )
       resp:
        ctx = ssl.create_default_context()
        ctx.check_hostname = 
        ctx.verify_mode = ssl.CERT_NONE
        sock = ctx.wrap_socket(sock, server_hostname=DOMAIN)
        
        sock.sendall(STREAM_HEADER.encode())
        features = recv_until(sock, )
        (, features[:])

    features:
    ()
    sock.close()
    sys.exit()


sock.sendall()
resp = recv_until(sock, )
(, resp)

   resp:
    ()
    
    sock.sendall(STREAM_HEADER.encode())
    features = recv_until(sock, )
    
    sock.sendall()
    resp = recv_until(sock, )
    (, resp)
:
    (, resp)

sock.close()

2b. In-Band Registration (XEP-0077)

If the stream features include <register>, test in-band registration:

#!/usr/bin/env python3
"""Test XEP-0077 in-band registration and register an account."""
import socket
import ssl
import sys

TARGET = sys.argv[1]
PORT = int(sys.argv[2]) if len(sys.argv) > 2 else 5222
DOMAIN = sys.argv[3] if len(sys.argv) > 3 else TARGET
USERNAME = sys.argv[4] if len(sys.argv) > 4 else "testuser123"
PASSWORD = sys.argv[5] if len(sys.argv) > 5 else "TestPass123!"

STREAM_HEADER = f'''<?xml version='1.0'?>
<stream:stream xmlns='jabber:client'
  xmlns:stream='http://etherx.jabber.org/streams'
  to='{DOMAIN}' version='1.0'>'''

def recv_until(sock, marker, timeout=10):
    sock.settimeout(timeout)
    data = b""
    while marker.encode() not in data:
        try:
            chunk = sock.recv(4096)
            if not chunk:
                break
            data += chunk
        except socket.timeout:
            break
    return data.decode(errors=)

sock = socket.create_connection((TARGET, PORT), timeout=)
sock.sendall(STREAM_HEADER.encode())
features = recv_until(sock, )


   features:
    sock.sendall()
    resp = recv_until(sock, )
       resp:
        ctx = ssl.create_default_context()
        ctx.check_hostname = 
        ctx.verify_mode = ssl.CERT_NONE
        sock = ctx.wrap_socket(sock, server_hostname=DOMAIN)
        sock.sendall(STREAM_HEADER.encode())
        features = recv_until(sock, )


sock.sendall(.encode())
resp = recv_until(sock, )
(, resp)

   resp.lower()    resp.lower():
    ()
    sock.close()
    sys.exit()


reg_iq = 
sock.sendall(reg_iq.encode())
resp = recv_until(sock, )
(, resp)

    resp:
    ()
   resp.lower():
    ()
    ()
:
    ()

sock.close()

If registration succeeds, you now have valid credentials. Report the registered account and proceed to user enumeration with authenticated access.

Step 3: User Enumeration

Enumerate valid usernames on the XMPP server. Multiple techniques available depending on access level.

3a. Registration Conflict Enumeration (Unauthenticated)

If in-band registration is enabled, you can enumerate users by attempting to register known usernames and checking for <conflict/> errors:

#!/usr/bin/env python3
"""Enumerate XMPP users via registration conflict errors."""
import socket
import ssl
import sys
import time

TARGET = sys.argv[1]
PORT = int(sys.argv[2]) if len(sys.argv) > 2 else 5222
DOMAIN = sys.argv[3] if len(sys.argv) > 3 else TARGET
USERFILE = sys.argv[4] if len(sys.argv) > 4 else "/usr/share/seclists/Usernames/xato-net-10-million-usernames-dup.txt"

STREAM_HEADER = f'''<?xml version='1.0'?>
<stream:stream xmlns='jabber:client'
  xmlns:stream='http://etherx.jabber.org/streams'
  to='{DOMAIN}' version='1.0'>'''

def recv_until(sock, marker, timeout=10):
    sock.settimeout(timeout)
    data = b""
    while marker.encode() not in data:
        try:
            chunk = sock.recv(4096)
            if not chunk:
                break
            data += chunk
        except socket.timeout:
            break
    return data.decode(errors="replace")

def connect_and_tls():
    sock = socket.create_connection((TARGET, PORT), timeout=)
    sock.sendall(STREAM_HEADER.encode())
    features = recv_until(sock, )
       features:
        sock.sendall()
        resp = recv_until(sock, )
           resp:
            ctx = ssl.create_default_context()
            ctx.check_hostname = 
            ctx.verify_mode = ssl.CERT_NONE
            sock = ctx.wrap_socket(sock, server_hostname=DOMAIN)
            sock.sendall(STREAM_HEADER.encode())
            recv_until(sock, )
     sock

found_users = []
 (USERFILE)  f:
    usernames = [line.strip()  line  f  line.strip()]

()

sock = connect_and_tls()
count = 

 username  usernames:
    :
        reg_iq = 
        sock.sendall(reg_iq.encode())
        resp = recv_until(sock, )

           resp.lower():
            ()
            found_users.append(username)
            resp:
            
            ()
            found_users.append(username)

        count += 
        
         count %  == :
            :
                sock.close()
             Exception:
                
            sock = connect_and_tls()
            ()

     Exception  e:
        ()
        :
            sock.close()
         Exception:
            
        sock = connect_and_tls()

sock.close()

()
 u  found_users:
    ()

Note: This technique registers accounts for non-existent usernames. Use a small, targeted username list (top 1000) rather than a massive wordlist to avoid creating thousands of accounts. If the engagement scope allows it, clean up registered accounts afterward.

3b. Roster/Contact Queries (Authenticated)

With authenticated access (from registration or provided credentials):

#!/usr/bin/env python3
"""Query user roster and service discovery with authenticated XMPP session."""
import socket
import ssl
import base64
import sys

TARGET = sys.argv[1]
PORT = int(sys.argv[2]) if len(sys.argv) > 2 else 5222
DOMAIN = sys.argv[3] if len(sys.argv) > 3 else TARGET
USERNAME = sys.argv[4]
PASSWORD = sys.argv[5]

STREAM_HEADER = f'''<?xml version='1.0'?>
<stream:stream xmlns='jabber:client'
  xmlns:stream='http://etherx.jabber.org/streams'
  to='{DOMAIN}' version='1.0'>'''

def recv_until(sock, marker, timeout=10):
    sock.settimeout(timeout)
    data = b""
    while marker.encode() not in data:
        try:
            chunk = sock.recv(4096)
            if not chunk:
                break
            data += chunk
        except socket.timeout:
            break
    return data.decode(errors="replace")

sock = socket.create_connection((TARGET, PORT), timeout=10)
sock.sendall(STREAM_HEADER.encode())
features = recv_until(sock, "</stream:features>")

# STARTTLS
if "<starttls"  features:
    sock.sendall()
    resp = recv_until(sock, )
       resp:
        ctx = ssl.create_default_context()
        ctx.check_hostname = 
        ctx.verify_mode = ssl.CERT_NONE
        sock = ctx.wrap_socket(sock, server_hostname=DOMAIN)
        sock.sendall(STREAM_HEADER.encode())
        features = recv_until(sock, )


auth_str = 
auth_b64 = base64.b64encode(auth_str.encode()).decode()
sock.sendall(.encode())
resp = recv_until(sock, )

    resp:
    ()
    sock.close()
    sys.exit()

()


sock.sendall(STREAM_HEADER.encode())
features = recv_until(sock, )


sock.sendall()
resp = recv_until(sock, )
()


sock.sendall()
resp = recv_until(sock, )
()


sock.sendall(.encode())
resp = recv_until(sock, )
()


sock.sendall(.encode())
resp = recv_until(sock, )
()


sock.sendall(.encode())
resp = recv_until(sock, )
()


    resp.lower():
    sock.sendall(.encode())
    resp = recv_until(sock, , timeout=)
    ()

sock.close()

Large user directories: If the search returns hundreds or thousands of users, save the full output to engagement/evidence/xmpp-users.txt and extract usernames for further use:

# Extract usernames from search results (adapt grep pattern to output format)
grep -oP 'value>\K[^<]+' engagement/evidence/xmpp-users.txt | sort -u > engagement/evidence/xmpp-usernames.txt

Step 4: MUC Room Discovery

Discover Multi-User Chat rooms and check for accessible rooms with message history.

#!/usr/bin/env python3
"""Discover MUC rooms and retrieve history from open rooms."""
import socket
import ssl
import base64
import sys
import re

TARGET = sys.argv[1]
PORT = int(sys.argv[2]) if len(sys.argv) > 2 else 5222
DOMAIN = sys.argv[3] if len(sys.argv) > 3 else TARGET
USERNAME = sys.argv[4]
PASSWORD = sys.argv[5]
MUC_SERVICE = sys.argv[6] if len(sys.argv) > 6 else f"conference.{DOMAIN}"

STREAM_HEADER = f'''<?xml version='1.0'?>
<stream:stream xmlns='jabber:client'
  xmlns:stream='http://etherx.jabber.org/streams'
  to='{DOMAIN}' version='1.0'>'''

def recv_until(sock, marker, timeout=10):
    sock.settimeout(timeout)
    data = b""
    while marker.encode() not in data:
        try:
            chunk = sock.recv(4096)
            if not chunk:
                break
            data += chunk
        except socket.timeout:
            break
    return data.decode(errors="replace")


sock = socket.create_connection((TARGET, PORT), timeout=)
sock.sendall(STREAM_HEADER.encode())
features = recv_until(sock, )

   features:
    sock.sendall()
    resp = recv_until(sock, )
       resp:
        ctx = ssl.create_default_context()
        ctx.check_hostname = 
        ctx.verify_mode = ssl.CERT_NONE
        sock = ctx.wrap_socket(sock, server_hostname=DOMAIN)
        sock.sendall(STREAM_HEADER.encode())
        features = recv_until(sock, )

auth_str = 
auth_b64 = base64.b64encode(auth_str.encode()).decode()
sock.sendall(.encode())
resp = recv_until(sock, )
    resp:
    ()
    sys.exit()

sock.sendall(STREAM_HEADER.encode())
recv_until(sock, )
sock.sendall()
recv_until(sock, )

()


sock.sendall(.encode())
resp = recv_until(sock, , timeout=)
()


rooms = re.findall(, resp)
  rooms:
    rooms = re.findall(, resp)

()


 room  rooms[:]:  
    ()
    
    join = 
    sock.sendall(join.encode())

    
    resp = recv_until(sock, , timeout=)
    
    :
        sock.settimeout()
        extra = sock.recv().decode(errors=)
        resp += extra
     socket.timeout:
        

       resp    resp:
        ()
    :
        msg_count = resp.count()
        ()
         msg_count > :
            ()

    
    sock.sendall(.encode())
    recv_until(sock, , timeout=)

sock.close()

Interesting findings in MUC rooms:

• Credentials shared in chat messages (passwords, tokens, API keys)
• Internal hostnames, IP addresses, and infrastructure details
• Employee names mapping to AD usernames
• Application URLs, deployment details, internal documentation links

Save any interesting room history to engagement/evidence/xmpp-room-<name>.txt.

Step 5: Information Gathering

5a. Server Version and Plugins

With authenticated access, query the server for detailed information:

# Query server version (XEP-0092)
sock.sendall(f"<iq type='get' id='ver1' to='{DOMAIN}'><query xmlns='jabber:iq:version'/></iq>".encode())
resp = recv_until(sock, "</iq>")
print(f"[*] Server version: {resp}")

# Query server uptime/stats (admin feature, may be restricted)
sock.sendall(f"<iq type='get' id='stats1' to='{DOMAIN}'><query xmlns='http://jabber.org/protocol/stats'/></iq>".encode())
resp = recv_until(sock, "</iq>")
print(f"[*] Server stats: {resp}")

5b. Known Vulnerabilities by Server Version

Server	Version	CVE	Impact
Openfire	< 4.7.5	CVE-2023-32315	Auth bypass → admin console → RCE
Openfire	< 4.6.8	CVE-2023-32315	Same — path traversal in admin
ejabberd	< 23.01	Various	Check NVD for version-specific
Prosody	< 0.12.3	CVE-2022-0217	Memory exhaustion DoS

If the server version matches a known vulnerable version, report it and return — do not attempt exploitation. The orchestrator will route to the appropriate technique skill.

5c. Admin Interface Detection

Openfire exposes admin consoles on separate ports:

• HTTP: http://<target>:9090
• HTTPS: https://<target>:9091

If these ports were found in the nmap scan, note them for web-discovery. Do not test the web admin interface from this skill.

Troubleshooting

Connection refused on 5222

• Server may only listen on 5223 (legacy TLS) — try direct TLS connection
• Check if a firewall is blocking — try from a different source IP
• Server may use non-standard ports — check nmap results for XMPP on other ports

STARTTLS fails

• Try direct TLS connection on port 5223
• Server may not support TLS — try plain connection (not recommended for auth)
• Certificate hostname mismatch — use -servername flag with openssl

Registration returns "not-allowed"

• In-band registration is disabled (server policy)
• Try SASL ANONYMOUS as an alternative for unauthenticated access
• Report finding and return — user enumeration requires credentials

Stream error: host-unknown

• Wrong domain in stream header — use the domain from TLS certificate or nmap
• Try variations: chat.corp.local, jabber.corp.local, xmpp.corp.local
• Check DNS/hosts file for the correct hostname mapping

Python socket timeout

• Increase timeout values in the scripts (default 10s may be too short)
• Server may be rate-limiting — add delays between requests
• Network latency — try direct TLS (5223) which skips the STARTTLS handshake

Large user directories (>1000 users)

• The search query may return paginated results — check for result set management
• Save output in chunks to avoid memory issues
• Focus on usernames that match AD naming patterns (first.last, flast, etc.)