Logging in¶

Now that we have the base URL for the homeserver, we can do some useful things. There are some things that you can do without logging in, but generally, you want to be logged in to a homeserver if you’re going to interact with it.

For now, we’ll assume that we already have an account on the homeserver somehow. We’ll look into registering an account later on. {TODO: link to section on registering}

There are different ways of logging in to a homeserver, such as logging in with a password that gets sent directly to the homeserver, or using a single sign-on (SSO) mechanism such as OpenID Connect. These methods are identified by a login type. For example, m.login.password indicates logging in with a password. To determine the login types that the homeserver supports, you can use the GET /login endpoint. Based on the response to this request, a client can present an appropriate login interface to the user.

Client class methods:¶

async def login_methods(self) -> list[str]:
    """Get a list of the login methods supported by the server."""
    async with self.http_session.get(self.url("v3/login")) as resp:
        code, body = await check_response(resp)
        try:
            return [flow["type"] for flow in body["flows"]]
        except KeyError:
            raise error.InvalidResponseError()

Tests

tests/test_login.py:¶

# {{copyright}}

import pytest

from matrixlib import client
from matrixlib import error


{{test login}}

test login:¶

@pytest.mark.asyncio
async def test_login_methods(mock_aioresponse):
    async with client.Client(
        storage={},
        callbacks={},
        base_client_url="https://matrix-client.example.org/_matrix/client/",
    ) as c:
        {{login methods tests}}

We first test that a properly-formatted response is parsed and returned correctly.

mock_aioresponse.get(
    "https://matrix-client.example.org/_matrix/client/v3/login",
    status=200,
    body='{"flows":[{"type":"m.login.password"}]}',
    headers={
        "content-type": "application/json",
    },
)
assert await c.login_methods() == ["m.login.password"]

And we test that an invalid response raises the appropriate exception.

mock_aioresponse.get(
    "https://matrix-client.example.org/_matrix/client/v3/login",
    status=200,
    body="{}",
    headers={
        "content-type": "application/json",
    },
)
with pytest.raises(error.InvalidResponseError):
    await c.login_methods()

Logging in with a password¶

Logging in with a password is the most common way to log in to a homeserver. If the login_methods method includes m.login.password in its return value, then the homeserver supports this method. To log in with a password, we need a password (obviously), and a way to identify the account that we wish to log into. Most often, the identifier will be either a full user ID or just the username. However, accounts can be linked to third-party identifiers (3PID), such as email addresses or phone numbers, and these can be used to identify the account. Clients can determine what types of identifiers to support; many clients do not support logging in using 3PIDs, and only use user IDs or usernames. Our log in function will take an identifier of some sort, along with an indication of what type of identifier it is, and a password.

Client class methods:¶

async def log_in_with_password(
    self,
    identifier: typing.Union[str, dict[str, typing.Any]],
    password: str,
    identifier_type: str = "user",  # FIXME: make this an Union[{enum}, str]
    **kwargs: dict[str, typing.Any],
) -> dict[str, typing.Any]:
    """Log in to the server

    Arguments:

    ``identifier``:
      the identifier to use.  The format depends on the identifier type, but in
      most cases will be a string.
    ``password``:
      the password to use.
    ``identifier_type``:
      the type of identifer to use.  The default, ``'user'`` indicates that the
      identifier is a username or user ID.  Other recognized values are
      ``'email'`` and ``'phone'``.  Any other values will be passed unchanged
      to the login endpoint.  For ``'phone'``, ``identifier`` is either a
      ``dict`` with ``country`` (the two-letter country code that the phone
      number is from) and ``phone`` (the actual phone number) properties,
      or a the canonicalised phone number as a string.  For unknown values,
      the ``identifier`` should be a ``dict``.
    """
    {{check if already logged in}}

    {{create login identifier}}

    {{perform login}}

Logging in is done by calling the POST /login endpoint with the login type, the identifier, and the password. We also need to indicate whether we support refresh tokens (we will discuss what these are later later) — our library will support them. We will also allow adding arbitrary data to requests. This can be used for specifying the device ID, the initial device display name, or any other future extensions. We will discuss the device ID later. The device display name is a human-readable name to identify the device.

The identifier that we send to the login endpoint is defined in the spec as a JSON object with a type property, and other properties depending on the type.

create login identifier:¶

login_identifier: typing.Union[str, dict[str, typing.Any]]

if identifier_type == "user":
    login_identifier = {
        "type": "m.id.user",
        "user": identifier,
    }
elif identifier_type == "email":
    login_identifier = {
        "type": "m.id.thirdparty",
        "medium": "email",
        "address": identifier,
    }
elif identifier_type == "phone":
    if type(identifier) == str:
        login_identifier = {
            "type": "m.id.thirdparty",
            "medium": "msisdn",
            "address": identifier,
        }
    else:
        login_identifier = typing.cast(dict[str, typing.Any], identifier)
        login_identifier["type"] = "m.id.phone"
elif type(identifier) == dict:
    login_identifier = identifier
    login_identifier["type"] = identifier_type
else:
    raise RuntimeError("Unsupported identifier")

After creating the identifier, we can create the request body and make the login request by calling the POST /login endpoint. We will have several functions that will need to make a call to this endpoint and to parse its result, so we will separate out a function that performs the actual login, given the request body.

perform login:¶

req_body = {
    "type": "m.login.password",
    "identifier": login_identifier,
    "password": password,
    "refresh_token": True,
}

for key in kwargs:
    req_body[key] = kwargs[key]

return await self._do_log_in(req_body)

Client class methods:¶

async def _do_log_in(
    self, req_body: dict[str, typing.Any]
) -> dict[str, typing.Any]:
    {{make login request}}

    {{parse login response}}

make login request:¶

async with self.http_session.post(self.url("v3/login"), json=req_body) as resp:
    code, resp_body = await check_response(resp)

The result of the login request has a lot of information, which we will record in our Client object’s storage. First we will ensure that the response is valid, and then we will extract and store the information.

parse login response:¶

schema.ensure_valid(
    resp_body,
    {
        {{login response body schema}}
    },
)

{{extract login response properties}}

return resp_body

The properties that we will look at are:

The access_token property gives us a unique string that we will use to authenticate our requests to the server. The details will be discussed later, but the idea is that we include the access token in our requests and the server will use it to determine which user and device made the request so that it can respond appropriately.

As mentioned earlier, we will support refresh tokens. This means that the access token may expire at some point, and we will need to obtain a new one. This is done by means of a refresh token (given in the refresh_token property). If the access token will expire, the server also informs us of its lifetime.
login response body schema:¶
```
"access_token": str,
"expires_in_ms": schema.Optional(int),
"refresh_token": schema.Optional(str),
```
extract login response properties:¶
```
self.storage["access_token"] = resp_body["access_token"]
if "expires_in_ms" in resp_body:
    self.storage["access_token_valid_until"] = (
        time.time_ns() + resp_body["expires_in_ms"] * 1_000_000
    )
    # if the access token expires, we need a refresh token
    if "refresh_token" not in resp_body:
        raise error.InvalidResponseError()
    self.storage["refresh_token"] = resp_body["refresh_token"]
```
The user_id property contains the full Matrix ID of the user. We need to know our own user ID for various things, and the user may have logged in using an email address or phone number, so the client would not have known the user ID. In addition, even if the user logged in using a user ID, the homeserver could technically log the user into a different account, though this setup is extremely rare. In any event, the response gives us the user ID, so we will record it.
login response body schema:¶
```
"user_id": str,
```
extract login response properties:¶
```
self.storage["user_id"] = resp_body["user_id"]
```
The device_id property contains an identifier for the device. Each time you log in, this creates what is referred to as a device. In many cases, the individual devices do not matter, as devices access data that is shared among all of a user’s devices (room, account data, etc.). However, in certain cases, such as end-to-end encryption or voice/video calls, individual devices need to be addressed. The device ID used to refer to specific devices.
login response body schema:¶
```
"device_id": str,
```
extract login response properties:¶
```
self.storage["device_id"] = resp_body["device_id"]
```

Finally, the optional well_known property contains information in the same format as the .well-known/matrix/client file that was used in Discovery. If this is present, then we will store it for later use, and also update the base client URL.

"well_known": schema.Optional(
    {
        "m.homeserver": {
            "base_url": str,
        },
    },
),

extract login response properties:¶

if "well_known" in resp_body:
    well_known = resp_body["well_known"]
    self.storage["well_known_from_login"] = well_known
    base_url = urlparse(well_known["m.homeserver"]["base_url"])
    if base_url.scheme == "http" or base_url.scheme == "https":
        # ensure nonempty path ends with a "/" so that URL joining works
        if base_url.path != "" and not base_url.path.endswith("/"):
            base_url.path += "/"

        self.storage["base_client_url"] = urljoin(
            base_url.geturl(), "_matrix/client/"
        )

Before we log in, we should check if the client is already logged in, and if so, throw an error. We can check if we’re already logged in by checking to see if the access token is set.

check if already logged in:¶

if "access_token" in self.storage:
    raise RuntimeError("Already logged in")  # FIXME: use a custom exception

Tests

test login:¶

@pytest.mark.asyncio
async def test_log_in_with_password(mock_aioresponse):
    {{log in with password tests}}

We first test that a properly-formatted response is parsed and returned correctly.

async with client.Client(
    storage={},
    callbacks={},
    base_client_url="https://matrix-client.example.org/_matrix/client/",
) as c:
    mock_aioresponse.post(
        "https://matrix-client.example.org/_matrix/client/v3/login",
        status=200,
        payload={
            "user_id": "@alice:example.org",
            "device_id": "DEVICEID",
            "access_token": "anopaquestring",
        },
        headers={
            "content-type": "application/json",
        },
    )
    await c.log_in_with_password("@alice:example.org", "password")
    mock_aioresponse.assert_called_with(
        "https://matrix-client.example.org/_matrix/client/v3/login",
        method="POST",
        json={
            "type": "m.login.password",
            "identifier": {
                "type": "m.id.user",
                "user": "@alice:example.org",
            },
            "password": "password",
            "refresh_token": True,
        },
    )

Todo

more tests

log in with email/phone number
invalid response

Once the user is logged in, the application may want to know the user and device IDs, so we will expose these as properties of the Client object.

Client class methods:¶

@property
def user_id(self) -> typing.Union[str, None]:
    """The user ID of the logged-in user, or ``None`` if not logged in"""
    return self.storage["user_id"] if "user_id" in self.storage else None

@property
def device_id(self) -> typing.Union[str, None]:
    """The device ID of the logged-in user, or ``None`` if not logged in"""
    return self.storage["device_id"] if "device_id" in self.storage else None

Single Sign-On¶

Todo

…

Example: login script¶

Some of our examples in future sections will assume that we have a logged in session. So we create a script that we can use to log in using a password so that we can use those scripts. For the client’s storage, we will use a dict, stored to disk as a JSON file. When the storage object is created, it will try to read the file and load the data from it. If the file does not exist, then an empty storage is created.

json file storage:¶

class JsonFileStorage:
    def __init__(self):
        try:
            self.file = open("data.json", "r+")
            self.data = json.load(self.file)
        except FileNotFoundError:
            self.file = open("data.json", "x")
            self.file.write("{}")
            self.data = {}

    def __getitem__(self, key: str) -> typing.Any:
        return self.data[key]

    def __setitem__(self, key: str, value: typing.Any) -> None:
        self.data[key] = value
        self.file.seek(0)
        self.file.truncate(0)
        json.dump(self.data, self.file)

    def __delitem__(self, key: str) -> None:
        del self.data[key]
        self.file.seek(0)
        self.file.truncate(0)
        json.dump(self.data, self.file)

    def __contains__(self, key: str) -> bool:
        return key in self.data

    def get(self, key: str, default: typing.Any = None) -> typing.Any:
        return self.data.get(key, default)

    def clear(self) -> None:
        self.data.clear()
        self.file.seek(0)
        self.file.truncate(0)
        self.file.write("{}")

Now we create a script that will log in with a password.

examples/login.py:¶

# {{copyright}}

"""Log in as a user to a homeserver"""

import argparse
import asyncio
import getpass
import json
import typing
from urllib.parse import urljoin

from matrixlib import client


{{json file storage}}


async def main():
    {{login example}}


asyncio.run(main())

The script will take as argument the user ID to log in as. It can optionally take the base URL of a homeserver.

parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("user_id")
parser.add_argument(
    "-H",
    "--homeserver-url",
    dest="hsurl",
    help="The base URL of the homeserver to connect to",
)

args = parser.parse_args()

If the homeserver base URL is not given, it will extract the server name from the user ID and perform discovery on it to find the base client URL. It will then use the discovery result to create a Client object.

Otherwise, it will create a Client object using the given URL.

if args.hsurl == None:
    [_, servername] = args.user_id.split(":", maxsplit=1)
    print(servername)
    res = await client.discover(servername)
    if res == None:
        print("No Matrix server found")
        exit(1)
    else:
        base, well_known, versions = res
        c = client.Client(
            storage=JsonFileStorage(),
            base_client_url=base,
            well_known=well_known,
            versions=versions,
        )
else:
    c = client.Client(
        storage=JsonFileStorage(),
        base_client_url=urljoin(args.hsurl, "_matrix/client/"),
    )

async with c:
    {{login example with client}}

Next, it will ensure that the server supports logging in with a password.

if "m.login.password" not in await c.login_methods():
    print("Error: server does not support logging in with password")
    exit(1)

If it does, then we can prompt for the password, and then log in.

password = getpass.getpass()
await c.log_in_with_password(args.user_id, password)
print(f"logged in as {c.user_id} with device ID {c.device_id}")

Note that if the homeserver URL is given, then the user ID can just be the localpart, rather than the full user ID, since the login endpoint accepts both full user IDs and localparts as the identifier.