Skip to content

fix(matrix): stop retry loop on token introspection auth errors #921

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hashbender wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(matrix): stop retry loop on token introspection auth errors #921

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
120 changes: 102 additions & 18 deletions plugins/platforms/matrix/adapter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -360,6 +360,16 @@ class _MatrixModelPickerPrompt:
"(requires libolm C library)"
)

_PERMANENT_AUTH_ERROR_MARKERS = (
"401",
"403",
"unauthorized",
"forbidden",
"m_unknown_token",
"unknown_token",
"unable to introspect",
)

_MATRIX_IMAGE_FILENAME_EXTS = frozenset({
".jpg",
".jpeg",
Expand All @@ -373,6 +383,13 @@ class _MatrixModelPickerPrompt:
".avif",
})


def _is_permanent_auth_error_text(text: Any) -> bool:
if not isinstance(text, str):
return False
lower = text.lower()
return any(marker in lower for marker in _PERMANENT_AUTH_ERROR_MARKERS)

_MATRIX_MODEL_PICKER_REACTIONS = (
"1\ufe0f\u20e3",
"2\ufe0f\u20e3",
Expand Down Expand Up @@ -806,6 +823,7 @@ def __init__(self, config: PlatformConfig):
self._device_id: str = config.extra.get("device_id", "") or os.getenv(
"MATRIX_DEVICE_ID", ""
)
self._device_id_unverified: bool = False

self._client: Any = None # mautrix.client.Client
self._crypto_db: Any = None # mautrix.util.async_db.Database
Expand Down Expand Up @@ -1039,6 +1057,12 @@ 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:
resp = await client.query_keys({client.mxid: [client.device_id]})
dk = getattr(resp, "device_keys", {}) or {}
Expand All @@ -1065,6 +1089,12 @@ async def _verify_device_keys_on_server(self, client: Any, olm: Any) -> bool:
Returns True if keys are valid or were successfully re-uploaded.
Returns False if verification fails (caller should 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:
resp = await client.query_keys({client.mxid: [client.device_id]})
except Exception as exc:
Expand Down Expand Up @@ -1139,6 +1169,13 @@ async def _verify_device_keys_on_server(self, client: Any, olm: Any) -> bool:

async def connect(self, *, is_reconnect: bool = False) -> bool:
"""Connect to the Matrix homeserver and start syncing."""
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
Expand Down Expand Up @@ -1189,6 +1226,36 @@ async def connect(self, *, is_reconnect: bool = False) -> bool:
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 len(all_devices) == 0:
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)",
Expand Down Expand Up @@ -1409,9 +1476,21 @@ async def connect(self, *, is_reconnect: bool = False) -> bool:
# Without this the INVITE handler below never fires.
client.add_dispatcher(MembershipEventDispatcher)

client.add_event_handler(EventType.ROOM_MESSAGE, self._on_room_message)
client.add_event_handler(EventType.REACTION, self._on_reaction)
client.add_event_handler(IntEvt.INVITE, self._on_invite)
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,
)

# Initial sync to catch up, then start background sync.
self._startup_ts = time.time()
Expand Down Expand Up @@ -2220,14 +2299,12 @@ async def _sync_loop(self) -> None:
# nio returns SyncError objects (not exceptions) for auth
# failures like M_UNKNOWN_TOKEN. Detect and stop immediately.
_sync_msg = getattr(sync_data, "message", None)
if _sync_msg and isinstance(_sync_msg, str):
_lower = _sync_msg.lower()
if "m_unknown_token" in _lower or "unknown_token" in _lower:
logger.error(
"Matrix: permanent auth error from sync: %s — stopping",
_sync_msg,
)
return
if _is_permanent_auth_error_text(_sync_msg):
logger.error(
"Matrix: permanent auth error from sync: %s — stopping",
_sync_msg,
)
return

if isinstance(sync_data, dict):
self._last_sync_ts = time.time()
Expand Down Expand Up @@ -2262,13 +2339,7 @@ async def _sync_loop(self) -> None:
if self._closing:
return
# Detect permanent auth/permission failures.
err_str = str(exc).lower()
if (
"401" in err_str
or "403" in err_str
or "unauthorized" in err_str
or "forbidden" in err_str
):
if _is_permanent_auth_error_text(str(exc)):
logger.error(
"Matrix: permanent auth error: %s — stopping sync", exc
)
Expand Down Expand Up @@ -2967,6 +3038,19 @@ async def _join_room_by_id(self, room_id: str) -> bool:
return True
except Exception as exc:
logger.warning("Matrix: error joining %s: %s", room_id, exc)
# Abandoned rooms (no current members) surface as "no servers
# in the room have been provided" or "room not found". The
# pending invite keeps retrying every startup unless we
# explicitly leave it. The match is narrow enough that
# transient failures still leave the invite untouched for the
# next try.
msg = str(exc).lower()
if ("no servers" in msg) or ("room not found" in msg):
try:
await self._client.leave_room(RoomID(room_id))
logger.info("Matrix: declined dead invite to %s", room_id)
except Exception:
pass
return False

def _schedule_invite_join(
Expand Down
35 changes: 35 additions & 0 deletions tests/gateway/test_ws_auth_retry.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -187,6 +187,41 @@ async def run():
asyncio.run(run())
assert call_count == 1

def test_introspection_auth_error_stops_loop(self):
"""mautrix token introspection failures should not retry forever."""
from plugins.platforms.matrix.adapter import MatrixAdapter
adapter = MatrixAdapter.__new__(MatrixAdapter)
adapter._closing = False

call_count = 0

async def fake_sync(timeout=30000, since=None):
nonlocal call_count
call_count += 1
raise RuntimeError("Unable to introspect the access token")

adapter._client = MagicMock()
adapter._client.sync = fake_sync
adapter._client.sync_store = MagicMock()
adapter._client.sync_store.get_next_batch = AsyncMock(return_value=None)
adapter._pending_megolm = []
adapter._joined_rooms = set()

async def run():
import types
nio_mock = types.ModuleType("nio")
nio_mock.SyncError = type("SyncError", (), {})

import sys
sys.modules["nio"] = nio_mock
try:
await adapter._sync_loop()
finally:
del sys.modules["nio"]

asyncio.run(run())
assert call_count == 1

def test_transient_error_retries(self):
"""A transient error should retry (not stop immediately)."""
from plugins.platforms.matrix.adapter import MatrixAdapter
Expand Down
Loading