"""Matrix gateway adapter (any homeserver, via mautrix; optional E2EE with ``mautrix[encryption]``). Env vars (config.yaml ``matrix:`` keys alias several — env wins): MATRIX_HOMESERVER, MATRIX_ACCESS_TOKEN (preferred) | MATRIX_USER_ID + MATRIX_PASSWORD; MATRIX_E2EE_MODE off|optional|required (legacy MATRIX_ENCRYPTION=true => required); MATRIX_DEVICE_ID (stable E2EE device), MATRIX_RECOVERY_KEY (cross-signing after key rotation), MATRIX_RECOVERY_KEY_OUTPUT_FILE (one-time 0600 write of a bootstrapped key), MATRIX_PROXY; MATRIX_ALLOWED_USERS, MATRIX_ALLOWED_ROOMS (whitelist; DMs exempt), MATRIX_IGNORE_USER_PATTERNS (regexes for bridge ghosts), MATRIX_HOME_ROOM (cron delivery), MATRIX_REACTIONS (default true); MATRIX_REQUIRE_MENTION (default true), MATRIX_THREAD_REQUIRE_MENTION, MATRIX_FREE_RESPONSE_ROOMS, MATRIX_PROCESS_NOTICES, MATRIX_ALLOW_ROOM_MENTIONS, MATRIX_ALLOW_PUBLIC_ROOMS (all default false); MATRIX_AUTO_THREAD (default true), MATRIX_DM_AUTO_THREAD, MATRIX_DM_MENTION_THREADS, MATRIX_SESSION_SCOPE auto|room|thread; MATRIX_MAX_MESSAGE_LENGTH (default 16000), MATRIX_MAX_MEDIA_BYTES, MATRIX_ROOM_IDENTITY_TTL_SECONDS; MATRIX_APPROVAL_REQUIRE_SENDER (default true), MATRIX_APPROVAL_TIMEOUT_SECONDS (default 300). Note: any room with <=2 joined members is auto-classified as a DM (see ``_resolve_room_identity``), regardless of ``m.direct`` account data or an explicit room name — clients auto-name DMs like "Alice & Bot", so name alone can't be trusted. A DM-classified room therefore bypasses MATRIX_ALLOWED_ROOMS, MATRIX_FREE_RESPONSE_ROOMS, and MATRIX_REQUIRE_MENTION, and follows MATRIX_DM_AUTO_THREAD / MATRIX_DM_MENTION_THREADS instead of MATRIX_AUTO_THREAD / MATRIX_SESSION_SCOPE. To make a deliberately-created 2-person room behave like a regular room, add a third member so it has >2 joined members. """ from __future__ import annotations import asyncio import array import inspect from contextlib import suppress import logging import mimetypes import os import re import shutil import subprocess import sys import time from urllib.parse import urljoin, urlsplit, urlunsplit from dataclasses import dataclass, field from html import escape as _html_escape from html.parser import HTMLParser from pathlib import Path from typing import Any, Dict, Optional, Set from agent.i18n import t from agent.secret_scope import get_secret from gateway.platforms._shared import ( apply_yaml_bridge as _apply_yaml_bridge, extra_or_secret as _extra_or_secret, get_scoped_secret as _get_scoped_secret, send_error ) try: from mautrix.types import ( ContentURI, EventID, EventType, PresenceState, RoomCreatePreset, RoomID, TrustState, UserID) except ImportError: # Import-safe stubs without mautrix: check_matrix_requirements() gates production use, but # tests exercise adapter methods so the attributes must exist. ContentURI = EventID = RoomID = UserID = str # type: ignore[misc,assignment] EventType = type("_EventTypeStub", (), { # type: ignore[misc,assignment] "ROOM_MESSAGE": "m.room.message", "REACTION": "m.reaction", "ROOM_ENCRYPTED": "m.room.encrypted", "ROOM_NAME": "m.room.name"}) PresenceState = type("_PresenceStateStub", (), { # type: ignore[misc,assignment] "ONLINE": "online", "OFFLINE": "offline", "UNAVAILABLE": "unavailable"}) RoomCreatePreset = type("_RoomCreatePresetStub", (), { # type: ignore[misc,assignment] "PRIVATE": "private_chat", "PUBLIC": "public_chat", "TRUSTED_PRIVATE": "trusted_private_chat"}) TrustState = type("_TrustStateStub", (), {"UNVERIFIED": 0, "VERIFIED": 1}) # type: ignore[misc,assignment] from gateway.config import Platform, PlatformConfig from gateway.platforms.base import ( gateway_trust_env, BasePlatformAdapter, ExecApprovalPrompt, SendResult, resolve_proxy_url, proxy_kwargs_for_aiohttp, _ssrf_redirect_guard, ) from gateway.platforms.base import transcode_to_ogg_opus from gateway.platforms.event import MessageEvent, MessageType, ProcessingOutcome from gateway.platforms.helpers import ThreadParticipationTracker from plugins.platforms.matrix.voice_mention import ParkedVoices, VoiceGate, has_voice_marker, is_voice_event logger = logging.getLogger(__name__) _MATRIX_VOICE_WAVEFORM_BINS = 30 def _run_media_tool(cmd: list, *, timeout: int, text: bool = False): """Run ffmpeg/ffprobe with captured output and no stdin.""" return subprocess.run(cmd, capture_output=True, text=text, timeout=timeout, stdin=subprocess.DEVNULL) def _matrix_voice_metadata_for_file(path: Path) -> Dict[str, Any]: """Best-effort duration + MSC1767 waveform for voice bubbles; must work without ffprobe/ffmpeg.""" metadata: Dict[str, Any] = {} ffprobe = shutil.which("ffprobe") if ffprobe: try: result = _run_media_tool( [ffprobe, "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", str(path)], timeout=10, text=True) if result.returncode == 0: duration = float((result.stdout or "").strip() or 0) if duration > 0: metadata["duration"] = int(duration * 1000) except Exception: logger.debug("Matrix: failed to probe voice duration for %s", path, exc_info=True) ffmpeg = shutil.which("ffmpeg") if ffmpeg: try: result = _run_media_tool( [ffmpeg, "-v", "error", "-i", str(path), "-ac", "1", "-ar", "8000", "-f", "s16le", "-"], timeout=15) if result.returncode == 0 and result.stdout: samples = array.array("h") samples.frombytes(result.stdout) if sys.byteorder != "little": samples.byteswap() if samples: count, bins = len(samples), _MATRIX_VOICE_WAVEFORM_BINS waveform = [] for idx in range(bins): start = idx * count // bins peak = max(abs(v) for v in samples[start:max(start + 1, (idx + 1) * count // bins)]) waveform.append(min(1024, int(peak / 32767 * 1024))) metadata["waveform"] = waveform except Exception: logger.debug("Matrix: failed to build voice waveform for %s", path, exc_info=True) return metadata _MATRIX_BANG_COMMAND_RE = re.compile(r"^!([A-Za-z][A-Za-z0-9_-]*)(?=$|\s)(.*)$", re.DOTALL) def _resolve_matrix_bang_command(name: str) -> str | None: """Resolve a ``!command`` token (Matrix clients reserve ``/``) to a dispatchable token. Only known gateway/skill commands resolve, so ordinary exclamations stay chat text. Returns whichever candidate resolved — raw lowercased first, then ``_``→``-`` — never a forced canonical form: aliases pass through for the dispatcher.""" if not name: return None candidates = list(dict.fromkeys((name.lower(), name.lower().replace("_", "-")))) try: from hermes_cli.commands import is_gateway_known_command for candidate in candidates: if is_gateway_known_command(candidate): return candidate except Exception: logger.debug("Matrix: is_gateway_known_command failed for %r", name, exc_info=True) try: from agent.skill_commands import get_skill_commands skill_commands = get_skill_commands() or {} # keys are slash-prefixed ("/arxiv") for candidate in candidates: if f"/{candidate}" in skill_commands: return candidate except Exception: logger.debug("Matrix: get_skill_commands failed for %r", name, exc_info=True) return None def _normalize_matrix_bang_command(text: str) -> str: """Convert Matrix ``!command`` aliases to normal Hermes ``/command`` text.""" if not text or not text.startswith("!"): return text match = _MATRIX_BANG_COMMAND_RE.match(text) resolved = _resolve_matrix_bang_command(match.group(1)) if match else None if resolved is None: return text return f"/{resolved}{match.group(2) or ''}" # Reply fallback prefix: "> <@alice:example.org> quoted\n> more\n\nactual reply". _MATRIX_REPLY_FALLBACK_PILL_RE = re.compile(r"^>\s*<(@[^>]+)>\s*(.*)$") def _extract_reply_fallback(body: str) -> tuple[Optional[str], Optional[str]]: """Return (quoted_text, author_mxid) from the inline reply fallback; author from the first-line pill.""" if not body or not body.startswith("> "): return None, None quoted_lines: list[str] = [] author_id: Optional[str] = None for line in body.split("\n"): if not line.startswith("> "): break content = line[2:] if author_id is None: pill_match = _MATRIX_REPLY_FALLBACK_PILL_RE.match(line) if pill_match: author_id = pill_match.group(1) content = pill_match.group(2) # drop the pill from the visible quote quoted_lines.append(content) quoted_text = "\n".join(quoted_lines).strip() or None return quoted_text, author_id def _strip_reply_fallback(body: str) -> str: """Strip the inline ``> quote\\n\\nreply`` fallback prefix; unchanged if absent.""" if not body or not body.startswith("> "): return body stripped = [] past_fallback = False for line in body.split("\n"): if not past_fallback: if line.startswith("> ") or line == ">": continue past_fallback = True if line == "": continue stripped.append(line) return "\n".join(stripped) if stripped else body # Auth errcodes that genuinely require re-authentication (never retried). _MATRIX_PERMANENT_ERRCODES = frozenset({ "m_unknown_token", "m_missing_token", "m_forbidden", }) def _is_permanent_matrix_auth_error(exc: BaseException) -> bool: """Return True only for genuine auth failures that must stop the sync loop. A transient homeserver outage surfaces as a 5xx whose body may be an HTML error page (Umbrel's app-proxy returns one). Naive substring checks like ``"403" in str(exc)`` false-positive on digits embedded in that HTML (an SVG path coordinate such as ``1403.2`` contains ``403``) or in the ``since`` token echoed by a timeout message, which stopped the sync loop permanently on a passing blip. mautrix raises ``MatrixRequestError`` with ``errcode`` and ``http_status`` for every non-2xx, so classify on those alone; anything without a structured auth signal (timeouts, dropped connections, 5xx) is retried. Deliberately not ``.status``/``.status_code``/``.code``: those belong to unrelated exception shapes (aiohttp responses, OS errno) and can misclassify on a coincidental integer. """ errcode = getattr(exc, "errcode", None) if isinstance(errcode, str) and errcode.strip().lower() in _MATRIX_PERMANENT_ERRCODES: return True status = getattr(exc, "http_status", None) return isinstance(status, int) and status in (401, 403) def _split_reply_fallback(body: str) -> tuple[str, str]: """Split ``> quote\\n\\nreply`` into ``(quote_block, reply_text)``; ``("", body)`` when absent. The two halves always concatenate back to *body* verbatim (``quote + reply == body``), so callers can transform one half and rebuild the body without disturbing the other. The blank separator line belongs to the quote block. Used to keep the ``> <@user:srv>`` reply pill — the only mention text in a reply-to-the-bot — out of whole-body rewrites. """ if not body or not body.startswith("> "): return "", body lines = body.split("\n") idx = 0 while idx < len(lines) and (lines[idx].startswith("> ") or lines[idx] == ">"): idx += 1 if idx < len(lines) and lines[idx] == "": idx += 1 # the blank line separating the quote from the reply belongs to the quote head = "\n".join(lines[:idx]) return (head, "") if idx >= len(lines) else (head + "\n", "\n".join(lines[idx:])) class _MatrixHtmlSanitizer(HTMLParser): """Allowlist sanitizer for Matrix-compatible formatted HTML.""" _ALLOWED_TAGS = { "a", "b", "blockquote", "br", "code", "del", "em", "h1", "h2", "h3", "h4", "h5", "h6", "hr", "i", "li", "ol", "p", "pre", "s", "strike", "strong", "table", "tbody", "td", "th", "thead", "tr", "ul"} _VOID_TAGS = {"br", "hr"} def __init__(self) -> None: super().__init__(convert_charrefs=False) self._parts: list[str] = [] self._skip_depth = 0 @staticmethod def _safe_url(value: str) -> str: stripped = re.sub(r"[\x00-\x1f\x7f]+", "", value or "").strip() match = re.match(r"^([A-Za-z][A-Za-z0-9+.-]*):", stripped) scheme = match.group(1).lower() if match else "" if scheme and scheme not in {"http", "https", "matrix", "mailto"}: return "" return stripped def _safe_attrs(self, tag: str, attrs: list[tuple[str, str | None]]) -> str: safe: list[str] = [] for key, value in attrs: attr = str(key or "").lower() raw_value = "" if value is None else str(value) if tag == "a" and attr == "href": href = self._safe_url(raw_value) if href: safe.append(f' href="{_html_escape(href, quote=True)}"') elif tag == "code" and attr == "class" and re.fullmatch(r"language-[A-Za-z0-9_+.-]{1,64}", raw_value): safe.append(f' class="{_html_escape(raw_value, quote=True)}"') return "".join(safe) def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: tag = tag.lower() if tag in {"script", "style"}: self._skip_depth += 1 elif not self._skip_depth and tag in self._ALLOWED_TAGS: self._parts.append(f"<{tag}>" if tag in self._VOID_TAGS else f"<{tag}{self._safe_attrs(tag, attrs)}>") def handle_endtag(self, tag: str) -> None: tag = tag.lower() if tag in {"script", "style"} and self._skip_depth: self._skip_depth -= 1 return if self._skip_depth or tag not in self._ALLOWED_TAGS or tag in self._VOID_TAGS: return self._parts.append(f"") def _emit(self, text: str) -> None: if not self._skip_depth: self._parts.append(text) def handle_data(self, data: str) -> None: self._emit(_html_escape(data)) def handle_entityref(self, name: str) -> None: self._emit(f"&{name};") def handle_charref(self, name: str) -> None: self._emit(f"&#{name};") def get_html(self) -> str: return "".join(self._parts) @dataclass(frozen=True) class MatrixRoomIdentity: """Resolved Matrix room identity for routing and prompt context.""" room_id: str room_name: str | None room_topic: str | None canonical_alias: str | None server_name: str | None joined_member_count: int | None is_direct_account_data: bool display_name: str has_explicit_name: bool chat_type: str conflict: bool = False @dataclass class _MatrixApprovalPrompt: """Pending reaction-based exec approval prompt.""" session_key: str chat_id: str message_id: str resolved: bool = False requester_user_id: str | None = None expires_at: float | None = None bot_reaction_events: dict[str, str] = field(default_factory=dict, init=False) # emoji -> event_id @dataclass class _MatrixPickerPrompt: """Pending reaction-based picker; ``choices`` maps emoji -> selection, ``on_selected`` is the callback.""" chat_id: str message_id: str session_key: str choices: dict on_selected: Any requester_user_id: str | None = None expires_at: float | None = None resolved: bool = False bot_reaction_events: dict[str, str] = field(default_factory=dict) _MatrixModelPickerPrompt = _MatrixChoicePickerPrompt = _MatrixPickerPrompt # Spec allows ~65 KB events; 4000 was too small (split Markdown tables mid-row). # Matrix message size limit. The spec allows large events (~65 KB), but very large bodies can render poorly # in some clients. The previous 4,000-char default was overly conservative and split Markdown tables mid-row # (#53026). DEFAULT_MAX_MESSAGE_LENGTH = 16000 MATRIX_MAX_MESSAGE_LENGTH_CEILING = 65535 def _resolve_max_message_length(config) -> int: """Resolve outbound chunk size from config, env, or plugin registry.""" raw = _extra_or_secret(getattr(config, "extra", None), "max_message_length", "MATRIX_MAX_MESSAGE_LENGTH", None) if raw is None or not str(raw).strip(): with suppress(Exception): from gateway.platform_registry import platform_registry entry = platform_registry.get("matrix") if entry and entry.max_message_length: raw = entry.max_message_length try: value = int(raw) except (TypeError, ValueError): return DEFAULT_MAX_MESSAGE_LENGTH return max(500, min(value, MATRIX_MAX_MESSAGE_LENGTH_CEILING)) # E2EE store dir is resolved per adapter in connect() (``_resolve_store_dir``), NOT at module scope: # the multiplex gateway imports this once and a module constant would collide every profile's Olm # identity in one crypto.db. # Store directory for E2EE keys and sync state. Mirrors the pairing-store fix (a6397c379). See #89168. from hermes_constants import get_hermes_dir as _get_hermes_dir _STARTUP_GRACE_SECONDS = 5 # ignore messages older than this many seconds before startup _OUTBOUND_MENTION_RE = re.compile(r"(? bool: """True when an m.image body is just the uploaded filename (no caption) — not user text.""" return _looks_like_transport_filename(text, "image/", _MATRIX_IMAGE_FILENAME_EXTS) def _looks_like_transport_filename(text: str, mime_prefixes, exts: frozenset, reject_spaces: bool = False) -> bool: """Bare single-token filename with a known media extension or a matching guessed MIME type.""" candidate = str(text or "").strip() if not candidate or "\n" in candidate or candidate.endswith("/"): return False # A genuine caption essentially always contains whitespace; a bare transport filename does not. if reject_spaces and any(ch.isspace() for ch in candidate): return False if Path(candidate).name != candidate: return False suffix = Path(candidate).suffix.lower() if not suffix: return False guessed_type, _ = mimetypes.guess_type(candidate) return bool(guessed_type and guessed_type.startswith(mime_prefixes)) or suffix in exts def _looks_like_matrix_media_filename(text: str) -> bool: """True when an m.audio/m.file/m.video body is just the uploaded filename (no caption).""" return _looks_like_transport_filename(text, ("audio/", "video/"), _MATRIX_MEDIA_FILENAME_EXTS, True) def _is_bare_media_filename(msgtype: str, body: str) -> bool: """True when a media event body is only the uploaded filename for its msgtype.""" if msgtype == "m.image": return _looks_like_matrix_image_filename(body) return msgtype in ("m.audio", "m.file", "m.video") and _looks_like_matrix_media_filename(body) def _matrix_event_timestamp_seconds(event: Any) -> float: """Return a Matrix event timestamp in seconds, accepting ms or sec values.""" try: ts = float(getattr(event, "timestamp", None) or getattr(event, "server_timestamp", None) or 0) except (TypeError, ValueError): return 0.0 # origin_server_ts is ms; some SDK objects/fakes expose seconds — keep both sane. return ts / 1000.0 if ts > 10_000_000_000 else ts def _create_matrix_session(proxy_url: str | None): """ClientSession whose proxy applies to *all* requests: mautrix's ``HTTPAPI._send()`` never forwards per-request ``proxy=``, so it must be session-level (``proxy=`` for HTTP(S), ``ProxyConnector`` for SOCKS); with no proxy, ``trust_env`` honours HTTP(S)_PROXY.""" import aiohttp if not proxy_url: return aiohttp.ClientSession(trust_env=gateway_trust_env()) if proxy_url.split("://")[0].lower().startswith("socks"): try: from aiohttp_socks import ProxyConnector return aiohttp.ClientSession(connector=ProxyConnector.from_url(proxy_url, rdns=True)) except ImportError: logger.warning( "aiohttp_socks not installed — SOCKS proxy %s ignored. Run: pip install aiohttp-socks", proxy_url) return aiohttp.ClientSession(trust_env=gateway_trust_env()) return aiohttp.ClientSession(proxy=proxy_url) def _check_e2ee_deps() -> bool: """True if all four E2EE deps import: olm, PgCryptoStore (also drives sqlite), asyncpg, aiosqlite. Without all four, encrypted rooms fail at connect with ``No module named 'asyncpg'``. Verifies python-olm (via mautrix.crypto.OlmMachine), the SQLite crypto store backend (mautrix.crypto.store.asyncpg.PgCryptoStore — yes, the PgCryptoStore class also drives the sqlite backend in mautrix 0.21), and the database drivers actually used at connect time (``asyncpg`` for the underlying upgrade_table machinery, ``aiosqlite`` for the ``sqlite:///`` URL we pass to ``Database.create``). See #31116. """ try: from mautrix.crypto import OlmMachine # noqa: F401 from mautrix.crypto.store.asyncpg import PgCryptoStore # noqa: F401 import asyncpg # noqa: F401 import aiosqlite # noqa: F401 return True except (ImportError, AttributeError): return False def _normalize_e2ee_mode(value: Any) -> str: raw = str(value or "").strip().lower() if raw in ("required", "require", "true", "1", "yes", "on"): return "required" if raw in ("optional", "prefer", "preferred"): return "optional" return "off" def _resolve_e2ee_mode(extra: Optional[Dict[str, Any]] = None) -> str: """Resolve E2EE mode with MATRIX_ENCRYPTION backwards compatibility.""" extra = extra or {} explicit = extra.get("e2ee_mode") or _get_scoped_secret("MATRIX_E2EE_MODE", "") if explicit: return _normalize_e2ee_mode(explicit) legacy_enabled = extra.get("encryption", _env_truthy("MATRIX_ENCRYPTION")) return "required" if legacy_enabled else "off" def _env_truthy(name: str, default: str = "") -> bool: """Return True when the env var is one of true/1/yes (case-insensitive).""" return str(_get_scoped_secret(name, default)).lower() in ("true", "1", "yes") def _env_number(name: str, default, cast): """Parse a numeric env var, falling back to *default* on ValueError.""" try: return cast(_get_scoped_secret(name, str(default))) except ValueError: return default def _csv_set(raw: Any) -> Set[str]: """Normalize a comma-separated string or list into a set of stripped tokens.""" if isinstance(raw, list): return {str(r).strip() for r in raw if str(r).strip()} return {r.strip() for r in str(raw).split(",") if r.strip()} def _thread_root(relates_to: dict) -> Optional[str]: """The m.thread root event_id an event belongs to, else None.""" return relates_to.get("event_id") if relates_to.get("rel_type") == "m.thread" else None def _extra_csv_set(config, key: str, env_name: str) -> Set[str]: """Resolve a room/user list: scoped env var → config.extra[key] → empty.""" return _csv_set(_extra_or_secret(config.extra, key, env_name, "", blank_is_unset=False)) def _recovery_key_output_path() -> Optional[Path]: """MATRIX_RECOVERY_KEY_OUTPUT_FILE via the profile-scoped reader: a bare os.getenv under multiplex resolves the default profile's path, writing/finding the wrong profile's file.""" output_file = _get_scoped_secret("MATRIX_RECOVERY_KEY_OUTPUT_FILE", "").strip() return Path(output_file).expanduser() if output_file else None def _write_matrix_recovery_key_output_file(recovery_key: str) -> Optional[Path]: """Write a generated recovery key to MATRIX_RECOVERY_KEY_OUTPUT_FILE (0600, never overwritten).""" path = _recovery_key_output_path() if path is None: return None path.parent.mkdir(parents=True, exist_ok=True) fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) try: with os.fdopen(fd, "w", encoding="utf-8") as fh: fh.write(recovery_key) fh.write("\n") except Exception: with suppress(OSError): os.close(fd) raise return path def _get_matrix_recovery_key_output_target() -> tuple[Optional[Path], str]: """Return a usable one-time recovery-key output path, or a redacted reason.""" path = _recovery_key_output_path() if path is None: return None, "not_configured" if path.exists(): return None, "exists" try: path.parent.mkdir(parents=True, exist_ok=True) except Exception as exc: return None, f"unusable: {exc}" return path, "" def _handle_generated_matrix_recovery_key(mxid: str, recovery_key: str) -> None: """Handle a freshly generated Matrix recovery key without logging it.""" try: output_path = _write_matrix_recovery_key_output_file(recovery_key) except FileExistsError: logger.warning( "Matrix: bootstrapped cross-signing for %s. Recovery key output file " "already exists; refusing to overwrite. Store the generated key " "securely and set MATRIX_RECOVERY_KEY for future restarts.", mxid) return except Exception as exc: logger.warning( "Matrix: bootstrapped cross-signing for %s, but failed to write " "MATRIX_RECOVERY_KEY_OUTPUT_FILE: %s. Store the generated key " "securely and set MATRIX_RECOVERY_KEY for future restarts.", mxid, exc) return if output_path: logger.warning( "Matrix: bootstrapped cross-signing for %s. A new recovery key was written to %s with mode 0600. Move it " "to your secret store and set MATRIX_RECOVERY_KEY for future restarts.", mxid, output_path) else: logger.warning( "Matrix: bootstrapped cross-signing for %s. A new recovery key was generated but will " "not be logged. Set MATRIX_RECOVERY_KEY_OUTPUT_FILE to write it once with mode 0600, " "or configure MATRIX_RECOVERY_KEY from your Matrix client before future restarts.", mxid) def _scoped_recovery_key() -> str: """MATRIX_RECOVERY_KEY via the profile-scoped reader: a bare os.getenv under multiplex resolves the default profile's key and verification fails with "Key MAC does not match".""" return _get_scoped_secret("MATRIX_RECOVERY_KEY", "").strip() # --- LaTeX math ($...$, $$...$$) -> Element data-mx-maths markup --- # Element (feature_latex_maths) typesets at display time. # Our sanitizer allowlists tags/attrs, so data-mx-maths cannot pass through HTML # sanitization directly. Instead, math is swapped for opaque sentinel tokens before # Markdown conversion (protecting TeX from escaping) and expanded back to math # markup after sanitization. Tokens are plain printable text with no special # HTML/Markdown meaning, so both the Markdown converter and the sanitizer # pass them through verbatim. _TEX_TOKEN_RE = re.compile(r"HERMESTEX(?:DISPLAY|INLINE)(\d+)HERMESTEXEND") _TEX_DISPLAY_TOKEN = "HERMESTEXDISPLAY%dHERMESTEXEND" _TEX_INLINE_TOKEN = "HERMESTEXINLINE%dHERMESTEXEND" def _latex_to_tokens(text: str) -> tuple[str, list[tuple[str, str]]]: """Replace ``$$...$$``/``$...$`` with sentinel tokens. Returns the tokenized text plus an ordered ``(tag, tex)`` store, where tag is ``div`` for display math and ``span`` for inline math. Dollars that do not form a pair (prices, literals) are left untouched. """ if not text or "$" not in text: return text, [] store: list[tuple[str, str]] = [] def _sub_display(match: re.Match[str]) -> str: store.append(("div", match.group(1).strip())) return _TEX_DISPLAY_TOKEN % (len(store) - 1) def _sub_inline(match: re.Match[str]) -> str: store.append(("span", match.group(1).strip())) return _TEX_INLINE_TOKEN % (len(store) - 1) text = re.sub(r"\$\$([^\n$]+?)\$\$", _sub_display, text) text = re.sub(r"(? str: """Expand sentinel tokens into ``data-mx-maths`` markup (TeX HTML-escaped).""" def _expand(match: re.Match[str]) -> str: idx = int(match.group(1)) if idx >= len(store): # Not one of our tokens (user-typed text that collides with the # sentinel format) — leave it verbatim. return match.group(0) tag, tex = store[idx] escaped = _html_escape(tex, quote=True) return f'<{tag} data-mx-maths="{escaped}">{escaped}' return _TEX_TOKEN_RE.sub(_expand, html) def _sanitize_matrix_html(html: str) -> str: sanitizer = _MatrixHtmlSanitizer() try: sanitizer.feed(html or "") sanitizer.close() return sanitizer.get_html() except Exception: return _html_escape(html or "") def _redact_url_for_log(url: str) -> str: """Strip query/fragment from URLs before logging signed media links.""" try: parts = urlsplit(str(url)) if not parts.scheme and not parts.netloc: return str(url).split("?", 1)[0].split("#", 1)[0] return urlunsplit((parts.scheme, parts.netloc, parts.path, "", "")) except Exception: return "" def _pre_sanitize_matrix_markdown(text: str) -> str: """Remove unsafe raw HTML before Markdown conversion can escape it.""" result = re.sub(r"(?is)<\s*(script|style)\b[^>]*>.*?<\s*/\s*\1\s*>", "", text or "") result = re.sub(r"""(?is)\s+on[a-z0-9_-]+\s*=\s*("[^"]*"|'[^']*'|[^\s>]+)""", "", result) return re.sub( r"""(?is)\s+(href|src)\s*=\s*("[^"]*(?:javascript|data|vbscript):[^"]*"|'[^']*(?:javascript|data|vbscript):[^']*'|[^\s>]*(?:javascript|data|vbscript):[^\s>]*)""", "", result) def matrix_deps_present() -> bool: """PASSIVE registry ``check_fn`` — must never install; ``ensure_matrix_deps`` is the installer. Registry ``check_fn`` — called from status displays and config loading, so it must never install anything. The ACTIVE lazy-installer (``check_matrix_requirements``) is registered as ``ensure_deps_fn`` and runs from ``create_adapter()`` when this returns False (#79812). """ try: from pm import available as is_available return is_available("matrix") except Exception: # pragma: no cover — defensive return False def check_matrix_requirements() -> bool: """Credentials + deps answer for setup/status callers (credentials must NOT gate the installer).""" token = _get_scoped_secret("MATRIX_ACCESS_TOKEN", "").strip() password = _get_scoped_secret("MATRIX_PASSWORD", "").strip() homeserver = _get_scoped_secret("MATRIX_HOMESERVER", "").strip() if not token and not password: logger.debug("Matrix: neither MATRIX_ACCESS_TOKEN nor MATRIX_PASSWORD set") return False if not homeserver: logger.warning("Matrix: MATRIX_HOMESERVER not set") return False return ensure_matrix_deps() def ensure_matrix_deps() -> bool: """ACTIVE deps-only installer (registry ``ensure_deps_fn``); rebinds the type globals. Installs the whole ``platform.matrix`` group when ANY declared package is missing — short-circuiting on ``import mautrix`` left asyncpg/aiosqlite uninstalled forever. Lazy-installs the full ``platform.matrix`` feature group via ``pm.extras.ensure_and_bind`` whenever any of the declared packages (mautrix, Markdown, aiosqlite, asyncpg, aiohttp-socks) is missing — not just mautrix itself. Previously this short-circuited on ``import mautrix``, which left the other four packages uninstalled forever and broke E2EE connect with ``No module named 'asyncpg'`` (#31116). Rebinds module-level type globals on success. """ from pm import extras def _import(): from mautrix.types import ( ContentURI, EventID, EventType, PresenceState, RoomCreatePreset, RoomID, TrustState, UserID) return { "ContentURI": ContentURI, "EventID": EventID, "EventType": EventType, "PresenceState": PresenceState, "RoomCreatePreset": RoomCreatePreset, "RoomID": RoomID, "TrustState": TrustState, "UserID": UserID, } # A complete install (module-level imports already bound the types) needs no sync; only a # partial one goes through ensure_and_bind, which rebinds after the install. if extras.missing("matrix") and not extras.ensure_and_bind("matrix", _import, globals()): logger.warning( "Matrix: required packages not installed or need a restart. " "Run `hermes pm install`, then restart Hermes." ) return False e2ee_mode = _resolve_e2ee_mode() if e2ee_mode == "required" and not _check_e2ee_deps(): logger.error( "Matrix: E2EE is required but dependencies are missing. %s. Without this, encrypted " "rooms will not work. Set MATRIX_E2EE_MODE=off to disable E2EE.", _E2EE_INSTALL_HINT) return False if e2ee_mode == "optional" and not _check_e2ee_deps(): logger.warning("Matrix: E2EE optional but dependencies are missing. %s", _E2EE_INSTALL_HINT) return True class _CryptoStateStore: """StateStore shim for OlmMachine (MemoryStateStore lacks is_encrypted/get_encryption_info/ find_shared_rooms); falls back to a homeserver state query when the store has no info.""" def __init__(self, client_state_store: Any, joined_rooms: set, client=None): self._ss = client_state_store self._joined_rooms = joined_rooms self._client = client # MemoryStateStore has no set_encryption_info, so cache homeserver answers here. self._enc_info_cache: dict = {} async def is_encrypted(self, room_id: str) -> bool: return (await self.get_encryption_info(room_id)) is not None async def get_encryption_info(self, room_id: str): info = await self._ss.get_encryption_info(room_id) if hasattr(self._ss, "get_encryption_info") else None if info is not None: return info if room_id in self._enc_info_cache: return self._enc_info_cache[room_id] if self._client is None: return None try: from mautrix.types import EventType as _ET, RoomEncryptionStateEventContent as _Enc, RoomID as _RID raw = await self._client.get_state_event(_RID(room_id), _ET.ROOM_ENCRYPTION) except Exception as exc: logger.debug("Matrix: homeserver encryption-info query failed for %s: %s", room_id, exc) return None if not raw: return None content = raw if isinstance(raw, _Enc) else _Enc.deserialize( raw.serialize() if hasattr(raw, "serialize") else raw) if hasattr(self._ss, "set_encryption_info"): with suppress(Exception): await self._ss.set_encryption_info(_RID(room_id), content) self._enc_info_cache[room_id] = content return content async def find_shared_rooms(self, user_id: str) -> list: return list(self._joined_rooms) # all joined rooms: correct for a single-user bot class MatrixAdapter(BasePlatformAdapter): """Gateway adapter for Matrix (any homeserver).""" supports_code_blocks = True # Matrix renders fenced code blocks (HTML/markdown) splits_long_messages = True # send() chunks via truncate_message(max_message_length) typed_command_prefix = "!" # clients reserve typed "/" for local commands; "!command" always reaches Hermes # Class-level defaults keep object.__new__-built test instances working. max_message_length = DEFAULT_MAX_MESSAGE_LENGTH _SPLIT_THRESHOLD = DEFAULT_MAX_MESSAGE_LENGTH - 100 def _resolve_store_dir(self) -> Path: """Pin the crypto-store dir to the active profile (connect() runs inside the profile scope); cached so later out-of-scope reads report the store actually in use.""" self._store_dir = _get_hermes_dir("platforms/matrix/store", "matrix/store") return self._store_dir @property def _crypto_db_path(self) -> Path: return (self._store_dir or _get_hermes_dir("platforms/matrix/store", "matrix/store")) / "crypto.db" def __init__(self, config: PlatformConfig): super().__init__(config, Platform.MATRIX) self.max_message_length = _resolve_max_message_length(config) self.MAX_MESSAGE_LENGTH = self.max_message_length # mirrors other adapters for tooling # A chunk near the outbound limit almost certainly has a continuation. self._SPLIT_THRESHOLD = max(100, self.max_message_length - 100) # Homeserver/user_id/device_id go through the same scoped reader as the token/password: # under multiplex os.environ holds the DEFAULT profile's identity, and pairing it with a # secondary's credential sends that credential to the wrong homeserver (or reuses the # default's E2EE device id). self._homeserver: str = (config.extra.get("homeserver", "") or _get_scoped_secret("MATRIX_HOMESERVER", "").strip()).rstrip("/") self._access_token: str = config.token or _get_scoped_secret("MATRIX_ACCESS_TOKEN", "").strip() self._user_id: str = config.extra.get("user_id", "") or _get_scoped_secret("MATRIX_USER_ID", "").strip() self._password: str = config.extra.get("password", "") or _get_scoped_secret("MATRIX_PASSWORD", "").strip() self._e2ee_mode: str = _resolve_e2ee_mode(config.extra) self._encryption: bool = self._e2ee_mode != "off" self._device_id: str = config.extra.get("device_id", "") or _get_scoped_secret("MATRIX_DEVICE_ID", "").strip() self._device_id_unverified: bool = False self._client: Any = None # mautrix.client.Client self._crypto_db: Any = None # mautrix.util.async_db.Database self._store_dir: Optional[Path] = None # pinned per profile in connect() self._sync_task: Optional[asyncio.Task] = None self._invite_join_tasks: Dict[str, asyncio.Task] = {} self._closing = False self._startup_ts: float = 0.0 self._reset_clock_skew_detector() self._last_sync_ts: float = 0.0 self._dm_rooms: Dict[str, bool] = {} self._room_identities: Dict[str, MatrixRoomIdentity] = {} self._room_identity_cached_at: Dict[str, float] = {} self._room_identity_ttl_seconds = _env_number("MATRIX_ROOM_IDENTITY_TTL_SECONDS", 60.0, float) self._room_identity_cache_max = 256 self._joined_rooms: Set[str] = set() from collections import deque self._processed_events: deque = deque(maxlen=1000) # event dedup, newest kept self._processed_events_set: set = set() self._threads = ThreadParticipationTracker("matrix") # require_mention bypass self._parked_voices = ParkedVoices() # unmentioned voice awaiting a bare @mention self._require_mention: bool = self._parse_require_mention(config) self._thread_require_mention: bool = self._parse_thread_require_mention(config) self._free_rooms: Set[str] = _extra_csv_set(config, "free_response_rooms", "MATRIX_FREE_RESPONSE_ROOMS") # If non-empty, bot ONLY responds in these rooms (whitelist); DMs exempt. self._allowed_rooms: Set[str] = _extra_csv_set(config, "allowed_rooms", "MATRIX_ALLOWED_ROOMS") self._allow_room_mentions: bool = _env_truthy("MATRIX_ALLOW_ROOM_MENTIONS", "false") # Extra-first: the YAML bridge seeds these into extra and skips the env write under a # multiplexed secondary scope, where os.environ holds the DEFAULT profile's flags. self._auto_thread: bool = self._extra_truthy(config, "auto_thread", "MATRIX_AUTO_THREAD", "true") self._dm_auto_thread: bool = _env_truthy("MATRIX_DM_AUTO_THREAD", "false") self._dm_mention_threads: bool = self._extra_truthy(config, "dm_mention_threads", "MATRIX_DM_MENTION_THREADS", "false") raw_session_scope = str(_extra_or_secret(config.extra, "session_scope", "MATRIX_SESSION_SCOPE", "auto")).strip().lower() self._matrix_session_scope = raw_session_scope if raw_session_scope in {"auto", "room", "thread"} else "auto" self._process_notices: bool = self._extra_truthy(config, "process_notices", "MATRIX_PROCESS_NOTICES", "false") self._reactions_enabled: bool = str(_extra_or_secret(config.extra, "reactions", "MATRIX_REACTIONS", "true")).lower() not in {"false", "0", "no"} self._pending_reactions: dict[tuple[str, str], str] = {} # Let the final message land before redacting reactions ("missing event" in some # clients). 5s is empirically safe; if it must be tunable, use config.yaml not env. self._reaction_redaction_delay_seconds = 5.0 self._reaction_redaction_tasks: Set[asyncio.Task] = set() self._proxy_url: str | None = resolve_proxy_url(platform_env_var="MATRIX_PROXY") if self._proxy_url: logger.info("Matrix: proxy configured — %s", self._proxy_url) self._max_media_bytes = _env_number("MATRIX_MAX_MEDIA_BYTES", 100 * 1024 * 1024, int) # Text batching merges client-side splits (~4000 chars) of one long message. self._text_batch_delay_seconds = float(os.getenv("HERMES_MATRIX_TEXT_BATCH_DELAY_SECONDS", "0.6")) self._text_batch_split_delay_seconds = float(os.getenv("HERMES_MATRIX_TEXT_BATCH_SPLIT_DELAY_SECONDS", "2.0")) self._approval_reaction_map = { "✅": "once", "🌀": "session", "♾️": "always", "♾": "always", "\u267e\ufe0f": "always", "\u267e": "always", "❌": "deny", "❎": "deny"} self._approval_prompts_by_event: Dict[str, _MatrixApprovalPrompt] = {} self._approval_prompt_by_session: Dict[str, str] = {} self._approval_require_sender: bool = _env_truthy("MATRIX_APPROVAL_REQUIRE_SENDER", "true") self._approval_timeout_seconds = _env_number("MATRIX_APPROVAL_TIMEOUT_SECONDS", 300, int) self._model_picker_prompts_by_event: Dict[str, _MatrixPickerPrompt] = {} self._choice_picker_prompts_by_event: Dict[str, _MatrixPickerPrompt] = {} # Authz lists: scoped env → this profile's YAML (``allowed_users`` / ``ignore_user_patterns``, # seeded by the bridge) → empty. Under multiplex os.environ is the DEFAULT profile's allowlist, # which must not decide who approves tool calls on a secondary bot. self._allowed_user_ids: Set[str] = _extra_csv_set(config, "allowed_users", "MATRIX_ALLOWED_USERS") self._allowed_room_ids: Set[str] = set(self._allowed_rooms) self._ignored_user_patterns: list[re.Pattern[str]] = [] for pattern in _csv_set(_extra_or_secret(config.extra, "ignore_user_patterns", "MATRIX_IGNORE_USER_PATTERNS", "")): try: self._ignored_user_patterns.append(re.compile(pattern)) except re.error as exc: logger.warning("Matrix: ignoring invalid MATRIX_IGNORE_USER_PATTERNS entry %r: %s", pattern, exc) def _is_duplicate_event(self, event_id) -> bool: """Return True if this event was already processed. Tracks the ID otherwise.""" if not event_id: return False if event_id in self._processed_events_set: return True if len(self._processed_events) == self._processed_events.maxlen: self._processed_events_set.discard(self._processed_events[0]) self._processed_events.append(event_id) self._processed_events_set.add(event_id) return False @staticmethod def _extra_truthy(config, key: str, env_name: str, default: str) -> bool: """Scoped env var → ``config.extra[key]`` (YAML, per profile) → ``default``; true/1/yes semantics.""" configured = _extra_or_secret(config.extra, key, env_name, default) return configured if isinstance(configured, bool) else str(configured).lower() in ("true", "1", "yes") @staticmethod def _configured_bool(config, key: str) -> Optional[bool]: """Parse a YAML bool / "true"/"off"-style string from config.extra; None if unset.""" configured = config.extra.get(key) if configured is None: return None if isinstance(configured, bool): return configured if isinstance(configured, str): return configured.lower() not in {"false", "0", "no", "off"} return bool(configured) @staticmethod def _parse_require_mention(config) -> bool: """MATRIX_REQUIRE_MENTION (scoped) → ``require_mention`` in config.extra → true.""" configured = _extra_or_secret(config.extra, "require_mention", "MATRIX_REQUIRE_MENTION", True) return configured if isinstance(configured, bool) else str(configured).lower() not in {"false", "0", "no", "off"} @staticmethod def _parse_thread_require_mention(config) -> bool: """MATRIX_THREAD_REQUIRE_MENTION (scoped) → ``thread_require_mention`` in config.extra → false.""" configured = _extra_or_secret(config.extra, "thread_require_mention", "MATRIX_THREAD_REQUIRE_MENTION", False) return configured if isinstance(configured, bool) else str(configured).lower() not in {"false", "0", "no", "off"} @staticmethod def _extract_server_ed25519(device_keys_obj: Any) -> Optional[str]: for kid, kval in (getattr(device_keys_obj, "keys", {}) or {}).items(): if str(kid).startswith("ed25519:"): return str(kval) return None @staticmethod async def _query_own_device_keys(client: Any): """query_keys for our own device; the DeviceKeys entry or None.""" resp = await client.query_keys({client.mxid: [client.device_id]}) our_user_devices = (getattr(resp, "device_keys", {}) or {}).get(str(client.mxid)) or {} return our_user_devices.get(str(client.device_id)) async def _reverify_keys_after_upload(self, client: Any, local_ed25519: str) -> bool: """Re-query the server after share_keys() and verify our ed25519 key matches.""" if not client.device_id or self._device_id_unverified: logger.warning("Matrix: skipping post-upload key verification — device_id not yet established") return True try: dev = await self._query_own_device_keys(client) if dev and self._extract_server_ed25519(dev) != local_ed25519: logger.error( "Matrix: device %s has immutable identity keys that don't match this " "installation. Generate a new access token with a fresh device.", client.device_id) return False except Exception as exc: logger.error("Matrix: post-upload key verification failed: %s", exc, exc_info=True) return False return True async def _reset_crypto_store_if_device_changed(self, crypto_store: Any, device_id: str) -> bool: """Reset the Olm account when the token's device changed; True if reset. The store is keyed by user ID, so a new device would inherit the old Olm account whose identity keys can never be published under the new device ID.""" if not device_id: return False try: stored_device_id = await crypto_store.get_device_id() except Exception as exc: logger.warning("Matrix: could not read stored device ID: %s", exc) return False if not stored_device_id or stored_device_id == device_id: return False logger.warning( "Matrix: access token belongs to a new device (%s -> %s) — resetting local Olm account " "so fresh identity keys are generated for this device", stored_device_id, device_id) await crypto_store.delete() return True async def _migrate_legacy_crypto_pickle( self, crypto_store: Any, crypto_db: Any, acct_id: str, pickle_key: str) -> bool: """Re-pickle the Olm account under the current pickle key when it changed. The key embeds the device ID; an account created before MATRIX_DEVICE_ID was set lives under ``:default`` and later fails with BAD_ACCOUNT_KEY (silently disabling optional E2EE). False only when an account exists but no key opens it.""" with suppress(Exception): await crypto_store.get_account() return True from mautrix.crypto.store.asyncpg import PgCryptoStore for legacy_key in (f"{acct_id}:default", acct_id): if legacy_key == pickle_key: continue try: account = await PgCryptoStore(account_id=acct_id, pickle_key=legacy_key, db=crypto_db).get_account() except Exception: account = None if account is None: continue # Sessions first, account last: the account is the commit marker (the fast path # above short-circuits once it reads), so an interrupted sweep is retried. try: await self._repickle_crypto_sessions(crypto_db, acct_id, legacy_key, pickle_key) except Exception as exc: logger.error( "Matrix: pickle key migration failed while re-pickling sessions (%s) — leaving " "the account under the legacy key so the migration is retried on the next start.", exc) return False await crypto_store.put_account(account) logger.info( "Matrix: re-pickled crypto store account and sessions under the current pickle key " "(device ID was configured after the account was created)") return True logger.error( "Matrix: crypto store account exists but cannot be unpickled with the current or any " "legacy pickle key. If MATRIX_DEVICE_ID was changed manually, restore its previous value.") return False async def _repickle_crypto_sessions(self, crypto_db: Any, acct_id: str, legacy_key: str, pickle_key: str) -> None: """Re-pickle olm/megolm sessions too — they share the key; account-only breaks key sharing.""" import olm as olm_lib tables = { "crypto_olm_session": olm_lib.Session, "crypto_megolm_inbound_session": olm_lib.InboundGroupSession, "crypto_megolm_outbound_session": olm_lib.OutboundGroupSession} for table, session_cls in tables.items(): rows = await crypto_db.fetch(f"SELECT session_id, session FROM {table} WHERE account_id=$1", acct_id) for row in rows: if row["session"] is None: continue pickled = bytes(row["session"]) with suppress(Exception): session_cls.from_pickle(pickled, pickle_key) continue # already readable with the current key try: session = session_cls.from_pickle(pickled, legacy_key) except Exception as exc: # Readable under neither key: leave it inert rather than delete crypto material. logger.warning( "Matrix: %s row %s cannot be unpickled with the current or legacy key; leaving " "it in place, its sessions are unrecoverable: %s", table, row["session_id"], exc) continue await crypto_db.execute( f"UPDATE {table} SET session=$1 WHERE account_id=$2 AND session_id=$3", session.pickle(pickle_key), acct_id, row["session_id"]) async def _verify_device_keys_on_server(self, client: Any, olm: Any) -> bool: """True if our device keys are on the server (or were re-uploaded); False ⇒ refuse E2EE.""" if not client.device_id or self._device_id_unverified: logger.warning("Matrix: skipping device key verification — device_id not yet established") return True try: our_keys = await self._query_own_device_keys(client) except Exception as exc: logger.error("Matrix: cannot verify device keys on server: %s — refusing E2EE", exc, exc_info=True) return False local_ed25519 = olm.account.identity_keys.get("ed25519") async def _reupload(error_fmt: str, *error_args) -> bool: try: await olm.share_keys() except Exception as exc: logger.error(error_fmt, *error_args, exc, exc_info=True) return False return await self._reverify_keys_after_upload(client, local_ed25519) if not our_keys: logger.warning("Matrix: device keys missing from server — re-uploading") olm.account.shared = False return await _reupload("Matrix: failed to re-upload device keys: %s") if self._extract_server_ed25519(our_keys) == local_ed25519: return True if olm.account.shared: logger.error( "Matrix: server has different identity keys for device %s — local crypto state is " "stale. Delete %s and restart.", client.device_id, str(self._crypto_db_path)) return False logger.warning("Matrix: server has stale keys for device %s — attempting re-upload", client.device_id) with suppress(Exception): await client.api.request( client.api.Method.DELETE if hasattr(client.api, "Method") else "DELETE", f"/_matrix/client/v3/devices/{client.device_id}") logger.info("Matrix: deleted stale device %s from server", client.device_id) return await _reupload( "Matrix: cannot upload device keys for %s: %s. Try generating a new access token to get a fresh device.", client.device_id) @staticmethod async def _abort_connect(api: Any, crypto_db: Any = None) -> bool: """Close what connect() opened so far; always False so callers can ``return await``.""" if crypto_db is not None: await crypto_db.stop() await api.session.close() return False async def _connect_authenticate(self, client: Any, api: Any) -> bool: """Authenticate via access token (whoami) or password login; resolve user/device IDs.""" if self._access_token: api.token = self._access_token try: resp = await client.whoami() resolved_user_id = getattr(resp, "user_id", "") or self._user_id resolved_device_id = str(getattr(resp, "device_id", "") or "") if resolved_user_id: self._user_id = str(resolved_user_id) client.mxid = UserID(self._user_id) # The configured device_id wins when whoami() reports none, but a token can # only upload keys for its own device — on conflict whoami() wins, loudly. if resolved_device_id and self._device_id and resolved_device_id != self._device_id: logger.error( "Matrix: MATRIX_DEVICE_ID=%s does not match the device this access token " "belongs to (%s). A token can only upload keys for its own device, so the " "configured value is being ignored. Unset MATRIX_DEVICE_ID, or use a token " "issued for %s.", self._device_id, resolved_device_id, self._device_id) effective_device_id = resolved_device_id else: effective_device_id = self._device_id or resolved_device_id if effective_device_id: client.device_id = effective_device_id if not client.device_id: try: dev_resp = await client.query_keys({client.mxid: []}) all_devices = (getattr(dev_resp, "device_keys", {}) or {}).get(str(client.mxid)) or {} if len(all_devices) == 1: client.device_id = next(iter(all_devices)) elif not all_devices: logger.warning( "Matrix: no devices found for %s — key verification will be skipped", client.mxid) except Exception as exc: logger.warning("Matrix: device list query failed: %s", exc) if not client.device_id: logger.warning( "Matrix: device_id could not be resolved for %s. Set MATRIX_DEVICE_ID for full " "key verification. E2EE will proceed without server-side device key confirmation.", client.mxid) self._device_id_unverified = True logger.info( "Matrix: using access token for %s%s", self._user_id or "(unknown user)", f" (device {effective_device_id})" if effective_device_id else "") except Exception as exc: logger.error( "Matrix: whoami failed — check MATRIX_ACCESS_TOKEN and MATRIX_HOMESERVER: %s", exc, exc_info=True) return await self._abort_connect(api) elif self._password and self._user_id: try: resp = await client.login( identifier=self._user_id, password=self._password, device_name="Hermes Agent", device_id=self._device_id or None) if resp and hasattr(resp, "device_id"): client.device_id = resp.device_id logger.info("Matrix: logged in as %s", self._user_id) except Exception as exc: logger.error("Matrix: login failed — %s", exc) return await self._abort_connect(api) else: logger.error("Matrix: need MATRIX_ACCESS_TOKEN or MATRIX_USER_ID + MATRIX_PASSWORD") return await self._abort_connect(api) return True async def _connect_setup_e2ee(self, client: Any, api: Any, state_store: Any) -> bool: """Set up the Olm machine + crypto store. Returns False when connect must abort.""" if not _check_e2ee_deps(): if self._e2ee_mode == "optional": logger.warning( "Matrix: E2EE optional but dependencies are missing. Continuing without " "encrypted-room support. %s", _E2EE_INSTALL_HINT) self._encryption = False else: logger.error( "Matrix: E2EE is required but dependencies are missing. %s. Refusing to connect — " "encrypted rooms would silently fail.", _E2EE_INSTALL_HINT) return await self._abort_connect(api) if not self._encryption: return True phase = "import" try: from mautrix.crypto import OlmMachine from mautrix.crypto.store.asyncpg import PgCryptoStore from mautrix.util.async_db import Database self._store_dir.mkdir(parents=True, exist_ok=True) phase = "create" if (self._store_dir / "crypto_store.pickle").exists(): # pre-SQLite era logger.info("Matrix: removing legacy crypto_store.pickle (migrated to SQLite)") (self._store_dir / "crypto_store.pickle").unlink() crypto_db = Database.create( f"sqlite:///{self._crypto_db_path}", upgrade_table=PgCryptoStore.upgrade_table) await crypto_db.start() self._crypto_db = crypto_db _acct_id = self._user_id or "hermes" # Key on the RESOLVED client.device_id (token's real device), not the configured # one, or the Olm account is stored under a key that can never be looked up. _pickle_key = f"{_acct_id}:{client.device_id or self._device_id or 'default'}" crypto_store = PgCryptoStore(account_id=_acct_id, pickle_key=_pickle_key, db=crypto_db) await crypto_store.open() _store_was_reset = False if client.device_id: _store_was_reset = await self._reset_crypto_store_if_device_changed(crypto_store, client.device_id) await crypto_store.put_device_id(client.device_id) # A just-deleted store has no account to migrate. if not _store_was_reset and not await self._migrate_legacy_crypto_pickle( crypto_store, crypto_db, _acct_id, _pickle_key): logger.warning("Matrix: crypto pickle migration failed — E2EE may not work correctly") crypto_state = _CryptoStateStore(state_store, self._joined_rooms, client) olm = OlmMachine(client, crypto_store, crypto_state) olm.share_keys_min_trust = TrustState.UNVERIFIED olm.send_keys_min_trust = TrustState.UNVERIFIED await olm.load() if not await self._verify_device_keys_on_server(client, olm): return await self._abort_connect(api, crypto_db) try: await olm.share_keys() except Exception as exc: if "already exists" in str(exc): logger.error( "Matrix: device %s has stale one-time keys on the server signed with a " "previous identity key. Delete the device from the homeserver and restart, " "or generate a new access token to get a fresh device ID.", client.device_id) return await self._abort_connect(api, crypto_db) logger.warning("Matrix: share_keys() warning during startup: %s", exc) await self._verify_or_bootstrap_cross_signing(olm, client) client.crypto = olm logger.info( "Matrix: E2EE enabled (store: %s%s)", str(self._crypto_db_path), f", device_id={client.device_id}" if client.device_id else "") except Exception as exc: return await self._e2ee_setup_failed(phase, exc, api) return True async def _e2ee_setup_failed(self, what: str, exc: Exception, api: Any) -> bool: """Optional mode: log + disable E2EE and return True; required mode: close + return False.""" if self._e2ee_mode == "optional": logger.warning( "Matrix: failed to %s optional E2EE client; continuing without encrypted-room " "support: %s. %s", what, exc, _E2EE_INSTALL_HINT) self._encryption = False return True logger.error("Matrix: failed to %s E2EE client: %s. %s", what, exc, _E2EE_INSTALL_HINT) return await self._abort_connect(api) async def _verify_or_bootstrap_cross_signing(self, olm: Any, client: Any) -> None: """Verify cross-signing via MATRIX_RECOVERY_KEY, or bootstrap a new key (non-fatal).""" # Honor the active profile's secret scope so a secondary profile under gateway.multiplex_profiles # resolves its own recovery key instead of the default profile's (which fails E2EE verification with # "Key MAC does not match", #69090). recovery_key = _scoped_recovery_key() if recovery_key: try: await olm.verify_with_recovery_key(recovery_key) logger.info("Matrix: cross-signing verified via recovery key") except Exception as exc: logger.warning("Matrix: recovery key verification failed: %s", exc) else: try: own_xsign = await olm.get_own_cross_signing_public_keys() except Exception as exc: own_xsign = None logger.warning("Matrix: cross-signing key lookup failed: %s", exc) if own_xsign is None: _, output_error = _get_matrix_recovery_key_output_target() if output_error: reason = { "not_configured": "is not configured. Configure MATRIX_RECOVERY_KEY from your Matrix client " "or set MATRIX_RECOVERY_KEY_OUTPUT_FILE to write a new recovery key once " "with mode 0600.", "exists": "already exists and will not be overwritten.", }.get(output_error, "is not usable: %s") logger.warning( "Matrix: cross-signing keys are missing, but automatic bootstrap is skipped because " "MATRIX_RECOVERY_KEY_OUTPUT_FILE " + reason, *([output_error] if output_error not in ("not_configured", "exists") else [])) else: try: new_recovery_key = await olm.generate_recovery_key() _handle_generated_matrix_recovery_key(str(client.mxid), new_recovery_key) except Exception as exc: logger.warning( "Matrix: cross-signing bootstrap failed (non-fatal — Element will show " "'not verified by its owner'): %s", exc) async def _connect_initial_sync(self, client: Any) -> None: """Full initial sync: seed joined rooms, DM cache, and dispatch queued to-device events.""" try: sync_data = await client.sync(timeout=10000, full_state=True) if isinstance(sync_data, dict): self._joined_rooms.clear() await self._absorb_sync(client, sync_data, initial=True) else: logger.warning("Matrix: initial sync returned unexpected type %s", type(sync_data).__name__) except Exception as exc: logger.warning("Matrix: initial sync error: %s", exc) async def connect(self, *, is_reconnect: bool = False) -> bool: self._device_id_unverified = False if self._client is not None: try: await self.disconnect() except Exception as exc: logger.warning("Matrix: error disconnecting before reconnect: %s", exc) from mautrix.api import HTTPAPI from mautrix.client import Client from mautrix.client.state_store import MemoryStateStore, MemorySyncStore if not self._homeserver: logger.error("Matrix: homeserver URL not configured") return False # Resolved here, inside the profile scope, so multiplexed profiles never share it. self._resolve_store_dir().mkdir(parents=True, exist_ok=True) client_session = _create_matrix_session(self._proxy_url) api = HTTPAPI(base_url=self._homeserver, token=self._access_token or "", client_session=client_session) state_store = MemoryStateStore() sync_store = MemorySyncStore() client = Client( mxid=UserID(self._user_id) if self._user_id else UserID(""), device_id=self._device_id or None, api=api, state_store=state_store, sync_store=sync_store) self._client = client if not await self._connect_authenticate(client, api): return False if self._encryption and not await self._connect_setup_e2ee(client, api, state_store): return False from mautrix.client import InternalEventType as IntEvt from mautrix.client.dispatcher import MembershipEventDispatcher client.add_dispatcher(MembershipEventDispatcher) # without this INVITE never fires client.add_event_handler(EventType.ROOM_MESSAGE, self._on_room_message, wait_sync=True) client.add_event_handler(EventType.REACTION, self._on_reaction, wait_sync=True) client.add_event_handler(IntEvt.INVITE, self._on_invite, wait_sync=True) self._startup_ts = time.time() self._reset_clock_skew_detector() # a reconnect after an NTP fix starts clean self._closing = False await self._connect_initial_sync(client) if self._encryption and getattr(client, "crypto", None): try: await client.crypto.share_keys() except Exception as exc: logger.warning("Matrix: initial key share failed: %s", exc) self._sync_task = asyncio.create_task(self._sync_loop()) self._mark_connected() self._wire_plugin_handlers(self._client) # plugin-registered native handlers return True async def disconnect(self) -> None: self._closing = True if self._sync_task and not self._sync_task.done(): self._sync_task.cancel() try: await self._sync_task except (asyncio.CancelledError, Exception): pass for tasks in (self._invite_join_tasks.values(), self._reaction_redaction_tasks): pending = list(tasks) for task in pending: if not task.done(): task.cancel() if pending: await asyncio.gather(*pending, return_exceptions=True) self._invite_join_tasks.clear() self._reaction_redaction_tasks.clear() if getattr(self, "_crypto_db", None): try: await self._crypto_db.stop() except Exception as exc: logger.debug("Matrix: could not close crypto DB on disconnect: %s", exc) if self._client: with suppress(Exception): await self._client.api.session.close() self._client = None logger.info("Matrix: disconnected") 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) last_event_id = None for chunk in self.truncate_message(self.format_message(content), self.max_message_length): msg_content = self._build_text_message_content(chunk) self._apply_relation_metadata(msg_content, reply_to=reply_to, metadata=metadata) try: last_event_id = await self._send_room_message(chat_id, msg_content) logger.info("Matrix: sent event %s to %s", last_event_id, chat_id) except Exception as exc: if not (self._encryption and getattr(self._client, "crypto", None)): logger.error("Matrix: failed to send to %s: %s", chat_id, exc) return SendResult(success=False, error=str(exc)) try: # E2EE error: retry once after sharing keys await self._client.crypto.share_keys() last_event_id = await self._send_room_message(chat_id, msg_content) logger.info("Matrix: sent event %s to %s (after key share)", last_event_id, chat_id) except Exception as retry_exc: logger.error("Matrix: failed to send to %s after retry: %s", chat_id, retry_exc) return SendResult(success=False, error=str(retry_exc)) return SendResult(success=True, message_id=last_event_id) async def _send_room_message(self, chat_id: str, msg_content: Dict[str, Any]) -> str: """Send one m.room.message event (45s cap) and return its event ID as str.""" event_id = await asyncio.wait_for( self._client.send_message_event(RoomID(chat_id), EventType.ROOM_MESSAGE, msg_content), timeout=45) return str(event_id) async def create_handoff_thread(self, parent_chat_id: str, name: str) -> Optional[str]: """Post a seed message and return its ``event_id`` as the handoff ``thread_id``. Matrix has no create-thread API: a thread is the events whose ``m.relates_to``/``rel_type: m.thread`` point at a root event (Slack-style), and ``_apply_relation_metadata`` already threads later sends off a supplied ``thread_id``. ``None`` when disconnected or the seed send failed. In-thread replies keep the ROOM's chat_type (``dm``/``group``) in the session key — the handoff watcher and the cron seeder mirror that shape rather than the shared ``thread`` slot.""" if self._client is None: return None result = await self.send(parent_chat_id, (name or "").strip() or t("platform.matrix.handoff.default_name")) root = result.message_id if result.success else None if not root: return None await self._threads.mark_async(str(root)) # replies in this thread bypass require_mention, like inbound roots return str(root) async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: identity = await self._resolve_room_identity(chat_id) return {"name": identity.display_name, "type": "dm" if identity.chat_type == "dm" else "group"} def get_diagnostics(self) -> Dict[str, Any]: now = time.time() token_present = bool(self._access_token) user_id = self._user_id or getattr(self._client, "mxid", "") or "" device_id = self._device_id or getattr(self._client, "device_id", "") or "" return { "platform": "matrix", "homeserver": self._homeserver, "auth": { "access_token_present": token_present, "password_present": bool(self._password), "token_preview": "***" if token_present else "", "user_id": user_id, "device_id_present": bool(device_id), "device_id_preview": "***" if str(device_id or "").strip() else ""}, "sync": { "connected": self._client is not None, "joined_room_count": len(self._joined_rooms), "last_sync_age_seconds": max(0.0, now - self._last_sync_ts) if self._last_sync_ts else None}, "e2ee": { "mode": self._e2ee_mode, "enabled": bool(self._encryption), "deps_available": _check_e2ee_deps(), "crypto_store_path": str(self._crypto_db_path), "recovery_key_configured": bool(_scoped_recovery_key().strip())}, "policy": { "allowed_user_count": len(self._allowed_user_ids), "allowed_room_count": len(self._allowed_room_ids), "ignored_user_pattern_count": len(self._ignored_user_patterns), "require_mention": self._require_mention, "free_response_room_count": len(self._free_rooms), "allow_room_mentions": self._allow_room_mentions, "process_notices": self._process_notices, "allow_public_rooms": _env_truthy("MATRIX_ALLOW_PUBLIC_ROOMS")}, "media": {"max_media_bytes": self._max_media_bytes}} async def _set_typing(self, chat_id: str, timeout: int) -> None: if self._client: with suppress(Exception): await self._client.set_typing(RoomID(chat_id), timeout=timeout) async def send_typing(self, chat_id: str, metadata: Optional[Dict[str, Any]] = None) -> None: await self._set_typing(chat_id, 30000) async def stop_typing(self, chat_id: str) -> None: await self._set_typing(chat_id, 0) async def edit_message(self, chat_id: str, message_id: str, content: str, *, finalize: bool = False) -> SendResult: formatted = self.format_message(content) new_content = self._build_text_message_content(formatted) msg_content: Dict[str, Any] = {"msgtype": "m.text", "body": f"* {formatted}", "m.new_content": new_content} if "m.mentions" in new_content: msg_content["m.mentions"] = new_content["m.mentions"] if "formatted_body" in new_content: msg_content["format"] = "org.matrix.custom.html" msg_content["formatted_body"] = f'* {new_content["formatted_body"]}' msg_content["m.relates_to"] = {"rel_type": "m.replace", "event_id": message_id} return await self._send_content_event(chat_id, msg_content) 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: from tools.url_safety import is_safe_url if not is_safe_url(image_url): logger.warning("Matrix: blocked unsafe image URL (SSRF protection)") return await super().send_image(chat_id, image_url, caption, reply_to, metadata=metadata) try: data, ct, fname = await self._download_external_media_with_cap(image_url) except Exception as exc: logger.warning("Matrix: failed to download image %s: %s", _redact_url_for_log(image_url), exc) fallback = t("platform.matrix.media.image_download_failed") return await self.emit_media_warning(chat_id, fallback, caption=caption, reply_to=reply_to, metadata=metadata) return await self._upload_and_send(chat_id, data, fname, ct, "m.image", caption, reply_to, metadata) async def _download_external_media_with_cap(self, url: str) -> tuple[bytes, str, str]: """Download external media while enforcing redirect safety and size caps.""" from tools.url_safety import is_safe_url if not is_safe_url(url): raise ValueError("blocked unsafe media URL") async def _read_capped(resp, chunks, content_type) -> tuple[bytes, str]: """Enforce Content-Length + streamed size caps, then require an image/* type.""" try: size = int(resp.headers.get("Content-Length") or resp.headers.get("content-length")) except Exception: size = None if size is not None and size > self._max_media_bytes: raise ValueError(f"media exceeds Matrix limit ({size} > {self._max_media_bytes} bytes)") parts: list[bytes] = [] total = 0 async for chunk in chunks: total += len(chunk) if total > self._max_media_bytes: raise ValueError(f"media exceeds Matrix limit (> {self._max_media_bytes} bytes)") parts.append(bytes(chunk)) content_type = str(content_type or "").split(";", 1)[0].strip().lower() if not content_type.startswith("image/"): raise ValueError("external media is not an image") return b"".join(parts), content_type fname = url.rsplit("/", 1)[-1].split("?")[0] or "image.png" try: import aiohttp as _aiohttp _sess_kw, _req_kw = proxy_kwargs_for_aiohttp(self._proxy_url) async with _aiohttp.ClientSession(**_sess_kw) as http: fetch_url = url for _ in range(20): async with http.get( fetch_url, timeout=_aiohttp.ClientTimeout(total=30), allow_redirects=False, **_req_kw) as resp: if resp.status in {301, 302, 303, 307, 308}: location = resp.headers.get("Location") if not location: raise ValueError("redirect missing Location") # Re-validate EVERY hop: a public URL can 302 toward loopback/metadata endpoints, # and checking only the final URL is too late (the hop already connected). fetch_url = urljoin(fetch_url, location) if not is_safe_url(fetch_url): raise ValueError("blocked unsafe redirect URL") continue resp.raise_for_status() data, ct = await _read_capped( resp, resp.content.iter_chunked(65536), getattr(resp, "content_type", None) or resp.headers.get("content-type", "application/octet-stream")) return data, ct, fname raise ValueError("too many redirects") except ImportError: from tools.url_safety import create_ssrf_safe_async_client _httpx_kw: dict = {"proxy": self._proxy_url} if self._proxy_url else {} _httpx_kw["event_hooks"] = {"response": [_ssrf_redirect_guard]} async with create_ssrf_safe_async_client(**_httpx_kw) as http: async with http.stream("GET", url, follow_redirects=True, timeout=30) as resp: resp.raise_for_status() data, ct = await _read_capped( resp, resp.aiter_bytes(), resp.headers.get("content-type", "application/octet-stream")) return data, ct, fname 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) -> SendResult: return await self._send_local_file(chat_id, image_path, "m.image", caption, reply_to, metadata=metadata) async def send_multiple_images( self, chat_id: str, images: list[tuple[str, str]], metadata: Optional[Dict[str, Any]] = None, human_delay: float = 0.0) -> SendResult: if not images: return SendResult(success=False, error="no images to send") from urllib.parse import unquote as _unquote total = len(images) delivered = False for idx, (image_url, alt_text) in enumerate(images, start=1): if human_delay > 0 and idx > 1: await asyncio.sleep(human_delay) caption = f"{alt_text} ({idx}/{total})" if alt_text and total > 1 else (alt_text or None) if image_url.startswith("file://"): result = await self.send_image_file( chat_id=chat_id, image_path=_unquote(image_url[7:]), caption=caption, metadata=metadata) else: result = await self.send_image(chat_id=chat_id, image_url=image_url, caption=caption, metadata=metadata) if not result.success: logger.warning("Matrix: failed to send image %d/%d: %s", idx, total, result.error) delivered = delivered or result.success return SendResult(success=delivered, error=None if delivered else "all images failed to send") 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) -> SendResult: return await self._send_local_file(chat_id, file_path, "m.file", caption, reply_to, file_name, 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, is_voice: Optional[bool] = None) -> SendResult: """Upload audio. The base media dispatch calls this with ``is_voice``: True for a voice-tagged attachment → MSC3245 voice bubble; False for an audio-ext MEDIA attachment → plain ``m.audio`` in the original format. Voice bubbles need Ogg/Opus but callers pass any format (e.g. TTS output), so transcode there — best-effort: without ffmpeg the original is sent. Callers that don't pass the flag (``play_audio``) keep the voice-bubble behavior this method was written for (#116776: the dispatch always passes ``is_voice``, and rejecting it dropped the file).""" if is_voice is False: return await self._send_local_file( chat_id, audio_path, "m.audio", caption, reply_to, metadata=metadata, is_voice=False) converted_path: Optional[str] = None if not str(audio_path).lower().endswith((".ogg", ".oga", ".opus")): # 48k (not the 32k default): Element renders voice bubbles at a higher quality tier. converted_path = await asyncio.to_thread(transcode_to_ogg_opus, audio_path, bitrate="48k", timeout=30) try: return await self._send_local_file( chat_id, converted_path or audio_path, "m.audio", caption, reply_to, # keep the caller's basename (the temp transcode file has a generated name) file_name=(Path(audio_path).with_suffix(".ogg").name if converted_path else None), metadata=metadata, is_voice=True) finally: if converted_path: with suppress(OSError): os.unlink(converted_path) 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) -> SendResult: return await self._send_local_file(chat_id, video_path, "m.video", caption, reply_to, metadata=metadata) # Template attrs for the shared _format_exec_approval core (header + fence + reason only; # the smart-deny/scope wording lives in the reaction legend below). _EA_CMD_BUDGET = 2000 @property def _EA_HEADER(self) -> str: # noqa: N802 — base class attr name; resolved per call for the active language return f"⚠️ **{t('gateway.exec_approval.header')}**\n" async def _send_reaction_prompt( self, chat_id: str, text: str, metadata: Optional[dict], make_prompt, registry: dict, emojis, label: str) -> SendResult: """Send *text*, register ``make_prompt(message_id, requester, expires_at)`` under the resulting event, then seed the bot's reaction controls (recording their IDs).""" result = await self.send(chat_id, text, metadata=metadata) if not result.success or not result.message_id: return result prompt = make_prompt( result.message_id, str((metadata or {}).get("requester_user_id") or "") or None, time.monotonic() + max(self._approval_timeout_seconds, 0)) registry[result.message_id] = prompt for emoji in emojis: try: reaction_event_id = await self._send_reaction(chat_id, result.message_id, emoji) if reaction_event_id: prompt.bot_reaction_events[emoji] = str(reaction_event_id) except Exception as exc: logger.debug("Matrix: failed to add %s reaction %s: %s", label, emoji, exc) return result _EA_REACTIONS = {"once": "✅", "session": "🌀", "always": "♾️", "deny": "❌"} _EA_LEGEND_KEYS = {"once": "platform.matrix.approval.legend_once", "session": "platform.matrix.approval.legend_session", "always": "platform.matrix.approval.legend_always", "deny": "platform.matrix.approval.legend_deny"} # Whole sentences per offered tier (the highest tier wins) so translations never splice fragments. _EA_TYPED_HINT_KEYS = {"once": "platform.matrix.approval.typed_hint_once", "session": "platform.matrix.approval.typed_hint_session", "always": "platform.matrix.approval.typed_hint_always"} async def _send_exec_approval_prompt(self, prompt: ExecApprovalPrompt) -> SendResult: """Reaction-driven approval: the bot seeds one reaction per offered choice.""" if not self._client: return SendResult(success=False, error="Not connected") choices = prompt.choices tier = "once" if not prompt.smart_denied: tier = "always" if "always" in choices else ("session" if "session" in choices else "once") text = ( f"{prompt.text}\n\n" f"{t(self._EA_TYPED_HINT_KEYS[tier])}\n\n" f"{t('platform.matrix.approval.legend_intro')}\n" + "\n".join(t(self._EA_LEGEND_KEYS[c]) for c in choices)) reactions = tuple(self._EA_REACTIONS[c] for c in choices) session_key, chat_id = prompt.session_key, prompt.chat_id def _make(message_id, requester, expires_at): old_event = self._approval_prompt_by_session.get(session_key) if old_event: self._approval_prompts_by_event.pop(old_event, None) self._approval_prompt_by_session[session_key] = message_id return _MatrixApprovalPrompt( session_key=session_key, chat_id=chat_id, message_id=message_id, requester_user_id=requester, expires_at=expires_at) return await self._send_reaction_prompt( chat_id, text, prompt.metadata, _make, self._approval_prompts_by_event, reactions, "approval") async def send_model_picker( self, chat_id: str, providers: list, current_model: str, current_provider: str, session_key: str, on_model_selected, metadata: Optional[Dict[str, Any]] = None) -> SendResult: if not self._client: return SendResult(success=False, error="Not connected") flat_choices = [ (str(model_id), str(p.get("slug") or ""), str(p.get("name") or p.get("slug") or "")) for p in providers or [] for model_id in (p.get("models") or [])][:len(_MATRIX_MODEL_PICKER_REACTIONS)] if not flat_choices: return await self.send( chat_id, t("platform.matrix.picker.no_models"), metadata=metadata) try: from hermes_cli.providers import get_label provider_label = get_label(current_provider) except Exception: provider_label = current_provider unknown = t("platform.shared.unknown") lines = [ t("platform.matrix.picker.title"), t("platform.matrix.picker.current_model", model=current_model or unknown), t("platform.matrix.picker.provider", provider=provider_label or unknown), "", t("platform.matrix.picker.react_model")] choices: dict[str, tuple[str, str]] = {} for emoji, (model_id, provider_slug, provider_name) in zip(_MATRIX_MODEL_PICKER_REACTIONS, flat_choices): choices[emoji] = (model_id, provider_slug) lines.append(f"{emoji} `{model_id}` — {provider_name}") return await self._send_picker( chat_id, lines, choices, session_key, on_model_selected, metadata, self._model_picker_prompts_by_event, "model picker") async def _send_picker( self, chat_id: str, lines: list, choices: dict, session_key: str, on_selected, metadata, registry: dict, label: str) -> SendResult: """Send picker *lines*, register a _MatrixPickerPrompt under the event, seed its reactions.""" return await self._send_reaction_prompt( chat_id, "\n".join(lines), metadata, lambda message_id, requester, expires_at: _MatrixPickerPrompt( chat_id=chat_id, message_id=message_id, session_key=session_key, choices=choices, on_selected=on_selected, requester_user_id=requester, expires_at=expires_at), registry, choices, label) async def send_choice_picker( self, chat_id: str, title: str, choices: list, session_key: str, on_choice_selected, metadata: Optional[Dict[str, Any]] = None) -> SendResult: """Reaction-based choice picker (/reasoning, /fast); choice = {value, label, is_current}.""" if not self._client: return SendResult(success=False, error="Not connected") emoji_choices: dict[str, str] = {} lines = [title, ""] for emoji, choice in zip(_MATRIX_CHOICE_PICKER_REACTIONS, choices): value = str(choice.get("value") or "") label = str(choice.get("label") or value) if choice.get("is_current"): label = t("platform.matrix.picker.current_suffix", label=label) emoji_choices[emoji] = value lines.append(f"{emoji} {label}") if not emoji_choices: return SendResult(success=False, error="No choices") lines += ["", t("platform.matrix.picker.react_choice")] return await self._send_picker( chat_id, lines, emoji_choices, session_key, on_choice_selected, metadata, self._choice_picker_prompts_by_event, "choice picker") def format_message(self, content: str) -> str: """Markdown passes through; strip image markdown (media is uploaded separately).""" return re.sub(r"!\[([^\]]*)\]\(([^)]+)\)", r"\2", content) async def _upload_and_send( self, room_id: str, data: bytes, filename: str, content_type: str, msgtype: str, caption: Optional[str] = None, reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, is_voice: bool = False, voice_metadata: Optional[Dict[str, Any]] = None) -> SendResult: if len(data) > self._max_media_bytes: return self._media_too_large(len(data)) upload_data = data encrypted_file = None if await self._room_needs_encrypted_upload(room_id): try: from mautrix.crypto.attachments import encrypt_attachment upload_data, encrypted_file = encrypt_attachment(data) except Exception as exc: logger.error("Matrix: attachment encryption failed: %s", exc) return SendResult(success=False, error=str(exc)) try: mxc_url = await self._client.upload_media( upload_data, mime_type=content_type, filename=filename, size=len(upload_data)) except Exception as exc: logger.error("Matrix: upload failed: %s", exc) return SendResult(success=False, error=str(exc)) msg_content: Dict[str, Any] = { "msgtype": msgtype, "body": caption or filename, "info": {"mimetype": content_type, "size": len(data)}} if encrypted_file is not None: msg_content["file"] = {**encrypted_file.serialize(), "url": str(mxc_url)} else: msg_content["url"] = str(mxc_url) if is_voice: # MSC3245 native voice flag + MSC1767 audio metadata msg_content["org.matrix.msc3245.voice"] = {} audio_metadata = { k: v for k in ("duration", "waveform") if (v := (voice_metadata or {}).get(k)) is not None} if "duration" in audio_metadata: msg_content["info"]["duration"] = audio_metadata["duration"] if audio_metadata: msg_content["org.matrix.msc1767.audio"] = audio_metadata self._apply_relation_metadata(msg_content, reply_to=reply_to, metadata=metadata) return await self._send_content_event(room_id, msg_content) async def _room_needs_encrypted_upload(self, room_id: str) -> bool: """E2EE on, Olm machine loaded, and the state store says the room is encrypted.""" if not (self._encryption and getattr(self._client, "crypto", None)): return False state_store = getattr(self._client, "state_store", None) if not state_store: return False try: return bool(await state_store.is_encrypted(RoomID(room_id))) except Exception: return False def _media_too_large(self, size: int) -> SendResult: return SendResult( success=False, error=f"Media file exceeds Matrix limit ({size} > {self._max_media_bytes} bytes)") async def _send_content_event(self, room_id: str, msg_content: Dict[str, Any]) -> SendResult: """Send a prebuilt m.room.message payload, mapping exceptions to SendResult.""" try: event_id = await self._client.send_message_event(RoomID(room_id), EventType.ROOM_MESSAGE, msg_content) return SendResult(success=True, message_id=str(event_id)) except Exception as exc: return SendResult(success=False, error=str(exc)) async def _send_local_file( self, room_id: str, file_path: str, msgtype: str, caption: Optional[str] = None, reply_to: Optional[str] = None, file_name: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, is_voice: bool = False) -> SendResult: p = Path(file_path).expanduser() if not p.exists(): # file_path is host-local; never echo it into chat. logger.warning("[%s] upload fallback: media file not found for %s", self.name, file_path) text = t("platform.shared.media.attachment_failed") return await self.emit_media_warning(room_id, text, caption=caption, reply_to=reply_to, metadata=metadata) try: file_size = p.stat().st_size except OSError: file_size = 0 if file_size > self._max_media_bytes: return self._media_too_large(file_size) fname = file_name or p.name # ffprobe/ffmpeg probing is blocking (subprocess timeouts up to 15s) — # run it off the event loop so voice uploads never stall the adapter. voice_metadata = await asyncio.to_thread(_matrix_voice_metadata_for_file, p) if is_voice else None return await self._upload_and_send( room_id, p.read_bytes(), fname, mimetypes.guess_type(fname)[0] or "application/octet-stream", msgtype, caption, reply_to, metadata, is_voice, voice_metadata) async def _sync_loop(self) -> None: client = self._client next_batch = await client.sync_store.get_next_batch() # resume from the initial sync while not self._closing: try: # 45s outer cap guards TCP-level hangs the 30s long-poll timeout cannot catch. # mautrix raises on every non-2xx, so a non-dict here is never an error object. sync_data = await asyncio.wait_for(client.sync(since=next_batch, timeout=30000), timeout=45.0) if isinstance(sync_data, dict): next_batch = await self._absorb_sync(client, sync_data) or next_batch await asyncio.sleep(0) # let fresh invite joins start before the next sync except asyncio.CancelledError: return except Exception as exc: if self._closing: return # Detect permanent auth/permission failures. Transient 5xx outages must retry. if _is_permanent_matrix_auth_error(exc): logger.error("Matrix: permanent auth error, stopping sync: %s", exc) return logger.warning("Matrix: sync error: %s — retrying in 5s", exc) await asyncio.sleep(5) async def _absorb_sync(self, client: Any, sync_data: Dict[str, Any], *, initial: bool = False) -> Optional[str]: """Apply one sync response: joined rooms, next_batch, event dispatch, pending invites. Returns next_batch. The initial (full-state) sync also seeds the DM cache and dispatches so the OlmMachine sees to-device key shares queued while offline.""" self._last_sync_ts = time.time() rooms_join = sync_data.get("rooms", {}).get("join", {}) if rooms_join or initial: self._joined_rooms.update(rooms_join.keys()) self._invalidate_room_identities() nb = sync_data.get("next_batch") # incremental syncs resume from here if nb: await client.sync_store.put_next_batch(nb) if initial: logger.info("Matrix: initial sync complete, joined %d rooms", len(self._joined_rooms)) await self._refresh_dm_cache() try: await self._dispatch_sync(sync_data) except Exception as exc: logger.warning("Matrix: %s: %s", "initial sync event dispatch error" if initial else "sync event dispatch error", exc) self._schedule_pending_invite_joins(sync_data) return nb async def _dispatch_sync(self, sync_data: Dict[str, Any]) -> None: """Dispatch a sync response through the mautrix event machinery.""" client = self._client if not client or not hasattr(client, "handle_sync"): return tasks = client.handle_sync(sync_data) if inspect.isawaitable(tasks): tasks = await tasks if tasks: # return_exceptions=True: one failing handler must not drop its SIBLING events. results = await asyncio.gather(*tasks, return_exceptions=True) for result in results: if isinstance(result, Exception): logger.warning("Matrix: event handler failed during sync dispatch: %s", result) def _is_self_sender(self, sender: str) -> bool: """True if *sender* is the bot itself (case-insensitive: homeservers vary localpart case). With no resolved user_id we can't prove a sender is NOT us, so return True — dropping our own events beats an echo loop ("hall of mirrors"). Matrix user IDs are byte-compared after trimming whitespace and lowercasing — some homeservers normalize the localpart case differently at different API surfaces, and the reply-loop tail of the "hall of mirrors" bug (#15763) has been observed with the bot's own account bypassing a case-sensitive equality check. """ own = (self._user_id or "").strip().lower() return not own or sender.strip().lower() == own @staticmethod def _is_system_or_bridge_sender(sender: str) -> bool: """True for appservice/bridge/system identities (``@_telegram_123:server``) or malformed IDs. Never offer these a pairing code: an approved bridge would relay every outbound message back as an "authorized user message" (echo loop). We treat these as system identities for pairing purposes: they should never be offered a pairing code, because an operator approving the code would hand the bridge itself permanent authorization — and every outbound message relayed by the bridge would then loop back into the agent as an "authorized user message", which is the root of issue #15763. """ localpart = (sender or "").strip().lstrip("@").partition(":")[0] return not localpart or localpart.startswith("_") async def _is_allowed_matrix_room_event(self, room_id: str) -> bool: """MATRIX_ALLOWED_ROOMS gate; DMs are exempt so personal chats survive a project allowlist.""" if not self._allowed_room_ids or room_id in self._allowed_room_ids: return True try: return await self._is_dm_room(room_id) except Exception as exc: logger.debug("Matrix: could not resolve room identity for allowlist check in %s: %s", room_id, exc) return False def _reset_clock_skew_detector(self) -> None: """State for _note_late_grace_drop: consecutive-drop count, their skew, and the once-only warning.""" # Clock-skew detection: count grace-check drops that happen well after startup (i.e. not # initial-sync backfill). If the host's system clock is set ahead of real time, the startup grace # check `event_ts < startup_ts - 5` silently drops every live message. See #12614 — the symptom is # "bot joins rooms but never replies". Drops only count when their skew matches the first sampled # drop (within 60s), so varied-age backfill from freshly-invited rooms doesn't trip the heuristic. self._late_grace_drops: int = 0 self._late_grace_skew: float = 0.0 self._clock_skew_warned: bool = False def _note_late_grace_drop(self, event_ts: float) -> None: """Clock-skew heuristic for grace-check drops well after startup. A host clock set ahead of real time makes every live event look "older than startup" and the bot silently never replies. Warn once when drops keep happening >30s after startup with a *consistent* skew — unlike backfill from a freshly invited room, whose event ages vary widely and reset the counter.""" if self._clock_skew_warned or time.time() - self._startup_ts <= 30: return skew = self._startup_ts - event_ts if not (5 < skew < 86400): # ignore malformed/absurd timestamps return if self._late_grace_drops and abs(skew - self._late_grace_skew) < 60: self._late_grace_drops += 1 else: self._late_grace_skew = skew self._late_grace_drops = 1 if self._late_grace_drops >= 3: logger.warning( "Matrix: dropped %d consecutive live events as 'too old' more than 30s after startup " "(skew ≈ %.0fs). The host system clock is likely set ahead of real time, which causes " "the startup grace filter to silently discard every incoming message. Run " "`timedatectl set-ntp true` (or sync NTP) and restart the bot.", self._late_grace_drops, skew) self._clock_skew_warned = True async def _on_room_message(self, event: Any) -> None: room_id = str(getattr(event, "room_id", "")) sender = str(getattr(event, "sender", "")) # DEBUG-level proof the callback fires at all (silent-inbound troubleshooting). logger.debug( "Matrix: callback fired — event %s from %s in %s", getattr(event, "event_id", "?"), sender, room_id) if self._is_self_sender(sender): return # Bridge/system identities must never reach the pairing flow (echo loop once paired). # Ignore own messages (case-insensitive; also drops when our own user_id hasn't been resolved yet — # see _is_self_sender docstring and issue #15763). # Once a bridge user is paired, every outbound message it relays would loop back as an authorized # user message (the "hall of mirrors" in #15763). if self._is_system_or_bridge_sender(sender): logger.debug("Matrix: ignoring system/bridge sender %s in %s", sender, room_id) return if any(pattern.search(sender or "") for pattern in self._ignored_user_patterns): logger.debug("Matrix: ignoring sender %s in %s due to configured ignore pattern", sender, room_id) return if not await self._is_allowed_matrix_room_event(room_id): logger.info("Matrix: ignoring message from unauthorized room %s", room_id) return event_id = str(getattr(event, "event_id", "")) if self._is_duplicate_event(event_id): return # Startup grace: ignore old messages replayed by the initial sync. event_ts = _matrix_event_timestamp_seconds(event) if event_ts and event_ts < self._startup_ts - _STARTUP_GRACE_SECONDS: self._note_late_grace_drop(event_ts) return content = getattr(event, "content", None) if content is None: return if isinstance(content, dict): source_content, msgtype = content, content.get("msgtype", "") else: source_content = content.serialize() if hasattr(content, "serialize") else {} msgtype = str(content.msgtype) if hasattr(content, "msgtype") else "" relates_to = source_content.get("m.relates_to", {}) if relates_to.get("rel_type") == "m.replace": # skip edits return # m.notice is the conventional bot-response msgtype; ignoring it prevents bot-to-bot loops. if msgtype == "m.notice" and not self._process_notices: return if msgtype in ("m.image", "m.audio", "m.video", "m.file"): await self._handle_media_message(room_id, sender, event_id, event_ts, source_content, relates_to, msgtype) elif msgtype in ("m.text", "m.notice"): await self._handle_text_message(room_id, sender, event_id, event_ts, source_content, relates_to) async def _resolve_message_context( self, room_id: str, sender: str, event_id: str, body: str, source_content: dict, relates_to: dict, mention_claimed: bool = False, voice_gate: Optional[VoiceGate] = None) -> Optional[tuple]: """Shared mention/thread/DM gating. Returns (body, is_dm, chat_type, thread_id, display_name, source) or None when the message should be dropped. ``mention_claimed`` marks a parked voice claimed by the sender's follow-up bare @mention; ``voice_gate`` is the in-flight mark of a parkable voice, released once the park decision is made.""" identity = await self._resolve_room_identity(room_id) is_dm = await self._is_dm_room(room_id) chat_type = "dm" if is_dm else "group" thread_id = _thread_root(relates_to) is_mentioned = mention_claimed or self._content_mentions_bot(body, source_content) if not is_dm: # Whitelist first: non-listed rooms are dropped even when @mentioned (DMs exempt). if self._allowed_rooms and room_id not in self._allowed_rooms: logger.debug( "Matrix: ignoring message %s in %s — room not in MATRIX_ALLOWED_ROOMS whitelist", event_id, room_id) return None is_free_room = room_id in self._free_rooms in_bot_thread = bool(thread_id and thread_id in self._threads) if self._require_mention and not is_free_room and not in_bot_thread: if not is_mentioned and not body.startswith("/"): if voice_gate is not None: # parkable voice: a bare @mention may follow (Element X) self._parked_voices.park(room_id, sender, voice_gate, event_id, source_content, relates_to) logger.debug( "Matrix: ignoring message %s in %s — no @mention " "(set MATRIX_REQUIRE_MENTION=false to disable)", event_id, room_id) return None # thread_require_mention: even inside a bot thread require @mention — prevents # infinite reply loops when several bots share one thread. elif self._thread_require_mention and in_bot_thread and not is_free_room and not is_mentioned: logger.debug( "Matrix: ignoring message %s in thread %s — no @mention (thread_require_mention=true)", event_id, thread_id) return None if is_mentioned and self._require_mention: # Strip the mention from the reply text only: the quote block carries the # ``> <@bot:srv> ...`` reply pill, which _extract_reply_context parses later # for reply_to_author_id. A whole-body replace rewrote the pill to ``> <>`` # and silently dropped the replied-to author (#111233). Only a real reply carries a # pill; a hand-typed blockquote in a plain message is stripped whole as before. if relates_to.get("m.in_reply_to"): quote_block, reply_text = _split_reply_fallback(body) body = quote_block + self._strip_mention(reply_text) else: body = self._strip_mention(body) # Real thread roots are preserved above; synthetic roots (this event) follow policy: DM # @mention threads / DM auto-thread, or room auto-thread unless session_scope pins the room. if not thread_id: if is_dm: synthetic = (self._dm_mention_threads and is_mentioned) or self._dm_auto_thread else: synthetic = self._matrix_session_scope == "thread" or ( self._matrix_session_scope != "room" and self._auto_thread) if synthetic: thread_id = event_id if voice_gate is not None: # decided (parked or passing): don't hold bare mentions any longer self._parked_voices.release(room_id, sender, voice_gate) display_name = await self._get_display_name(room_id, sender) source = self.build_source( chat_id=room_id, chat_name=identity.display_name, chat_type=chat_type, user_id=sender, user_name=display_name, thread_id=thread_id, chat_topic=identity.room_topic, guild_id=identity.server_name, parent_chat_id=room_id if thread_id else None, message_id=event_id) if thread_id: await self._threads.mark_async(thread_id) # covers real roots and synthetic ones alike self._background_read_receipt(room_id, event_id) return body, is_dm, chat_type, thread_id, display_name, source async def _extract_reply_context( self, room_id: str, body: str, relates_to: dict ) -> tuple[str, Optional[str], Optional[str], Optional[str], Optional[str]]: """Return (body, reply_to, reply_to_text, reply_to_author_id, reply_to_author_name). Captures the inline reply fallback (``> <@user:srv> text\\n\\nreply``) BEFORE stripping it, so the prompt layer can render "[Replying to: ...]" like Signal/Slack/Telegram.""" reply_to = (relates_to.get("m.in_reply_to") or {}).get("event_id") reply_to_text = reply_to_author_id = reply_to_author_name = None if reply_to and body.startswith("> "): reply_to_text, reply_to_author_id = _extract_reply_fallback(body) body = _strip_reply_fallback(body) # Resolve the replied-to author's display name (falls back to localpart). if reply_to_author_id: reply_to_author_name = await self._get_display_name(room_id, reply_to_author_id) return body, reply_to, reply_to_text, reply_to_author_id, reply_to_author_name async def _build_inbound_event( self, room_id: str, sender: str, event_id: str, body: str, source_content: dict, relates_to: dict, ctx: Optional[tuple] = None, **extra) -> Optional[MessageEvent]: """Gate + normalise an inbound event into a MessageEvent (None => drop). Text body may still change (reply-fallback strip); ``extra`` carries media fields / message_type. ``ctx`` is a pre-resolved ``_resolve_message_context`` result (media path gates before downloading); resolving it twice would double the read receipt / thread mark.""" if ctx is None: ctx = await self._resolve_message_context(room_id, sender, event_id, body, source_content, relates_to) if ctx is None: return None body, _is_dm, _chat_type, _thread_id, display_name, source = ctx body, reply_to, reply_to_text, reply_to_author_id, reply_to_author_name = ( await self._extract_reply_context(room_id, body, relates_to)) media_msgtype = extra.pop("media_msgtype", None) if media_msgtype is None: # Re-normalize after reply stripping so ``> quoted\n\n!model`` is still a command. body = _normalize_matrix_bang_command(body) extra["message_type"] = MessageType.COMMAND if body.startswith("/") else MessageType.TEXT elif _is_bare_media_filename(media_msgtype, body): body = "" # transport filename, not user text return MessageEvent( text=body, source=source, raw_message=source_content, message_id=event_id, reply_to_message_id=reply_to, reply_to_text=reply_to_text, reply_to_author_id=reply_to_author_id, reply_to_author_name=reply_to_author_name, # Top-level sender fields mirror source.* — downstream prompt code reads them. user_id=sender, user_name=display_name, **extra) async def _handle_text_message( self, room_id: str, sender: str, event_id: str, event_ts: float, source_content: dict, relates_to: dict) -> None: body = source_content.get("body", "") or "" if not body: return # Dict lookup first: the mention regexes only run when a voice is parked or being gated # (both only happen under require_mention). if (self._parked_voices.pending(room_id, sender) and not self._strip_mention(body).strip() and self._content_mentions_bot(body, source_content)): limit = self._parked_voices.mark() # never claim a voice sent after this mention await self._parked_voices.settle(room_id, sender) # same-/sync-batch voice still gating parked = self._parked_voices.claim(room_id, sender, before=limit) if parked: # answer the voice this bare mention was typed for, not an empty text voice_id, voice_content, voice_relates = parked await self._handle_media_message( room_id, sender, voice_id, event_ts, voice_content, voice_relates, "m.audio", mention_claimed=True) self._background_read_receipt(room_id, event_id) # the claim receipted the voice return msg_event = await self._build_inbound_event( room_id, sender, event_id, _normalize_matrix_bang_command(body), source_content, relates_to) if msg_event is None: return if msg_event.message_type == MessageType.TEXT and self._text_batch_delay_seconds > 0: self._enqueue_text_event(msg_event) else: await self.handle_message(msg_event) async def _handle_media_message( self, room_id: str, sender: str, event_id: str, event_ts: float, source_content: dict, relates_to: dict, msgtype: str, mention_claimed: bool = False) -> None: body = source_content.get("body", "") or "" url = source_content.get("url", "") if url and not str(url).startswith("mxc://"): logger.warning("[Matrix] Rejecting inbound media %s with non-MXC URL", event_id) return content_info = source_content.get("info", {}) if not isinstance(content_info, dict): content_info = {} event_mimetype = content_info.get("mimetype", "") try: event_size_int = int(content_info.get("size") or 0) except (TypeError, ValueError): event_size_int = 0 if event_size_int and event_size_int > self._max_media_bytes: logger.warning( "[Matrix] Rejecting oversized inbound media %s (%d > %d bytes)", event_id, event_size_int, self._max_media_bytes) return file_content = source_content.get("file", {}) # encrypted media carries file.url if not url and isinstance(file_content, dict): url = file_content.get("url", "") or "" if url and not str(url).startswith("mxc://"): logger.warning("[Matrix] Rejecting inbound encrypted media %s with non-MXC URL", event_id) return is_encrypted_media = bool(file_content and isinstance(file_content, dict) and file_content.get("url")) msg_type, media_type, is_voice_message = self._classify_inbound_media(msgtype, event_mimetype, source_content) # Gate (require_mention / allowed rooms) BEFORE the download: an unmentioned or # non-allowlisted room must not pull media onto the host only to drop it. # First await: mark a voice that may park in-flight so a concurrent bare mention waits for it. gate = self._parked_voices.begin(room_id, sender) if self._voice_may_park( room_id, body, source_content, relates_to, mention_claimed) else None try: ctx = await self._resolve_message_context( room_id, sender, event_id, body, source_content, relates_to, mention_claimed=mention_claimed, voice_gate=gate) finally: if gate is not None: self._parked_voices.release(room_id, sender, gate) if ctx is None: return # Cache locally so downstream tools get a real file path. cached_path = None if url: try: cached_path = await self._download_and_cache_media( url, event_id, file_content if is_encrypted_media else None, msg_type, media_type, is_voice_message, body) except Exception as e: logger.warning("[Matrix] Failed to cache media: %s", e) # Unencrypted media may fall back to the HTTP download URL when caching failed. http_url = self._mxc_to_http(url) if url and not is_encrypted_media else "" media_urls = [cached_path] if cached_path else ([http_url] if http_url else None) msg_event = await self._build_inbound_event( room_id, sender, event_id, body, source_content, relates_to, ctx=ctx, message_type=msg_type, media_urls=media_urls, media_types=[media_type] if media_urls else None, media_msgtype=msgtype) if msg_event is not None: await self.handle_message(msg_event) @staticmethod def _classify_inbound_media( msgtype: str, event_mimetype: str, source_content: dict) -> tuple[MessageType, str, bool]: """Map a Matrix media msgtype to (MessageType, mime type, is_voice_message).""" if msgtype == "m.image": return MessageType.PHOTO, event_mimetype or "image/png", False if msgtype == "m.audio": is_voice = has_voice_marker(source_content) return (MessageType.VOICE if is_voice else MessageType.AUDIO), event_mimetype or "audio/ogg", is_voice if msgtype == "m.video": return MessageType.VIDEO, event_mimetype or "video/mp4", False return MessageType.DOCUMENT, event_mimetype or "application/octet-stream", False async def _download_and_cache_media( self, url: str, event_id: str, encrypted_file: Optional[dict], msg_type: MessageType, media_type: str, is_voice_message: bool, body: str) -> Optional[str]: """Download (and decrypt, when *encrypted_file* is given) media into the local cache.""" file_bytes = await self._client.download_media(ContentURI(url)) if file_bytes is None: return None if encrypted_file is not None: from mautrix.crypto.attachments import decrypt_attachment hashes_value, key_value = encrypted_file.get("hashes"), encrypted_file.get("key") hash_value = hashes_value.get("sha256") if isinstance(hashes_value, dict) else None key_value = key_value.get("k") if isinstance(key_value, dict) else key_value iv_value = encrypted_file.get("iv") if not (key_value and hash_value and iv_value): logger.warning("[Matrix] Encrypted media event missing decryption metadata for %s", event_id) return None file_bytes = decrypt_attachment(file_bytes, key_value, hash_value, iv_value) from gateway.platforms.base import ( cache_audio_from_bytes_async, cache_document_from_bytes_async, cache_image_from_bytes_async, ) if msg_type == MessageType.PHOTO: ext_map = {"image/jpeg": ".jpg", "image/png": ".png", "image/gif": ".gif", "image/webp": ".webp"} cached_path = await cache_image_from_bytes_async(file_bytes, ext=ext_map.get(media_type, ".jpg")) logger.info("[Matrix] Cached user image at %s", cached_path) return cached_path if msg_type in {MessageType.AUDIO, MessageType.VOICE}: ext = Path(body or ("voice.ogg" if is_voice_message else "audio.ogg")).suffix or ".ogg" return await cache_audio_from_bytes_async(file_bytes, ext=ext) filename = body or ("video.mp4" if msg_type == MessageType.VIDEO else "document") return await cache_document_from_bytes_async(file_bytes, filename) async def _on_invite(self, event: Any) -> None: """Auto-join rooms when invited, recording DM rooms in m.direct.""" room_id = str(getattr(event, "room_id", "")) is_direct = bool(getattr(getattr(event, "content", None), "is_direct", False)) inviter = str(getattr(event, "sender", "")) # Only authorized inviters — otherwise any federated user could pull the bot into rooms. if not self._is_authorized_user(inviter, str(room_id)): logger.warning("Matrix: rejecting invite to %s from unauthorized user %s", room_id, inviter) return logger.info("Matrix: invited to %s — joining (is_direct=%s)", room_id, is_direct) # Join off the sync path; a declared DM is recorded in m.direct once the join lands. self._schedule_invite_join(room_id, is_direct=is_direct and bool(inviter), inviter=inviter) async def _join_room_by_id(self, room_id: str) -> bool: if not room_id or room_id in self._joined_rooms: return bool(room_id) try: await self._client.join_room(RoomID(room_id)) self._joined_rooms.add(room_id) self._invalidate_room_identities(room_id) logger.info("Matrix: joined %s", room_id) await self._refresh_dm_cache() return True except Exception as exc: logger.warning("Matrix: error joining %s: %s", room_id, exc) # Abandoned rooms ("no servers ..." / "room not found") would retry every startup # unless we leave the invite; the match is narrow so transient errors keep retrying. msg = str(exc).lower() if ("no servers" in msg) or ("room not found" in msg): with suppress(Exception): await self._client.leave_room(RoomID(room_id)) logger.info("Matrix: declined dead invite to %s", room_id) return False def _schedule_invite_join(self, room_id: str, *, is_direct: bool = False, inviter: str = "") -> None: """Schedule an invite join without blocking sync or gateway readiness.""" existing = self._invite_join_tasks.get(room_id) if not room_id or room_id in self._joined_rooms or (existing and not existing.done()): return async def _join_invite() -> None: try: joined = await asyncio.wait_for(self._join_room_by_id(room_id), timeout=45.0) if joined and is_direct and inviter: await self._record_dm_room(room_id, inviter) except asyncio.TimeoutError: logger.warning("Matrix: timed out joining invite %s", room_id) finally: self._invite_join_tasks.pop(room_id, None) self._invite_join_tasks[room_id] = asyncio.create_task(_join_invite()) def _schedule_pending_invite_joins(self, sync_data: Dict[str, Any]) -> None: """Join rooms still present in rooms.invite after sync processing.""" invites = (sync_data.get("rooms", {}) if isinstance(sync_data, dict) else {}).get("invite", {}) if not isinstance(invites, dict): return for room_id, invited_room in invites.items(): if room_id in self._joined_rooms: continue # This reconcile pass runs after _dispatch_sync and sees every # rooms.invite entry, whether _on_invite joined it, rejected # it, or (for invites that arrived while the gateway was down) # is only now seeing it. The invite event object is gone by # this point, so the DM signal must be read from the stripped # invite state; without it a direct invite joined here is never # recorded in m.direct and gets misclassified as a group. is_direct, inviter = self._extract_invite_dm_signal(invited_room) # The inviter allowlist gate from _on_invite must apply here # too: an unconditional join would re-admit a live invite that # _on_invite just rejected milliseconds earlier, and would # auto-join any invite from an arbitrary federated user on # restart. An inviter missing from the stripped invite state # fails closed, like an empty sender in _on_invite. if not self._is_authorized_user(inviter, str(room_id)): logger.warning( "Matrix: rejecting invite to %s from unauthorized user %s", room_id, inviter, ) continue logger.info( "Matrix: reconciling pending invite for %s (is_direct=%s)", room_id, is_direct, ) self._schedule_invite_join(str(room_id), is_direct=is_direct, inviter=inviter) def _extract_invite_dm_signal(self, invited_room: Any) -> tuple[bool, str]: """Read the is_direct flag and inviter from a room's invite_state. The stripped ``m.room.member`` event for our own user carries the ``is_direct`` flag from the original invite; its sender is the inviter. Returns ``(False, "")`` when the signal is absent. """ if not self._user_id: return False, "" if not isinstance(invited_room, dict): return False, "" invite_state = invited_room.get("invite_state", {}) if not isinstance(invite_state, dict): return False, "" events = invite_state.get("events", []) if not isinstance(events, list): return False, "" for event in events: if not isinstance(event, dict): continue if event.get("type") != "m.room.member": continue if event.get("state_key") != self._user_id: continue content = event.get("content", {}) if not isinstance(content, dict): continue if content.get("membership") != "invite": continue return bool(content.get("is_direct")), str(event.get("sender", "")) return False, "" async def _send_reaction(self, room_id: str, event_id: str, emoji: str) -> Optional[str]: """Send an emoji reaction; returns the reaction event_id, or None on failure.""" if not self._client: return None content = {"m.relates_to": {"rel_type": "m.annotation", "event_id": event_id, "key": emoji}} try: resp_event_id = await self._client.send_message_event(RoomID(room_id), EventType.REACTION, content) logger.debug("Matrix: sent reaction %s to %s", emoji, event_id) return str(resp_event_id) except Exception as exc: logger.debug("Matrix: reaction send error: %s", exc) return None async def _redact_reaction(self, room_id: str, reaction_event_id: str, reason: str = "") -> bool: return await self.redact_message(room_id, reaction_event_id, reason) def _schedule_reaction_redaction(self, room_id: str, reaction_event_id: str, reason: str = "") -> None: """Redact a reaction after a short delay so message delivery settles.""" async def _redact_later() -> None: try: if self._reaction_redaction_delay_seconds: await asyncio.sleep(self._reaction_redaction_delay_seconds) if not await self._redact_reaction(room_id, reaction_event_id, reason): logger.debug("Matrix: failed to redact reaction %s", reaction_event_id) except asyncio.CancelledError: raise except Exception as exc: logger.debug("Matrix: delayed reaction redaction failed for %s: %s", reaction_event_id, exc) task = asyncio.create_task(_redact_later()) self._reaction_redaction_tasks.add(task) task.add_done_callback(self._reaction_redaction_tasks.discard) async def on_processing_start(self, event: MessageEvent) -> None: msg_id, room_id = event.message_id, event.source.chat_id if self._reactions_enabled and msg_id and room_id: reaction_event_id = await self._send_reaction(room_id, msg_id, "\U0001f440") if reaction_event_id: self._pending_reactions[(room_id, msg_id)] = reaction_event_id async def on_processing_complete(self, event: MessageEvent, outcome: ProcessingOutcome) -> None: msg_id, room_id = event.message_id, event.source.chat_id if not self._reactions_enabled or not msg_id or not room_id or outcome == ProcessingOutcome.CANCELLED: return eyes_event_id = self._pending_reactions.pop((room_id, msg_id), None) if eyes_event_id: self._schedule_reaction_redaction(room_id, eyes_event_id, "processing complete") await self._send_reaction(room_id, msg_id, "\u2705" if outcome == ProcessingOutcome.SUCCESS else "\u274c") async def _on_reaction(self, event: Any) -> None: sender = str(getattr(event, "sender", "")) if self._is_self_sender(sender): return event_id = str(getattr(event, "event_id", "")) if self._is_duplicate_event(event_id): return room_id = str(getattr(event, "room_id", "")) content = getattr(event, "content", None) if not content: return relates_to = (content.get("m.relates_to", {}) if isinstance(content, dict) else getattr(content, "relates_to", {})) reacts_to = key = "" if isinstance(relates_to, dict): reacts_to = relates_to.get("event_id", "") key = relates_to.get("key", "") elif hasattr(relates_to, "event_id"): reacts_to = str(getattr(relates_to, "event_id", "")) key = str(getattr(relates_to, "key", "")) logger.info("Matrix: reaction %s from %s on %s in %s", key, sender, reacts_to, room_id) for handler in (self._handle_approval_reaction, self._handle_model_picker_reaction, self._handle_choice_picker_reaction): if await handler(room_id, reacts_to, key, sender): return async def _claim_reaction_prompt( self, registry: dict, room_id: str, reacts_to: str, key: str, sender: str, label: str, invalid_text: str, on_expired, choices: Optional[dict] = None) -> tuple[bool, Any, Any]: """Shared gate for reaction prompts: (handled, prompt, selection). handled=False => not our prompt; selection=None with handled=True => consumed without action (wrong room, expired, unauthorized reactor, or a key that is not a choice). ``choices`` defaults to ``prompt.choices``.""" prompt = registry.get(reacts_to) if not prompt or prompt.resolved: return False, None, None if room_id != prompt.chat_id: return True, prompt, None if self._matrix_prompt_expired(prompt): await on_expired(room_id, reacts_to, prompt) return True, prompt, None if not await self._validate_matrix_prompt_reactor(room_id, reacts_to, sender, prompt, label): return True, prompt, None selection = (prompt.choices if choices is None else choices).get(key) if selection is None: await self._send_invalid_reaction_feedback(room_id, reacts_to, invalid_text) return True, prompt, selection async def _handle_approval_reaction(self, room_id: str, reacts_to: str, key: str, sender: str) -> bool: """Resolve a pending exec-approval prompt from a reaction. True if it was the target.""" handled, prompt, choice = await self._claim_reaction_prompt( self._approval_prompts_by_event, room_id, reacts_to, key, sender, "approval", t("platform.matrix.approval.invalid_reaction"), self._expire_matrix_approval_prompt, choices=self._approval_reaction_map) if choice is None: return handled try: from tools.approval import resolve_gateway_approval count = resolve_gateway_approval(prompt.session_key, choice) if count: prompt.resolved = True self._approval_prompts_by_event.pop(reacts_to, None) self._approval_prompt_by_session.pop(prompt.session_key, None) logger.info( "Matrix reaction resolved %d approval(s) for session %s (choice=%s, user=%s)", count, prompt.session_key, choice, sender) await self._redact_bot_approval_reactions(room_id, prompt) except Exception as exc: logger.error("Failed to resolve gateway approval from Matrix reaction: %s", exc) return True async def _handle_model_picker_reaction(self, room_id: str, reacts_to: str, key: str, sender: str) -> bool: """Apply a model-picker reaction. True if the reaction targeted a pending picker.""" return await self._handle_picker_reaction( self._model_picker_prompts_by_event, room_id, reacts_to, key, sender, "model picker", t("platform.matrix.picker.invalid_model_reaction"), self._expire_matrix_model_picker_prompt, ("switch model", "platform.matrix.picker.verb_switch_model"), redact_bot_reactions=True) async def _handle_choice_picker_reaction(self, room_id: str, reacts_to: str, key: str, sender: str) -> bool: """Apply a choice-picker reaction. True if the reaction targeted a pending picker.""" async def _expire(_room_id, target_event_id, _prompt): self._choice_picker_prompts_by_event.pop(target_event_id, None) return await self._handle_picker_reaction( self._choice_picker_prompts_by_event, room_id, reacts_to, key, sender, "choice picker", t("platform.matrix.picker.invalid_choice_reaction"), _expire, ("apply choice", "platform.matrix.picker.verb_apply_selection")) async def _handle_picker_reaction( self, registry: dict, room_id: str, reacts_to: str, key: str, sender: str, label: str, invalid_text: str, on_expired, verbs: tuple[str, str], *, redact_bot_reactions: bool = False) -> bool: """Claim the picker, fire ``on_selected(room_id, *selection)`` and post its confirmation (or the error). ``verbs`` = (log verb, catalog key of the user-facing verb).""" handled, prompt, selection = await self._claim_reaction_prompt( registry, room_id, reacts_to, key, sender, label, invalid_text, on_expired) if selection is None: return handled prompt.resolved = True registry.pop(reacts_to, None) args = selection if isinstance(selection, tuple) else (selection,) try: confirmation = await prompt.on_selected(room_id, *args) if redact_bot_reactions: await self._redact_bot_model_picker_reactions(room_id, prompt) if confirmation: await self.send(room_id, confirmation, reply_to=reacts_to) except Exception as exc: logger.error("Failed to %s from Matrix reaction: %s", verbs[0], exc) await self.send(room_id, t("platform.matrix.picker.failed", action=t(verbs[1]), error=str(exc)), reply_to=reacts_to) return True def _matrix_prompt_expired(self, prompt: Any) -> bool: expires_at = getattr(prompt, "expires_at", None) return expires_at is not None and time.monotonic() > float(expires_at) def _is_authorized_user(self, user_id: str, room_id: str | None = None) -> bool: """Resolve live gateway authorization, falling back to the startup snapshot when unwired.""" if getattr(self, "_authorization_check", None) is not None: return self._is_sender_authorized(user_id, chat_id=room_id) is True # Scoped read — the DEFAULT profile's os.environ opt-in must not authorize on a secondary bot. return _get_scoped_secret("GATEWAY_ALLOW_ALL_USERS", "").strip().lower() in ("true", "1", "yes") or bool( self._allowed_user_ids and user_id in self._allowed_user_ids) async def _validate_matrix_prompt_reactor( self, room_id: str, target_event_id: str, sender: str, prompt: Any, prompt_label: str) -> bool: if not self._is_authorized_user(sender, room_id): logger.info( "Matrix: ignoring %s reaction from unauthorized user %s on %s", prompt_label, sender, target_event_id) await self._send_invalid_reaction_feedback( room_id, target_event_id, t("platform.matrix.reaction.unauthorized")) return False requester = getattr(prompt, "requester_user_id", None) # getattr: object.__new__-built test doubles may lack the attribute. if getattr(self, "_approval_require_sender", True) and requester and sender != requester: logger.info("Matrix: ignoring %s reaction from %s; requester is %s", prompt_label, sender, requester) await self._send_invalid_reaction_feedback( room_id, target_event_id, t("platform.matrix.reaction.not_requester")) return False return True async def _send_invalid_reaction_feedback(self, room_id: str, target_event_id: str, text: str) -> None: try: await self.send(room_id, text, reply_to=target_event_id) except Exception as exc: logger.debug("Matrix: failed to send invalid reaction feedback: %s", exc) async def _expire_matrix_approval_prompt(self, room_id: str, target_event_id: str, prompt: Any) -> None: prompt.resolved = True self._approval_prompts_by_event.pop(target_event_id, None) self._approval_prompt_by_session.pop(prompt.session_key, None) await self._redact_bot_approval_reactions(room_id, prompt) await self._send_invalid_reaction_feedback( room_id, target_event_id, t("platform.matrix.approval.expired")) async def _expire_matrix_model_picker_prompt(self, room_id: str, target_event_id: str, prompt: Any) -> None: prompt.resolved = True self._model_picker_prompts_by_event.pop(target_event_id, None) await self._redact_bot_model_picker_reactions(room_id, prompt) await self._send_invalid_reaction_feedback( room_id, target_event_id, t("platform.shared.model_picker_expired")) async def _redact_bot_approval_reactions(self, room_id: str, prompt: Any) -> None: """Redact the bot's seeded approval reactions (delayed), leaving only the user's reaction.""" for emoji, evt_id in prompt.bot_reaction_events.items(): self._schedule_reaction_redaction(room_id, evt_id, "approval resolved") logger.debug("Matrix: scheduled bot reaction redaction %s (%s)", emoji, evt_id) async def _redact_bot_model_picker_reactions(self, room_id: str, prompt: Any) -> None: for emoji, evt_id in prompt.bot_reaction_events.items(): try: await self.redact_message(room_id, evt_id, "model picker resolved") logger.debug("Matrix: redacted model picker reaction %s (%s)", emoji, evt_id) except Exception as exc: logger.debug("Matrix: failed to redact model picker reaction %s: %s", emoji, exc) def _background_read_receipt(self, room_id: str, event_id: str) -> None: async def _send() -> None: try: await self.send_read_receipt(room_id, event_id) except Exception as exc: # pragma: no cover — defensive logger.debug("Matrix: background read receipt failed: %s", exc) asyncio.ensure_future(_send()) async def send_read_receipt(self, room_id: str, event_id: str) -> bool: if not self._client: return False try: room, event = RoomID(room_id), EventID(event_id) if hasattr(self._client, "set_fully_read_marker"): await self._client.set_fully_read_marker(room, event, event) elif hasattr(self._client, "send_receipt"): await self._client.send_receipt(room, event) elif hasattr(self._client, "set_read_markers"): await self._client.set_read_markers(room, fully_read_event=event, read_receipt=event) else: logger.debug("Matrix: client has no read receipt method") return False logger.debug("Matrix: sent read receipt for %s in %s", event_id, room_id) return True except Exception as exc: logger.debug("Matrix: read receipt failed: %s", exc) return False async def _client_op(self, coro_factory, ok_msg: tuple, err_msg: str, *, level: str = "warning") -> bool: """Run one client call when connected: log *ok_msg* and return True, or log the error and return False.""" if not self._client: return False try: await coro_factory() getattr(logger, "debug" if level == "debug" else "info")(*ok_msg) return True except Exception as exc: getattr(logger, level)(err_msg, exc) return False async def redact_message(self, room_id: str, event_id: str, reason: str = "") -> bool: return await self._client_op( lambda: self._client.redact(RoomID(room_id), EventID(event_id), reason=reason or None), ("Matrix: redacted %s in %s", event_id, room_id), "Matrix: redact error: %s") async def create_room( self, name: str = "", topic: str = "", invite: Optional[list] = None, is_direct: bool = False, preset: str = "private_chat") -> Optional[str]: if not self._client: return None if preset == "public_chat" and not _env_truthy("MATRIX_ALLOW_PUBLIC_ROOMS"): logger.warning("Matrix: refusing to create public room without MATRIX_ALLOW_PUBLIC_ROOMS=true") return None try: preset_enum = { "private_chat": RoomCreatePreset.PRIVATE, "public_chat": RoomCreatePreset.PUBLIC, "trusted_private_chat": RoomCreatePreset.TRUSTED_PRIVATE}.get(preset, RoomCreatePreset.PRIVATE) room_id = await self._client.create_room( name=name or None, topic=topic or None, invitees=[UserID(u) for u in (invite or [])], is_direct=is_direct, preset=preset_enum) room_id_str = str(room_id) self._joined_rooms.add(room_id_str) logger.info("Matrix: created room %s (%s)", room_id_str, name or "unnamed") return room_id_str except Exception as exc: logger.warning("Matrix: create_room error: %s", exc) return None async def invite_user(self, room_id: str, user_id: str) -> bool: return await self._client_op( lambda: self._client.invite_user(RoomID(room_id), UserID(user_id)), ("Matrix: invited %s to %s", user_id, room_id), "Matrix: invite error: %s") _VALID_PRESENCE_STATES = frozenset(("online", "offline", "unavailable")) async def set_presence(self, state: str = "online", status_msg: str = "") -> bool: if not self._client: return False if state not in self._VALID_PRESENCE_STATES: logger.warning("Matrix: invalid presence state %r", state) return False presence_map = { "online": PresenceState.ONLINE, "offline": PresenceState.OFFLINE, "unavailable": PresenceState.UNAVAILABLE} return await self._client_op( lambda: self._client.set_presence(presence=presence_map[state], status=status_msg or None), ("Matrix: presence set to %s", state), "Matrix: set_presence failed: %s", level="debug") @staticmethod def _state_event_value(event: Any, key: str) -> Optional[str]: """Extract a simple value from a Matrix state event object or dict (top-level, then .content).""" if event is None: return None for obj in (event, event.get("content") if isinstance(event, dict) else getattr(event, "content", None)): value = obj.get(key) if isinstance(obj, dict) else getattr(obj, key, None) if value: return str(value) return None async def _get_room_member_count(self, room_id: str) -> Optional[int]: """state_store first (cached), then a direct joined_members API query.""" state_store = getattr(self._client, "state_store", None) if self._client else None if state_store: with suppress(Exception): members = await state_store.get_members(room_id) if members is not None: return len(members) client = getattr(self, "_client", None) # object.__new__-built test doubles may lack it if client is not None and hasattr(client, "joined_members"): with suppress(Exception): resp = await client.joined_members(room_id) if getattr(resp, "members", None) is not None: return len(resp.members) return None async def _get_room_state_value(self, room_id: str, event_type: str, key: str) -> Optional[str]: """Fetch a stripped string field from a room state event, or None.""" if not self._client or not hasattr(self._client, "get_state_event"): return None try: event = await self._client.get_state_event(RoomID(room_id), event_type) except Exception: return None value = (self._state_event_value(event, key) or "").strip() return value or None def _invalidate_room_identities(self, room_id: str | None = None) -> None: """Drop one cached room identity (or all when *room_id* is None).""" if room_id is None: self._room_identities.clear() self._room_identity_cached_at.clear() else: self._room_identities.pop(room_id, None) self._room_identity_cached_at.pop(room_id, None) async def _resolve_room_identity(self, room_id: str, *, force_refresh: bool = False) -> MatrixRoomIdentity: """Resolve room identity; member count is the primary DM signal (see below).""" cached = self._room_identities.get(room_id) ttl = self._room_identity_ttl_seconds cache_fresh = ttl <= 0 or time.monotonic() - self._room_identity_cached_at.get(room_id, 0.0) <= ttl if cached is not None and cache_fresh and not force_refresh: return cached room_name = await self._get_room_state_value(room_id, "m.room.name", "name") room_topic = await self._get_room_state_value(room_id, "m.room.topic", "topic") canonical_alias = await self._get_room_state_value(room_id, "m.room.canonical_alias", "alias") member_count = await self._get_room_member_count(room_id) has_explicit_name = bool(room_name) is_direct = bool(self._dm_rooms.get(room_id, False)) # <=2 members is necessarily a DM regardless of m.direct/name (clients auto-name DMs # like "Alice & Bot"); fall back to m.direct + unnamed only when the count is unknown. is_likely_dm = (member_count is not None and member_count <= 2) or (is_direct and not has_explicit_name) identity = MatrixRoomIdentity( room_id=room_id, room_name=room_name, room_topic=room_topic, canonical_alias=canonical_alias, server_name=(room_id.rsplit(":", 1)[-1].strip() or None) if ":" in room_id else None, joined_member_count=member_count, is_direct_account_data=is_direct, display_name=room_name or canonical_alias or room_id, has_explicit_name=has_explicit_name, chat_type="dm" if is_likely_dm else "room", conflict=bool(is_direct and has_explicit_name and (member_count is None or member_count > 2))) if len(self._room_identities) >= self._room_identity_cache_max: oldest = min(self._room_identity_cached_at, key=self._room_identity_cached_at.get, default=None) if oldest: self._invalidate_room_identities(oldest) self._room_identities[room_id] = identity self._room_identity_cached_at[room_id] = time.monotonic() return identity async def _is_dm_room(self, room_id: str) -> bool: return (await self._resolve_room_identity(room_id)).chat_type == "dm" async def _fetch_m_direct(self, *, log_failure: bool = False, require_dict: bool = False): """Return the m.direct account-data mapping, or None when absent/unreadable.""" try: resp = await self._client.get_account_data("m.direct") except Exception as exc: if log_failure: logger.debug("Matrix: get_account_data('m.direct') failed: %s", exc) return None if hasattr(resp, "content") and (not require_dict or isinstance(resp.content, dict)): return resp.content return resp if isinstance(resp, dict) else None async def _refresh_dm_cache(self) -> None: if not self._client: return dm_data = await self._fetch_m_direct(log_failure=True) if dm_data is None: return dm_room_ids = {str(r) for rooms in dm_data.values() if isinstance(rooms, list) for r in rooms if isinstance(r, str)} self._dm_rooms = {rid: (rid in dm_room_ids) for rid in self._joined_rooms} self._invalidate_room_identities() async def _record_dm_room(self, room_id: str, inviter: str) -> None: """Persist a room as DM in m.direct account data after an invite. ``m.direct`` is absent (404) until the account has had a DM; fetch the current mapping (if any), append *room_id* under *inviter*, write it back so ``_refresh_dm_cache`` sees the DM.""" if not self._client: return dm_data: Dict[str, list] = await self._fetch_m_direct(require_dict=True) or {} rooms_for_user = dm_data.get(inviter, []) rooms_for_user = rooms_for_user if isinstance(rooms_for_user, list) else [] if room_id not in rooms_for_user: rooms_for_user.append(room_id) dm_data[inviter] = rooms_for_user try: await self._client.set_account_data("m.direct", dm_data) logger.info("Matrix: recorded %s as DM room (inviter=%s)", room_id, inviter) except Exception as exc: logger.warning("Matrix: failed to update m.direct: %s", exc) # Local cache so _resolve_room_identity sees it immediately. self._dm_rooms[room_id] = True self._invalidate_room_identities(room_id) def _build_text_message_content(self, text: str, msgtype: str = "m.text") -> Dict[str, Any]: """Build Matrix text content with HTML and outbound mention metadata.""" msg_content: Dict[str, Any] = {"msgtype": msgtype, "body": text} mention_user_ids = self._extract_outbound_mentions(text) if mention_user_ids: msg_content["m.mentions"] = {"user_ids": mention_user_ids} if self._allow_room_mentions and self._has_outbound_room_mention(text): msg_content.setdefault("m.mentions", {})["room"] = True html = self._markdown_to_html(self._inject_outbound_mention_links(text)) if html and html != text: msg_content["format"] = "org.matrix.custom.html" msg_content["formatted_body"] = html return msg_content def _apply_relation_metadata( self, msg_content: Dict[str, Any], *, reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> None: """Apply Matrix reply/thread relation metadata to an outbound payload.""" thread_id = str((metadata or {}).get("thread_id") or "") if reply_to: msg_content["m.relates_to"] = {"m.in_reply_to": {"event_id": reply_to}} if thread_id: relates_to = msg_content.get("m.relates_to", {}) relates_to["rel_type"] = "m.thread" relates_to["event_id"] = thread_id relates_to["is_falling_back"] = True # Non-thread clients render the reply fallback; default it to the thread root. relates_to.setdefault("m.in_reply_to", {"event_id": reply_to or thread_id}) msg_content["m.relates_to"] = relates_to def _extract_outbound_mentions(self, text: str) -> list[str]: protected, _ = self._protect_outbound_mention_regions(text) return list(dict.fromkeys(m.group(1) for m in _OUTBOUND_MENTION_RE.finditer(protected))) def _has_outbound_room_mention(self, text: str) -> bool: """Return True when outbound text contains @room outside protected spans.""" protected, _ = self._protect_outbound_mention_regions(text) return bool(re.search(r"(? str: """Wrap outbound Matrix mentions in markdown links outside code spans.""" if not text: return text protected, placeholders = self._protect_outbound_mention_regions(text) linked = _OUTBOUND_MENTION_RE.sub(lambda m: f"[{m.group(1)}](https://matrix.to/#/{m.group(1)})", protected) for idx, original in enumerate(placeholders): linked = linked.replace(f"\x00MENTION_PROTECTED{idx}\x00", original) return linked def _protect_outbound_mention_regions(self, text: str) -> tuple[str, list[str]]: """Protect markdown regions where outbound mentions should stay literal.""" placeholders: list[str] = [] def _protect(fragment: str) -> str: idx = len(placeholders) placeholders.append(fragment) return f"\x00MENTION_PROTECTED{idx}\x00" protected = text or "" for pattern in (r"```[\s\S]*?```", r"`[^`\n]+`", r"\[[^\]]+\]\([^)]+\)"): protected = re.sub(pattern, lambda match: _protect(match.group(0)), protected) return protected, placeholders def _is_bot_mentioned( self, body: str, formatted_body: Optional[str] = None, mention_user_ids: Optional[list] = None) -> bool: """True if the bot is mentioned; ``m.mentions.user_ids`` (MSC3952) is authoritative even when the body has no ``@bot`` text (pills may live only in formatted_body).""" if mention_user_ids and self._user_id and self._user_id in mention_user_ids: return True if not body and not formatted_body: return False if self._user_id and self._user_id in body: return True localpart = self._user_localpart() if localpart and re.search(r"\b" + re.escape(localpart) + r"\b", body, re.IGNORECASE): return True return bool(formatted_body and self._user_id and f"matrix.to/#/{self._user_id}" in formatted_body) def _voice_may_park(self, room_id: str, body: str, content: dict, relates_to: dict, mention_claimed: bool) -> bool: """Synchronous pre-check of the ``_resolve_message_context`` park branch (only DM-ness needs an await; a DM voice's mark is released as soon as that is known).""" if mention_claimed or not self._require_mention or not is_voice_event(content): return False if room_id in self._free_rooms or (self._allowed_rooms and room_id not in self._allowed_rooms): return False thread_id = _thread_root(relates_to) if thread_id and thread_id in self._threads: return False return not body.startswith("/") and not self._content_mentions_bot(body, content) def _content_mentions_bot(self, body: str, content: dict) -> bool: """``_is_bot_mentioned`` fed from an event's content (MSC3952 ``m.mentions`` is authoritative).""" mentions = content.get("m.mentions") or {} return self._is_bot_mentioned( body, content.get("formatted_body"), mentions.get("user_ids") if isinstance(mentions, dict) else None) def _user_localpart(self) -> str: """``@bot:server`` -> ``bot``; empty when the user ID has no server part.""" return self._user_id.split(":")[0].lstrip("@") if self._user_id and ":" in self._user_id else "" def _strip_mention(self, body: str) -> str: """Strip explicit ``@user:server`` / ``@localpart`` tokens only — never bare localpart words, or "Hermes Agent" would become "Agent".""" if not body: return "" if self._user_id: body = body.replace(self._user_id, "") localpart = self._user_localpart() if localpart: body = re.sub(r'(? str: """Get a user's display name in a room, falling back to user_id.""" state_store = getattr(self._client, "state_store", None) if self._client else None if state_store: with suppress(Exception): member = await state_store.get_member(room_id, user_id) if member and getattr(member, "displayname", None): return member.displayname if user_id.startswith("@") and ":" in user_id: return user_id[1:].split(":")[0] return user_id def _mxc_to_http(self, mxc_url: str) -> str: if not mxc_url.startswith("mxc://"): return mxc_url return f"{self._homeserver}/_matrix/client/v1/media/download/{mxc_url[6:]}" def _markdown_to_html(self, text: str) -> str: """Markdown → org.matrix.custom.html via ``markdown`` when installed, else the regex fallback.""" text = _pre_sanitize_matrix_markdown(text) text, tex_store = _latex_to_tokens(text) with suppress(ImportError): import markdown as _md md = _md.Markdown(extensions=["fenced_code", "tables", "nl2br", "sane_lists"]) if "html_block" in md.preprocessors: md.preprocessors.deregister("html_block") html = md.convert(text) md.reset() if html.count("

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

", "").replace("

", "") return _tokens_to_mx_maths(_sanitize_matrix_html(html), tex_store) return _tokens_to_mx_maths(_sanitize_matrix_html(self._markdown_to_html_fallback(text)), tex_store) @staticmethod def _sanitize_link_url(url: str) -> str: stripped = url.strip() if ":" in stripped and stripped.split(":", 1)[0].lower().strip() in {"javascript", "data", "vbscript"}: return "" return stripped.replace('"', """) @staticmethod def _markdown_to_html_fallback(text: str) -> str: """Comprehensive regex Markdown-to-HTML for Matrix.""" placeholders: list = [] def _is_bq_line(ln: str) -> bool: return ln.startswith(("> ", "> ")) or ln in (">", ">") def _protect_html(html_fragment: str) -> str: idx = len(placeholders) placeholders.append(html_fragment) return f"\x00PROTECTED{idx}\x00" result = re.sub( r"```(\w*)\n(.*?)```", lambda m: _protect_html( f'
{_html_escape(m.group(2))}
' if m.group(1) else f"
{_html_escape(m.group(2))}
"), text, flags=re.DOTALL) result = re.sub(r"`([^`\n]+)`", lambda m: _protect_html(f"{_html_escape(m.group(1))}"), result) # Protect markdown links before escaping. result = re.sub( r"\[([^\]]+)\]\(([^)]+)\)", lambda m: _protect_html( f'{_html_escape(m.group(1))}'), result) result = "".join(p if p.startswith("\x00PROTECTED") else _html_escape(p) for p in re.split(r"(\x00PROTECTED\d+\x00)", result)) # Block-level transforms (line-oriented): hr, headers, blockquote, lists. lines = result.split("\n") out_lines: list = [] i = 0 while i < len(lines): line = lines[i] if re.match(r"^[\s]*([-*_])\s*\1\s*\1[\s\-*_]*$", line): out_lines.append("
") i += 1 continue hdr = re.match(r"^(#{1,6})\s+(.+)$", line) if hdr: level = len(hdr.group(1)) out_lines.append(f"{hdr.group(2).strip()}") i += 1 continue if _is_bq_line(line): bq_lines = [] while i < len(lines) and _is_bq_line(lines[i]): ln = lines[i] bq_lines.append(ln[5:] if ln.startswith("> ") else ln[2:] if ln.startswith("> ") else "") i += 1 out_lines.append(f"
{'
'.join(bq_lines)}
") continue for item_re, tag in ((r"^[\s]*[-*+]\s+(.+)$", "ul"), (r"^[\s]*\d+[.)]\s+(.+)$", "ol")): if re.match(item_re, line): items = [] while i < len(lines) and re.match(item_re, lines[i]): items.append(re.match(item_re, lines[i]).group(1)) i += 1 out_lines.append(f"<{tag}>{''.join(f'
  • {item}
    • ' for item in items)}") break else: out_lines.append(line) i += 1 result = "\n".join(out_lines) for pattern, repl in ( (r"\*\*(.+?)\*\*", r"\1"), (r"__(.+?)__", r"\1"), (r"\*(.+?)\*", r"\1"), (r"(?\1"), (r"~~(.+?)~~", r"\1")): result = re.sub(pattern, repl, result, flags=re.DOTALL) result = re.sub(r"\n", "
    \n", result) result = re.sub(r"
    \n()
    ", r"\1", result) for idx, original in enumerate(placeholders): result = result.replace(f"\x00PROTECTED{idx}\x00", original) return result async def _standalone_send(pconfig, chat_id, message, *, thread_id=None, media_files=None, force_document=False): """standalone_sender_fn: out-of-process delivery via the Client-Server API (cron without gateway).""" extra = getattr(pconfig, "extra", {}) or {} try: import aiohttp except ImportError: return send_error("aiohttp not installed. Run: pip install aiohttp") try: # In-turn reads inside an installed secret scope: honor get_secret, no env fallback — for the # homeserver too, so the scoped token is never sent to the default profile's server. homeserver = (extra.get("homeserver") or get_secret("MATRIX_HOMESERVER", "") or "").rstrip("/") token = getattr(pconfig, "token", None) or get_secret("MATRIX_ACCESS_TOKEN", "") or "" if not homeserver or not token: return send_error("Matrix not configured (MATRIX_HOMESERVER, MATRIX_ACCESS_TOKEN required)") txn_id = f"hermes_{int(time.time() * 1000)}_{os.urandom(4).hex()}" from urllib.parse import quote url = f"{homeserver}/_matrix/client/v3/rooms/{quote(chat_id, safe='')}/send/m.room.message/{txn_id}" headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} payload = {"msgtype": "m.text", "body": message} with suppress(ImportError): import markdown as _md tokenized, tex_store = _latex_to_tokens(message) html = _md.markdown(tokenized, extensions=["fenced_code", "tables"]) payload["format"] = "org.matrix.custom.html" payload["formatted_body"] = _tokens_to_mx_maths( re.sub(r"(.*?)", r"\1", html), tex_store) # asyncio.wait_for, not aiohttp.ClientTimeout: cron invokes this via # run_coroutine_threadsafe ("Timeout context manager should be used inside a task"). async with aiohttp.ClientSession() as session: async def _do_send(): async with session.put(url, headers=headers, json=payload) as resp: if resp.status not in {200, 201}: return send_error(f"Matrix API error ({resp.status}): {await resp.text()}") data = await resp.json() return {"success": True, "platform": "matrix", "chat_id": chat_id, "message_id": data.get("event_id")} try: return await asyncio.wait_for(_do_send(), timeout=30) except asyncio.TimeoutError: return send_error("Matrix API timeout (30s)") except Exception as e: return send_error(f"Matrix send failed: {e}") def interactive_setup() -> None: """Interactive credential setup (setup_fn); CLI helpers are lazy-imported.""" from hermes_cli.config import get_env_value, remove_env_value, save_env_value from hermes_cli.cli_output import prompt, prompt_yes_no, print_header, print_info, print_success, print_warning from hermes_cli.setup_platforms import declines_reconfigure print_header("Matrix") if declines_reconfigure("Matrix", "Reconfigure Matrix?", "MATRIX_ACCESS_TOKEN", "MATRIX_PASSWORD"): return for line in ("Works with any Matrix homeserver (Synapse, Conduit, Dendrite, or matrix.org).", " 1. Create a bot user on your homeserver, or use your own account", " 2. Get an access token from Element, or provide user ID + password"): print_info(line) def _ask(key: str, question: str, **kw) -> str: value = prompt(question, **kw) if value: save_env_value(key, value.rstrip("/") if key == "MATRIX_HOMESERVER" else value) return value _ask("MATRIX_HOMESERVER", "Homeserver URL (e.g. https://matrix.example.org)") print_info("Auth: provide an access token (recommended), or user ID + password.") token = _ask("MATRIX_ACCESS_TOKEN", "Access token (leave empty for password login)", password=True) if token: _ask("MATRIX_USER_ID", "User ID (@bot:server — optional, will be auto-detected)") print_success("Matrix access token saved") else: _ask("MATRIX_USER_ID", "User ID (@bot:server)") if _ask("MATRIX_PASSWORD", "Password", password=True): print_success("Matrix credentials saved") if token or get_env_value("MATRIX_PASSWORD"): want_e2ee = prompt_yes_no("Enable end-to-end encryption (E2EE)?", False) if want_e2ee: save_env_value("MATRIX_ENCRYPTION", "true") print_success("E2EE enabled") try: from pm import sync_venv print_info("Preparing Matrix dependencies...") sync_venv(["matrix"], explicit=True) print_success("Matrix dependencies prepared. Restart Hermes to use them.") except Exception as exc: print_warning(f"Matrix dependencies could not be prepared: {exc}") print_info("Run `hermes pm install`, then restart Hermes.") print_info("🔒 Security: Restrict who can use your bot") print_info(" Matrix user IDs look like @username:server") allowed_users = prompt("Allowed user IDs (comma-separated, leave empty for open access)") if allowed_users: save_env_value("MATRIX_ALLOWED_USERS", allowed_users.replace(" ", "")) print_success("Matrix allowlist configured") else: print_info("⚠️ No allowlist set - anyone who can message the bot can use it!") for line in ("📬 Home Room: where Hermes delivers cron job results and notifications.", " Room IDs look like !abc123:server (shown in Element room settings)", " You can also set this later by typing /set-home in a Matrix room.", "Leave blank to clear a previously saved home room (cron / notifications)."): print_info(line) home_room = prompt("Home room ID (leave empty to set later with /set-home)").strip() if home_room: save_env_value("MATRIX_HOME_ROOM", home_room) elif remove_env_value("MATRIX_HOME_ROOM"): print_info("Home room cleared.") _YAML_BRIDGE = ( # (yaml key, env var, kind) for apply_yaml_bridge ("require_mention", "MATRIX_REQUIRE_MENTION", "lower"), ("process_notices", "MATRIX_PROCESS_NOTICES", "lower"), ("session_scope", "MATRIX_SESSION_SCOPE", "lower"), ("auto_thread", "MATRIX_AUTO_THREAD", "lower"), ("dm_mention_threads", "MATRIX_DM_MENTION_THREADS", "lower"), ("allowed_users", "MATRIX_ALLOWED_USERS", "csv"), ("free_response_rooms", "MATRIX_FREE_RESPONSE_ROOMS", "csv"), ("allowed_rooms", "MATRIX_ALLOWED_ROOMS", "csv"), ("ignore_user_patterns", "MATRIX_IGNORE_USER_PATTERNS", "csv"), ("max_message_length", "MATRIX_MAX_MESSAGE_LENGTH", "str"), ) def _apply_yaml_config(yaml_cfg: dict, matrix_cfg: dict) -> dict | None: """``apply_yaml_config_fn`` (#24849): config.yaml matrix: keys → MATRIX_* env (env wins; skipped under a multiplexed secondary profile's scope) + ``PlatformConfig.extra`` (extra-first readers).""" return _apply_yaml_bridge(matrix_cfg, _YAML_BRIDGE) def _is_connected(config) -> bool: """Connected = homeserver + token (or password). Reads via hermes_cli.gateway.get_env_value so setup-status callers that patch it see the same value; PlatformConfig extras are honored.""" extra = getattr(config, "extra", {}) or {} import hermes_cli.gateway as gateway_mod homeserver = extra.get("homeserver") or gateway_mod.get_env_value("MATRIX_HOMESERVER") or "" token = (getattr(config, "token", None) or gateway_mod.get_env_value("MATRIX_ACCESS_TOKEN") or gateway_mod.get_env_value("MATRIX_PASSWORD") or "") return bool(str(homeserver).strip() and str(token).strip()) def register(ctx) -> None: ctx.register_platform( name="matrix", label="Matrix", adapter_factory=MatrixAdapter, check_fn=matrix_deps_present, ensure_deps_fn=ensure_matrix_deps, is_connected=_is_connected, required_env=["MATRIX_HOMESERVER", "MATRIX_ACCESS_TOKEN"], install_hint="pip install 'mautrix[encryption]'", setup_fn=interactive_setup, apply_yaml_config_fn=_apply_yaml_config, allowed_users_env="MATRIX_ALLOWED_USERS", allow_all_env="MATRIX_ALLOW_ALL_USERS", cron_deliver_env_var="MATRIX_HOME_ROOM", standalone_sender_fn=_standalone_send, max_message_length=DEFAULT_MAX_MESSAGE_LENGTH, emoji="🔐", allow_update_command=True)