Test XMPP Chatbot Connections

xmpp-debugger-skill

Minimal Python scripts for XMPP debugging: connect, send a chat message, and listen for replies.

Requirements

• Python 3.10+ (standard library only)
• Access to an XMPP server and valid account

Quick Start

export XMPP_PASSWORD='...'

python3 scripts/send_message.py \
  --jid user@example.com \
  --to bot@example.com \
  --body 'ping'

python3 scripts/listen_messages.py \
  --jid user@example.com \
  --filter-from bot@example.com \
  --timeout 20

Scripts

• scripts/xmpp_client.py: shared minimal XMPP client.
• scripts/send_message.py: sends one message and exits.
• scripts/listen_messages.py: listens for incoming messages and prints them.

Optional Flags

• --host: override XMPP host (otherwise taken from JID domain)
• --port: server port (default 5222)
• --resource: XMPP resource (default codex)
• --direct-tls: use direct TLS instead of STARTTLS
• --debug: print raw XMPP XML to stderr

Publish to GitHub

git init
git add .
git commit -m "Prepare xmpp debugger skill for public publishing"
git branch -M main
git remote add origin https://github.com/MaxeeTig/xmpp-debugger-skill.git
git push -u origin main

scripts/send_message.py

#!/usr/bin/env python3
"""Send one XMPP chat message using a configured account."""

from __future__ import annotations

import argparse

from xmpp_client import add_common_args, build_client


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    add_common_args(parser)
    parser.add_argument("--to", required=True, help="recipient JID")
    parser.add_argument("--body", required=True, help="message body")
    args = parser.parse_args()

    client = build_client(args)
    try:
        client.connect()
        client.send_message(args.to, args.body)
        print(f"message sent to {args.to}")
    finally:
        client.close()


if __name__ == "__main__":
    main()

scripts/listen_messages.py

#!/usr/bin/env python3
"""Listen for incoming XMPP chat messages and print matching replies."""

from __future__ import annotations

import argparse
import time

from xmpp_client import add_common_args, build_client


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    add_common_args(parser)
    parser.add_argument("--filter-from", help="only print messages from this JID")
    parser.add_argument(
        "--listen-for",
        type=float,
        default=None,
        help="seconds to keep listening; defaults to --timeout",
    )
    args = parser.parse_args()

    listen_for = args.listen_for if args.listen_for is not None else args.timeout
    deadline = time.time() + listen_for
    client = build_client(args)
    found = 0

    try:
        client.connect()
        client.send_presence()
        while time.time() < deadline:
            remaining = max(0.1, deadline - time.time())
            message = client.next_message(timeout=remaining, filter_from=args.filter_from)
            if message is None:
                continue
            found += 1
            print(f"from={message.sender}")
            print(f"body={message.body}")
            print("---")
    finally:
        client.close()

    if found == 0:
        print("no messages received")


if __name__ == "__main__":
    main()

scripts/xmpp_client.py

#!/usr/bin/env python3
"""Minimal XMPP client for simple send/listen debugging flows."""

from __future__ import annotations

import argparse
import base64
import os
import select
import socket
import ssl
import sys
import time
import uuid
from dataclasses import dataclass
from html import escape
from typing import Optional
from xml.etree import ElementTree as ET


JABBER_CLIENT_NS = "jabber:client"
SASL_NS = "urn:ietf:params:xml:ns:xmpp-sasl"
BIND_NS = "urn:ietf:params:xml:ns:xmpp-bind"
SESSION_NS = "urn:ietf:params:xml:ns:xmpp-session"
TLS_NS = "urn:ietf:params:xml:ns:xmpp-tls"
STREAM_NS = "http://etherx.jabber.org/streams"
XML_NS = "http://www.w3.org/XML/1998/namespace"
NS = {
    "jc": JABBER_CLIENT_NS,
    "sasl": SASL_NS,
    "bind": BIND_NS,
    "sess": SESSION_NS,
    "tls": TLS_NS,
}


def _require_password(env_name: str) -> str:
    value = os.environ.get(env_name)
    if not value:
        raise SystemExit(f"missing password in environment variable {env_name}")
    return value


def _split_jid(jid: str) -> tuple[str, str]:
    if "@" not in jid:
        raise SystemExit(f"invalid JID: {jid}")
    user, domain = jid.split("@", 1)
    if not user or not domain:
        raise SystemExit(f"invalid JID: {jid}")
    return user, domain


def _extract_message_stanza(buffer: str) -> tuple[Optional[str], str]:
    start = buffer.find("<message")
    if start == -1:
        return None, buffer[-8192:]
    end = buffer.find("</message>", start)
    if end == -1:
        return None, buffer[start:]
    end += len("</message>")
    return buffer[start:end], buffer[end:]


def _extract_iq_stanza(buffer: str) -> tuple[Optional[str], str]:
    start = buffer.find("<iq")
    if start == -1:
        return None, buffer[-8192:]
    end = buffer.find("</iq>", start)
    if end == -1:
        return None, buffer[start:]
    end += len("</iq>")
    return buffer[start:end], buffer[end:]


@dataclass
class ChatMessage:
    sender: str
    body: str
    stanza: str


class XmppClient:
    def __init__(
        self,
        jid: str,
        password: str,
        host: Optional[str] = None,
        port: int = 5222,
        resource: str = "codex",
        debug: bool = False,
        timeout: float = 10.0,
        starttls: bool = True,
    ) -> None:
        self.jid = jid
        self.password = password
        self.host = host or _split_jid(jid)[1]
        self.port = port
        self.resource = resource
        self.debug = debug
        self.timeout = timeout
        self.starttls = starttls
        self.sock: Optional[socket.socket] = None
        self.stream_buffer = ""
        self.bound_jid: Optional[str] = None

    def _log(self, prefix: str, payload: str) -> None:
        if self.debug:
            print(f"{prefix} {payload}", file=sys.stderr)

    def _send_xml(self, xml: str) -> None:
        assert self.sock is not None
        self._log(">>>", xml)
        self.sock.sendall(xml.encode("utf-8"))

    def _recv_text(self, deadline: float) -> str:
        assert self.sock is not None
        while time.time() < deadline:
            timeout = max(0.0, deadline - time.time())
            ready, _, _ = select.select([self.sock], [], [], timeout)
            if not ready:
                continue
            chunk = self.sock.recv(65535)
            if not chunk:
                raise RuntimeError("server closed the XMPP stream")
            text = chunk.decode("utf-8", errors="replace")
            self._log("<<<", text)
            self.stream_buffer += text
            return text
        raise TimeoutError("timed out waiting for XMPP data")

    def _wait_for(self, *needles: str, timeout: Optional[float] = None) -> str:
        deadline = time.time() + (timeout or self.timeout)
        while time.time() < deadline:
            if any(needle in self.stream_buffer for needle in needles):
                data = self.stream_buffer
                self.stream_buffer = ""
                return data
            self._recv_text(deadline)
        raise TimeoutError(f"timed out waiting for: {', '.join(needles)}")

    def _open_stream(self) -> str:
        domain = _split_jid(self.jid)[1]
        xml = (
            "<?xml version='1.0'?>"
            f"<stream:stream to='{escape(domain, quote=True)}' "
            f"xmlns='{JABBER_CLIENT_NS}' "
            f"xmlns:stream='{STREAM_NS}' version='1.0'>"
        )
        self._send_xml(xml)
        return self._wait_for("<stream:features", "<stream:stream")

    def connect(self) -> None:
        raw = socket.create_connection((self.host, self.port), timeout=self.timeout)
        self.sock = raw
        self.stream_buffer = ""
        features = self._open_stream()
        if self.starttls:
            self._starttls(features)
            self._open_stream()
        self._authenticate_plain()
        self.stream_buffer = self._open_stream()
        self._bind_resource()
        self._maybe_start_session()

    def close(self) -> None:
        if self.sock is None:
            return
        try:
            self._send_xml("</stream:stream>")
        except Exception:
            pass
        try:
            self.sock.close()
        finally:
            self.sock = None

    def _authenticate_plain(self) -> None:
        user, _domain = _split_jid(self.jid)
        token = base64.b64encode(f"\x00{user}\x00{self.password}".encode("utf-8")).decode("ascii")
        self._send_xml(
            f"<auth xmlns='{SASL_NS}' mechanism='PLAIN'>{token}</auth>"
        )
        data = self._wait_for("<success", "<failure")
        if "<failure" in data:
            raise RuntimeError("XMPP authentication failed")

    def _starttls(self, features: str) -> None:
        if "<starttls" not in features:
            raise RuntimeError("XMPP server does not advertise STARTTLS")
        self._send_xml(f"<starttls xmlns='{TLS_NS}'/>")
        data = self._wait_for("<proceed", "<failure")
        if "<failure" in data:
            raise RuntimeError("XMPP STARTTLS negotiation failed")
        assert self.sock is not None
        context = ssl.create_default_context()
        self.sock = context.wrap_socket(self.sock, server_hostname=self.host)
        self.stream_buffer = ""

    def _bind_resource(self) -> None:
        iq_id = f"bind-{uuid.uuid4().hex[:8]}"
        self._send_xml(
            f"<iq type='set' id='{iq_id}' xmlns='{JABBER_CLIENT_NS}'>"
            f"<bind xmlns='{BIND_NS}'><resource>{escape(self.resource)}</resource></bind>"
            "</iq>"
        )
        stanza = self._wait_for_iq(iq_id)
        element = ET.fromstring(stanza)
        if element.attrib.get("type") != "result":
            raise RuntimeError("resource binding failed")
        jid_node = element.find(".//bind:jid", NS)
        if jid_node is None or not jid_node.text:
            raise RuntimeError("server did not return a bound JID")
        self.bound_jid = jid_node.text

    def _maybe_start_session(self) -> None:
        if "xmpp-session" not in self.stream_buffer and SESSION_NS not in self.stream_buffer:
            return
        iq_id = f"sess-{uuid.uuid4().hex[:8]}"
        self._send_xml(
            f"<iq to='{escape(_split_jid(self.jid)[1], quote=True)}' type='set' id='{iq_id}' xmlns='{JABBER_CLIENT_NS}'>"
            f"<session xmlns='{SESSION_NS}'/>"
            "</iq>"
        )
        stanza = self._wait_for_iq(iq_id)
        element = ET.fromstring(stanza)
        if element.attrib.get("type") != "result":
            raise RuntimeError("XMPP session start failed")

    def _wait_for_iq(self, iq_id: str) -> str:
        deadline = time.time() + self.timeout
        buffer = self.stream_buffer
        self.stream_buffer = ""
        while time.time() < deadline:
            stanza, buffer = _extract_iq_stanza(buffer)
            if stanza is None:
                self._recv_text(deadline)
                buffer += self.stream_buffer
                self.stream_buffer = ""
                continue
            element = ET.fromstring(stanza)
            if element.attrib.get("id") == iq_id:
                self.stream_buffer = buffer
                return stanza
        raise TimeoutError(f"timed out waiting for iq id {iq_id}")

    def send_presence(self) -> None:
        self._send_xml("<presence/>")

    def send_message(self, recipient: str, body: str, message_type: str = "chat") -> None:
        xml = (
            f"<message to='{escape(recipient, quote=True)}' type='{escape(message_type, quote=True)}'>"
            f"<body>{escape(body)}</body>"
            "</message>"
        )
        self._send_xml(xml)

    def next_message(
        self,
        timeout: float,
        filter_from: Optional[str] = None,
    ) -> Optional[ChatMessage]:
        deadline = time.time() + timeout
        buffer = self.stream_buffer
        self.stream_buffer = ""
        while time.time() < deadline:
            stanza, buffer = _extract_message_stanza(buffer)
            if stanza is None:
                self._recv_text(deadline)
                buffer += self.stream_buffer
                self.stream_buffer = ""
                continue
            message = self._parse_message(stanza)
            if message is None:
                continue
            if filter_from and _bare_jid(message.sender) != _bare_jid(filter_from):
                continue
            self.stream_buffer = buffer
            return message
        self.stream_buffer = buffer
        return None

    def _parse_message(self, stanza: str) -> Optional[ChatMessage]:
        element = ET.fromstring(stanza)
        if element.tag not in {f"{{{JABBER_CLIENT_NS}}}message", "message"}:
            return None
        body_node = element.find("jc:body", NS)
        if body_node is None:
            body_node = element.find("body")
        if body_node is None or body_node.text is None:
            return None
        sender = element.attrib.get("from", "")
        return ChatMessage(sender=sender, body=body_node.text, stanza=stanza)


def _bare_jid(jid: str) -> str:
    return jid.split("/", 1)[0]


def add_common_args(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("--jid", required=True, help="login JID (for example: user@example.com)")
    parser.add_argument(
        "--password-env",
        default="XMPP_PASSWORD",
        help="environment variable containing the account password",
    )
    parser.add_argument("--host", help="XMPP server host; defaults to the JID domain")
    parser.add_argument("--port", type=int, default=5222, help="XMPP server port")
    parser.add_argument("--resource", default="codex", help="resource to bind")
    parser.add_argument("--debug", action="store_true", help="print raw XML to stderr")
    parser.add_argument("--timeout", type=float, default=10.0, help="network timeout in seconds")
    parser.add_argument(
        "--direct-tls",
        action="store_true",
        help="use direct TLS immediately after TCP connect instead of STARTTLS",
    )


def build_client(args: argparse.Namespace) -> XmppClient:
    return XmppClient(
        jid=args.jid,
        password=_require_password(args.password_env),
        host=args.host,
        port=args.port,
        resource=args.resource,
        debug=args.debug,
        timeout=args.timeout,
        starttls=not args.direct_tls,
    )