matrix-oauth-login

Log a Python Matrix bot or client into a homeserver that uses the new OAuth 2.0 login (Matrix Authentication Service, the "next generation auth" of MSC3861, stable in the Matrix spec since v1.15), then keep it logged in by refreshing its token. It is for people who write bots, bridges and integrations with matrix-nio or mautrix-python and hit a server where password login is gone (SSO only) or where tokens expire after a few minutes.

It does the whole flow and nothing else: server discovery, dynamic client registration, the device code login (for headless bots), the browser login with PKCE (for desktop apps and CLIs), token refresh, revocation and storage, with two small adapters for matrix-nio and mautrix-python. One dependency: httpx.

Install

pip install matrix-oauth-login

Python 3.9 or newer.

Quickstart (runs offline)

This runs as written, with no server: FakeAuthServer is an in-memory homeserver and authorization server shipped in matrix_oauth_login.testing, the same one the test suite uses. Swap it for a real homeserver in the next section.

import asyncio

from matrix_oauth_login import ClientMetadata, OAuthClient
from matrix_oauth_login.testing import FakeAuthServer


async def main() -> None:
    server = FakeAuthServer()  # https://matrix.example.org, user @alice:example.org
    app = ClientMetadata(client_name="My bot", client_uri="https://example.org")
    async with OAuthClient(server.homeserver, app, transport=server.transport) as client:
        flow = await client.start_device_login()
        print("Open", flow.verification_uri_complete, "code", flow.user_code)
        server.approve(flow.user_code)  # here the user approves in the browser
        session = await client.finish_device_login(flow)
        print("Logged in as", session.user_id, "on device", session.device_id)


asyncio.run(main())

Output (the codes and the device id are random):

Open https://auth.example.org/link?code=A3FB40 code A3FB40
Logged in as @alice:example.org on device LJ4SK66D9Q

Device code login (bots, servers, anything headless)

The user opens a link on any device and approves; the bot polls until they do. Polling honours slow_down (5 more seconds), and rides out network errors, HTTP 429 and 5xx answers by doubling the interval (at most 60 seconds) until the code expires; access_denied, expired_token and any other OAuth error stop it.

import asyncio

from matrix_oauth_login import ClientMetadata, FileTokenStore, OAuthClient, resolve_homeserver


async def main() -> None:
    homeserver = await resolve_homeserver("matrix.org")  # reads .well-known/matrix/client
    app = ClientMetadata(client_name="My bot", client_uri="https://example.org")
    async with OAuthClient(homeserver, app) as client:
        flow = await client.start_device_login()
        print("Open", flow.verification_uri_complete or flow.verification_uri)
        print("and enter", flow.user_code)
        session = await client.finish_device_login(flow)  # polls until approved
    FileTokenStore("session.json").save(session)  # written atomically, mode 0600


asyncio.run(main())

Browser login with PKCE (desktop apps, CLIs)

loopback_login listens on 127.0.0.1 (or ::1, nothing else) on a free port, opens the browser on the server's login page (SSO included), catches the redirect, checks state and exchanges the code. If the browser cannot be opened, it prints the URL to stderr and keeps waiting. A request to the local server with a wrong state gets a 400 page and is ignored.

The client is registered with the port-less redirect URI http://127.0.0.1/callback (http://[::1]/callback for ::1), and each login sends that URI with the actual port: the server ignores the port of a loopback redirect only when the registered URI has none (RFC 8252 section 7.3). If you pass a client_id registered earlier, that client must have the port-less URI among its redirect_uris (matrix_oauth_login.loopback.registered_redirect_uri() returns it).

import asyncio

from matrix_oauth_login import ClientMetadata, OAuthClient, loopback_login


async def main() -> None:
    app = ClientMetadata(client_name="My CLI", client_uri="https://example.org")
    async with OAuthClient("https://matrix-client.matrix.org", app) as client:
        session = await loopback_login(client)  # opens the default browser
    print(session.user_id, session.device_id)


asyncio.run(main())

For a web app with its own redirect page, use the two halves yourself: request = await client.start_authorization(redirect_uri) gives request.url to send the user to, and await client.finish_authorization(request, callback_url) turns the URL they come back on into a Session. Keep request (state, PKCE verifier, device id, client id) between the two: request.to_dict() gives a JSON-ready dict for your server side session store, and AuthorizationRequest.from_dict(data) rebuilds it, so another worker with a fresh OAuthClient on the same homeserver can finish the login. A device code flow carries its client id the same way.

Refresh and storage

Access tokens on these servers live minutes, not months. ensure_fresh refreshes the session when it expires within margin seconds and saves the new one (servers rotate the refresh token, so the saved copy must be replaced each time). Concurrent ensure_fresh calls on the same OAuthClient refresh once and share the result.

from matrix_oauth_login import FileTokenStore, OAuthClient, ensure_fresh

store = FileTokenStore("session.json")
session = store.load()  # None when the file does not exist
async with OAuthClient(session.homeserver, app, client_id=session.client_id) as client:
    session = await ensure_fresh(client, session, store=store, margin=60)
    ...
    await client.revoke(session)  # log out: revokes the access and refresh tokens

TokenStore is a two method protocol (load, save): put the session in a database or a secret manager by implementing it. Session.to_dict() and Session.from_dict() serialize it, and its repr masks the tokens.

With matrix-nio

from nio import AsyncClient

from matrix_oauth_login.nio import apply_session

nio_client = apply_session(AsyncClient(session.homeserver), session)
await nio_client.sync(timeout=30000)
# after each refresh: apply_session(nio_client, fresh_session)

With mautrix-python

from mautrix.api import HTTPAPI
from mautrix.client import ClientAPI

from matrix_oauth_login.mautrix import apply_session

api = ClientAPI(api=HTTPAPI(base_url=session.homeserver))
apply_session(api, session)  # sets api.token, mxid and device_id

Testing your own code

FakeAuthServer checks what a real server checks (known client id, registered redirect URI, PKCE verifier, device scope, bearer token on whoami), so your tests catch a wrong request.

import asyncio

from matrix_oauth_login import AccessDenied, ClientMetadata, OAuthClient
from matrix_oauth_login.testing import FakeAuthServer


async def denied() -> None:
    server = FakeAuthServer()
    app = ClientMetadata(client_name="test", client_uri="https://example.org")
    async with OAuthClient(server.homeserver, app, transport=server.transport) as client:
        flow = await client.start_device_login()
        server.deny(flow.user_code)
        try:
            await client.finish_device_login(flow)
        except AccessDenied:
            print("denied, as expected")
    print(server.paths())  # every request the client made


asyncio.run(denied())

It can also serve only the unstable discovery path (discovery="unstable"), the old auth_issuer plus OpenID configuration path (discovery="issuer"), or none at all (discovery="none", the client raises OAuthNotSupported), and force slow_down or expired_token on a pending device code, and inject an HTTP error or a dropped connection on the next request to an endpoint (server.fail_next("/oauth2/token", status=503), server.disconnect_next("/oauth2/token")). Pending device code polls get HTTP 403 like MAS. It announces a polling interval of 0 so tests do not wait; pass interval=5 to mirror MAS.

What it checks for you

• Discovery: GET /_matrix/client/v1/auth_metadata, then the MSC2965 unstable path, then auth_issuer plus /.well-known/openid-configuration (the issuer must match). A server that answers none raises OAuthNotSupported, so you can fall back to password login.
• The homeserver URL, the .well-known base_url and every endpoint must be https (allow_insecure=True for local development), and the server must offer the S256 PKCE method.
• Registration follows MSC2966: native application type for loopback and custom scheme redirects, web for https ones, public client (token_endpoint_auth_method: none), only the grant types the server offers.
• Scopes follow MSC2967 (urn:matrix:client:api:* and urn:matrix:client:device:<id>; unstable_scopes=True for the org.matrix.msc2967 names on older servers). The granted scope must contain the requested device, and whoami must return that device: otherwise the new tokens are revoked and IdentityMismatch is raised.
• The callback state is compared in constant time (StateMismatch otherwise), and an error in the callback raises OAuthError with the server's description.

Why this library

Option	OAuth login	Token refresh	Language
matrix-nio 0.26.0	no	no	Python
mautrix-python 0.21.1	no	no	Python
matrix-rust-sdk	yes	yes	Rust (bindings for Kotlin and Swift, not Python)
Password login through the MAS compatibility layer	only where the server keeps local passwords	no, and compatibility tokens can expire	any
Writing the flow by hand	what you write	what you write	any
matrix-oauth-login	device code and browser (PKCE)	yes, with rotation and storage	Python, adapters for nio and mautrix

The two Python SDKs log in with m.login.password or a token you paste. That stops working on servers that delegate login to an SSO provider (Home Assistant's Matrix integration cannot log in there, home-assistant/core#104138), and a bot on a compatibility token dies when the token expires because nothing refreshes it. Projects that needed it wrote the flow themselves, for example meshtastic-matrix-relay. This package is that flow, tested against the specs, in one place you can depend on.

Moving an existing project to it

• A project with its own OAuth module, such as meshtastic-matrix-relay's matrix/oauth.py (discovery, registration, device grant, refresh, revocation, whoami check): the same steps map to OAuthClient.discover, register, start_device_login / finish_device_login, refresh, revoke, and the stored credentials to Session.to_dict(). Its tests can run against FakeAuthServer instead of hand-written HTTP mocks.
• A matrix-nio integration that only knows password login, such as Home Assistant's Matrix integration (issue #104138): when OAuthClient.discover() succeeds, offer the device code login (the user approves on any device, no redirect to host), keep the Session in the integration's config entry, call ensure_fresh before use and matrix_oauth_login.nio. apply_session after each refresh. When it raises OAuthNotSupported, keep the password path as it is.

Limits

• Async only (httpx AsyncClient); call it from asyncio.run in sync code.
• It logs in and keeps the token fresh. It does not sync, send messages or handle encryption: that stays the job of matrix-nio or mautrix-python.
• QR code login (MSC4108) is not implemented.

Development

pip install -e ".[dev]"
pytest -q && ruff check . && ruff format --check . && mypy src

See CONTRIBUTING.md and CHANGELOG.md.

License

MIT, see LICENSE.