diff --git a/plugins/platforms/xmpp/__init__.py b/plugins/platforms/xmpp/__init__.py new file mode 100644 index 000000000000..d4f1d7bf0e3f --- /dev/null +++ b/plugins/platforms/xmpp/__init__.py @@ -0,0 +1,3 @@ +from .adapter import register + +__all__ = ["register"] diff --git a/plugins/platforms/xmpp/adapter.py b/plugins/platforms/xmpp/adapter.py new file mode 100644 index 000000000000..afce9855b66d --- /dev/null +++ b/plugins/platforms/xmpp/adapter.py @@ -0,0 +1,1344 @@ +"""XMPP platform adapter (Hermes plugin). + +Connects to a self-hosted or hosted XMPP server using slixmpp. Inbound +```` and addressed MUC messages are delivered to the +Hermes agent; outbound replies are sent back as the same stanza type. + +Plugin layout follows the SimpleX / IRC / LINE conventions: + +* ``check_requirements`` gates on ``XMPP_JID`` + ``XMPP_PASSWORD`` + the + ``slixmpp`` package importing. Missing any of the three keeps the + platform out of ``get_connected_platforms()`` so the gateway never + instantiates the adapter. +* ``_env_enablement`` seeds ``PlatformConfig.extra`` from env so + ``hermes gateway status`` reflects env-only setups without spinning + up the XMPP client. +* ``_standalone_send`` opens an ephemeral session for cron jobs that run + separately from the gateway. + +Required environment variables: + XMPP_JID Bare JID the bot logs in as + XMPP_PASSWORD Password for the bot account + +Optional environment variables: + XMPP_HOST Server host override (default: SRV lookup) + XMPP_PORT Server port override (default: 5222) + XMPP_FORCE_STARTTLS Require STARTTLS (default: true) + XMPP_NICKNAME MUC nickname (default: JID local part) + XMPP_ROOMS Comma-separated MUC JIDs to join + XMPP_ALLOWED_USERS Comma-separated bare JIDs allowlist + XMPP_ALLOW_ALL_USERS true = bypass the allowlist (dev only) + XMPP_HOME_CHANNEL Default target for cron deliveries + XMPP_HOME_CHANNEL_NAME Display name for the home channel + XMPP_UPLOAD_SERVICE Pin the XEP-0363 upload service JID + (default: SRV-style auto-discovery) + XMPP_HTML_FORMATTING Emit XEP-0071 XHTML-IM dual body (default: true) + +The ``slixmpp`` Python package is imported lazily — the plugin stays +importable for discovery and setup-time prompts even when slixmpp is not +installed. +""" + +from __future__ import annotations + +import asyncio +import logging +import os +import re +import time +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +from gateway.config import Platform, PlatformConfig +from gateway.platforms.base import ( + BasePlatformAdapter, + MessageEvent, + MessageType, + SendResult, +) + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- +MAX_MESSAGE_LENGTH = 16_000 # XMPP has no hard limit; cap stanza body sanely +DEFAULT_PORT = 5222 +CONNECT_TIMEOUT_SECONDS = 20.0 +DISCONNECT_TIMEOUT_SECONDS = 5.0 +MUC_PREFIX = "muc:" + +_TRUTHY = frozenset({"1", "true", "yes", "on"}) + + +def _parse_bool(value: Any, default: bool) -> bool: + if isinstance(value, bool): + return value + if value is None: + return default + return str(value).strip().lower() in _TRUTHY + + +def _parse_comma_list(value: str) -> List[str]: + return [v.strip() for v in (value or "").split(",") if v.strip()] + + +def _strip_resource(jid: str) -> str: + """Return the bare JID (``user@server``) by dropping any ``/resource``.""" + return (jid or "").split("/", 1)[0] + + +def _strip_markdown(text: str) -> str: + """Convert markdown to plain text for the XMPP ```` fallback. + + XHTML-IM aware clients render the parallel ```` element produced + by :func:`_format_html`; this strip keeps the plain-body fallback + readable on clients that do not. + """ + # Bold / italic / strikethrough + text = re.sub(r"\*\*(.+?)\*\*", r"\1", text) + text = re.sub(r"__(.+?)__", r"\1", text) + text = re.sub(r"\*(.+?)\*", r"\1", text) + text = re.sub(r"(? str: + """Strip the few characters XML 1.0 cannot encode in a message body. + + Slixmpp serializes stanzas as XML 1.0; raw NUL or stray vertical-tab + bytes get rejected by the underlying parser. CR/LF are valid in a + body, but normalizing them to ``\\n`` keeps logs readable. + """ + cleaned = [] + for ch in text or "": + cp = ord(ch) + if cp == 0x00: + continue + if cp < 0x20 and ch not in ("\t", "\n", "\r"): + continue + cleaned.append(ch) + return "".join(cleaned).replace("\r\n", "\n").replace("\r", "\n") + + +def _split_message(content: str, max_bytes: int) -> List[str]: + """Split ``content`` on paragraph boundaries to keep each chunk under + ``max_bytes`` of UTF-8. No code-block awareness — XMPP renders plain + text bodies and the stripper above has already removed fences.""" + chunks: List[str] = [] + for paragraph in content.split("\n"): + if not paragraph.strip(): + continue + while True: + data = paragraph.encode("utf-8") + if len(data) <= max_bytes: + chunks.append(paragraph) + break + # Binary search for the largest prefix that fits within max_bytes + low, high, best = 1, len(paragraph), 0 + while low <= high: + mid = (low + high) // 2 + if len(paragraph[:mid].encode("utf-8")) <= max_bytes: + best = mid + low = mid + 1 + else: + high = mid - 1 + split_at = best + space = paragraph.rfind(" ", 0, split_at) + if space > split_at // 3: + split_at = space + chunks.append(paragraph[:split_at].rstrip()) + paragraph = paragraph[split_at:].lstrip() + return chunks if chunks else [""] + + +# --------------------------------------------------------------------------- +# XHTML-IM helpers (XEP-0071) +# --------------------------------------------------------------------------- + +# XHTML-IM allows a restricted subset of HTML. We target what the popular +# mobile + desktop clients (Conversations, Dino, Gajim, Profanity) all +# render: p, br, strong, em, code, pre, a, ul, ol, li, blockquote, span. +_XHTML_IM_NS = "http://www.w3.org/1999/xhtml" + + +def _html_escape(text: str) -> str: + return ( + text.replace("&", "&") + .replace("<", "<") + .replace(">", ">") + .replace('"', """) + ) + + +def _sanitize_link_url(url: str) -> str: + """Reject dangerous schemes (``javascript:``, ``data:``, ``vbscript:``).""" + stripped = (url or "").strip() + scheme = stripped.split(":", 1)[0].lower().strip() if ":" in stripped else "" + if scheme in {"javascript", "data", "vbscript"}: + return "" + return stripped.replace('"', """) + + +def _markdown_to_xhtml_im(text: str) -> str: + """Convert markdown to a safe XHTML-IM subset. + + Uses the ``markdown`` library when installed (matches the Matrix + adapter's path; ``markdown`` is pinned for the ``matrix`` extra and + is therefore commonly available in Hermes installs). Falls back to a + regex pipeline when not. + + Returns an XHTML fragment suitable for assignment to + ``msg['html']['body']`` in slixmpp. The caller wraps it in a single + ``

...

`` envelope so the result is well-formed even when the + markdown source is empty. + """ + text = text or "" + if not text.strip(): + return "" + + try: + import markdown as _md # type: ignore[import-untyped] + + md = _md.Markdown(extensions=["fenced_code", "nl2br", "sane_lists"]) + if "html_block" in md.preprocessors: + md.preprocessors.deregister("html_block") + html = md.convert(text) + md.reset() + # ``markdown`` wraps single-paragraph output in a stray

; XMPP + # clients render dual ``

`` blocks with a leading blank line, so + # strip the wrapper when there is only one. + if html.count("

") == 1: + html = html.replace("

", "", 1).replace("

", "", 1) + return html.strip() + except ImportError: + return _markdown_to_xhtml_im_fallback(text) + + +def _markdown_to_xhtml_im_fallback(text: str) -> str: + """Regex Markdown-to-XHTML-IM for installs without the ``markdown`` package.""" + placeholders: List[str] = [] + + def _protect(html_fragment: str) -> str: + placeholders.append(html_fragment) + return f"\x00P{len(placeholders) - 1}\x00" + + # Fenced code blocks: ```lang\n...\n``` + result = re.sub( + r"```[a-zA-Z0-9_+\-]*\n?(.*?)```", + lambda m: _protect(f"
{_html_escape(m.group(1))}
"), + text, + flags=re.DOTALL, + ) + # Inline code + result = re.sub( + r"`([^`\n]+)`", + lambda m: _protect(f"{_html_escape(m.group(1))}"), + result, + ) + # Images (extract URL only — XHTML-IM ```` is poorly supported) + result = re.sub( + r"!\[([^\]]*)\]\(([^)]+)\)", + lambda m: _protect(_html_escape(m.group(2))), + result, + ) + # Links: [text](url) + result = re.sub( + r"\[([^\]]+)\]\(([^)]+)\)", + lambda m: _protect( + '{}'.format( + _sanitize_link_url(m.group(2)), + _html_escape(m.group(1)), + ) + ) + if _sanitize_link_url(m.group(2)) + else _protect(_html_escape(m.group(1))), + result, + ) + # HTML-escape the rest + parts = re.split(r"(\x00P\d+\x00)", result) + for idx, part in enumerate(parts): + if not part.startswith("\x00P"): + parts[idx] = _html_escape(part) + result = "".join(parts) + + # Inline emphasis on the escaped+protected string + result = re.sub(r"\*\*(.+?)\*\*", r"\1", result) + result = re.sub(r"__(.+?)__", r"\1", result) + result = re.sub(r"(?\1", result) + result = re.sub(r"(?\1", result) + + # Block-level: headings → ; lists; blockquotes; paragraphs + lines = result.split("\n") + out_lines: List[str] = [] + i = 0 + while i < len(lines): + line = lines[i] + hdr = re.match(r"^(#{1,6})\s+(.+)$", line) + if hdr: + out_lines.append(f"

{hdr.group(2).strip()}

") + i += 1 + continue + if re.match(r"^(\s*[-*]\s+)", line): + bullets = [] + while i < len(lines) and re.match(r"^(\s*[-*]\s+)", lines[i]): + bullets.append(re.sub(r"^\s*[-*]\s+", "", lines[i])) + i += 1 + out_lines.append("") + continue + if re.match(r"^\s*\d+\.\s+", line): + items = [] + while i < len(lines) and re.match(r"^\s*\d+\.\s+", lines[i]): + items.append(re.sub(r"^\s*\d+\.\s+", "", lines[i])) + i += 1 + out_lines.append("
    " + "".join(f"
  1. {it}
    2. " for it in items) + "
") + continue + if line.startswith("> ") or line == ">": + bq = [] + while i < len(lines) and (lines[i].startswith("> ") or lines[i] == ">"): + bq.append(lines[i][5:] if lines[i].startswith("> ") else "") + i += 1 + out_lines.append("
" + "
".join(bq) + "
") + continue + if not line.strip(): + i += 1 + continue + # Plain paragraph + para = [line] + i += 1 + while i < len(lines) and lines[i].strip() and not _starts_block(lines[i]): + para.append(lines[i]) + i += 1 + out_lines.append("

" + "
".join(para) + "

") + + rendered = "".join(out_lines) + # Restore protected fragments + rendered = re.sub( + r"\x00P(\d+)\x00", + lambda m: placeholders[int(m.group(1))], + rendered, + ) + return rendered + + +def _starts_block(line: str) -> bool: + return bool( + re.match(r"^(#{1,6})\s+", line) + or re.match(r"^(\s*[-*]\s+)", line) + or re.match(r"^\s*\d+\.\s+", line) + or line.startswith("> ") + or line == ">" + ) + + +# --------------------------------------------------------------------------- +# Media / file helpers +# --------------------------------------------------------------------------- + +_IMAGE_EXTS = frozenset({".png", ".jpg", ".jpeg", ".gif", ".webp", ".heic"}) +_AUDIO_EXTS = frozenset({".ogg", ".opus", ".mp3", ".wav", ".m4a", ".flac", ".aac"}) +_VIDEO_EXTS = frozenset({".mp4", ".webm", ".mov", ".mkv", ".avi"}) + +_CONTENT_TYPE_BY_EXT = { + ".png": "image/png", + ".jpg": "image/jpeg", + ".jpeg": "image/jpeg", + ".gif": "image/gif", + ".webp": "image/webp", + ".heic": "image/heic", + ".ogg": "audio/ogg", + ".opus": "audio/ogg", + ".mp3": "audio/mpeg", + ".wav": "audio/wav", + ".m4a": "audio/mp4", + ".flac": "audio/flac", + ".aac": "audio/aac", + ".mp4": "video/mp4", + ".webm": "video/webm", + ".mov": "video/quicktime", + ".mkv": "video/x-matroska", + ".avi": "video/x-msvideo", + ".pdf": "application/pdf", + ".zip": "application/zip", + ".txt": "text/plain", +} + + +def _content_type_for(path: str) -> str: + _, dot, ext = path.rpartition(".") + if not dot: + return "application/octet-stream" + return _CONTENT_TYPE_BY_EXT.get("." + ext.lower(), "application/octet-stream") + + +class _UploadUnavailable(Exception): + """Raised when HTTP Upload is not usable for the current request.""" + + +# --------------------------------------------------------------------------- +# Adapter +# --------------------------------------------------------------------------- + + +class XMPPAdapter(BasePlatformAdapter): + """XMPP adapter using slixmpp's asyncio ClientXMPP. + + Instantiated by the ``adapter_factory`` passed to + ``ctx.register_platform()`` in :func:`register`. + """ + + def __init__(self, config: PlatformConfig, **kwargs): + platform = Platform("xmpp") + super().__init__(config=config, platform=platform) + + extra = getattr(config, "extra", {}) or {} + + self.jid = (os.getenv("XMPP_JID") or extra.get("jid") or "").strip() + self.password = os.getenv("XMPP_PASSWORD") or extra.get("password", "") + self.host = (os.getenv("XMPP_HOST") or extra.get("host") or "").strip() or None + + port_value = os.getenv("XMPP_PORT") or extra.get("port") or DEFAULT_PORT + try: + self.port = int(port_value) + except (TypeError, ValueError): + self.port = DEFAULT_PORT + + self.force_starttls = _parse_bool( + os.getenv("XMPP_FORCE_STARTTLS", extra.get("force_starttls", True)), + default=True, + ) + + bare = _strip_resource(self.jid) + local_part = bare.split("@", 1)[0] if "@" in bare else (bare or "hermes") + self.nickname = ( + os.getenv("XMPP_NICKNAME") + or extra.get("nickname") + or local_part + ).strip() or "hermes" + + rooms_raw = os.getenv("XMPP_ROOMS") or extra.get("rooms", "") + if isinstance(rooms_raw, list): + self.rooms = [str(r).strip() for r in rooms_raw if str(r).strip()] + else: + self.rooms = _parse_comma_list(rooms_raw) + + self.max_message_length = int(extra.get("max_message_length") or MAX_MESSAGE_LENGTH) + + self.upload_service = ( + os.getenv("XMPP_UPLOAD_SERVICE") or extra.get("upload_service") or "" + ).strip() or None + self.html_formatting = _parse_bool( + os.getenv("XMPP_HTML_FORMATTING", extra.get("html_formatting", True)), + default=True, + ) + + # Runtime state + self._client = None # slixmpp.ClientXMPP + self._lock_key: Optional[str] = None + self._connected_event = asyncio.Event() + self._closing = False + # Upload service discovery: None = not probed yet, False = unavailable, + # str = service JID. Cached on session_start to avoid per-send discovery. + self._upload_service_resolved: Any = None + + @property + def name(self) -> str: + return "XMPP" + + # ------------------------------------------------------------------ + # Lifecycle + # ------------------------------------------------------------------ + + async def connect(self) -> bool: + if not self.jid or not self.password: + logger.error("XMPP: XMPP_JID and XMPP_PASSWORD are required") + self._set_fatal_error( + "config_missing", + "XMPP_JID and XMPP_PASSWORD must be set", + retryable=False, + ) + return False + + try: + import slixmpp # noqa: F401 + except ImportError: + logger.error("XMPP: 'slixmpp' package not installed. Run: pip install slixmpp") + self._set_fatal_error( + "missing_dependency", + "slixmpp package is not installed", + retryable=False, + ) + return False + + # Prevent two profiles from logging in as the same JID. + try: + from gateway.status import acquire_scoped_lock + if not acquire_scoped_lock("xmpp", _strip_resource(self.jid)): + logger.error("XMPP: %s already in use by another profile", _strip_resource(self.jid)) + self._set_fatal_error( + "lock_conflict", + "XMPP identity in use by another profile", + retryable=False, + ) + return False + self._lock_key = _strip_resource(self.jid) + except ImportError: + self._lock_key = None # status module not available (tests) + + from slixmpp import ClientXMPP + + client = ClientXMPP(self.jid, self.password) + client.register_plugin("xep_0030") # Service Discovery + client.register_plugin("xep_0199") # XMPP Ping + client.register_plugin("xep_0045") # Multi-User Chat + client.register_plugin("xep_0066") # Out-of-band data (attachment URLs) + client.register_plugin("xep_0363") # HTTP File Upload + if self.html_formatting: + client.register_plugin("xep_0071") # XHTML-IM + + # TLS toggles. ``enable_starttls`` controls whether slixmpp negotiates + # STARTTLS during stream negotiation; ``enable_direct_tls`` covers + # direct-TLS ports (5223). When ``XMPP_FORCE_STARTTLS=false`` we + # disable both so plaintext local-loopback dev servers work. + client.enable_starttls = self.force_starttls + client.enable_direct_tls = self.force_starttls + if not self.force_starttls: + logger.warning( + "XMPP: TLS disabled (XMPP_FORCE_STARTTLS=false). " + "Only do this on trusted local networks." + ) + + client.add_event_handler("session_start", self._on_session_start) + client.add_event_handler("message", self._on_message) + client.add_event_handler("groupchat_message", self._on_groupchat_message) + client.add_event_handler("disconnected", self._on_disconnected) + client.add_event_handler("failed_auth", self._on_failed_auth) + + self._client = client + self._closing = False + self._connected_event = asyncio.Event() + + host = self.host + port = self.port if host else None # slixmpp does SRV when host omitted + client.connect(host=host, port=port) + + try: + await asyncio.wait_for(self._connected_event.wait(), timeout=CONNECT_TIMEOUT_SECONDS) + except asyncio.TimeoutError: + logger.error("XMPP: connect timed out after %.0fs", CONNECT_TIMEOUT_SECONDS) + await self._safe_disconnect() + self._set_fatal_error("connect_timeout", "XMPP session_start not received", retryable=True) + return False + + if self.has_fatal_error: + return False + + self._mark_connected() + logger.info("XMPP: connected as %s", _strip_resource(self.jid)) + return True + + async def disconnect(self) -> None: + self._closing = True + if self._lock_key: + try: + from gateway.status import release_scoped_lock + release_scoped_lock("xmpp", self._lock_key) + except Exception: + pass + self._lock_key = None + await self._safe_disconnect() + self._client = None + self._mark_disconnected() + + async def _safe_disconnect(self) -> None: + client = self._client + if client is None: + return + try: + future = client.disconnect(wait=DISCONNECT_TIMEOUT_SECONDS) + if future is not None: + await asyncio.wait_for(asyncio.shield(asyncio.ensure_future(future)), timeout=DISCONNECT_TIMEOUT_SECONDS + 2.0) + except (asyncio.TimeoutError, Exception): + # Best-effort: never raise out of disconnect. + pass + + # ------------------------------------------------------------------ + # Event handlers + # ------------------------------------------------------------------ + + async def _on_session_start(self, event): + client = self._client + if client is None: + return + try: + client.send_presence() + await client.get_roster() + except Exception: + logger.exception("XMPP: error during session start") + + # Join any pre-configured MUC rooms. + muc = client.plugin.get("xep_0045") + if muc is not None and self.rooms: + for room in self.rooms: + try: + muc.join_muc(room, self.nickname, wait=False) + logger.info("XMPP: joining MUC %s as %s", room, self.nickname) + except Exception: + logger.exception("XMPP: failed to join MUC %s", room) + + # Pre-warm the HTTP Upload service so the first attachment send does + # not pay the discovery round-trip. Failure (no upload component on + # the server) becomes a sticky False so subsequent send_image_file / + # send_document calls fall back to text immediately. + await self._resolve_upload_service() + + self._connected_event.set() + + async def _resolve_upload_service(self) -> None: + client = self._client + if client is None: + self._upload_service_resolved = False + return + if self.upload_service: + self._upload_service_resolved = self.upload_service + logger.info("XMPP: using pinned upload service %s", self.upload_service) + return + try: + upload = client.plugin["xep_0363"] + iq = await upload.find_upload_service() + if iq is None: + logger.info( + "XMPP: no HTTP Upload service advertised by %s; " + "media will be sent as text", _strip_resource(self.jid).split("@", 1)[-1] + ) + self._upload_service_resolved = False + return + service_jid = str(iq["from"]) if iq.get("from") else True + self._upload_service_resolved = service_jid + logger.info("XMPP: discovered HTTP Upload service: %s", service_jid) + except Exception: + logger.warning( + "XMPP: HTTP Upload discovery failed; media will be sent as text", + exc_info=True, + ) + self._upload_service_resolved = False + + def _on_failed_auth(self, event): + logger.error("XMPP: authentication failed for %s", _strip_resource(self.jid)) + self._set_fatal_error("auth_failed", "XMPP authentication failed", retryable=False) + # Release the wait_for in connect() + self._connected_event.set() + + async def _on_disconnected(self, event): + if self._closing: + return + if self.is_connected: + logger.warning("XMPP: lost connection, marking disconnected") + self._set_fatal_error("connection_lost", "XMPP connection closed unexpectedly", retryable=True) + try: + await self._notify_fatal_error() + except Exception: + logger.exception("XMPP: error notifying fatal error") + + async def _on_message(self, msg): + """Handle a 1:1 ````.""" + if msg["type"] not in ("chat", "normal"): + return + text = msg.get("body") or "" + if not text.strip(): + return # ignore chat-state / typing-only stanzas + + sender_jid_full = str(msg["from"]) + sender_jid = _strip_resource(sender_jid_full) + if sender_jid == _strip_resource(self.jid): + return # own echo (shouldn't happen for type=chat but be defensive) + + await self._dispatch( + text=text, + chat_id=sender_jid, + chat_name=sender_jid, + chat_type="dm", + user_id=sender_jid, + user_name=sender_jid, + message_id=str(msg.get("id") or ""), + ) + + async def _on_groupchat_message(self, msg): + """Handle a MUC ````.""" + if msg["type"] != "groupchat": + return + text = msg.get("body") or "" + if not text.strip(): + return + + room_jid = _strip_resource(str(msg["from"])) + sender_nick = msg.get("mucnick") or "" + if not sender_nick or sender_nick == self.nickname: + return # own echo / system message + + # Only respond when addressed by our nickname (matches IRC behavior). + prefixes = ( + f"{self.nickname}:", + f"{self.nickname},", + f"{self.nickname} ", + f"@{self.nickname} ", + ) + addressed = False + for prefix in prefixes: + if text.lower().startswith(prefix.lower()): + text = text[len(prefix):].strip() + addressed = True + break + if not addressed: + return + + chat_id = f"{MUC_PREFIX}{room_jid}" + await self._dispatch( + text=text, + chat_id=chat_id, + chat_name=room_jid, + chat_type="group", + user_id=sender_nick, + user_name=sender_nick, + message_id=str(msg.get("id") or ""), + ) + + async def _dispatch( + self, + *, + text: str, + chat_id: str, + chat_name: str, + chat_type: str, + user_id: str, + user_name: str, + message_id: str = "", + ) -> None: + if not self._message_handler: + return + source = self.build_source( + chat_id=chat_id, + chat_name=chat_name, + chat_type=chat_type, + user_id=user_id, + user_name=user_name, + ) + event = MessageEvent( + text=text, + message_type=MessageType.TEXT, + source=source, + message_id=message_id or str(int(time.time() * 1000)), + timestamp=datetime.now(tz=timezone.utc), + ) + await self.handle_message(event) + + # ------------------------------------------------------------------ + # Outbound + # ------------------------------------------------------------------ + + async def send( + self, + chat_id: str, + content: str, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + if self._client is None: + return SendResult(success=False, error="Not connected") + + target, mtype = self._resolve_target(chat_id) + if not target: + return SendResult(success=False, error=f"Invalid XMPP target: {chat_id!r}") + + plain = _strip_control_chars(_strip_markdown(content or "")) + if not plain.strip(): + return SendResult(success=True, message_id=str(int(time.time() * 1000))) + + chunks = _split_message(plain, self.max_message_length) + # XHTML-IM only on single-chunk messages: splitting HTML across + # stanzas at arbitrary byte boundaries would produce invalid + # fragments, and most XMPP clients reject those. + emit_html = ( + self.html_formatting + and len(chunks) == 1 + and "xep_0071" in self._client.plugin + ) + html_body = "" + if emit_html: + html_body = _markdown_to_xhtml_im(content or "") + # Skip the html element when the rendered fragment is identical + # to the plain body (no formatting present): nothing to gain. + if html_body and html_body.strip() == _html_escape(plain).strip(): + html_body = "" + + last_id = "" + try: + for idx, chunk in enumerate(chunks): + stanza = self._client.make_message(mto=target, mbody=chunk, mtype=mtype) + if emit_html and idx == 0 and html_body: + try: + stanza["html"]["body"] = html_body + except Exception: + logger.debug("XMPP: failed to attach XHTML-IM body", exc_info=True) + stanza.send() + last_id = str(int(time.time() * 1000)) + await asyncio.sleep(0.05) + except Exception as e: + return SendResult(success=False, error=str(e)) + return SendResult(success=True, message_id=last_id) + + async def send_typing(self, chat_id: str, metadata=None) -> None: + """XEP-0085 chat states — best-effort, ignored if the peer does not support it.""" + if self._client is None: + return + target, mtype = self._resolve_target(chat_id) + if not target or mtype == "groupchat": + return # composing in MUC is noisy and rarely useful + try: + stanza = self._client.make_message(mto=target, mtype=mtype) + stanza["chat_state"] = "composing" + stanza.send() + except Exception: + # Chat-state plugin may not be registered; that's fine. + pass + + async def send_image( + self, + chat_id: str, + image_url: str, + caption: Optional[str] = None, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + """Send an HTTPS image URL with an OOB extension. + + Clients that understand XEP-0066 (Conversations, Gajim, Dino) treat + ``…`` as an inline + attachment and render the image; clients that ignore the extension + still see the URL in ```` and can tap through. + """ + return await self._send_url_with_oob( + chat_id=chat_id, + url=image_url, + caption=caption, + ) + + async def send_image_file( + self, + chat_id: str, + image_path: str, + caption: Optional[str] = None, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + **kwargs, + ) -> SendResult: + return await self._upload_then_send(chat_id, image_path, caption=caption) + + async def send_animation( + self, + chat_id: str, + animation_url: str, + caption: Optional[str] = None, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + # The gateway routes local-file animated GIFs through + # ``send_image_file`` already; this path is for hosted URLs. + return await self._send_url_with_oob(chat_id, animation_url, caption=caption) + + async def send_voice( + self, + chat_id: str, + audio_path: str, + caption: Optional[str] = None, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + **kwargs, + ) -> SendResult: + return await self._upload_then_send(chat_id, audio_path, caption=caption) + + async def send_video( + self, + chat_id: str, + video_path: str, + caption: Optional[str] = None, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + **kwargs, + ) -> SendResult: + return await self._upload_then_send(chat_id, video_path, caption=caption) + + async def send_document( + self, + chat_id: str, + file_path: str, + caption: Optional[str] = None, + file_name: Optional[str] = None, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + **kwargs, + ) -> SendResult: + return await self._upload_then_send(chat_id, file_path, caption=caption) + + async def _send_url_with_oob( + self, + chat_id: str, + url: str, + *, + caption: Optional[str] = None, + ) -> SendResult: + """Send a stanza whose ```` is *url* plus an OOB extension.""" + if self._client is None: + return SendResult(success=False, error="Not connected") + target, mtype = self._resolve_target(chat_id) + if not target: + return SendResult(success=False, error=f"Invalid XMPP target: {chat_id!r}") + clean_url = _strip_control_chars(url or "").strip() + if not clean_url: + return SendResult(success=False, error="Empty URL") + + body_lines = [] + if caption: + body_lines.append(_strip_control_chars(_strip_markdown(caption))) + body_lines.append(clean_url) + body = "\n".join(line for line in body_lines if line.strip()) + + try: + stanza = self._client.make_message(mto=target, mbody=body, mtype=mtype) + # Attach OOB only for http(s) URLs — file:// or relative paths + # would let a misconfigured caller leak local paths to peers. + if clean_url.startswith(("https://", "http://")): + try: + stanza["oob"]["url"] = clean_url + except Exception: + logger.debug("XMPP: failed to attach OOB extension", exc_info=True) + stanza.send() + except Exception as e: + return SendResult(success=False, error=str(e)) + return SendResult(success=True, message_id=str(int(time.time() * 1000))) + + async def _upload_then_send( + self, + chat_id: str, + path: str, + *, + caption: Optional[str] = None, + ) -> SendResult: + """Upload *path* via HTTP Upload then send the resulting URL via OOB. + + Falls back to a text message describing the file when the upload + component is unavailable or the upload fails. + """ + if self._client is None: + return SendResult(success=False, error="Not connected") + if not path or not os.path.isfile(path): + return SendResult(success=False, error=f"File not found: {path!r}") + + try: + url = await self._upload_file(path) + except _UploadUnavailable as exc: + logger.info("XMPP: falling back to text for %s — %s", path, exc) + return await self._send_upload_fallback(chat_id, path, caption, str(exc)) + + return await self._send_url_with_oob(chat_id, url, caption=caption) + + async def _upload_file(self, path: str) -> str: + """Upload a local file via XEP-0363 and return the HTTPS URL. + + Raises :class:`_UploadUnavailable` when the upload service is not + available, the file exceeds the server's advertised limit, or the + HTTP PUT itself fails. Callers turn the exception into a text + fallback so a missing upload component never breaks message + delivery. + """ + if self._upload_service_resolved is None: + await self._resolve_upload_service() + if not self._upload_service_resolved: + raise _UploadUnavailable("no HTTP Upload service available") + + client = self._client + if client is None: + raise _UploadUnavailable("not connected") + + from pathlib import Path as _Path + + size = os.path.getsize(path) + content_type = _content_type_for(path) + domain_kwarg: Dict[str, Any] = {} + if isinstance(self._upload_service_resolved, str): + domain_kwarg["domain"] = self._upload_service_resolved + + try: + upload = client.plugin["xep_0363"] + url = await upload.upload_file( + _Path(path), + size=size, + content_type=content_type, + **domain_kwarg, + ) + except Exception as exc: # UploadServiceNotFound / FileTooBig / HTTPError + raise _UploadUnavailable(str(exc)) from exc + return str(url) + + async def _send_upload_fallback( + self, + chat_id: str, + path: str, + caption: Optional[str], + reason: str, + ) -> SendResult: + """Send a text bubble when an upload could not happen.""" + filename = os.path.basename(path) or path + body = f"📎 {filename} (upload failed: {reason})" + if caption: + body = f"{caption}\n{body}" + return await self.send(chat_id, body) + + async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: + if chat_id.startswith(MUC_PREFIX): + jid = chat_id[len(MUC_PREFIX):] + return {"chat_id": chat_id, "type": "group", "name": jid} + return {"chat_id": chat_id, "type": "dm", "name": chat_id} + + def _resolve_target(self, chat_id: str) -> tuple[str, str]: + """Return ``(jid, mtype)`` for a ``chat_id`` string.""" + if not chat_id: + return "", "chat" + if chat_id.startswith(MUC_PREFIX): + return chat_id[len(MUC_PREFIX):], "groupchat" + return chat_id, "chat" + + +# --------------------------------------------------------------------------- +# Plugin entry-point hooks +# --------------------------------------------------------------------------- + +def check_requirements() -> bool: + """Plugin gate: require XMPP_JID + XMPP_PASSWORD + slixmpp importable.""" + if not os.getenv("XMPP_JID") or not os.getenv("XMPP_PASSWORD"): + return False + try: + import slixmpp # noqa: F401 + except ImportError: + return False + return True + + +def validate_config(config) -> bool: + extra = getattr(config, "extra", {}) or {} + jid = os.getenv("XMPP_JID") or extra.get("jid", "") + password = os.getenv("XMPP_PASSWORD") or extra.get("password", "") + return bool(jid and password) + + +def is_connected(config) -> bool: + return validate_config(config) + + +def _env_enablement() -> Optional[dict]: + jid = (os.getenv("XMPP_JID") or "").strip() + if not jid: + return None + if not (os.getenv("XMPP_PASSWORD") or "").strip(): + return None + + seed: dict = {"jid": jid} + host = (os.getenv("XMPP_HOST") or "").strip() + if host: + seed["host"] = host + port = (os.getenv("XMPP_PORT") or "").strip() + if port: + try: + seed["port"] = int(port) + except ValueError: + pass + force_tls = (os.getenv("XMPP_FORCE_STARTTLS") or "").strip() + if force_tls: + seed["force_starttls"] = force_tls.lower() in _TRUTHY + nickname = (os.getenv("XMPP_NICKNAME") or "").strip() + if nickname: + seed["nickname"] = nickname + rooms = (os.getenv("XMPP_ROOMS") or "").strip() + if rooms: + seed["rooms"] = _parse_comma_list(rooms) + upload_service = (os.getenv("XMPP_UPLOAD_SERVICE") or "").strip() + if upload_service: + seed["upload_service"] = upload_service + html_formatting = (os.getenv("XMPP_HTML_FORMATTING") or "").strip() + if html_formatting: + seed["html_formatting"] = html_formatting.lower() in _TRUTHY + + home = (os.getenv("XMPP_HOME_CHANNEL") or "").strip() + if home: + seed["home_channel"] = { + "chat_id": home, + "name": (os.getenv("XMPP_HOME_CHANNEL_NAME") or "").strip() or home, + } + return seed + + +async def _standalone_send( + pconfig, + chat_id: str, + message: str, + *, + thread_id: Optional[str] = None, + media_files: Optional[List[str]] = None, + force_document: bool = False, +) -> Dict[str, Any]: + """Open an ephemeral XMPP session, send, and disconnect. + + Used by ``tools/send_message_tool._send_via_adapter`` when the gateway + runner is not in this process (e.g. ``hermes cron`` running separately). + Without this hook, ``deliver=xmpp`` cron jobs fail with "No live + adapter for platform". + + ``thread_id``, ``media_files`` and ``force_document`` are accepted for + signature parity with other plugin standalone senders but are not + meaningful here: XMPP has no native thread primitive and media + delivery would require HTTP Upload (XEP-0363) and a reachable + upload component, which a one-shot session cannot guarantee. + """ + del thread_id, media_files, force_document + + try: + import slixmpp + except ImportError: + return {"error": "slixmpp not installed. Run: pip install slixmpp"} + + extra = getattr(pconfig, "extra", {}) or {} + jid = (os.getenv("XMPP_JID") or extra.get("jid") or "").strip() + password = os.getenv("XMPP_PASSWORD") or extra.get("password", "") + if not jid or not password: + return {"error": "XMPP standalone send: XMPP_JID and XMPP_PASSWORD are required"} + + raw_target = (chat_id or extra.get("home_channel", {}).get("chat_id") or "").strip() + if not raw_target: + return {"error": "XMPP standalone send: missing target JID"} + if any(ch in raw_target for ch in ("\r", "\n", "\x00", " ")): + return {"error": "XMPP standalone send: chat_id contains illegal characters"} + + if raw_target.startswith(MUC_PREFIX): + target = raw_target[len(MUC_PREFIX):] + mtype = "groupchat" + else: + target = raw_target + mtype = "chat" + + host = (os.getenv("XMPP_HOST") or extra.get("host") or "").strip() or None + port_raw = os.getenv("XMPP_PORT") or extra.get("port") or DEFAULT_PORT + try: + port = int(port_raw) + except (TypeError, ValueError): + port = DEFAULT_PORT + force_tls = _parse_bool( + os.getenv("XMPP_FORCE_STARTTLS", extra.get("force_starttls", True)), + default=True, + ) + + body = _strip_control_chars(_strip_markdown(message or "")) + if not body.strip(): + return {"error": "XMPP standalone send: empty message after stripping"} + + # Distinct resource so we don't collide with a live gateway adapter on + # the same identity. + full_jid = f"{_strip_resource(jid)}/hermes-cron-{int(time.time() * 1000) % 100000}" + + client = slixmpp.ClientXMPP(full_jid, password) + client.register_plugin("xep_0030") + client.register_plugin("xep_0045") + client.enable_starttls = force_tls + client.enable_direct_tls = force_tls + + ready = asyncio.Event() + error: Dict[str, str] = {} + + async def _on_start(_event): + try: + client.send_presence() + if mtype == "groupchat": + muc = client.plugin["xep_0045"] + nick = (os.getenv("XMPP_NICKNAME") or extra.get("nickname") or jid.split("@", 1)[0]).strip() or "hermes" + muc.join_muc(target, nick, wait=False) + # Give the server a moment to process the JOIN before we + # send to a +n-style room. Real protocol ack would be + # ``groupchat_subject``; a short sleep avoids us holding + # the event loop here just to listen for one stanza. + await asyncio.sleep(0.8) + for chunk in _split_message(body, MAX_MESSAGE_LENGTH): + client.send_message(mto=target, mbody=chunk, mtype=mtype) + await asyncio.sleep(0.05) + except Exception as exc: # noqa: BLE001 + error["msg"] = str(exc) + finally: + ready.set() + + def _on_auth_fail(_event): + error["msg"] = "authentication failed" + ready.set() + + client.add_event_handler("session_start", _on_start) + client.add_event_handler("failed_auth", _on_auth_fail) + + try: + client.connect(host=host, port=port if host else None) + except Exception as exc: # noqa: BLE001 + return {"error": f"XMPP standalone connect failed: {exc}"} + + try: + await asyncio.wait_for(ready.wait(), timeout=CONNECT_TIMEOUT_SECONDS) + except asyncio.TimeoutError: + try: + client.disconnect(wait=1.0) + except Exception: + pass + return {"error": "XMPP standalone send: timed out before session_start"} + + try: + fut = client.disconnect(wait=DISCONNECT_TIMEOUT_SECONDS) + if fut is not None: + await asyncio.wait_for(asyncio.shield(asyncio.ensure_future(fut)), timeout=DISCONNECT_TIMEOUT_SECONDS + 2.0) + except Exception: + pass + + if error: + return {"error": f"XMPP standalone send failed: {error['msg']}"} + return {"success": True, "platform": "xmpp", "chat_id": chat_id, "message_id": str(int(time.time() * 1000))} + + +def interactive_setup() -> None: + """Minimal stdin wizard for ``hermes setup gateway`` → XMPP.""" + try: + from hermes_cli.setup import ( + prompt, + prompt_yes_no, + save_env_value, + get_env_value, + print_header, + print_info, + print_warning, + print_success, + ) + except ImportError: + print() + print("hermes_cli.setup not available; set XMPP_* vars manually in ~/.hermes/.env") + return + + print_header("XMPP") + existing = get_env_value("XMPP_JID") + if existing: + print_info(f"XMPP: already configured (JID: {existing})") + if not prompt_yes_no("Reconfigure XMPP?", False): + return + + print_info("Connect Hermes to an XMPP server (Prosody, ejabberd, hosted, etc.).") + print_info(" Requires the slixmpp Python package: pip install slixmpp") + print() + + jid = prompt("Bot JID (e.g. hermes@chat.example.org)", default=existing or "") + if not jid: + print_warning("JID is required — skipping XMPP setup") + return + save_env_value("XMPP_JID", jid.strip()) + + password = prompt("Password for this account", password=True) + if password: + save_env_value("XMPP_PASSWORD", password) + elif not get_env_value("XMPP_PASSWORD"): + print_warning("Password is required — skipping XMPP setup") + return + + host = prompt("Server host (blank = SRV lookup on the JID domain)", + default=get_env_value("XMPP_HOST") or "") + save_env_value("XMPP_HOST", host.strip()) + + port = prompt("Server port (default 5222)", default=get_env_value("XMPP_PORT") or "") + if port: + try: + save_env_value("XMPP_PORT", str(int(port))) + except ValueError: + print_warning("Invalid port — using default 5222") + + use_tls = prompt_yes_no("Require STARTTLS (recommended)?", True) + save_env_value("XMPP_FORCE_STARTTLS", "true" if use_tls else "false") + + nickname = prompt( + "MUC nickname (blank = JID local part)", + default=get_env_value("XMPP_NICKNAME") or "", + ) + save_env_value("XMPP_NICKNAME", nickname.strip()) + + rooms = prompt( + "MUC rooms to auto-join (comma-separated, or blank)", + default=get_env_value("XMPP_ROOMS") or "", + ) + save_env_value("XMPP_ROOMS", rooms.replace(" ", "")) + + print() + print_info("🔒 Access control: restrict which JIDs can DM the bot") + allow_all = prompt_yes_no("Allow any JID to talk to the bot?", False) + if allow_all: + save_env_value("XMPP_ALLOW_ALL_USERS", "true") + save_env_value("XMPP_ALLOWED_USERS", "") + print_warning("⚠️ Open access — any JID can command the bot.") + else: + save_env_value("XMPP_ALLOW_ALL_USERS", "false") + allowed = prompt( + "Allowed bare JIDs (comma-separated, blank to deny everyone)", + default=get_env_value("XMPP_ALLOWED_USERS") or "", + ) + save_env_value("XMPP_ALLOWED_USERS", allowed.replace(" ", "")) + if allowed: + print_success("Allowlist configured") + else: + print_info("No JIDs allowed — the bot will ignore all DMs until you add some.") + + print() + print_success("XMPP configuration saved to ~/.hermes/.env") + print_info("Restart the gateway for changes to take effect: hermes gateway restart") + + +def register(ctx) -> None: + """Plugin entry point — called by the Hermes plugin system at startup.""" + ctx.register_platform( + name="xmpp", + label="XMPP", + adapter_factory=lambda cfg: XMPPAdapter(cfg), + check_fn=check_requirements, + validate_config=validate_config, + is_connected=is_connected, + required_env=["XMPP_JID", "XMPP_PASSWORD"], + install_hint="pip install slixmpp # XMPP adapter requires slixmpp", + setup_fn=interactive_setup, + env_enablement_fn=_env_enablement, + cron_deliver_env_var="XMPP_HOME_CHANNEL", + standalone_sender_fn=_standalone_send, + allowed_users_env="XMPP_ALLOWED_USERS", + allow_all_env="XMPP_ALLOW_ALL_USERS", + max_message_length=MAX_MESSAGE_LENGTH, + emoji="✉️", + pii_safe=False, + allow_update_command=True, + platform_hint=( + "You are chatting via XMPP. JIDs look like email addresses " + "(user@server). MUC rooms are identified by a ``muc:`` " + "prefix; in rooms, users address you by your nickname. " + "You can use markdown sparingly (bold, italic, code, links, " + "lists) — the adapter renders it via XHTML-IM for clients " + "that support it, with a plain-text fallback for the rest. " + "Files attach natively when the server has an HTTP Upload " + "component: include MEDIA:/absolute/path/to/file in your " + "response and the adapter uploads it and sends the URL with " + "an OOB attachment hint. If the server has no upload " + "component the message arrives as text describing the file." + ), + ) diff --git a/plugins/platforms/xmpp/plugin.yaml b/plugins/platforms/xmpp/plugin.yaml new file mode 100644 index 000000000000..7b6cb8a8daf2 --- /dev/null +++ b/plugins/platforms/xmpp/plugin.yaml @@ -0,0 +1,69 @@ +name: xmpp-platform +label: XMPP +kind: platform +version: 1.0.0 +description: > + XMPP gateway adapter for Hermes Agent. + Connects to a self-hosted or hosted XMPP server (Prosody, ejabberd, etc.) + and relays messages between XMPP contacts/rooms and the Hermes agent. + Built on slixmpp; STARTTLS is on by default. JIDs in an allowlist are + authorized to chat with the bot; MUC rooms join with a configurable + nickname and reply only when addressed by that nickname. +author: alien2003 +# ``requires_env`` and ``optional_env`` entries are surfaced in the +# ``hermes config`` UI via the platform-plugin env var injector in +# ``hermes_cli/config.py``. +requires_env: + - name: XMPP_JID + description: "Bare JID the bot logs in as (e.g. hermes@chat.example.org)" + prompt: "Bot JID" + password: false + - name: XMPP_PASSWORD + description: "Password for the bot account" + prompt: "XMPP password" + password: true +optional_env: + - name: XMPP_HOST + description: "Server host override (default: SRV lookup on the JID domain)" + prompt: "XMPP host (or empty)" + password: false + - name: XMPP_PORT + description: "Server port override (default: 5222)" + prompt: "XMPP port (or empty)" + password: false + - name: XMPP_FORCE_STARTTLS + description: "Require STARTTLS (default: true). Set 'false' only for trusted dev servers." + prompt: "Force STARTTLS? (true/false)" + password: false + - name: XMPP_NICKNAME + description: "MUC nickname (default: the JID's local part)" + prompt: "MUC nickname (or empty)" + password: false + - name: XMPP_ROOMS + description: "Comma-separated MUC JIDs to auto-join (e.g. ops@conference.example.org)" + prompt: "Rooms to join (or empty)" + password: false + - name: XMPP_ALLOWED_USERS + description: "Comma-separated bare JIDs allowed to DM the bot" + prompt: "Allowed JIDs (comma-separated)" + password: false + - name: XMPP_ALLOW_ALL_USERS + description: "Allow any JID to talk to the bot (dev only — disables allowlist)" + prompt: "Allow all users? (true/false)" + password: false + - name: XMPP_HOME_CHANNEL + description: "Default JID for cron / notification delivery (DM JID or MUC JID prefixed with 'muc:')" + prompt: "Home channel JID (or empty)" + password: false + - name: XMPP_HOME_CHANNEL_NAME + description: "Human label for the home channel (defaults to the JID)" + prompt: "Home channel display name (or empty)" + password: false + - name: XMPP_UPLOAD_SERVICE + description: "Pin the XEP-0363 HTTP Upload component JID (default: auto-discover)" + prompt: "Upload service JID (or empty)" + password: false + - name: XMPP_HTML_FORMATTING + description: "Emit XEP-0071 XHTML-IM dual body so clients can render bold/italic/links (default: true)" + prompt: "Send XHTML-IM dual body? (true/false)" + password: false diff --git a/scripts/release.py b/scripts/release.py index 177009ee5489..ccf4e259942c 100755 --- a/scripts/release.py +++ b/scripts/release.py @@ -1270,6 +1270,7 @@ "120500656+oooindefatigable@users.noreply.github.com": "ooovenenoso", "vanthinh6886@gmail.com": "vanthinh6886", # PR #28018 salvage (yaml/flock/atomic write guards) "erik.engervall@gmail.com": "erikengervall", # PR #28774 (firecrawl integration tag) + "alien2003@protonmail.ch": "alien2003", # XMPP platform plugin } diff --git a/tests/gateway/test_xmpp_plugin.py b/tests/gateway/test_xmpp_plugin.py new file mode 100644 index 000000000000..44d8409ca82c --- /dev/null +++ b/tests/gateway/test_xmpp_plugin.py @@ -0,0 +1,953 @@ +"""Tests for the XMPP platform-plugin adapter. + +Loaded via the ``_plugin_adapter_loader`` helper so this lives under +``plugin_adapter_xmpp`` in ``sys.modules`` and cannot collide with +sibling platform-plugin tests on the same xdist worker. +""" + +from __future__ import annotations + +import asyncio +import os +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from tests.gateway._plugin_adapter_loader import load_plugin_adapter + +_xmpp = load_plugin_adapter("xmpp") + +XMPPAdapter = _xmpp.XMPPAdapter +check_requirements = _xmpp.check_requirements +validate_config = _xmpp.validate_config +is_connected = _xmpp.is_connected +register = _xmpp.register +_env_enablement = _xmpp._env_enablement +_standalone_send = _xmpp._standalone_send +_strip_resource = _xmpp._strip_resource +_strip_markdown = _xmpp._strip_markdown +_strip_control_chars = _xmpp._strip_control_chars +_split_message = _xmpp._split_message +_parse_bool = _xmpp._parse_bool +_parse_comma_list = _xmpp._parse_comma_list +_markdown_to_xhtml_im = _xmpp._markdown_to_xhtml_im +_markdown_to_xhtml_im_fallback = _xmpp._markdown_to_xhtml_im_fallback +_html_escape = _xmpp._html_escape +_sanitize_link_url = _xmpp._sanitize_link_url +_content_type_for = _xmpp._content_type_for +_UploadUnavailable = _xmpp._UploadUnavailable +MUC_PREFIX = _xmpp.MUC_PREFIX + + +class _FakeStanza: + """Minimal slixmpp stanza stand-in that records ``stanza['html']['body']``.""" + + class _Slot: + def __init__(self, parent: "_FakeStanza", key: str): + self._parent = parent + self._key = key + + def __setitem__(self, k, v): + if self._key == "html" and k == "body": + self._parent._html_body = v + elif self._key == "oob" and k == "url": + self._parent._oob_url = v + + def __getitem__(self, k): # pragma: no cover - not used by adapter + return None + + def __init__(self, mto: str, mbody: str, mtype: str): + self.mto = mto + self.mbody = mbody + self.mtype = mtype + self._html_body = None + self._oob_url = None + self._sent = False + self._sent_by: list = [] + + def __getitem__(self, key): + return _FakeStanza._Slot(self, key) + + def send(self): + self._sent = True + self._sent_by.append(self) + + +def _make_mock_client(*, has_html: bool = False): + """Build a MagicMock that mirrors slixmpp's surface for tests. + + The adapter calls ``client.make_message(mto=, mbody=, mtype=).send()`` + (returning the stanza for ``stanza["html"]["body"]`` assignment), and + consults ``"xep_0071" in client.plugin`` to decide whether to attach + XHTML-IM. + """ + client = MagicMock() + client.plugin = {"xep_0071": MagicMock()} if has_html else {} + sent_stanzas: list = [] + + def _make_message(*, mto, mbody, mtype, **_kwargs): + stanza = _FakeStanza(mto=mto, mbody=mbody, mtype=mtype) + original_send = stanza.send + + def _record(): + sent_stanzas.append(stanza) + return original_send() + + stanza.send = _record # type: ignore[assignment] + return stanza + + client.make_message.side_effect = _make_message + client._sent_stanzas = sent_stanzas + return client + + +# --------------------------------------------------------------------------- +# 1. Platform enum (plugin-discovered, not bundled) +# --------------------------------------------------------------------------- + +def test_platform_enum_resolves_via_plugin_scan(): + from gateway.config import Platform + p = Platform("xmpp") + assert p.value == "xmpp" + assert Platform("xmpp") is p + + +# --------------------------------------------------------------------------- +# 2. check_requirements / validate_config / is_connected +# --------------------------------------------------------------------------- + +def test_check_requirements_needs_credentials(monkeypatch): + monkeypatch.delenv("XMPP_JID", raising=False) + monkeypatch.delenv("XMPP_PASSWORD", raising=False) + assert check_requirements() is False + + monkeypatch.setenv("XMPP_JID", "hermes@example.org") + assert check_requirements() is False # password still missing + + +def test_check_requirements_true_when_fully_configured(monkeypatch): + monkeypatch.setenv("XMPP_JID", "hermes@example.org") + monkeypatch.setenv("XMPP_PASSWORD", "secret") + try: + import slixmpp # noqa: F401 + slixmpp_present = True + except ImportError: + slixmpp_present = False + assert check_requirements() is slixmpp_present + + +def test_validate_config_uses_env_or_extra(monkeypatch): + from gateway.config import PlatformConfig + + monkeypatch.delenv("XMPP_JID", raising=False) + monkeypatch.delenv("XMPP_PASSWORD", raising=False) + cfg = PlatformConfig(enabled=True) + assert validate_config(cfg) is False + + cfg2 = PlatformConfig(enabled=True, extra={"jid": "hermes@example.org", "password": "secret"}) + assert validate_config(cfg2) is True + + +def test_is_connected_mirrors_validate(monkeypatch): + from gateway.config import PlatformConfig + monkeypatch.delenv("XMPP_JID", raising=False) + monkeypatch.delenv("XMPP_PASSWORD", raising=False) + cfg = PlatformConfig(enabled=True, extra={"jid": "x@x", "password": "y"}) + assert is_connected(cfg) is True + assert is_connected(PlatformConfig(enabled=True)) is False + + +# --------------------------------------------------------------------------- +# 3. _env_enablement seeds PlatformConfig.extra +# --------------------------------------------------------------------------- + +def test_env_enablement_none_when_unset(monkeypatch): + monkeypatch.delenv("XMPP_JID", raising=False) + monkeypatch.delenv("XMPP_PASSWORD", raising=False) + assert _env_enablement() is None + + +def test_env_enablement_needs_password(monkeypatch): + monkeypatch.setenv("XMPP_JID", "hermes@example.org") + monkeypatch.delenv("XMPP_PASSWORD", raising=False) + assert _env_enablement() is None + + +def test_env_enablement_seeds_minimal(monkeypatch): + monkeypatch.setenv("XMPP_JID", "hermes@example.org") + monkeypatch.setenv("XMPP_PASSWORD", "secret") + for var in ("XMPP_HOST", "XMPP_PORT", "XMPP_FORCE_STARTTLS", + "XMPP_NICKNAME", "XMPP_ROOMS", "XMPP_HOME_CHANNEL"): + monkeypatch.delenv(var, raising=False) + seed = _env_enablement() + assert seed == {"jid": "hermes@example.org"} + + +def test_env_enablement_seeds_full(monkeypatch): + monkeypatch.setenv("XMPP_JID", "hermes@example.org") + monkeypatch.setenv("XMPP_PASSWORD", "secret") + monkeypatch.setenv("XMPP_HOST", "chat.example.org") + monkeypatch.setenv("XMPP_PORT", "5223") + monkeypatch.setenv("XMPP_FORCE_STARTTLS", "false") + monkeypatch.setenv("XMPP_NICKNAME", "h2") + monkeypatch.setenv("XMPP_ROOMS", "ops@conf, dev@conf") + monkeypatch.setenv("XMPP_HOME_CHANNEL", "alice@example.org") + monkeypatch.setenv("XMPP_HOME_CHANNEL_NAME", "Alice") + + seed = _env_enablement() + assert seed["jid"] == "hermes@example.org" + assert seed["host"] == "chat.example.org" + assert seed["port"] == 5223 + assert seed["force_starttls"] is False + assert seed["nickname"] == "h2" + assert seed["rooms"] == ["ops@conf", "dev@conf"] + assert seed["home_channel"] == {"chat_id": "alice@example.org", "name": "Alice"} + + +def test_env_enablement_home_channel_defaults_name_to_id(monkeypatch): + monkeypatch.setenv("XMPP_JID", "hermes@example.org") + monkeypatch.setenv("XMPP_PASSWORD", "secret") + monkeypatch.setenv("XMPP_HOME_CHANNEL", "alice@example.org") + monkeypatch.delenv("XMPP_HOME_CHANNEL_NAME", raising=False) + + seed = _env_enablement() + assert seed["home_channel"] == { + "chat_id": "alice@example.org", + "name": "alice@example.org", + } + + +# --------------------------------------------------------------------------- +# 4. Helper functions +# --------------------------------------------------------------------------- + +def test_strip_resource(): + assert _strip_resource("user@example.org/laptop") == "user@example.org" + assert _strip_resource("user@example.org") == "user@example.org" + assert _strip_resource("") == "" + + +def test_strip_markdown_removes_basic_formatting(): + out = _strip_markdown("**bold** *italic* `code` ~~strike~~") + assert "*" not in out + assert "`" not in out + assert "~~" not in out + assert "bold" in out and "italic" in out and "code" in out and "strike" in out + + +def test_strip_markdown_keeps_link_target(): + out = _strip_markdown("see [docs](https://example.org/x)") + assert "https://example.org/x" in out + assert "[" not in out + + +def test_strip_markdown_strips_code_fences(): + out = _strip_markdown("```python\nprint(1)\n```") + assert "```" not in out + assert "print(1)" in out + + +def test_strip_control_chars_drops_null_and_normalizes_crlf(): + out = _strip_control_chars("a\x00b\r\nc\rd") + assert "\x00" not in out + assert "\r" not in out + assert out == "ab\nc\nd" + + +def test_split_message_short_returns_single_chunk(): + chunks = _split_message("hello world", max_bytes=100) + assert chunks == ["hello world"] + + +def test_split_message_long_paragraph_splits_under_limit(): + para = "a" * 10_000 + chunks = _split_message(para, max_bytes=500) + assert len(chunks) >= 20 + for chunk in chunks: + assert len(chunk.encode("utf-8")) <= 500 + + +def test_split_message_multi_paragraph(): + text = "para one\npara two" + chunks = _split_message(text, max_bytes=100) + assert chunks == ["para one", "para two"] + + +def test_parse_bool_handles_strings_and_booleans(): + assert _parse_bool("true", default=False) is True + assert _parse_bool("FALSE", default=True) is False + assert _parse_bool(None, default=True) is True + assert _parse_bool(True, default=False) is True + + +def test_parse_comma_list_strips_and_drops_empty(): + assert _parse_comma_list("a, b , ,c") == ["a", "b", "c"] + assert _parse_comma_list("") == [] + + +# --------------------------------------------------------------------------- +# 5. Adapter init +# --------------------------------------------------------------------------- + +def test_adapter_init_reads_extra(monkeypatch): + from gateway.config import PlatformConfig + for var in ("XMPP_JID", "XMPP_PASSWORD", "XMPP_HOST", "XMPP_PORT", + "XMPP_FORCE_STARTTLS", "XMPP_NICKNAME", "XMPP_ROOMS"): + monkeypatch.delenv(var, raising=False) + + cfg = PlatformConfig(enabled=True, extra={ + "jid": "hermes@example.org", + "password": "s3cret", + "host": "chat.example.org", + "port": 5223, + "force_starttls": False, + "nickname": "h", + "rooms": "ops@conf, dev@conf", + }) + adapter = XMPPAdapter(cfg) + assert adapter.jid == "hermes@example.org" + assert adapter.host == "chat.example.org" + assert adapter.port == 5223 + assert adapter.force_starttls is False + assert adapter.nickname == "h" + assert adapter.rooms == ["ops@conf", "dev@conf"] + + +def test_adapter_init_env_overrides_extra(monkeypatch): + from gateway.config import PlatformConfig + monkeypatch.setenv("XMPP_JID", "env@example.org") + monkeypatch.setenv("XMPP_PASSWORD", "envpass") + monkeypatch.setenv("XMPP_PORT", "5223") + monkeypatch.setenv("XMPP_FORCE_STARTTLS", "false") + + cfg = PlatformConfig(enabled=True, extra={ + "jid": "extra@example.org", + "password": "xpass", + "port": 5222, + "force_starttls": True, + }) + adapter = XMPPAdapter(cfg) + assert adapter.jid == "env@example.org" + assert adapter.password == "envpass" + assert adapter.port == 5223 + assert adapter.force_starttls is False + + +def test_adapter_init_defaults(monkeypatch): + from gateway.config import PlatformConfig + for var in ("XMPP_JID", "XMPP_PASSWORD", "XMPP_HOST", "XMPP_PORT", + "XMPP_FORCE_STARTTLS", "XMPP_NICKNAME", "XMPP_ROOMS"): + monkeypatch.delenv(var, raising=False) + + cfg = PlatformConfig(enabled=True, extra={"jid": "bot@example.org", "password": "x"}) + adapter = XMPPAdapter(cfg) + assert adapter.port == 5222 + assert adapter.force_starttls is True + assert adapter.host is None + assert adapter.nickname == "bot" # local part of the JID + assert adapter.rooms == [] + + +def test_adapter_init_invalid_port_falls_back(monkeypatch): + from gateway.config import PlatformConfig + monkeypatch.delenv("XMPP_PORT", raising=False) + cfg = PlatformConfig(enabled=True, extra={"jid": "x@x", "password": "y", "port": "not-a-port"}) + adapter = XMPPAdapter(cfg) + assert adapter.port == 5222 + + +def test_adapter_platform_identity(): + from gateway.config import Platform, PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"jid": "x@x", "password": "y"}) + adapter = XMPPAdapter(cfg) + assert adapter.platform is Platform("xmpp") + + +def test_resolve_target_dm(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"jid": "x@x", "password": "y"}) + adapter = XMPPAdapter(cfg) + target, mtype = adapter._resolve_target("alice@example.org") + assert target == "alice@example.org" + assert mtype == "chat" + + +def test_resolve_target_muc(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"jid": "x@x", "password": "y"}) + adapter = XMPPAdapter(cfg) + target, mtype = adapter._resolve_target(f"{MUC_PREFIX}ops@conf.example.org") + assert target == "ops@conf.example.org" + assert mtype == "groupchat" + + +def test_resolve_target_empty(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"jid": "x@x", "password": "y"}) + adapter = XMPPAdapter(cfg) + target, mtype = adapter._resolve_target("") + assert target == "" + assert mtype == "chat" + + +# --------------------------------------------------------------------------- +# 6. Outbound send (mocked client) +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_send_dm_strips_markdown_in_body(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"jid": "bot@x", "password": "y"}) + adapter = XMPPAdapter(cfg) + adapter._client = _make_mock_client() + + result = await adapter.send("alice@example.org", "**hi**") + assert result.success is True + sent = adapter._client._sent_stanzas + assert len(sent) == 1 + assert sent[0].mto == "alice@example.org" + assert sent[0].mtype == "chat" + assert "**" not in sent[0].mbody + + +@pytest.mark.asyncio +async def test_send_muc_uses_groupchat(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"jid": "bot@x", "password": "y"}) + adapter = XMPPAdapter(cfg) + adapter._client = _make_mock_client() + + result = await adapter.send(f"{MUC_PREFIX}ops@conf", "hello") + assert result.success is True + sent = adapter._client._sent_stanzas + assert sent[0].mto == "ops@conf" + assert sent[0].mtype == "groupchat" + + +@pytest.mark.asyncio +async def test_send_when_disconnected_returns_error(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"jid": "bot@x", "password": "y"}) + adapter = XMPPAdapter(cfg) + result = await adapter.send("alice@example.org", "hi") + assert result.success is False + assert "Not connected" in (result.error or "") + + +@pytest.mark.asyncio +async def test_send_empty_body_after_strip_succeeds_without_call(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"jid": "bot@x", "password": "y"}) + adapter = XMPPAdapter(cfg) + adapter._client = _make_mock_client() + + result = await adapter.send("alice@example.org", "\x00 \r\n ") + assert result.success is True + assert adapter._client.make_message.call_count == 0 + + +@pytest.mark.asyncio +async def test_send_splits_long_message(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"jid": "bot@x", "password": "y", "max_message_length": 100}) + adapter = XMPPAdapter(cfg) + adapter._client = _make_mock_client() + + body = "lorem ipsum " * 50 + result = await adapter.send("alice@example.org", body) + assert result.success is True + assert len(adapter._client._sent_stanzas) >= 2 + + +@pytest.mark.asyncio +async def test_send_emits_xhtml_im_for_single_chunk(): + """Single-chunk markdown messages should carry both and .""" + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"jid": "bot@x", "password": "y"}) + adapter = XMPPAdapter(cfg) + adapter._client = _make_mock_client(has_html=True) + + result = await adapter.send("alice@example.org", "**hello** _world_") + assert result.success is True + sent = adapter._client._sent_stanzas + assert len(sent) == 1 + assert sent[0]._html_body, "single-chunk markdown should set the html body" + assert "" in sent[0]._html_body or "" in sent[0]._html_body + + +@pytest.mark.asyncio +async def test_send_skips_xhtml_im_when_no_formatting(): + """Plain text messages should not waste bytes on an empty block.""" + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"jid": "bot@x", "password": "y"}) + adapter = XMPPAdapter(cfg) + adapter._client = _make_mock_client(has_html=True) + + result = await adapter.send("alice@example.org", "plain text reply") + assert result.success is True + sent = adapter._client._sent_stanzas + assert sent[0]._html_body in (None, "") + + +@pytest.mark.asyncio +async def test_send_skips_xhtml_im_on_multi_chunk(): + """Chunked sends should fall through to plain text only.""" + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"jid": "bot@x", "password": "y", "max_message_length": 80}) + adapter = XMPPAdapter(cfg) + adapter._client = _make_mock_client(has_html=True) + + result = await adapter.send("alice@example.org", "**bold** " + ("lorem " * 30)) + assert result.success is True + sent = adapter._client._sent_stanzas + assert len(sent) >= 2 + for stanza in sent: + assert stanza._html_body in (None, "") + + +@pytest.mark.asyncio +async def test_send_respects_html_formatting_disabled(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={ + "jid": "bot@x", "password": "y", "html_formatting": False, + }) + adapter = XMPPAdapter(cfg) + adapter._client = _make_mock_client(has_html=True) + + result = await adapter.send("alice@example.org", "**hi**") + assert result.success is True + assert adapter._client._sent_stanzas[0]._html_body in (None, "") + + +# --------------------------------------------------------------------------- +# 7. get_chat_info +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_get_chat_info_dm(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"jid": "bot@x", "password": "y"}) + adapter = XMPPAdapter(cfg) + info = await adapter.get_chat_info("alice@example.org") + assert info["type"] == "dm" + assert info["chat_id"] == "alice@example.org" + + +@pytest.mark.asyncio +async def test_get_chat_info_muc(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"jid": "bot@x", "password": "y"}) + adapter = XMPPAdapter(cfg) + info = await adapter.get_chat_info(f"{MUC_PREFIX}ops@conf") + assert info["type"] == "group" + assert info["name"] == "ops@conf" + + +# --------------------------------------------------------------------------- +# 8. Inbound message dispatch +# --------------------------------------------------------------------------- + +class _FakeMsg: + def __init__(self, *, mtype="chat", body="", frm="", mid="", mucnick=""): + self._data = { + "type": mtype, + "body": body, + "from": frm, + "id": mid, + "mucnick": mucnick, + } + + def __getitem__(self, key): + return self._data.get(key, "") + + def get(self, key, default=None): + v = self._data.get(key) + return v if v else default + + +@pytest.mark.asyncio +async def test_on_message_dispatches_chat(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"jid": "bot@example.org", "password": "y"}) + adapter = XMPPAdapter(cfg) + adapter.handle_message = AsyncMock() + adapter._message_handler = lambda evt: None + + msg = _FakeMsg(mtype="chat", body="hello", frm="alice@example.org/laptop", mid="m1") + await adapter._on_message(msg) + assert adapter.handle_message.await_count == 1 + event = adapter.handle_message.await_args.args[0] + assert event.text == "hello" + assert event.source.chat_id == "alice@example.org" + assert event.source.chat_type == "dm" + + +@pytest.mark.asyncio +async def test_on_message_ignores_wrong_type(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"jid": "bot@x", "password": "y"}) + adapter = XMPPAdapter(cfg) + adapter.handle_message = AsyncMock() + adapter._message_handler = lambda evt: None + + await adapter._on_message(_FakeMsg(mtype="error", body="oops", frm="alice@x")) + await adapter._on_message(_FakeMsg(mtype="groupchat", body="hi", frm="ops@conf")) + adapter.handle_message.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_on_message_ignores_empty_body(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"jid": "bot@x", "password": "y"}) + adapter = XMPPAdapter(cfg) + adapter.handle_message = AsyncMock() + adapter._message_handler = lambda evt: None + + await adapter._on_message(_FakeMsg(mtype="chat", body="", frm="alice@x")) + adapter.handle_message.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_on_message_ignores_self_echo(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"jid": "bot@example.org", "password": "y"}) + adapter = XMPPAdapter(cfg) + adapter.handle_message = AsyncMock() + adapter._message_handler = lambda evt: None + + await adapter._on_message(_FakeMsg(mtype="chat", body="hi", frm="bot@example.org/desk")) + adapter.handle_message.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_on_groupchat_requires_addressed_prefix(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"jid": "bot@example.org", "password": "y", "nickname": "hermes"}) + adapter = XMPPAdapter(cfg) + adapter.handle_message = AsyncMock() + adapter._message_handler = lambda evt: None + + # Not addressed → ignored + await adapter._on_groupchat_message(_FakeMsg( + mtype="groupchat", body="hello world", frm="ops@conf/alice", mucnick="alice", + )) + adapter.handle_message.assert_not_awaited() + + # Addressed → dispatched and prefix stripped + await adapter._on_groupchat_message(_FakeMsg( + mtype="groupchat", body="hermes: status?", frm="ops@conf/alice", mucnick="alice", + )) + assert adapter.handle_message.await_count == 1 + event = adapter.handle_message.await_args.args[0] + assert event.text == "status?" + assert event.source.chat_id == f"{MUC_PREFIX}ops@conf" + assert event.source.chat_type == "group" + assert event.source.user_id == "alice" + + +@pytest.mark.asyncio +async def test_on_groupchat_ignores_own_nick(): + from gateway.config import PlatformConfig + cfg = PlatformConfig(enabled=True, extra={"jid": "bot@x", "password": "y", "nickname": "hermes"}) + adapter = XMPPAdapter(cfg) + adapter.handle_message = AsyncMock() + adapter._message_handler = lambda evt: None + + await adapter._on_groupchat_message(_FakeMsg( + mtype="groupchat", body="hermes: ping", frm="ops@conf/hermes", mucnick="hermes", + )) + adapter.handle_message.assert_not_awaited() + + +# --------------------------------------------------------------------------- +# 9. Standalone (out-of-process) send for cron +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_standalone_send_missing_slixmpp(monkeypatch): + """When slixmpp is unimportable, return a clean error dict.""" + import sys + saved = {name: sys.modules.pop(name) + for name in list(sys.modules) if name == "slixmpp" or name.startswith("slixmpp.")} + saved_meta = list(sys.meta_path) + + class _Blocker: + @staticmethod + def find_spec(name, path=None, target=None): + if name == "slixmpp" or name.startswith("slixmpp."): + raise ImportError("slixmpp blocked for test") + return None + + sys.meta_path.insert(0, _Blocker()) + try: + pconfig = MagicMock() + pconfig.extra = {"jid": "bot@x", "password": "y"} + result = await _standalone_send(pconfig, "alice@x", "hi") + assert isinstance(result, dict) + assert "error" in result + assert "slixmpp" in result["error"].lower() + finally: + sys.meta_path[:] = saved_meta + sys.modules.update(saved) + + +@pytest.mark.asyncio +async def test_standalone_send_missing_credentials(monkeypatch): + monkeypatch.delenv("XMPP_JID", raising=False) + monkeypatch.delenv("XMPP_PASSWORD", raising=False) + pconfig = MagicMock() + pconfig.extra = {} + try: + import slixmpp # noqa: F401 + except ImportError: + pytest.skip("slixmpp not installed") + result = await _standalone_send(pconfig, "alice@x", "hi") + assert isinstance(result, dict) + assert "error" in result + assert "XMPP_JID" in result["error"] or "required" in result["error"].lower() + + +@pytest.mark.asyncio +async def test_standalone_send_rejects_injection_chars(monkeypatch): + monkeypatch.setenv("XMPP_JID", "bot@example.org") + monkeypatch.setenv("XMPP_PASSWORD", "secret") + try: + import slixmpp # noqa: F401 + except ImportError: + pytest.skip("slixmpp not installed") + pconfig = MagicMock() + pconfig.extra = {} + result = await _standalone_send(pconfig, "alice@x\nINJECT", "hi") + assert "error" in result + assert "illegal" in result["error"].lower() + + +# --------------------------------------------------------------------------- +# 10. register() — plugin-side metadata +# --------------------------------------------------------------------------- + +def test_register_calls_register_platform(): + ctx = MagicMock() + register(ctx) + ctx.register_platform.assert_called_once() + kwargs = ctx.register_platform.call_args.kwargs + + assert kwargs["name"] == "xmpp" + assert kwargs["label"] == "XMPP" + assert kwargs["required_env"] == ["XMPP_JID", "XMPP_PASSWORD"] + assert kwargs["allowed_users_env"] == "XMPP_ALLOWED_USERS" + assert kwargs["allow_all_env"] == "XMPP_ALLOW_ALL_USERS" + assert kwargs["cron_deliver_env_var"] == "XMPP_HOME_CHANNEL" + assert kwargs["max_message_length"] == _xmpp.MAX_MESSAGE_LENGTH + assert kwargs["pii_safe"] is False + assert callable(kwargs["check_fn"]) + assert callable(kwargs["validate_config"]) + assert callable(kwargs["is_connected"]) + assert callable(kwargs["env_enablement_fn"]) + assert callable(kwargs["standalone_sender_fn"]) + assert callable(kwargs["adapter_factory"]) + assert callable(kwargs["setup_fn"]) + assert "XMPP" in kwargs["platform_hint"] + # Adapter advertises media support via XEP-0363; the hint should + # tell the agent it can emit MEDIA: tags. + assert "MEDIA" in kwargs["platform_hint"] + + +# --------------------------------------------------------------------------- +# 11. XHTML-IM (XEP-0071) helpers +# --------------------------------------------------------------------------- + +def test_html_escape_basic(): + assert _html_escape("&\"x") == "<a>&"x" + + +def test_sanitize_link_url_rejects_dangerous_schemes(): + assert _sanitize_link_url("javascript:alert(1)") == "" + assert _sanitize_link_url("DATA:text/html,`") + assert "<script>" in html + assert "