From 3568148f2a73e5d6b55bbbd85d2db8a8becd2b0e Mon Sep 17 00:00:00 2001 From: Kaitlyn Concilio Date: Tue, 28 Jul 2026 17:33:47 +0000 Subject: [PATCH] feat(gateway): add native Fluxer platform adapter --- plugins/platforms/fluxer/__init__.py | 3 + plugins/platforms/fluxer/adapter.py | 1496 +++++++++++++++++ plugins/platforms/fluxer/plugin.yaml | 56 + tests/gateway/test_fluxer.py | 1029 ++++++++++++ tests/gateway/test_fluxer_integration.py | 157 ++ tests/gateway/test_fluxer_plugin_setup.py | 58 + .../gateway/test_plugin_platform_interface.py | 30 +- .../docs/reference/environment-variables.md | 13 + website/docs/user-guide/messaging/fluxer.md | 118 ++ website/docs/user-guide/messaging/index.md | 5 +- 10 files changed, 2958 insertions(+), 7 deletions(-) create mode 100644 plugins/platforms/fluxer/__init__.py create mode 100644 plugins/platforms/fluxer/adapter.py create mode 100644 plugins/platforms/fluxer/plugin.yaml create mode 100644 tests/gateway/test_fluxer.py create mode 100644 tests/gateway/test_fluxer_integration.py create mode 100644 tests/gateway/test_fluxer_plugin_setup.py create mode 100644 website/docs/user-guide/messaging/fluxer.md diff --git a/plugins/platforms/fluxer/__init__.py b/plugins/platforms/fluxer/__init__.py new file mode 100644 index 0000000000000..d4f1d7bf0e3fc --- /dev/null +++ b/plugins/platforms/fluxer/__init__.py @@ -0,0 +1,3 @@ +from .adapter import register + +__all__ = ["register"] diff --git a/plugins/platforms/fluxer/adapter.py b/plugins/platforms/fluxer/adapter.py new file mode 100644 index 0000000000000..0ef8e61bb2672 --- /dev/null +++ b/plugins/platforms/fluxer/adapter.py @@ -0,0 +1,1496 @@ +"""Native Fluxer messaging platform adapter for Hermes Agent. + +Fluxer exposes a Discord-shaped JSON API, but it is a distinct service with +its own REST origin and Gateway. This adapter talks to those interfaces +directly through aiohttp; it does not patch or impersonate discord.py. + +Environment variables: + FLUXER_BOT_TOKEN Bot token from Fluxer Developer Settings + FLUXER_API_URL REST base (default https://api.fluxer.app/v1) + FLUXER_GATEWAY_URL Optional Gateway URL override + FLUXER_ALLOWED_USERS Comma-separated Fluxer user IDs + FLUXER_ALLOW_ALL_USERS Allow any user (development only) + FLUXER_ALLOWED_CHANNELS Optional guild-channel allowlist + FLUXER_FREE_RESPONSE_CHANNELS Guild channels that do not require mention + FLUXER_REQUIRE_MENTION Require bot mention in guild channels (true) + FLUXER_HOME_CHANNEL Default cron/notification channel + FLUXER_PROXY HTTP/SOCKS proxy override +""" + +from __future__ import annotations + +import asyncio +import hashlib +import ipaddress +import json +import logging +import mimetypes +import os +import random +import re +import uuid +from pathlib import Path +from typing import Any, Dict, List, Optional, Sequence, Tuple +from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit + +from gateway.config import Platform, PlatformConfig +from gateway.platforms.base import ( + BasePlatformAdapter, + MessageEvent, + MessageType, + SendResult, + _ssrf_redirect_guard, + cache_audio_from_bytes, + cache_document_from_bytes, + cache_image_from_bytes, + proxy_kwargs_for_aiohttp, + resolve_channel_prompt, + resolve_proxy_url, +) +from gateway.platforms.helpers import MessageDeduplicator + +logger = logging.getLogger(__name__) + +DEFAULT_API_URL = "https://api.fluxer.app/v1" +MAX_MESSAGE_LENGTH = 4000 +MAX_ATTACHMENTS = 10 +_DEFAULT_DOWNLOAD_LIMIT = 25 * 1024 * 1024 +_DEFAULT_UPLOAD_LIMIT = 25 * 1024 * 1024 +_RECONNECT_BASE_DELAY = 2.0 +_RECONNECT_MAX_DELAY = 60.0 +_RECONNECT_JITTER = 0.2 +_SAFE_ALLOWED_MENTIONS = {"parse": [], "replied_user": False} +_TEXT_MESSAGE_TYPES = {0, 19} # default and reply +_CHANNEL_TYPE_MAP = { + 0: "channel", # guild text + 1: "dm", + 2: "channel", # guild voice text chat + 3: "group", + 999: "dm", # personal notes +} + + +class _ReconnectRequested(RuntimeError): + """Internal signal used to restart the Gateway socket.""" + + def __init__(self, message: str, retry_delay: Optional[float] = None): + super().__init__(message) + self.retry_delay = retry_delay + + +class _PermanentGatewayError(RuntimeError): + """Gateway failure that reconnecting cannot repair.""" + + +class _UploadTooLarge(ValueError): + """Outbound file exceeded the configured bounded-read limit.""" + + +def _read_file_bounded(path: Path, limit: int) -> bytes: + """Read at most ``limit`` bytes without unbounded buffering.""" + with path.open("rb") as handle: + data = handle.read(limit + 1) + if len(data) > limit: + raise _UploadTooLarge(f"File exceeds Fluxer upload limit of {limit} bytes") + return data + + +def _csv_set(value: Any) -> set[str]: + if value is None: + return set() + if isinstance(value, (list, tuple, set)): + return {str(item).strip() for item in value if str(item).strip()} + return {part.strip() for part in str(value).split(",") if part.strip()} + + +def _truthy(value: Any, default: bool = False) -> bool: + if value is None or str(value).strip() == "": + return default + return str(value).strip().lower() in {"1", "true", "yes", "on"} + + +def _normalise_api_url(value: str) -> str: + raw = (value or DEFAULT_API_URL).strip().rstrip("/") + try: + parsed = urlsplit(raw) + except ValueError as exc: + raise ValueError( + "Fluxer API URL must include a valid hostname and port" + ) from exc + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise ValueError("Fluxer API URL must be an absolute HTTP(S) URL") + _validate_endpoint_host(parsed, "Fluxer API") + if parsed.username or parsed.password or parsed.query or parsed.fragment: + raise ValueError("Fluxer API URL must not include userinfo, query, or fragment") + if parsed.scheme == "http" and not _is_loopback_host(parsed.hostname): + raise ValueError("Fluxer API URL must use HTTPS (HTTP is loopback-only)") + return raw + + +def _is_loopback_host(hostname: Optional[str]) -> bool: + if not hostname: + return False + host = hostname.rstrip(".").lower() + if host == "localhost": + return True + try: + return ipaddress.ip_address(host).is_loopback + except ValueError: + return False + + +def _validate_endpoint_host(parsed: Any, label: str) -> str: + """Reject malformed token-bearing endpoint authorities before any I/O.""" + try: + hostname = parsed.hostname + # Accessing ``port`` forces urllib to reject non-numeric/out-of-range ports. + parsed.port + except ValueError as exc: + raise ValueError(f"{label} URL must include a valid hostname and port") from exc + if not hostname: + raise ValueError(f"{label} URL must include a valid hostname and port") + + host = hostname.rstrip(".") + try: + ipaddress.ip_address(host) + return hostname + except ValueError: + pass + + try: + ascii_host = host.encode("idna").decode("ascii") + except UnicodeError as exc: + raise ValueError(f"{label} URL must include a valid hostname and port") from exc + labels = ascii_host.split(".") + if ( + not ascii_host + or len(ascii_host) > 253 + or any( + not part + or len(part) > 63 + or part.startswith("-") + or part.endswith("-") + or re.fullmatch(r"[A-Za-z0-9-]+", part) is None + for part in labels + ) + ): + raise ValueError(f"{label} URL must include a valid hostname and port") + return hostname + + +def _normalise_gateway_url(value: str) -> str: + raw = value.strip() + try: + parsed = urlsplit(raw) + except ValueError as exc: + raise ValueError( + "Fluxer Gateway URL must include a valid hostname and port" + ) from exc + if parsed.scheme not in {"ws", "wss"} or not parsed.netloc: + raise ValueError("Fluxer Gateway URL must be an absolute WS(S) URL") + _validate_endpoint_host(parsed, "Fluxer Gateway") + if parsed.username or parsed.password or parsed.fragment: + raise ValueError("Fluxer Gateway URL must not include userinfo or fragment") + if parsed.scheme == "ws" and not _is_loopback_host(parsed.hostname): + raise ValueError("Fluxer Gateway URL must use WSS (WS is loopback-only)") + return raw + + +def _valid_http_base(value: str) -> bool: + try: + _normalise_api_url(value) + except ValueError: + return False + return True + + +def check_fluxer_requirements() -> bool: + try: + import aiohttp # noqa: F401 + + return True + except ImportError: + logger.warning("Fluxer: aiohttp is not installed") + return False + + +def validate_fluxer_config(config: PlatformConfig) -> bool: + extra = getattr(config, "extra", {}) or {} + token = ( + getattr(config, "token", None) + or extra.get("token") + or os.getenv("FLUXER_BOT_TOKEN", "") + ) + gateway_url = str( + extra.get("gateway_url") or os.getenv("FLUXER_GATEWAY_URL", "") + ).strip() + try: + api_url = _normalise_api_url( + extra.get("api_url") or os.getenv("FLUXER_API_URL", DEFAULT_API_URL) + ) + if gateway_url: + _normalise_gateway_url(gateway_url) + except ValueError: + return False + return bool(str(token).strip()) and _valid_http_base(api_url) + + +class FluxerAdapter(BasePlatformAdapter): + """Hermes gateway adapter for Fluxer's REST API and real-time Gateway.""" + + splits_long_messages = True + + def __init__(self, config: PlatformConfig): + super().__init__(config, Platform("fluxer")) + extra = config.extra or {} + self._token = str( + config.token or extra.get("token") or os.getenv("FLUXER_BOT_TOKEN", "") + ).strip() + self._api_url = _normalise_api_url( + str(extra.get("api_url") or os.getenv("FLUXER_API_URL", DEFAULT_API_URL)) + ) + gateway_url = str( + extra.get("gateway_url") or os.getenv("FLUXER_GATEWAY_URL", "") + ).strip() + self._gateway_url = _normalise_gateway_url(gateway_url) if gateway_url else "" + + self._session: Any = None + self._ws: Any = None + self._gateway_task: Optional[asyncio.Task] = None + self._heartbeat_task: Optional[asyncio.Task] = None + self._heartbeat_acknowledged = True + self._gateway_was_ready = False + self._ready_event = asyncio.Event() + self._closing = False + + self._bot_user_id = "" + self._bot_username = "" + self._session_id = "" + self._resume_gateway_url = "" + self._sequence = 0 + self._channel_cache: Dict[str, Dict[str, Any]] = {} + self._dedup = MessageDeduplicator() + + self._last_http_status: Optional[int] = None + self._last_http_error = "" + self._last_retry_after: Optional[float] = None + try: + self._max_upload_bytes = int( + extra.get("max_upload_bytes") + or os.getenv("FLUXER_MAX_UPLOAD_BYTES", _DEFAULT_UPLOAD_LIMIT) + ) + except (TypeError, ValueError): + self._max_upload_bytes = _DEFAULT_UPLOAD_LIMIT + if self._max_upload_bytes <= 0: + self._max_upload_bytes = _DEFAULT_UPLOAD_LIMIT + + # ------------------------------------------------------------------ + # REST helpers + # ------------------------------------------------------------------ + + def _headers(self) -> Dict[str, str]: + return { + "Authorization": f"Bot {self._token}", + "Content-Type": "application/json", + "User-Agent": "Hermes-Agent/Fluxer", + } + + def _reset_http_result(self) -> None: + self._last_http_status = None + self._last_http_error = "" + self._last_retry_after = None + + async def _api( + self, + method: str, + path: str, + *, + json: Optional[Dict[str, Any]] = None, + params: Optional[Dict[str, Any]] = None, + ) -> Optional[Any]: + """Call a Fluxer REST endpoint and return decoded JSON on success.""" + import aiohttp + + self._reset_http_result() + if self._session is None: + self._last_http_error = "Fluxer client is not connected" + return None + if ".." in path: + self._last_http_status = 400 + self._last_http_error = "Unsafe Fluxer API path" + return None + + url = f"{self._api_url}/{path.lstrip('/')}" + try: + async with self._session.request( + method.upper(), + url, + headers=self._headers(), + json=json, + params=params, + timeout=aiohttp.ClientTimeout(total=30), + **(getattr(self, "_request_proxy_kwargs", {}) or {}), + ) as response: + self._last_http_status = response.status + if response.status == 204: + return {} + if response.status >= 400: + body = await response.text() + self._last_http_error = body[:1000] + if response.status == 429: + try: + data = json_module_loads(body) + except (TypeError, ValueError): + data = {} + retry = ( + data.get("retry_after") if isinstance(data, dict) else None + ) + if retry is None: + retry = response.headers.get("Retry-After") + try: + self._last_retry_after = ( + float(retry) if retry is not None else None + ) + except (TypeError, ValueError): + self._last_retry_after = None + logger.warning( + "Fluxer REST %s %s returned HTTP %s: %s", + method.upper(), + path, + response.status, + body[:300], + ) + return None + if response.content_length == 0: + return {} + return await response.json() + except (aiohttp.ClientError, asyncio.TimeoutError) as exc: + self._last_http_error = str(exc) + logger.warning("Fluxer REST %s %s failed: %s", method.upper(), path, exc) + return None + + async def _api_multipart( + self, + method: str, + path: str, + *, + payload: Dict[str, Any], + files: Sequence[Tuple[str, bytes, str]], + ) -> Optional[Dict[str, Any]]: + """Send Fluxer's Discord-compatible multipart message shape.""" + import aiohttp + + self._reset_http_result() + if self._session is None: + self._last_http_error = "Fluxer client is not connected" + return None + if ".." in path: + self._last_http_status = 400 + self._last_http_error = "Unsafe Fluxer API path" + return None + + form = aiohttp.FormData() + form.add_field( + "payload_json", json.dumps(payload), content_type="application/json" + ) + for index, (filename, data, content_type) in enumerate(files[:MAX_ATTACHMENTS]): + form.add_field( + f"files[{index}]", + data, + filename=filename, + content_type=content_type or "application/octet-stream", + ) + + headers = { + "Authorization": f"Bot {self._token}", + "User-Agent": "Hermes-Agent/Fluxer", + } + url = f"{self._api_url}/{path.lstrip('/')}" + try: + async with self._session.request( + method.upper(), + url, + headers=headers, + data=form, + timeout=aiohttp.ClientTimeout(total=90), + **(getattr(self, "_request_proxy_kwargs", {}) or {}), + ) as response: + self._last_http_status = response.status + if response.status >= 400: + body = await response.text() + self._last_http_error = body[:1000] + if response.status == 429: + retry: Any = None + try: + body_data = json_module_loads(body) + if isinstance(body_data, dict): + retry = body_data.get("retry_after") + except (TypeError, ValueError): + pass + if retry is None: + retry = response.headers.get("Retry-After") + try: + self._last_retry_after = ( + float(retry) if retry is not None else None + ) + except (TypeError, ValueError): + self._last_retry_after = None + return None + return await response.json() + except (aiohttp.ClientError, asyncio.TimeoutError) as exc: + self._last_http_error = str(exc) + return None + + def _send_failure(self, action: str) -> SendResult: + status = self._last_http_status + retryable = False + retry_after = None + if status == 429: + kind = "rate_limited" + retryable = True + retry_after = self._last_retry_after + elif status in {401, 403}: + kind = "forbidden" + elif status == 404: + kind = "not_found" + elif status is not None and status >= 500: + kind = "transient" + retryable = True + elif status is None: + kind = "transient" + retryable = True + else: + kind = "unknown" + detail = self._last_http_error or ( + f"HTTP {status}" if status is not None else "network error" + ) + return SendResult( + success=False, + error=f"Fluxer {action} failed: {detail}", + error_kind=kind, + retryable=retryable, + retry_after=retry_after, + ) + + @staticmethod + def _message_payload( + content: str, + chat_id: str, + reply_to: Optional[str] = None, + ) -> Dict[str, Any]: + payload: Dict[str, Any] = { + "content": content, + "nonce": uuid.uuid4().hex, + "allowed_mentions": dict(_SAFE_ALLOWED_MENTIONS), + } + if reply_to: + payload["message_reference"] = { + "message_id": str(reply_to), + "channel_id": str(chat_id), + "type": 0, + } + return payload + + # ------------------------------------------------------------------ + # Base adapter lifecycle + # ------------------------------------------------------------------ + + async def connect(self, *, is_reconnect: bool = False) -> bool: + import aiohttp + + if not self._token or not _valid_http_base(self._api_url): + self._set_fatal_error( + "fluxer_config", + "FLUXER_BOT_TOKEN and a valid FLUXER_API_URL are required", + retryable=False, + ) + return False + + lock_identity = hashlib.sha256(self._token.encode()).hexdigest()[:24] + if not self._acquire_platform_lock("fluxer", lock_identity, "Fluxer bot token"): + return False + + try: + proxy = resolve_proxy_url( + platform_env_var="FLUXER_PROXY", + target_hosts=["api.fluxer.app", "gateway.fluxer.app"], + ) + session_kwargs, request_kwargs = proxy_kwargs_for_aiohttp(proxy) + # aiohttp's ws_connect does not inherit per-request proxy kwargs, so + # retain them and pass them explicitly for both REST and Gateway. + self._request_proxy_kwargs = request_kwargs + self._session = aiohttp.ClientSession( + timeout=aiohttp.ClientTimeout(total=30), **session_kwargs + ) + self._closing = False + self._ready_event.clear() + + me = await self._api("GET", "users/@me") + if not isinstance(me, dict) or not me.get("id"): + self._set_fatal_error( + "fluxer_auth", + "Fluxer authentication failed; check FLUXER_BOT_TOKEN", + retryable=False, + ) + await self.disconnect() + return False + self._bot_user_id = str(me["id"]) + self._bot_username = str(me.get("username") or me.get("global_name") or "") + + if not self._gateway_url: + gateway = await self._api("GET", "gateway/bot") + if not isinstance(gateway, dict) or not gateway.get("url"): + self._set_fatal_error( + "fluxer_gateway_discovery", + "Fluxer did not return a Gateway URL", + retryable=True, + ) + await self.disconnect() + return False + try: + self._gateway_url = _normalise_gateway_url(str(gateway["url"])) + except ValueError as exc: + self._set_fatal_error( + "fluxer_gateway_url_invalid", str(exc), retryable=False + ) + await self.disconnect() + return False + + self._gateway_task = asyncio.create_task( + self._gateway_loop(), name="fluxer-gateway" + ) + try: + await asyncio.wait_for(self._ready_event.wait(), timeout=20.0) + except asyncio.TimeoutError: + self._set_fatal_error( + "fluxer_gateway_timeout", + "Timed out waiting for Fluxer Gateway READY", + retryable=True, + ) + await self.disconnect() + return False + + if self._fatal_error_code and not self._fatal_error_retryable: + await self.disconnect() + return False + + self._mark_connected() + logger.info( + "Fluxer: connected as %s (%s)", + self._bot_username or "bot", + self._bot_user_id, + ) + return True + except asyncio.CancelledError: + # Cancellation is not an Exception on modern Python. Clean up the + # HTTP session, Gateway task, and machine-local token lock before + # preserving cancellation semantics for the gateway runner. + await asyncio.shield(self.disconnect()) + raise + except Exception as exc: + self._set_fatal_error( + "fluxer_connect_error", f"Fluxer startup failed: {exc}", retryable=True + ) + logger.error("Fluxer startup failed: %s", exc, exc_info=True) + await self.disconnect() + return False + + async def disconnect(self) -> None: + self._closing = True + current = asyncio.current_task() + for task in (self._heartbeat_task, self._gateway_task): + if task and task is not current and not task.done(): + task.cancel() + pending = [ + task + for task in (self._heartbeat_task, self._gateway_task) + if task and task is not current + ] + if pending: + await asyncio.gather(*pending, return_exceptions=True) + self._heartbeat_task = None + self._gateway_task = None + + if self._ws is not None: + try: + await self._ws.close(code=1000, message=b"Hermes shutdown") + except Exception: + pass + self._ws = None + if self._session is not None and not self._session.closed: + await self._session.close() + self._session = None + + try: + self._release_platform_lock() + except Exception: + logger.warning("Fluxer: failed to release token lock", exc_info=True) + self._mark_disconnected() + + # ------------------------------------------------------------------ + # Outbound messages + # ------------------------------------------------------------------ + + async def send( + self, + chat_id: str, + content: str, + reply_to: Optional[str] = None, + metadata: Optional[Dict[str, Any]] = None, + ) -> SendResult: + if not content: + return SendResult(success=True) + chunks = self.truncate_message(self.format_message(content), MAX_MESSAGE_LENGTH) + message_ids: List[str] = [] + for chunk in chunks: + data = await self._api( + "POST", + f"channels/{chat_id}/messages", + json=self._message_payload(chunk, chat_id, reply_to), + ) + if not isinstance(data, dict) or not data.get("id"): + return self._send_failure("message send") + message_ids.append(str(data["id"])) + return SendResult( + success=True, + message_id=message_ids[-1], + continuation_message_ids=tuple(message_ids[:-1]), + ) + + async def edit_message( + self, + chat_id: str, + message_id: str, + content: str, + *, + finalize: bool = False, + ) -> SendResult: + payload = { + "content": self.format_message(content)[:MAX_MESSAGE_LENGTH], + "allowed_mentions": dict(_SAFE_ALLOWED_MENTIONS), + } + data = await self._api( + "PATCH", f"channels/{chat_id}/messages/{message_id}", json=payload + ) + if not isinstance(data, dict) or not data.get("id"): + return self._send_failure("message edit") + return SendResult(success=True, message_id=str(data["id"])) + + async def send_typing( + self, chat_id: str, metadata: Optional[Dict[str, Any]] = None + ) -> None: + await self._api("POST", f"channels/{chat_id}/typing", json={}) + + async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: + channel = await self._get_channel(chat_id) + channel_type = int(channel.get("type", 0)) if channel else 0 + return { + "name": channel.get("name") or channel.get("display_name") or chat_id, + "type": _CHANNEL_TYPE_MAP.get(channel_type, "channel"), + } + + async def _send_files( + self, + chat_id: str, + files: Sequence[Tuple[str, bytes, str]], + caption: Optional[str], + reply_to: Optional[str], + ) -> SendResult: + attachments = [ + {"id": index, "filename": filename, "content_type": content_type} + for index, (filename, _data, content_type) in enumerate( + files[:MAX_ATTACHMENTS] + ) + ] + payload = self._message_payload( + (caption or "")[:MAX_MESSAGE_LENGTH], chat_id, reply_to + ) + payload["attachments"] = attachments + data = await self._api_multipart( + "POST", + f"channels/{chat_id}/messages", + payload=payload, + files=files[:MAX_ATTACHMENTS], + ) + if not isinstance(data, dict) or not data.get("id"): + return self._send_failure("file send") + return SendResult(success=True, message_id=str(data["id"])) + + 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: Any, + ) -> SendResult: + path = Path(file_path) + filename = file_name or path.name + content_type = mimetypes.guess_type(filename)[0] or "application/octet-stream" + try: + data = await asyncio.to_thread( + _read_file_bounded, path, self._max_upload_bytes + ) + except (FileNotFoundError, IsADirectoryError): + return SendResult(success=False, error=f"File not found: {file_path}") + except OSError as exc: + return SendResult(success=False, error=f"Could not read file: {exc}") + except _UploadTooLarge as exc: + return SendResult( + success=False, + error=str(exc), + error_kind="file_too_large", + retryable=False, + ) + return await self._send_files( + chat_id, [(filename, data, content_type)], caption, reply_to + ) + + 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: Any, + ) -> SendResult: + return await self.send_document( + chat_id, image_path, caption=caption, reply_to=reply_to, metadata=metadata + ) + + 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: Any, + ) -> SendResult: + return await self.send_document( + chat_id, audio_path, caption=caption, reply_to=reply_to, metadata=metadata + ) + + 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: Any, + ) -> SendResult: + return await self.send_document( + chat_id, video_path, caption=caption, reply_to=reply_to, metadata=metadata + ) + + 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: + cached = await self._download_attachment({ + "url": image_url, + "filename": "image", + "content_type": "image/png", + }) + if not cached: + return await self.send( + chat_id, f"{caption or ''}\n{image_url}".strip(), reply_to, metadata + ) + local_path, _mime = cached + return await self.send_document( + chat_id, + local_path, + caption=caption, + reply_to=reply_to, + metadata=metadata, + ) + + # ------------------------------------------------------------------ + # Gateway protocol + # ------------------------------------------------------------------ + + def _gateway_connect_url(self) -> str: + gateway_url = ( + self._resume_gateway_url + if self._session_id and self._resume_gateway_url + else self._gateway_url + ) + parsed = urlsplit(_normalise_gateway_url(gateway_url)) + query = dict(parse_qsl(parsed.query, keep_blank_values=True)) + query.update({"v": "1", "encoding": "json", "compress": "none"}) + return urlunsplit(( + parsed.scheme, + parsed.netloc, + parsed.path or "/", + urlencode(query), + parsed.fragment, + )) + + async def _gateway_loop(self) -> None: + delay = _RECONNECT_BASE_DELAY + while not self._closing: + self._gateway_was_ready = False + sleep_delay = delay + apply_backoff = True + try: + await self._gateway_once() + self._mark_disconnected() + delay = _RECONNECT_BASE_DELAY + except asyncio.CancelledError: + return + except _PermanentGatewayError as exc: + was_ready = self._ready_event.is_set() + self._mark_disconnected() + self._set_fatal_error("fluxer_gateway_auth", str(exc), retryable=False) + # Wake ``connect()`` immediately so a permanent close is not + # misreported 20 seconds later as a retryable READY timeout. + self._ready_event.set() + if was_ready: + await self._notify_fatal_error() + logger.error("Fluxer Gateway permanent failure: %s", exc) + return + except _ReconnectRequested as exc: + if self._closing: + return + self._mark_disconnected() + if self._gateway_was_ready: + delay = _RECONNECT_BASE_DELAY + if exc.retry_delay is not None: + sleep_delay = exc.retry_delay + apply_backoff = False + else: + sleep_delay = delay + logger.warning( + "Fluxer Gateway disconnected: %s; reconnecting in %.1fs", + exc, + sleep_delay, + ) + except Exception as exc: + if self._closing: + return + self._mark_disconnected() + if self._gateway_was_ready: + delay = _RECONNECT_BASE_DELAY + logger.warning( + "Fluxer Gateway disconnected: %s; reconnecting in %.1fs", exc, delay + ) + if self._closing: + return + jitter = ( + random.random() * sleep_delay * _RECONNECT_JITTER + if apply_backoff + else 0.0 + ) + await asyncio.sleep(sleep_delay + jitter) + if apply_backoff: + delay = min(delay * 2, _RECONNECT_MAX_DELAY) + + async def _gateway_once(self) -> None: + import aiohttp + + url = self._gateway_connect_url() + kwargs = dict(getattr(self, "_request_proxy_kwargs", {}) or {}) + ws_timeout_factory = getattr(aiohttp, "ClientWSTimeout") + ws = await self._session.ws_connect( + url, + headers={"User-Agent": "Hermes-Agent/Fluxer"}, + heartbeat=30.0, + timeout=ws_timeout_factory(ws_close=10.0), + **kwargs, + ) + self._ws = ws + try: + async for message in ws: + if self._closing: + return + if message.type == aiohttp.WSMsgType.TEXT: + try: + payload = json.loads(message.data) + except (TypeError, ValueError): + continue + await self._handle_gateway_payload(payload, ws) + elif message.type == aiohttp.WSMsgType.BINARY: + try: + payload = json.loads(message.data.decode("utf-8")) + except (AttributeError, UnicodeDecodeError, ValueError): + continue + await self._handle_gateway_payload(payload, ws) + elif message.type in { + aiohttp.WSMsgType.CLOSE, + aiohttp.WSMsgType.CLOSED, + aiohttp.WSMsgType.CLOSING, + aiohttp.WSMsgType.ERROR, + }: + break + close_code = ws.close_code + if close_code in {4004, 4010, 4011, 4012}: + raise _PermanentGatewayError( + f"Fluxer Gateway rejected the connection (close {close_code})" + ) + if close_code in {4007, 4009}: + self._session_id = "" + self._resume_gateway_url = "" + self._sequence = 0 + raise _ReconnectRequested(f"Gateway closed ({close_code})") + finally: + if self._heartbeat_task and not self._heartbeat_task.done(): + self._heartbeat_task.cancel() + await asyncio.gather(self._heartbeat_task, return_exceptions=True) + self._heartbeat_task = None + if not ws.closed: + try: + await ws.close(code=1000, message=b"Hermes reconnect") + except Exception: + logger.debug("Fluxer Gateway socket close failed", exc_info=True) + if self._ws is ws: + self._ws = None + + def _start_heartbeat(self, ws: Any, interval_ms: Any) -> None: + try: + interval = max(1.0, float(interval_ms) / 1000.0) + except (TypeError, ValueError): + interval = 45.0 + if self._heartbeat_task and not self._heartbeat_task.done(): + self._heartbeat_task.cancel() + self._heartbeat_acknowledged = True + self._heartbeat_task = asyncio.create_task( + self._heartbeat_loop(ws, interval), name="fluxer-heartbeat" + ) + + async def _heartbeat_loop(self, ws: Any, interval: float) -> None: + try: + while not self._closing and not ws.closed: + await asyncio.sleep(max(1.0, interval * 0.8)) + if not self._heartbeat_acknowledged: + logger.warning("Fluxer Gateway missed heartbeat ACK; reconnecting") + await ws.close(code=4000, message=b"Heartbeat ACK timeout") + return + self._heartbeat_acknowledged = False + try: + await ws.send_json({"op": 1, "d": self._sequence}) + except Exception: + if not ws.closed: + await ws.close(code=4000, message=b"Heartbeat send failed") + return + except asyncio.CancelledError: + raise + except Exception as exc: + logger.debug("Fluxer heartbeat stopped: %s", exc) + + async def _handle_gateway_payload(self, payload: Dict[str, Any], ws: Any) -> None: + opcode = payload.get("op") + if opcode == 0: + sequence = payload.get("s") + if isinstance(sequence, int): + self._sequence = sequence + event = payload.get("t") + data = payload.get("d") or {} + if event == "READY": + self._session_id = str(data.get("session_id") or "") + self._resume_gateway_url = "" + resume_url = str(data.get("resume_gateway_url") or "").strip() + if resume_url: + try: + self._resume_gateway_url = _normalise_gateway_url(resume_url) + except ValueError: + self._resume_gateway_url = "" + logger.warning( + "Fluxer Gateway supplied an invalid resume URL; ignoring it" + ) + ready_user = data.get("user") or {} + if ready_user.get("id"): + self._bot_user_id = str(ready_user["id"]) + self._ready_event.set() + self._gateway_was_ready = True + self._mark_connected() + elif event == "RESUMED": + self._ready_event.set() + self._gateway_was_ready = True + self._mark_connected() + elif event == "MESSAGE_CREATE" and isinstance(data, dict): + await self._handle_message_create(data) + return + + if opcode == 10: + hello = payload.get("d") or {} + self._start_heartbeat(ws, hello.get("heartbeat_interval", 45000)) + if self._session_id: + await ws.send_json({ + "op": 6, + "d": { + "token": self._token, + "session_id": self._session_id, + "seq": self._sequence, + }, + }) + else: + await ws.send_json({ + "op": 2, + "d": { + "token": self._token, + "properties": { + "os": os.name, + "browser": "hermes-agent", + "device": "hermes-agent", + }, + "presence": None, + # Hermes only consumes new user messages. Asking the + # Gateway not to emit noisy high-volume events keeps + # the bot's connection lightweight. + "ignored_events": [ + "MESSAGE_UPDATE", + "MESSAGE_DELETE", + "TYPING_START", + "PRESENCE_UPDATE", + ], + "flags": 0, + }, + }) + return + + if opcode == 1: + self._heartbeat_acknowledged = False + await ws.send_json({"op": 1, "d": self._sequence}) + elif opcode == 11: + self._heartbeat_acknowledged = True + elif opcode == 7: + raise _ReconnectRequested("Gateway requested reconnect", retry_delay=0.0) + elif opcode == 9: + if payload.get("d") is not True: + self._session_id = "" + self._resume_gateway_url = "" + self._sequence = 0 + raise _ReconnectRequested( + "Gateway invalidated session", retry_delay=random.uniform(2.5, 3.5) + ) + + # ------------------------------------------------------------------ + # Inbound messages + # ------------------------------------------------------------------ + + async def _get_channel(self, channel_id: str) -> Dict[str, Any]: + cached = self._channel_cache.get(str(channel_id)) + if cached is not None: + return cached + channel = await self._api("GET", f"channels/{channel_id}") + if not isinstance(channel, dict): + # Do not cache a synthetic type: a transient lookup failure must not + # permanently turn a DM into a mention-gated guild channel. + return {"id": str(channel_id), "_lookup_failed": True} + self._channel_cache[str(channel_id)] = channel + return channel + + def _extra_or_env_set(self, extra_key: str, env_key: str) -> set[str]: + extra = self.config.extra or {} + value = extra.get(extra_key) + if value is None: + value = os.getenv(env_key, "") + return _csv_set(value) + + def _requires_mention(self) -> bool: + extra = self.config.extra or {} + value = extra.get("require_mention") + if value is None: + value = os.getenv("FLUXER_REQUIRE_MENTION", "true") + return _truthy(value, default=True) + + async def _handle_message_create(self, message: Dict[str, Any]) -> None: + message_id = str(message.get("id") or "") + channel_id = str(message.get("channel_id") or "") + author = message.get("author") or {} + author_id = str(author.get("id") or "") + if not message_id or not channel_id or not author_id: + return + if author_id == self._bot_user_id or bool(author.get("bot")): + return + if message.get("webhook_id"): + return + try: + message_type_value = int(message.get("type", 0)) + except (TypeError, ValueError): + return + if message_type_value not in _TEXT_MESSAGE_TYPES: + return + if self._dedup.is_duplicate(message_id): + return + + channel = await self._get_channel(channel_id) + try: + channel_type_value = int(channel.get("type", 0)) + except (TypeError, ValueError): + channel_type_value = 0 + guild_id = str(message.get("guild_id") or channel.get("guild_id") or "") or None + if channel.get("_lookup_failed"): + chat_type = "channel" if guild_id else "dm" + else: + chat_type = _CHANNEL_TYPE_MAP.get(channel_type_value, "channel") + text = str(message.get("content") or "") + + if chat_type == "channel": + allowed = self._extra_or_env_set( + "allowed_channels", "FLUXER_ALLOWED_CHANNELS" + ) + if allowed and channel_id not in allowed: + return + free = self._extra_or_env_set( + "free_response_channels", "FLUXER_FREE_RESPONSE_CHANNELS" + ) + mentions = message.get("mentions") or [] + mentioned_ids = { + str(item.get("id")) + for item in mentions + if isinstance(item, dict) and item.get("id") + } + mention_pattern = re.compile(rf"<@!?{re.escape(self._bot_user_id)}>") + has_mention = self._bot_user_id in mentioned_ids or bool( + self._bot_user_id and mention_pattern.search(text) + ) + if self._requires_mention() and channel_id not in free and not has_mention: + return + if has_mention: + text = mention_pattern.sub("", text).strip() + + # GatewayRunner registers the same platform-bound authorization check + # used by central ingress. Apply it before downloading attacker-controlled + # attachments so denied users cannot consume bandwidth or cache space. + sender_authorized = self._is_sender_authorized(author_id, chat_type, channel_id) + # Unknown DM senders must still reach central ingress so it can issue a + # pairing code. Keep their attachments out of the download path until + # that authorization succeeds. + attachments = ( + (message.get("attachments") or []) if sender_authorized is True else [] + ) + media_urls: List[str] = [] + media_types: List[str] = [] + for attachment in attachments[:MAX_ATTACHMENTS]: + if not isinstance(attachment, dict) or not attachment.get("url"): + continue + cached = await self._download_attachment(attachment) + if cached: + local_path, mime = cached + media_urls.append(local_path) + media_types.append(mime) + + if not text and not media_urls and sender_authorized is True: + return + if text[:1].isspace() and text.lstrip().startswith("/"): + text = text.lstrip() + normalized_type = ( + MessageType.COMMAND if text.startswith("/") else MessageType.TEXT + ) + if normalized_type == MessageType.TEXT and media_types: + if any(mime.startswith("image/") for mime in media_types): + normalized_type = MessageType.PHOTO + elif any(mime.startswith("video/") for mime in media_types): + normalized_type = MessageType.VIDEO + elif any(mime.startswith("audio/") for mime in media_types): + normalized_type = MessageType.VOICE + else: + normalized_type = MessageType.DOCUMENT + + reference = message.get("message_reference") or {} + referenced = message.get("referenced_message") or {} + referenced_author = referenced.get("author") or {} + reply_id = str(reference.get("message_id") or "") or None + + source = self.build_source( + chat_id=channel_id, + chat_name=channel.get("name") or channel.get("display_name"), + chat_type=chat_type, + user_id=author_id, + user_name=author.get("global_name") or author.get("username") or author_id, + scope_id=guild_id, + guild_id=guild_id, + message_id=message_id, + ) + channel_prompt = resolve_channel_prompt( + self.config.extra or {}, channel_id, None + ) + event = MessageEvent( + text=text, + message_type=normalized_type, + source=source, + raw_message=message, + message_id=message_id, + media_urls=media_urls, + media_types=media_types, + reply_to_message_id=reply_id, + reply_to_text=( + str(referenced.get("content")) + if referenced.get("content") is not None + else None + ), + reply_to_author_id=( + str(referenced_author.get("id")) + if referenced_author.get("id") + else None + ), + reply_to_author_name=( + referenced_author.get("global_name") + or referenced_author.get("username") + or None + ), + reply_to_is_own_message=bool( + referenced_author.get("id") + and str(referenced_author.get("id")) == self._bot_user_id + ), + channel_prompt=channel_prompt, + ) + await self.handle_message(event) + + async def _download_attachment( + self, attachment: Dict[str, Any] + ) -> Optional[Tuple[str, str]]: + import httpx + from tools.url_safety import create_ssrf_safe_async_client, is_safe_url + + url = str(attachment.get("url") or "") + if not url or not is_safe_url(url): + logger.warning("Fluxer: blocked unsafe attachment URL") + return None + filename = Path(str(attachment.get("filename") or "attachment")).name + declared_mime = str( + attachment.get("content_type") + or mimetypes.guess_type(filename)[0] + or "application/octet-stream" + ) + try: + limit = int(os.getenv("FLUXER_MAX_DOWNLOAD_BYTES", _DEFAULT_DOWNLOAD_LIMIT)) + except (TypeError, ValueError): + limit = _DEFAULT_DOWNLOAD_LIMIT + try: + async with create_ssrf_safe_async_client( + timeout=60.0, + follow_redirects=True, + event_hooks={"response": [_ssrf_redirect_guard]}, + ) as client: + async with client.stream("GET", url) as response: + response.raise_for_status() + content_length = response.headers.get("Content-Length") + if ( + content_length + and content_length.isdigit() + and int(content_length) > limit + ): + logger.warning( + "Fluxer: attachment exceeds download limit: %s", filename + ) + return None + chunks: List[bytes] = [] + size = 0 + async for chunk in response.aiter_bytes(64 * 1024): + size += len(chunk) + if size > limit: + logger.warning( + "Fluxer: attachment exceeded download limit: %s", + filename, + ) + return None + chunks.append(chunk) + data = b"".join(chunks) + mime = ( + response.headers.get("Content-Type", "").split(";", 1)[0] + or declared_mime + ) + except (httpx.HTTPError, ValueError, asyncio.TimeoutError) as exc: + logger.warning( + "Fluxer: failed to download attachment %s: %s", filename, exc + ) + return None + + suffix = Path(filename).suffix + if mime.startswith("image/"): + return cache_image_from_bytes(data, suffix or ".png"), mime + if mime.startswith("audio/"): + return cache_audio_from_bytes(data, suffix or ".ogg"), mime + return cache_document_from_bytes(data, filename), mime + + +# Keep json parsing mockable without shadowing the ``json=`` keyword in _api. +def json_module_loads(value: str) -> Any: + return json.loads(value) + + +# --------------------------------------------------------------------------- +# Plugin configuration, standalone delivery, and setup +# --------------------------------------------------------------------------- + + +def _env_enablement() -> Optional[Dict[str, Any]]: + token = os.getenv("FLUXER_BOT_TOKEN", "").strip() + if not token: + return None + seed: Dict[str, Any] = {"token": token} + api_url = os.getenv("FLUXER_API_URL", "").strip() + if api_url: + seed["api_url"] = _normalise_api_url(api_url) + gateway_url = os.getenv("FLUXER_GATEWAY_URL", "").strip() + if gateway_url: + seed["gateway_url"] = gateway_url + home = os.getenv("FLUXER_HOME_CHANNEL", "").strip() + if home: + seed["home_channel"] = { + "chat_id": home, + "name": os.getenv("FLUXER_HOME_CHANNEL_NAME", "Home").strip() or "Home", + } + return seed + + +def _is_connected(config: PlatformConfig) -> bool: + return bool(getattr(config, "enabled", False)) and validate_fluxer_config(config) + + +def _apply_yaml_config(_yaml_cfg: dict, fluxer_cfg: dict) -> Optional[dict]: + extras: Dict[str, Any] = {} + for key in ( + "api_url", + "gateway_url", + "allowed_channels", + "free_response_channels", + "require_mention", + "max_upload_bytes", + ): + if key in fluxer_cfg: + extras[key] = fluxer_cfg[key] + return extras or None + + +async def _standalone_send( + pconfig: PlatformConfig, + chat_id: str, + message: str, + *, + thread_id: Optional[str] = None, + media_files: Optional[list] = None, + force_document: bool = False, +) -> Dict[str, Any]: + import aiohttp + + adapter = FluxerAdapter(pconfig) + proxy = resolve_proxy_url( + platform_env_var="FLUXER_PROXY", + target_hosts=["api.fluxer.app", "gateway.fluxer.app"], + ) + session_kwargs, request_kwargs = proxy_kwargs_for_aiohttp(proxy) + adapter._request_proxy_kwargs = request_kwargs + adapter._session = aiohttp.ClientSession( + timeout=aiohttp.ClientTimeout(total=90), **session_kwargs + ) + try: + files: List[Tuple[str, bytes, str]] = [] + for media in (media_files or [])[:MAX_ATTACHMENTS]: + path_value = media.get("path") if isinstance(media, dict) else media + if not path_value: + continue + path = Path(str(path_value)) + if not path.is_file(): + continue + if path.stat().st_size > adapter._max_upload_bytes: + return { + "error": ( + "File exceeds Fluxer upload limit of " + f"{adapter._max_upload_bytes} bytes: {path.name}" + ) + } + mime = mimetypes.guess_type(path.name)[0] or "application/octet-stream" + try: + data = await asyncio.to_thread( + _read_file_bounded, path, adapter._max_upload_bytes + ) + except _UploadTooLarge as exc: + return {"error": f"{exc}: {path.name}"} + files.append((path.name, data, mime)) + if files: + result = await adapter._send_files(chat_id, files, message, thread_id) + else: + result = await adapter.send(chat_id, message, reply_to=thread_id) + if not result.success: + return {"error": result.error or "Fluxer send failed"} + return { + "success": True, + "platform": "fluxer", + "chat_id": chat_id, + "message_id": result.message_id, + } + finally: + await adapter._session.close() + adapter._session = None + + +def interactive_setup() -> None: + from hermes_cli.cli_output import ( + print_header, + print_info, + print_success, + prompt, + prompt_yes_no, + ) + from hermes_cli.config import get_env_value, remove_env_value, save_env_value + + print_header("Fluxer") + if get_env_value("FLUXER_BOT_TOKEN"): + print_info("Fluxer is already configured") + if not prompt_yes_no("Reconfigure Fluxer?", False): + return + + print_info("Create a bot in Fluxer Developer Settings and copy its bot token.") + print_info("Do not paste the token into a chat message.") + token = prompt("Fluxer bot token", password=True) + if not token: + return + save_env_value("FLUXER_BOT_TOKEN", token.strip()) + print_success("Fluxer bot token saved") + + # Guided setup is fail-closed. Clear any stale broad-access override before + # applying the selected allowlist/default pairing policy. + remove_env_value("FLUXER_ALLOW_ALL_USERS") + allowed = prompt( + "Allowed Fluxer user IDs (comma-separated, empty for default pairing policy)" + ) + if allowed.strip(): + save_env_value("FLUXER_ALLOWED_USERS", allowed.replace(" ", "")) + print_success("Fluxer user allowlist configured") + else: + remove_env_value("FLUXER_ALLOWED_USERS") + print_info("No Fluxer user allowlist configured") + + home = prompt("Home channel ID (empty to set later with /set-home)").strip() + if home: + save_env_value("FLUXER_HOME_CHANNEL", home) + else: + remove_env_value("FLUXER_HOME_CHANNEL") + + +def register(ctx) -> None: + ctx.register_platform( + name="fluxer", + label="Fluxer", + adapter_factory=lambda config: FluxerAdapter(config), + check_fn=check_fluxer_requirements, + validate_config=validate_fluxer_config, + is_connected=_is_connected, + required_env=["FLUXER_BOT_TOKEN"], + install_hint="pip install aiohttp", + setup_fn=interactive_setup, + env_enablement_fn=_env_enablement, + apply_yaml_config_fn=_apply_yaml_config, + allowed_users_env="FLUXER_ALLOWED_USERS", + allow_all_env="FLUXER_ALLOW_ALL_USERS", + cron_deliver_env_var="FLUXER_HOME_CHANNEL", + standalone_sender_fn=_standalone_send, + max_message_length=MAX_MESSAGE_LENGTH, + emoji="🟣", + platform_hint=( + "You are chatting via Fluxer. Markdown, replies, and file attachments " + "are supported. Do not emit mass mentions unless the user explicitly asks." + ), + allow_update_command=True, + ) diff --git a/plugins/platforms/fluxer/plugin.yaml b/plugins/platforms/fluxer/plugin.yaml new file mode 100644 index 0000000000000..f8dfcc57a653f --- /dev/null +++ b/plugins/platforms/fluxer/plugin.yaml @@ -0,0 +1,56 @@ +name: fluxer-platform +label: Fluxer +kind: platform +version: 1.0.0 +description: > + Native Fluxer gateway adapter for Hermes Agent. Connects directly to + Fluxer's REST API and real-time Gateway for DMs and server channels, with + replies, typing indicators, file attachments, allowlists, mention gating, + reconnect/resume, and home-channel cron delivery. +author: NousResearch +requires_env: + - name: FLUXER_BOT_TOKEN + description: "Fluxer bot token" + prompt: "Fluxer bot token" + password: true +optional_env: + - name: FLUXER_API_URL + description: "Fluxer REST API base; defaults to https://api.fluxer.app/v1" + prompt: "Fluxer API URL" + password: false + - name: FLUXER_GATEWAY_URL + description: "Optional Fluxer Gateway WebSocket URL override" + prompt: "Fluxer Gateway URL" + password: false + - name: FLUXER_ALLOWED_USERS + description: "Comma-separated Fluxer user IDs allowed to talk to the bot" + prompt: "Allowed users (comma-separated)" + password: false + - name: FLUXER_ALLOW_ALL_USERS + description: "Allow any Fluxer user to trigger the bot (development only)" + prompt: "Allow all users? (true/false)" + password: false + - name: FLUXER_HOME_CHANNEL + description: "Default Fluxer channel ID for cron and notification delivery" + prompt: "Home channel ID" + password: false + - name: FLUXER_REQUIRE_MENTION + description: "Require bot mention in server channels (default true)" + prompt: "Require mention? (true/false)" + password: false + - name: FLUXER_FREE_RESPONSE_CHANNELS + description: "Comma-separated server channel IDs where mention is not required" + prompt: "Free-response channel IDs (comma-separated)" + password: false + - name: FLUXER_ALLOWED_CHANNELS + description: "Optional comma-separated server-channel allowlist" + prompt: "Allowed channel IDs (comma-separated)" + password: false + - name: FLUXER_PROXY + description: "Optional HTTP/SOCKS proxy for Fluxer REST and Gateway traffic" + prompt: "Proxy URL" + password: false + - name: FLUXER_MAX_UPLOAD_BYTES + description: "Maximum bytes read for one outbound upload (default 25 MiB)" + prompt: "Maximum upload bytes" + password: false diff --git a/tests/gateway/test_fluxer.py b/tests/gateway/test_fluxer.py new file mode 100644 index 0000000000000..d425c8f22cf3f --- /dev/null +++ b/tests/gateway/test_fluxer.py @@ -0,0 +1,1029 @@ +"""Tests for the bundled Fluxer messaging platform adapter.""" + +from __future__ import annotations + +import asyncio +import os +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from gateway.config import Platform, PlatformConfig +from gateway.platforms.base import MessageType + + +def _make_adapter(**extra): + from plugins.platforms.fluxer.adapter import FluxerAdapter + + config = PlatformConfig( + enabled=True, + token="test-token", + extra={ + "api_url": "https://api.example.test/v1", + "gateway_url": "wss://gateway.example.test", + **extra, + }, + ) + return FluxerAdapter(config) + + +class TestFluxerPluginRegistration: + def test_dynamic_platform_member_is_discoverable(self): + assert Platform("fluxer").value == "fluxer" + + def test_register_exposes_full_platform_integration(self): + from plugins.platforms.fluxer.adapter import register + + ctx = MagicMock() + register(ctx) + kwargs = ctx.register_platform.call_args.kwargs + assert kwargs["name"] == "fluxer" + assert kwargs["required_env"] == ["FLUXER_BOT_TOKEN"] + assert kwargs["allowed_users_env"] == "FLUXER_ALLOWED_USERS" + assert kwargs["allow_all_env"] == "FLUXER_ALLOW_ALL_USERS" + assert kwargs["cron_deliver_env_var"] == "FLUXER_HOME_CHANNEL" + assert kwargs["max_message_length"] == 4000 + assert callable(kwargs["env_enablement_fn"]) + assert callable(kwargs["standalone_sender_fn"]) + + def test_env_enablement_seeds_token_urls_and_home(self, monkeypatch): + from plugins.platforms.fluxer.adapter import _env_enablement + + monkeypatch.setenv("FLUXER_BOT_TOKEN", "secret") + monkeypatch.setenv("FLUXER_API_URL", "https://self.example/v1/") + monkeypatch.setenv("FLUXER_GATEWAY_URL", "wss://gw.self.example/") + monkeypatch.setenv("FLUXER_HOME_CHANNEL", "123") + monkeypatch.setenv("FLUXER_HOME_CHANNEL_NAME", "Ops") + + seed = _env_enablement() + assert seed == { + "token": "secret", + "api_url": "https://self.example/v1", + "gateway_url": "wss://gw.self.example/", + "home_channel": {"chat_id": "123", "name": "Ops"}, + } + + def test_env_enablement_requires_token(self, monkeypatch): + from plugins.platforms.fluxer.adapter import _env_enablement + + monkeypatch.delenv("FLUXER_BOT_TOKEN", raising=False) + assert _env_enablement() is None + + def test_env_enablement_does_not_override_yaml_api_url_with_default( + self, monkeypatch + ): + from plugins.platforms.fluxer.adapter import _env_enablement + + monkeypatch.setenv("FLUXER_BOT_TOKEN", "secret") + monkeypatch.delenv("FLUXER_API_URL", raising=False) + + seed = _env_enablement() + + assert seed is not None + assert seed["token"] == "secret" + assert "api_url" not in seed + + +class TestFluxerConfiguration: + def test_defaults_to_production_endpoints(self): + from plugins.platforms.fluxer.adapter import FluxerAdapter + + adapter = FluxerAdapter(PlatformConfig(enabled=True, token="tok", extra={})) + assert adapter._api_url == "https://api.fluxer.app/v1" + assert adapter._gateway_url == "" + + def test_bot_authorization_header_uses_bot_scheme(self): + adapter = _make_adapter() + assert adapter._headers()["Authorization"] == "Bot test-token" + + def test_insecure_remote_api_and_gateway_urls_are_rejected(self): + with pytest.raises(ValueError, match="HTTPS"): + _make_adapter(api_url="http://api.attacker.invalid/v1") + + with pytest.raises(ValueError, match="WSS"): + _make_adapter(gateway_url="ws://gateway.attacker.invalid/") + + @pytest.mark.parametrize( + ("field", "url"), + [ + ("api_url", "https://:443/v1"), + ("api_url", "https://example.com:bad/v1"), + ("api_url", "https://../v1"), + ("gateway_url", "wss://:443/gateway"), + ("gateway_url", "wss://example.com:bad/gateway"), + ("gateway_url", "wss://../gateway"), + ], + ) + def test_malformed_token_bearing_endpoints_are_rejected(self, field, url): + with pytest.raises(ValueError, match="valid hostname and port"): + _make_adapter(**{field: url}) + + def test_loopback_http_and_websocket_are_allowed_for_local_development(self): + adapter = _make_adapter( + api_url="http://127.0.0.1:9000/v1", + gateway_url="ws://localhost:9001/gateway", + ) + + assert adapter._api_url == "http://127.0.0.1:9000/v1" + assert adapter._gateway_url == "ws://localhost:9001/gateway" + + def test_validate_requires_token_and_https_api(self): + from plugins.platforms.fluxer.adapter import validate_fluxer_config + + assert validate_fluxer_config( + PlatformConfig( + enabled=True, + token="tok", + extra={"api_url": "https://api.fluxer.app/v1"}, + ) + ) + assert not validate_fluxer_config( + PlatformConfig( + enabled=True, token="", extra={"api_url": "https://api.fluxer.app/v1"} + ) + ) + assert not validate_fluxer_config( + PlatformConfig( + enabled=True, token="tok", extra={"api_url": "file:///tmp/nope"} + ) + ) + + def test_validate_rejects_insecure_gateway_from_environment(self, monkeypatch): + from plugins.platforms.fluxer.adapter import validate_fluxer_config + + monkeypatch.setenv("FLUXER_GATEWAY_URL", "ws://gateway.attacker.invalid/") + config = PlatformConfig( + enabled=True, + token="tok", + extra={"api_url": "https://api.fluxer.app/v1"}, + ) + + assert not validate_fluxer_config(config) + + +class TestFluxerOutbound: + def test_bounded_file_reader_rejects_growth_past_limit(self, tmp_path): + from plugins.platforms.fluxer.adapter import ( + _UploadTooLarge, + _read_file_bounded, + ) + + path = tmp_path / "grew-during-read.bin" + path.write_bytes(b"12345") + + with pytest.raises(_UploadTooLarge): + _read_file_bounded(path, 4) + + @pytest.mark.asyncio + async def test_standalone_send_rejects_oversized_media(self, monkeypatch, tmp_path): + import aiohttp + from plugins.platforms.fluxer.adapter import FluxerAdapter, _standalone_send + + path = tmp_path / "large.bin" + path.write_bytes(b"12345") + fake_session = MagicMock() + fake_session.close = AsyncMock() + monkeypatch.setattr(aiohttp, "ClientSession", lambda *_a, **_kw: fake_session) + send_files = AsyncMock() + monkeypatch.setattr(FluxerAdapter, "_send_files", send_files) + + result = await _standalone_send( + PlatformConfig( + enabled=True, + token="test-token", + extra={ + "api_url": "https://api.example.test/v1", + "max_upload_bytes": 4, + }, + ), + "chan", + "caption", + media_files=[str(path)], + ) + + assert "upload limit" in result["error"] + send_files.assert_not_awaited() + + @pytest.mark.asyncio + async def test_send_posts_safe_mentions_and_reply_reference(self): + adapter = _make_adapter() + adapter._api = AsyncMock(return_value={"id": "msg-2"}) + + result = await adapter.send("chan-1", "hello @everyone", reply_to="msg-1") + + assert result.success is True + assert result.message_id == "msg-2" + method, path = adapter._api.call_args.args[:2] + payload = adapter._api.call_args.kwargs["json"] + assert (method, path) == ("POST", "channels/chan-1/messages") + assert payload["content"] == "hello @everyone" + assert payload["allowed_mentions"] == {"parse": [], "replied_user": False} + assert payload["message_reference"] == { + "message_id": "msg-1", + "channel_id": "chan-1", + "type": 0, + } + assert isinstance(payload["nonce"], str) and payload["nonce"] + + @pytest.mark.asyncio + async def test_send_chunks_at_fluxer_limit(self): + adapter = _make_adapter() + adapter._api = AsyncMock(side_effect=[{"id": "one"}, {"id": "two"}]) + + result = await adapter.send("chan", "x" * 5000) + + assert result.success is True + assert result.message_id == "two" + assert result.continuation_message_ids == ("one",) + assert adapter._api.await_count == 2 + for call in adapter._api.await_args_list: + assert len(call.kwargs["json"]["content"]) <= 4000 + + @pytest.mark.asyncio + async def test_send_maps_rate_limit_to_retryable_result(self): + adapter = _make_adapter() + adapter._api = AsyncMock(return_value=None) + adapter._last_http_status = 429 + adapter._last_http_error = "rate limited" + adapter._last_retry_after = 2.5 + + result = await adapter.send("chan", "hello") + + assert result.success is False + assert result.error_kind == "rate_limited" + assert result.retryable is True + assert result.retry_after == 2.5 + + @pytest.mark.asyncio + async def test_multipart_rate_limit_honors_retry_after_header(self): + adapter = _make_adapter() + response = MagicMock() + response.status = 429 + response.headers = {"Retry-After": "3.5"} + response.text = AsyncMock(return_value="{}") + context = MagicMock() + context.__aenter__ = AsyncMock(return_value=response) + context.__aexit__ = AsyncMock(return_value=False) + adapter._session = MagicMock() + adapter._session.request.return_value = context + + result = await adapter._api_multipart( + "POST", + "channels/chan/messages", + payload={"content": "x"}, + files=[("x.txt", b"x", "text/plain")], + ) + + assert result is None + assert adapter._last_retry_after == 3.5 + + @pytest.mark.asyncio + async def test_typing_and_edit_use_fluxer_rest_routes(self): + adapter = _make_adapter() + adapter._api = AsyncMock(side_effect=[{}, {"id": "edited"}]) + + await adapter.send_typing("chan") + result = await adapter.edit_message("chan", "msg", "new text") + + assert result.success is True + assert adapter._api.await_args_list[0].args[:2] == ( + "POST", + "channels/chan/typing", + ) + assert adapter._api.await_args_list[1].args[:2] == ( + "PATCH", + "channels/chan/messages/msg", + ) + assert adapter._api.await_args_list[1].kwargs["json"]["allowed_mentions"] == { + "parse": [], + "replied_user": False, + } + + @pytest.mark.asyncio + async def test_local_file_uses_fluxer_multipart_shape(self, tmp_path): + adapter = _make_adapter() + adapter._api_multipart = AsyncMock(return_value={"id": "file-msg"}) + path = tmp_path / "report.txt" + path.write_text("report body") + + result = await adapter.send_document("chan", str(path), caption="Report") + + assert result.success is True + assert result.message_id == "file-msg" + payload = adapter._api_multipart.call_args.kwargs["payload"] + files = adapter._api_multipart.call_args.kwargs["files"] + assert payload["content"] == "Report" + assert payload["attachments"][0]["id"] == 0 + assert payload["attachments"][0]["filename"] == "report.txt" + assert files == [("report.txt", b"report body", "text/plain")] + + @pytest.mark.asyncio + async def test_multipart_caption_is_capped_at_fluxer_limit(self): + adapter = _make_adapter() + adapter._api_multipart = AsyncMock(return_value={"id": "file-msg"}) + + result = await adapter._send_files( + "chan", [("x.txt", b"x", "text/plain")], "c" * 5000, None + ) + + assert result.success is True + payload = adapter._api_multipart.call_args.kwargs["payload"] + assert len(payload["content"]) == 4000 + + @pytest.mark.asyncio + async def test_document_rejects_file_above_upload_limit(self, tmp_path): + adapter = _make_adapter() + adapter._max_upload_bytes = 4 + adapter._api_multipart = AsyncMock() + path = tmp_path / "large.bin" + path.write_bytes(b"12345") + + result = await adapter.send_document("chan", str(path)) + + assert result.success is False + assert result.error_kind == "file_too_large" + adapter._api_multipart.assert_not_awaited() + + @pytest.mark.asyncio + async def test_document_reads_file_off_event_loop(self, monkeypatch, tmp_path): + adapter = _make_adapter() + adapter._api_multipart = AsyncMock(return_value={"id": "file-msg"}) + path = tmp_path / "report.txt" + path.write_text("report body") + to_thread = AsyncMock(return_value=b"report body") + monkeypatch.setattr(asyncio, "to_thread", to_thread) + + result = await adapter.send_document("chan", str(path)) + + assert result.success is True + to_thread.assert_awaited_once() + assert to_thread.await_args is not None + read_callable, read_path, read_limit = to_thread.await_args.args + assert read_callable.__name__ == "_read_file_bounded" + assert read_path == path + assert read_limit == adapter._max_upload_bytes + + +class TestFluxerGatewayProtocol: + @pytest.mark.asyncio + async def test_gateway_socket_is_closed_before_reconnect(self): + from plugins.platforms.fluxer.adapter import _ReconnectRequested + + adapter = _make_adapter() + ws = MagicMock() + ws.__aiter__.return_value = [] + ws.closed = False + ws.close_code = 1000 + ws.close = AsyncMock() + adapter._session = MagicMock() + adapter._session.ws_connect = AsyncMock(return_value=ws) + + with pytest.raises(_ReconnectRequested): + await adapter._gateway_once() + + ws.close.assert_awaited_once() + + @pytest.mark.asyncio + async def test_invalid_sequence_close_clears_resume_state(self): + from plugins.platforms.fluxer.adapter import _ReconnectRequested + + adapter = _make_adapter() + adapter._session_id = "stale-session" + adapter._sequence = 42 + ws = MagicMock() + ws.__aiter__.return_value = [] + ws.closed = False + ws.close_code = 4007 + ws.close = AsyncMock() + adapter._session = MagicMock() + adapter._session.ws_connect = AsyncMock(return_value=ws) + + with pytest.raises(_ReconnectRequested): + await adapter._gateway_once() + + assert adapter._session_id == "" + assert adapter._sequence == 0 + + @pytest.mark.asyncio + @pytest.mark.parametrize("close_code", [4004, 4010, 4011, 4012]) + async def test_configuration_close_is_permanent(self, close_code): + from plugins.platforms.fluxer.adapter import _PermanentGatewayError + + adapter = _make_adapter() + ws = MagicMock() + ws.__aiter__.return_value = [] + ws.closed = False + ws.close_code = close_code + ws.close = AsyncMock() + adapter._session = MagicMock() + adapter._session.ws_connect = AsyncMock(return_value=ws) + + with pytest.raises(_PermanentGatewayError): + await adapter._gateway_once() + + @pytest.mark.asyncio + @pytest.mark.parametrize("close_code", [4003, 4005, 4013]) + async def test_recoverable_protocol_close_reconnects(self, close_code): + from plugins.platforms.fluxer.adapter import _ReconnectRequested + + adapter = _make_adapter() + ws = MagicMock() + ws.__aiter__.return_value = [] + ws.closed = False + ws.close_code = close_code + ws.close = AsyncMock() + adapter._session = MagicMock() + adapter._session.ws_connect = AsyncMock(return_value=ws) + + with pytest.raises(_ReconnectRequested): + await adapter._gateway_once() + + @pytest.mark.asyncio + async def test_missing_heartbeat_ack_closes_socket(self, monkeypatch): + adapter = _make_adapter() + adapter._heartbeat_acknowledged = False + ws = MagicMock() + ws.closed = False + + async def close(*_args, **_kwargs): + ws.closed = True + + ws.close = AsyncMock(side_effect=close) + ws.send_json = AsyncMock() + monkeypatch.setattr(asyncio, "sleep", AsyncMock()) + + await adapter._heartbeat_loop(ws, 45.0) + + ws.close.assert_awaited_once() + ws.send_json.assert_not_awaited() + + @pytest.mark.asyncio + async def test_heartbeat_ack_is_tracked(self): + adapter = _make_adapter() + adapter._heartbeat_acknowledged = False + + await adapter._handle_gateway_payload({"op": 11, "d": None}, AsyncMock()) + + assert adapter._heartbeat_acknowledged is True + + @pytest.mark.asyncio + async def test_server_reconnect_opcode_requests_immediate_retry(self): + from plugins.platforms.fluxer.adapter import _ReconnectRequested + + adapter = _make_adapter() + with pytest.raises(_ReconnectRequested) as raised: + await adapter._handle_gateway_payload({"op": 7, "d": None}, AsyncMock()) + + assert raised.value.retry_delay == 0.0 + + @pytest.mark.asyncio + async def test_invalid_session_requests_protocol_delay(self, monkeypatch): + from plugins.platforms.fluxer.adapter import _ReconnectRequested + + adapter = _make_adapter() + monkeypatch.setattr( + "plugins.platforms.fluxer.adapter.random.uniform", lambda _a, _b: 3.0 + ) + with pytest.raises(_ReconnectRequested) as raised: + await adapter._handle_gateway_payload({"op": 9, "d": False}, AsyncMock()) + + assert raised.value.retry_delay == 3.0 + assert adapter._session_id == "" + assert adapter._sequence == 0 + + @pytest.mark.asyncio + async def test_ready_reconnect_resets_backoff(self, monkeypatch): + from plugins.platforms.fluxer.adapter import ( + _PermanentGatewayError, + _ReconnectRequested, + ) + + adapter = _make_adapter() + attempts = 0 + + async def gateway_once(): + nonlocal attempts + attempts += 1 + if attempts <= 2: + adapter._gateway_was_ready = True + raise _ReconnectRequested("test reconnect") + raise _PermanentGatewayError("stop") + + adapter._gateway_once = gateway_once + sleep = AsyncMock() + monkeypatch.setattr(asyncio, "sleep", sleep) + monkeypatch.setattr( + "plugins.platforms.fluxer.adapter.random.random", lambda: 0.0 + ) + + await adapter._gateway_loop() + + assert [call.args[0] for call in sleep.await_args_list] == [2.0, 2.0] + + @pytest.mark.asyncio + async def test_connect_cancellation_releases_session_and_token_lock( + self, monkeypatch + ): + import aiohttp + import gateway.status as status_mod + + adapter = _make_adapter() + expected_lock_id = __import__("hashlib").sha256(b"test-token").hexdigest()[:24] + fake_session = MagicMock() + fake_session.closed = False + fake_session.close = AsyncMock() + release = MagicMock() + monkeypatch.setattr( + status_mod, "acquire_scoped_lock", lambda *_a, **_kw: (True, None) + ) + monkeypatch.setattr(status_mod, "release_scoped_lock", release) + monkeypatch.setattr(aiohttp, "ClientSession", lambda *_a, **_kw: fake_session) + adapter._api = AsyncMock(side_effect=asyncio.CancelledError) + + with pytest.raises(asyncio.CancelledError): + await adapter.connect() + + fake_session.close.assert_awaited_once() + release.assert_called_once_with("fluxer", expected_lock_id) + + @pytest.mark.asyncio + async def test_connect_uses_base_platform_lock_helper(self, monkeypatch): + import aiohttp + import gateway.status as status_mod + + adapter = _make_adapter() + expected_lock_id = __import__("hashlib").sha256(b"test-token").hexdigest()[:24] + adapter._acquire_platform_lock = MagicMock(return_value=True) + fake_session = MagicMock() + fake_session.closed = False + fake_session.close = AsyncMock() + monkeypatch.setattr( + status_mod, "acquire_scoped_lock", lambda *_a, **_kw: (True, None) + ) + monkeypatch.setattr(status_mod, "release_scoped_lock", MagicMock()) + monkeypatch.setattr(aiohttp, "ClientSession", lambda *_a, **_kw: fake_session) + adapter._api = AsyncMock(side_effect=asyncio.CancelledError) + + with pytest.raises(asyncio.CancelledError): + await adapter.connect() + + adapter._acquire_platform_lock.assert_called_once_with( + "fluxer", expected_lock_id, "Fluxer bot token" + ) + + @pytest.mark.asyncio + async def test_hello_identifies_with_bot_token(self): + adapter = _make_adapter() + ws = AsyncMock() + adapter._start_heartbeat = MagicMock() + + await adapter._handle_gateway_payload( + {"op": 10, "d": {"heartbeat_interval": 45000}}, ws + ) + + adapter._start_heartbeat.assert_called_once_with(ws, 45000) + identify = ws.send_json.await_args.args[0] + assert identify["op"] == 2 + assert identify["d"]["token"] == "test-token" + assert identify["d"]["properties"]["browser"] == "hermes-agent" + assert "MESSAGE_UPDATE" in identify["d"]["ignored_events"] + + @pytest.mark.asyncio + async def test_gateway_permanent_failure_marks_adapter_disconnected(self): + from plugins.platforms.fluxer.adapter import _PermanentGatewayError + + adapter = _make_adapter() + adapter._gateway_once = AsyncMock( + side_effect=_PermanentGatewayError("bad token") + ) + adapter._ready_event.set() + adapter._mark_disconnected = MagicMock() + adapter._notify_fatal_error = AsyncMock() + + await adapter._gateway_loop() + + adapter._mark_disconnected.assert_called_once() + adapter._notify_fatal_error.assert_awaited_once() + + @pytest.mark.asyncio + async def test_connect_preserves_permanent_gateway_failure(self, monkeypatch): + import aiohttp + from plugins.platforms.fluxer.adapter import _PermanentGatewayError + + adapter = _make_adapter() + fake_session = MagicMock() + fake_session.closed = False + fake_session.close = AsyncMock() + monkeypatch.setattr(aiohttp, "ClientSession", lambda *_a, **_kw: fake_session) + adapter._api = AsyncMock(return_value={"id": "bot-1", "username": "Hermes"}) + adapter._gateway_once = AsyncMock( + side_effect=_PermanentGatewayError("bad gateway credential") + ) + adapter._notify_fatal_error = AsyncMock() + + real_wait_for = asyncio.wait_for + + async def wait_for_gateway(awaitable, timeout): + gateway_task = adapter._gateway_task + assert gateway_task is not None + await gateway_task + if hasattr(awaitable, "close"): + awaitable.close() + if adapter._ready_event.is_set(): + return True + raise asyncio.TimeoutError + + monkeypatch.setattr(asyncio, "wait_for", wait_for_gateway) + try: + assert await adapter.connect() is False + finally: + monkeypatch.setattr(asyncio, "wait_for", real_wait_for) + + assert adapter._fatal_error_code == "fluxer_gateway_auth" + assert adapter._fatal_error_retryable is False + adapter._notify_fatal_error.assert_not_awaited() + fatal_message = adapter._fatal_error_message + assert fatal_message is not None + assert "bad gateway credential" in fatal_message + + @pytest.mark.asyncio + async def test_hello_resumes_existing_session(self): + adapter = _make_adapter() + adapter._session_id = "session-1" + adapter._sequence = 42 + ws = AsyncMock() + adapter._start_heartbeat = MagicMock() + + await adapter._handle_gateway_payload( + {"op": 10, "d": {"heartbeat_interval": 45000}}, ws + ) + + resume = ws.send_json.await_args.args[0] + assert resume == { + "op": 6, + "d": {"token": "test-token", "session_id": "session-1", "seq": 42}, + } + + @pytest.mark.asyncio + async def test_server_heartbeat_request_is_acknowledged_with_sequence(self): + adapter = _make_adapter() + adapter._sequence = 9 + ws = AsyncMock() + + await adapter._handle_gateway_payload({"op": 1, "d": None}, ws) + + ws.send_json.assert_awaited_once_with({"op": 1, "d": 9}) + + @pytest.mark.asyncio + async def test_ready_records_session_and_signals_connection(self): + adapter = _make_adapter() + adapter._ready_event = asyncio.Event() + ws = AsyncMock() + + await adapter._handle_gateway_payload( + {"op": 0, "s": 7, "t": "READY", "d": {"session_id": "sid"}}, ws + ) + + assert adapter._sequence == 7 + assert adapter._session_id == "sid" + assert adapter._ready_event.is_set() + + @pytest.mark.asyncio + async def test_ready_retains_valid_resume_gateway_url(self): + adapter = _make_adapter() + + await adapter._handle_gateway_payload( + { + "op": 0, + "s": 7, + "t": "READY", + "d": { + "session_id": "sid", + "resume_gateway_url": "wss://resume.example.test/socket", + }, + }, + AsyncMock(), + ) + + assert adapter._resume_gateway_url == "wss://resume.example.test/socket" + assert adapter._gateway_connect_url().startswith( + "wss://resume.example.test/socket?" + ) + + @pytest.mark.asyncio + async def test_ready_rejects_insecure_resume_gateway_url(self): + adapter = _make_adapter() + + await adapter._handle_gateway_payload( + { + "op": 0, + "s": 7, + "t": "READY", + "d": { + "session_id": "sid", + "resume_gateway_url": "ws://attacker.invalid/socket", + }, + }, + AsyncMock(), + ) + + assert adapter._resume_gateway_url == "" + + @pytest.mark.asyncio + async def test_ready_without_resume_url_clears_stale_route(self): + adapter = _make_adapter() + adapter._resume_gateway_url = "wss://stale.example.test/socket" + + await adapter._handle_gateway_payload( + {"op": 0, "s": 8, "t": "READY", "d": {"session_id": "new-sid"}}, + AsyncMock(), + ) + + assert adapter._resume_gateway_url == "" + + @pytest.mark.asyncio + async def test_message_create_dispatches_to_ingress(self): + adapter = _make_adapter() + adapter._handle_message_create = AsyncMock() + ws = AsyncMock() + payload = {"id": "m1"} + + await adapter._handle_gateway_payload( + {"op": 0, "s": 8, "t": "MESSAGE_CREATE", "d": payload}, ws + ) + + adapter._handle_message_create.assert_awaited_once_with(payload) + + +class TestFluxerInbound: + @pytest.mark.asyncio + async def test_transient_channel_lookup_failure_is_not_cached_as_guild_channel( + self, + ): + adapter = _make_adapter() + adapter._bot_user_id = "bot-1" + adapter._api = AsyncMock( + side_effect=[None, {"id": "dm-1", "type": 1, "name": "Direct message"}] + ) + adapter.handle_message = AsyncMock() + + base_message = { + "channel_id": "dm-1", + "type": 0, + "content": "hello", + "author": {"id": "user-1", "username": "Kait"}, + "attachments": [], + } + await adapter._handle_message_create({**base_message, "id": "msg-lookup-1"}) + await adapter._handle_message_create({**base_message, "id": "msg-lookup-2"}) + + assert adapter._api.await_count == 2 + assert adapter.handle_message.await_count == 2 + assert adapter._channel_cache["dm-1"]["type"] == 1 + + @pytest.mark.asyncio + async def test_unauthorized_sender_is_rejected_before_attachment_download(self): + adapter = _make_adapter() + adapter._bot_user_id = "bot-1" + adapter._get_channel = AsyncMock(return_value={"id": "dm-1", "type": 1}) + adapter._download_attachment = AsyncMock() + adapter._authorization_check = lambda user_id, chat_type, chat_id: False + adapter.handle_message = AsyncMock() + + await adapter._handle_message_create({ + "id": "msg-blocked", + "channel_id": "dm-1", + "type": 0, + "content": "", + "author": {"id": "attacker", "username": "Mallory"}, + "attachments": [ + {"url": "https://cdn.example.test/large.bin", "filename": "large.bin"} + ], + }) + + adapter._download_attachment.assert_not_awaited() + adapter.handle_message.assert_awaited_once() + call = adapter.handle_message.await_args + assert call is not None + event = call.args[0] + assert event.source.user_id == "attacker" + assert event.media_urls == [] + assert event.media_types == [] + + @pytest.mark.asyncio + @pytest.mark.parametrize("authorization_state", ["missing", "raises"]) + async def test_indeterminate_authorization_never_downloads_attachment( + self, authorization_state + ): + adapter = _make_adapter() + adapter._bot_user_id = "bot-1" + adapter._get_channel = AsyncMock(return_value={"id": "dm-1", "type": 1}) + adapter._download_attachment = AsyncMock() + if authorization_state == "raises": + + def broken_check(*_args): + raise RuntimeError("authorization backend unavailable") + + adapter._authorization_check = broken_check + adapter.handle_message = AsyncMock() + + await adapter._handle_message_create({ + "id": f"msg-{authorization_state}", + "channel_id": "dm-1", + "type": 0, + "content": "", + "author": {"id": "unknown", "username": "Unknown"}, + "attachments": [ + {"url": "https://cdn.example.test/large.bin", "filename": "large.bin"} + ], + }) + + adapter._download_attachment.assert_not_awaited() + adapter.handle_message.assert_awaited_once() + call = adapter.handle_message.await_args + assert call is not None + event = call.args[0] + assert event.source.user_id == "unknown" + assert event.media_urls == [] + + @pytest.mark.asyncio + async def test_dm_message_becomes_normalized_event(self): + adapter = _make_adapter() + adapter._bot_user_id = "bot-1" + adapter._get_channel = AsyncMock( + return_value={"id": "chan", "type": 1, "name": "DM"} + ) + adapter.handle_message = AsyncMock() + + await adapter._handle_message_create({ + "id": "m1", + "channel_id": "chan", + "content": "hello", + "type": 0, + "author": {"id": "user-1", "username": "kait", "bot": False}, + "attachments": [], + "mentions": [], + }) + + call = adapter.handle_message.await_args + assert call is not None + event = call.args[0] + assert event.text == "hello" + assert event.message_type == MessageType.TEXT + assert event.source.platform.value == "fluxer" + assert event.source.chat_type == "dm" + assert event.source.user_id == "user-1" + assert event.message_id == "m1" + + @pytest.mark.asyncio + async def test_guild_channel_requires_and_strips_bot_mention(self, monkeypatch): + adapter = _make_adapter() + adapter._bot_user_id = "bot-1" + adapter._get_channel = AsyncMock( + return_value={"id": "chan", "type": 0, "name": "general", "guild_id": "g1"} + ) + adapter.handle_message = AsyncMock() + base = { + "channel_id": "chan", + "type": 0, + "author": {"id": "user-1", "username": "kait", "bot": False}, + "attachments": [], + } + + await adapter._handle_message_create({ + **base, + "id": "m1", + "content": "ignored", + "mentions": [], + }) + adapter.handle_message.assert_not_awaited() + + await adapter._handle_message_create({ + **base, + "id": "m2", + "content": "<@bot-1> diagnose this", + "mentions": [{"id": "bot-1"}], + }) + call = adapter.handle_message.await_args + assert call is not None + event = call.args[0] + assert event.text == "diagnose this" + assert event.source.chat_type == "channel" + assert event.source.guild_id == "g1" + + @pytest.mark.asyncio + async def test_free_response_channel_bypasses_mention(self, monkeypatch): + monkeypatch.setenv("FLUXER_FREE_RESPONSE_CHANNELS", "chan") + adapter = _make_adapter() + adapter._bot_user_id = "bot-1" + adapter._get_channel = AsyncMock( + return_value={"id": "chan", "type": 0, "guild_id": "g1"} + ) + adapter.handle_message = AsyncMock() + + await adapter._handle_message_create({ + "id": "m1", + "channel_id": "chan", + "content": "hello room", + "type": 0, + "author": {"id": "user-1", "username": "kait"}, + "attachments": [], + "mentions": [], + }) + + adapter.handle_message.assert_awaited_once() + + @pytest.mark.asyncio + async def test_allowed_channel_whitelist_blocks_other_channels(self, monkeypatch): + monkeypatch.setenv("FLUXER_ALLOWED_CHANNELS", "allowed") + adapter = _make_adapter() + adapter._bot_user_id = "bot-1" + adapter._get_channel = AsyncMock( + return_value={"id": "blocked", "type": 0, "guild_id": "g1"} + ) + adapter.handle_message = AsyncMock() + + await adapter._handle_message_create({ + "id": "m1", + "channel_id": "blocked", + "content": "<@bot-1> hello", + "type": 0, + "author": {"id": "user-1", "username": "kait"}, + "attachments": [], + "mentions": [{"id": "bot-1"}], + }) + + adapter.handle_message.assert_not_awaited() + + @pytest.mark.asyncio + async def test_ignores_own_messages_other_bots_and_duplicates(self): + adapter = _make_adapter() + adapter._bot_user_id = "bot-1" + adapter._get_channel = AsyncMock(return_value={"id": "chan", "type": 1}) + adapter.handle_message = AsyncMock() + base = { + "channel_id": "chan", + "content": "hello", + "type": 0, + "attachments": [], + "mentions": [], + } + + await adapter._handle_message_create({ + **base, + "id": "own", + "author": {"id": "bot-1"}, + }) + await adapter._handle_message_create({ + **base, + "id": "other", + "author": {"id": "bot-2", "bot": True}, + }) + duplicate = {**base, "id": "dup", "author": {"id": "user"}} + await adapter._handle_message_create(duplicate) + await adapter._handle_message_create(duplicate) + + assert adapter.handle_message.await_count == 1 + + @pytest.mark.asyncio + async def test_attachment_is_cached_and_reply_context_is_preserved(self): + adapter = _make_adapter() + adapter._bot_user_id = "bot-1" + adapter._authorization_check = lambda *_args: True + adapter._get_channel = AsyncMock(return_value={"id": "chan", "type": 1}) + adapter._download_attachment = AsyncMock( + return_value=("/tmp/image.png", "image/png") + ) + adapter.handle_message = AsyncMock() + + await adapter._handle_message_create({ + "id": "m2", + "channel_id": "chan", + "content": "see this", + "type": 19, + "author": {"id": "user", "username": "kait"}, + "mentions": [], + "attachments": [ + { + "url": "https://cdn.example/x", + "filename": "x.png", + "content_type": "image/png", + } + ], + "message_reference": {"message_id": "m1", "channel_id": "chan"}, + "referenced_message": { + "content": "previous", + "author": {"id": "other", "username": "alex"}, + }, + }) + + call = adapter.handle_message.await_args + assert call is not None + event = call.args[0] + assert event.message_type == MessageType.PHOTO + assert event.media_urls == ["/tmp/image.png"] + assert event.media_types == ["image/png"] + assert event.reply_to_message_id == "m1" + assert event.reply_to_text == "previous" + assert event.reply_to_author_id == "other" + assert event.reply_to_author_name == "alex" diff --git a/tests/gateway/test_fluxer_integration.py b/tests/gateway/test_fluxer_integration.py new file mode 100644 index 0000000000000..bc4e333ce962a --- /dev/null +++ b/tests/gateway/test_fluxer_integration.py @@ -0,0 +1,157 @@ +"""Local end-to-end tests for the Fluxer REST + Gateway integration.""" + +from __future__ import annotations + +import asyncio +from contextlib import asynccontextmanager +from typing import Any, AsyncIterator, TypedDict + +import pytest +from aiohttp import web + +from gateway.config import PlatformConfig +from gateway.platforms.base import MessageEvent + + +class _StubState(TypedDict): + identify: dict[str, Any] | None + outbound: dict[str, Any] | None + authorization: list[str | None] + api_url: str + gateway_url: str + + +@asynccontextmanager +async def _fluxer_stub() -> AsyncIterator[tuple[_StubState, asyncio.Event]]: + state: _StubState = { + "identify": None, + "outbound": None, + "authorization": [], + "api_url": "", + "gateway_url": "", + } + inbound_sent = asyncio.Event() + + async def users_me(request: web.Request) -> web.Response: + state["authorization"].append(request.headers.get("Authorization")) + return web.json_response({"id": "bot-1", "username": "Hermes"}) + + async def gateway_bot(request: web.Request) -> web.Response: + state["authorization"].append(request.headers.get("Authorization")) + return web.json_response({"url": state["gateway_url"]}) + + async def channel(request: web.Request) -> web.Response: + return web.json_response({"id": request.match_info["channel_id"], "type": 1}) + + async def create_message(request: web.Request) -> web.Response: + state["authorization"].append(request.headers.get("Authorization")) + state["outbound"] = await request.json() + return web.json_response({"id": "outbound-1"}) + + async def gateway(request: web.Request) -> web.WebSocketResponse: + ws = web.WebSocketResponse() + await ws.prepare(request) + await ws.send_json({"op": 10, "d": {"heartbeat_interval": 60_000}}) + identify = await ws.receive_json(timeout=5) + state["identify"] = identify + await ws.send_json({ + "op": 0, + "s": 1, + "t": "READY", + "d": { + "session_id": "session-1", + "user": {"id": "bot-1", "username": "Hermes"}, + }, + }) + await ws.send_json({ + "op": 0, + "s": 2, + "t": "MESSAGE_CREATE", + "d": { + "id": "inbound-1", + "channel_id": "dm-1", + "content": "hello from Fluxer", + "type": 0, + "author": {"id": "user-1", "username": "Kait", "bot": False}, + "attachments": [], + "mentions": [], + }, + }) + inbound_sent.set() + async for _message in ws: + pass + return ws + + app = web.Application() + app.router.add_get("/v1/users/@me", users_me) + app.router.add_get("/v1/gateway/bot", gateway_bot) + app.router.add_get("/v1/channels/{channel_id}", channel) + app.router.add_post("/v1/channels/{channel_id}/messages", create_message) + app.router.add_get("/gateway", gateway) + + runner = web.AppRunner(app) + await runner.setup() + site = web.TCPSite(runner, "127.0.0.1", 0) + await site.start() + addresses = runner.addresses + assert addresses + port = addresses[0][1] + state["api_url"] = f"http://127.0.0.1:{port}/v1" + state["gateway_url"] = f"ws://127.0.0.1:{port}/gateway" + try: + yield state, inbound_sent + finally: + await runner.cleanup() + + +@pytest.mark.asyncio +async def test_real_rest_and_gateway_round_trip(monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.delenv("HTTPS_PROXY", raising=False) + monkeypatch.delenv("HTTP_PROXY", raising=False) + monkeypatch.delenv("ALL_PROXY", raising=False) + + async with _fluxer_stub() as (state, inbound_sent): + from plugins.platforms.fluxer.adapter import FluxerAdapter + + adapter = FluxerAdapter( + PlatformConfig( + enabled=True, + token="integration-token", + extra={"api_url": state["api_url"]}, + ) + ) + received = asyncio.Event() + events: list[MessageEvent] = [] + + async def capture(event: MessageEvent) -> None: + events.append(event) + received.set() + + setattr(adapter, "handle_message", capture) + try: + assert await adapter.connect() is True + await asyncio.wait_for(inbound_sent.wait(), timeout=5) + await asyncio.wait_for(received.wait(), timeout=5) + + result = await adapter.send("dm-1", "hello back", reply_to="inbound-1") + assert result.success is True + assert result.message_id == "outbound-1" + finally: + await adapter.disconnect() + + assert state["authorization"] == [ + "Bot integration-token", + "Bot integration-token", + "Bot integration-token", + ] + identify = state["identify"] + assert identify is not None + assert identify["op"] == 2 + assert identify["d"]["token"] == "integration-token" + assert events[0].text == "hello from Fluxer" + assert events[0].source.chat_type == "dm" + outbound = state["outbound"] + assert outbound is not None + assert outbound["content"] == "hello back" + assert outbound["message_reference"]["message_id"] == "inbound-1" diff --git a/tests/gateway/test_fluxer_plugin_setup.py b/tests/gateway/test_fluxer_plugin_setup.py new file mode 100644 index 0000000000000..3b59750dd7d72 --- /dev/null +++ b/tests/gateway/test_fluxer_plugin_setup.py @@ -0,0 +1,58 @@ +"""Tests for Fluxer's interactive gateway setup.""" + +import hermes_cli.cli_output as cli_output_mod +import hermes_cli.config as config_mod + + +def test_interactive_setup_saves_token_allowlist_and_home(monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + saved = {} + removed = [] + prompts = iter(["bot-token", "user-1,user-2", "channel-1"]) + + monkeypatch.setattr(config_mod, "get_env_value", lambda _key: "") + monkeypatch.setattr(config_mod, "save_env_value", lambda k, v: saved.update({k: v})) + monkeypatch.setattr( + config_mod, "remove_env_value", lambda key: removed.append(key) or False + ) + monkeypatch.setattr(cli_output_mod, "prompt", lambda *_a, **_kw: next(prompts)) + monkeypatch.setattr(cli_output_mod, "prompt_yes_no", lambda *_a, **_kw: False) + for name in ("print_header", "print_info", "print_success", "print_warning"): + monkeypatch.setattr(cli_output_mod, name, lambda *_a, **_kw: None) + + from plugins.platforms.fluxer.adapter import interactive_setup + + interactive_setup() + + assert saved["FLUXER_BOT_TOKEN"] == "bot-token" + assert saved["FLUXER_ALLOWED_USERS"] == "user-1,user-2" + assert saved["FLUXER_HOME_CHANNEL"] == "channel-1" + assert "FLUXER_API_URL" not in saved + assert "FLUXER_ALLOW_ALL_USERS" in removed + + +def test_interactive_setup_blank_home_clears_existing(monkeypatch, tmp_path): + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + removed = [] + prompts = iter(["bot-token", "", ""]) + + monkeypatch.setattr( + config_mod, + "get_env_value", + lambda key: "old-channel" if key == "FLUXER_HOME_CHANNEL" else "", + ) + monkeypatch.setattr(config_mod, "save_env_value", lambda *_a, **_kw: None) + monkeypatch.setattr( + config_mod, "remove_env_value", lambda key: removed.append(key) or True + ) + monkeypatch.setattr(cli_output_mod, "prompt", lambda *_a, **_kw: next(prompts)) + monkeypatch.setattr(cli_output_mod, "prompt_yes_no", lambda *_a, **_kw: False) + for name in ("print_header", "print_info", "print_success", "print_warning"): + monkeypatch.setattr(cli_output_mod, name, lambda *_a, **_kw: None) + + from plugins.platforms.fluxer.adapter import interactive_setup + + interactive_setup() + + assert "FLUXER_HOME_CHANNEL" in removed + assert "FLUXER_ALLOW_ALL_USERS" in removed diff --git a/tests/gateway/test_plugin_platform_interface.py b/tests/gateway/test_plugin_platform_interface.py index c2392cf8279c1..3302ee2a80718 100644 --- a/tests/gateway/test_plugin_platform_interface.py +++ b/tests/gateway/test_plugin_platform_interface.py @@ -14,7 +14,7 @@ import pytest -PROJECT_ROOT = Path(__file__).parent.parent.resolve() +PROJECT_ROOT = Path(__file__).resolve().parents[2] PLATFORMS_DIR = PROJECT_ROOT / "plugins" / "platforms" @@ -55,6 +55,12 @@ class _MockPluginContext: def __init__(self): self.registered_names: list[str] = [] + def register_cli_command(self, **_kwargs: Any) -> None: + """Accept optional CLI registration performed by platform plugins.""" + + def register_hook(self, *_args: Any, **_kwargs: Any) -> None: + """Accept optional lifecycle-hook registration performed by plugins.""" + def register_platform( self, *, @@ -104,6 +110,7 @@ def test_plugin_registers_valid_platform_entry(platform_name: str, clean_registr assert platform_name in ctx.registered_names from gateway.platform_registry import platform_registry + entry = platform_registry.get(platform_name) assert entry is not None, f"{platform_name} did not register an entry" assert entry.name == platform_name @@ -120,6 +127,7 @@ def test_platform_entry_has_required_fields(platform_name: str, clean_registry): module.register(ctx) from gateway.platform_registry import platform_registry + entry = platform_registry.get(platform_name) assert entry is not None @@ -139,13 +147,16 @@ def test_platform_entry_has_required_fields(platform_name: str, clean_registry): @pytest.mark.parametrize("platform_name", _PLATFORM_NAMES) -def test_adapter_factory_produces_valid_adapter(platform_name: str, clean_registry): +def test_adapter_factory_produces_valid_adapter( + platform_name: str, clean_registry, monkeypatch +): """The adapter factory must return an object with the base interface.""" module = _import_platform_module(platform_name) ctx = _MockPluginContext() module.register(ctx) from gateway.platform_registry import platform_registry + entry = platform_registry.get(platform_name) assert entry is not None @@ -158,6 +169,9 @@ def test_adapter_factory_produces_valid_adapter(platform_name: str, clean_regist mock_config.home_channel = None mock_config.reply_to_mode = "first" + for env_name in entry.required_env: + monkeypatch.setenv(env_name, "contract-test-placeholder") + adapter = entry.adapter_factory(mock_config) assert adapter is not None, f"{platform_name} adapter_factory returned None" @@ -170,6 +184,7 @@ def test_adapter_factory_produces_valid_adapter(platform_name: str, clean_regist # Should be a BasePlatformAdapter subclass if importable try: from gateway.platforms.base import BasePlatformAdapter + assert isinstance(adapter, BasePlatformAdapter) except Exception: pytest.skip("BasePlatformAdapter not available for isinstance check") @@ -183,11 +198,14 @@ def test_check_fn_returns_bool(platform_name: str, clean_registry): module.register(ctx) from gateway.platform_registry import platform_registry + entry = platform_registry.get(platform_name) assert entry is not None result = entry.check_fn() - assert isinstance(result, bool), f"{platform_name}.check_fn() returned {type(result)}, expected bool" + assert isinstance(result, bool), ( + f"{platform_name}.check_fn() returned {type(result)}, expected bool" + ) @pytest.mark.parametrize("platform_name", _PLATFORM_NAMES) @@ -198,11 +216,12 @@ def test_validate_config_if_present(platform_name: str, clean_registry): module.register(ctx) from gateway.platform_registry import platform_registry + entry = platform_registry.get(platform_name) assert entry is not None if entry.validate_config is None: - pytest.skip("No validate_config provided") + return mock_config = MagicMock() mock_config.extra = {} @@ -218,11 +237,12 @@ def test_is_connected_if_present(platform_name: str, clean_registry): module.register(ctx) from gateway.platform_registry import platform_registry + entry = platform_registry.get(platform_name) assert entry is not None if entry.is_connected is None: - pytest.skip("No is_connected provided") + return mock_config = MagicMock() mock_config.extra = {} diff --git a/website/docs/reference/environment-variables.md b/website/docs/reference/environment-variables.md index 948a118c97f8e..d2b570d26063c 100644 --- a/website/docs/reference/environment-variables.md +++ b/website/docs/reference/environment-variables.md @@ -454,6 +454,19 @@ These are set automatically by the Docker terminal backend when `proxy.enabled: | `MATTERMOST_REQUIRE_MENTION` | Require `@mention` in channels (default: `true`). Set to `false` to respond to all messages. | | `MATTERMOST_FREE_RESPONSE_CHANNELS` | Comma-separated channel IDs where bot responds without `@mention` | | `MATTERMOST_REPLY_MODE` | Reply style: `thread` (threaded replies) or `off` (flat messages, default) | +| `FLUXER_BOT_TOKEN` | Fluxer bot token from User Settings → Applications | +| `FLUXER_API_URL` | Fluxer REST API base (default: `https://api.fluxer.app/v1`) | +| `FLUXER_GATEWAY_URL` | Optional Fluxer Gateway WebSocket URL override; normally discovered from `/gateway/bot` | +| `FLUXER_ALLOWED_USERS` | Comma-separated Fluxer user IDs allowed to message the bot | +| `FLUXER_ALLOW_ALL_USERS` | Allow any Fluxer user without an allowlist (development only) | +| `FLUXER_ALLOWED_CHANNELS` | Optional comma-separated server-channel allowlist; DMs remain available | +| `FLUXER_REQUIRE_MENTION` | Require the bot to be mentioned in server channels (default: `true`) | +| `FLUXER_FREE_RESPONSE_CHANNELS` | Comma-separated server channels where a mention is not required | +| `FLUXER_HOME_CHANNEL` | Fluxer channel ID for cron and notification delivery | +| `FLUXER_HOME_CHANNEL_NAME` | Display name for the Fluxer home channel | +| `FLUXER_PROXY` | Optional HTTP/SOCKS proxy for Fluxer REST and Gateway traffic | +| `FLUXER_MAX_DOWNLOAD_BYTES` | Maximum bytes buffered for one inbound Fluxer attachment (default: 25 MiB) | +| `FLUXER_MAX_UPLOAD_BYTES` | Maximum bytes read for one outbound Fluxer upload (default: 25 MiB) | | `MATRIX_HOMESERVER` | Matrix homeserver URL (e.g. `https://matrix.org`) | | `MATRIX_ACCESS_TOKEN` | Matrix access token for bot authentication | | `MATRIX_USER_ID` | Matrix user ID (e.g. `@hermes:matrix.org`) — required for password login, optional with access token | diff --git a/website/docs/user-guide/messaging/fluxer.md b/website/docs/user-guide/messaging/fluxer.md new file mode 100644 index 0000000000000..9c700cb087422 --- /dev/null +++ b/website/docs/user-guide/messaging/fluxer.md @@ -0,0 +1,118 @@ +--- +sidebar_position: 9 +title: "Fluxer" +description: "Run Hermes Agent as a native Fluxer bot" +--- + +# Fluxer Setup + +Hermes connects directly to Fluxer's REST API and real-time Gateway. The adapter does not route through Discord or require `discord.py`; Fluxer is a separate messaging platform with its own bot token and endpoints. + +## Supported behavior + +| Context or feature | Behavior | +|---|---| +| Direct messages | Hermes responds without a mention. | +| Server channels | Hermes requires a bot mention by default. | +| Text and replies | Incoming reply context is preserved; outgoing responses can reply to the triggering message. | +| Images and files | Incoming attachments are downloaded with SSRF and size protections; outgoing files use Fluxer's multipart API. | +| Typing and streaming | Native typing indicators and progressive edits are supported. | +| Connection recovery | Gateway sessions heartbeat, reconnect with backoff, and resume when Fluxer permits it. | +| Proactive delivery | Cron jobs and notifications can use a configured home channel. | + +Fluxer voice-channel participation and message reactions are not currently implemented. Audio files and voice-message attachments are supported as ordinary media. + +## 1. Create a Fluxer bot + +1. Open Fluxer **User Settings** and select **Applications**. +2. Create an application, add a bot user, and copy its bot token. +3. Add the bot to each server where Hermes should operate. +4. Copy your Fluxer user ID and any channel IDs you want to allow or use for proactive delivery. + +:::warning +Treat the bot token as a password. Do not post it in chat, put it in `config.yaml`, or commit it to source control. Each simultaneously running Hermes profile must use a different Fluxer bot token. +::: + +## 2. Configure Hermes + +Run the guided gateway setup: + +```bash +hermes gateway setup +``` + +Select **Fluxer**, enter the bot token locally, and optionally enter an allowed-user list and home channel. The setup writes secrets to the active profile's `.env` file. + +You can also configure the active profile manually: + +```bash +FLUXER_BOT_TOKEN=*** +FLUXER_ALLOWED_USERS=123456789012345678 +FLUXER_HOME_CHANNEL=234567890123456789 +``` + +Multiple allowed users are comma-separated. `FLUXER_ALLOW_ALL_USERS=true` permits any sender and is intended only for controlled development environments. + +## 3. Start and verify the gateway + +```bash +hermes gateway restart +hermes gateway status +``` + +Send the bot a DM. In a server channel, mention the bot in the message unless that channel is configured for free response. + +## Channel controls + +```bash +# Bot responds only in these server channels; DMs are unaffected. +FLUXER_ALLOWED_CHANNELS=234567890123456789,345678901234567890 + +# These channels do not require a bot mention. +FLUXER_FREE_RESPONSE_CHANNELS=234567890123456789 + +# Disable mention gating in every allowed server channel. +FLUXER_REQUIRE_MENTION=false +``` + +The bot still applies the normal Hermes user authorization policy through `FLUXER_ALLOWED_USERS` and `FLUXER_ALLOW_ALL_USERS`. + +## Self-hosted or proxied Fluxer + +The production defaults are: + +```bash +FLUXER_API_URL=https://api.fluxer.app/v1 +``` + +Fluxer's `/gateway/bot` endpoint supplies the Gateway URL automatically. Self-hosted deployments can override either endpoint: + +```bash +FLUXER_API_URL=https://chat.example.com/api/v1 +FLUXER_GATEWAY_URL=wss://gateway.chat.example.com/ +FLUXER_MAX_UPLOAD_BYTES=26214400 +``` + +Hermes requires HTTPS/WSS for remote endpoints so the bot token cannot cross a +plaintext connection. Plain HTTP/WS is accepted only for loopback development +endpoints such as `127.0.0.1` or `localhost`. + +Use `FLUXER_PROXY` for an adapter-specific HTTP or SOCKS proxy. + +## Troubleshooting + +### Authentication failed + +Regenerate the bot token in Fluxer's Applications settings and rerun `hermes gateway setup`. Fluxer REST requests use `Authorization: Bot `; a normal user session token is not accepted as a bot credential. + +### Bot responds in DMs but not a server channel + +Verify that the bot belongs to the server, the channel is included in `FLUXER_ALLOWED_CHANNELS` when that variable is set, and the message mentions the bot unless the channel is in `FLUXER_FREE_RESPONSE_CHANNELS`. + +### Reconnect loop + +Check that HTTPS and WebSocket traffic can reach the configured API and Gateway hosts. If you set `FLUXER_GATEWAY_URL`, it must use `wss://` unless it points to loopback development. The adapter reconnects with exponential backoff and attempts Gateway resume after transient disconnects. + +### Two profiles cannot use the same bot + +This is intentional. Hermes places a machine-local lock on each Fluxer bot identity to prevent duplicate Gateway consumers. Create a distinct Fluxer application and bot token for every concurrently running profile. diff --git a/website/docs/user-guide/messaging/index.md b/website/docs/user-guide/messaging/index.md index da107c4ae613b..f4d4534d5b343 100644 --- a/website/docs/user-guide/messaging/index.md +++ b/website/docs/user-guide/messaging/index.md @@ -1,12 +1,12 @@ --- sidebar_position: 1 title: "Messaging Gateway" -description: "Chat with Hermes from Telegram, Discord, Slack, WhatsApp, Signal, SMS, Email, Home Assistant, Mattermost, Matrix, DingTalk, Yuanbao, Microsoft Teams, LINE, Raft, Webhooks, or any OpenAI-compatible frontend via the API server — architecture and setup overview" +description: "Chat with Hermes from Telegram, Discord, Fluxer, Slack, WhatsApp, Signal, SMS, Email, Home Assistant, Mattermost, Matrix, DingTalk, Yuanbao, Microsoft Teams, LINE, Raft, Webhooks, or any OpenAI-compatible frontend via the API server — architecture and setup overview" --- # Messaging Gateway -Chat with Hermes from Telegram, Discord, Slack, WhatsApp, Signal, SMS, Email, Home Assistant, Mattermost, Matrix, DingTalk, Feishu/Lark, WeCom, Weixin, BlueBubbles (iMessage), QQ, Yuanbao, Microsoft Teams, LINE, ntfy, or your browser. The gateway is a single background process that connects to all your configured platforms, handles sessions, runs cron jobs, and delivers voice messages. +Chat with Hermes from Telegram, Discord, Fluxer, Slack, WhatsApp, Signal, SMS, Email, Home Assistant, Mattermost, Matrix, DingTalk, Feishu/Lark, WeCom, Weixin, BlueBubbles (iMessage), QQ, Yuanbao, Microsoft Teams, LINE, ntfy, or your browser. The gateway is a single background process that connects to all your configured platforms, handles sessions, runs cron jobs, and delivers voice messages. For the full voice feature set — including CLI microphone mode, spoken replies in messaging, and Discord voice-channel conversations — see [Voice Mode](/user-guide/features/voice-mode) and [Use Voice Mode with Hermes](/guides/use-voice-mode-with-hermes). @@ -20,6 +20,7 @@ Bots need both a model provider and tool providers (TTS, web). A [Nous Portal](/ |----------|:-----:|:------:|:-----:|:-------:|:---------:|:------:|:---------:| | Telegram | ✅ | ✅ | ✅ | ✅ | — | ✅ | ✅ | | Discord | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | +| Fluxer | ✅ | ✅ | ✅ | — | — | ✅ | ✅ | | Slack | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | | Google Chat | — | ✅ | ✅ | ✅ | — | ✅ | — | | WhatsApp | — | ✅ | ✅ | — | — | ✅ | ✅ |