Source code for matridge.gateway

import asyncio
import logging
from shutil import rmtree
from typing import TYPE_CHECKING, Optional

from nio.responses import LoginError
from slidge import BaseGateway, FormField, global_config
from slixmpp import JID
from slixmpp.exceptions import XMPPError

from . import config, reactions
from .matrix import AuthenticationClient
from .util import Credentials

if TYPE_CHECKING:
    from .session import Session



[docs]
class Gateway(BaseGateway):

[docs]
    REGISTRATION_FIELDS = [
        FormField(var="homeserver", label="Home Server", required=True),
        FormField(var="username", label="User name", required=True),
        FormField(var="password", label="Password", required=True, private=True),
        FormField(
            var="device",
            label="Device name",
            value=f"matridge on {getattr(global_config, 'JID', 'dev')}",
            required=True,
        ),
    ]


[docs]
    REGISTRATION_INSTRUCTIONS: str = "Enter your credentials"



[docs]
    COMPONENT_NAME = "Matrix (slidge)"


[docs]
    COMPONENT_TYPE = "matrix"



[docs]
    COMPONENT_AVATAR = (
        "https://codeberg.org/slidge/matridge/raw/branch/main/assets/matrix_outline.png"
    )



[docs]
    ROSTER_GROUP: str = "matrix"



[docs]
    MARK_ALL_MESSAGES = False



[docs]
    PROPER_RECEIPTS = True



[docs]
    GROUPS = True



[docs]
    DB_PURGE_SLEEP = 3600 * 24



[docs]
    PREFERENCES = BaseGateway.PREFERENCES + [
        FormField(
            var="trust_everything",
            label="Always trust new encryption keys. "
            "This is unsafe if you need perfect bridge to end encryption.",
            value="false",
            required=True,
            type="boolean",
        )
    ]


    def __init__(self):
        super().__init__()
        if not config.NIO_VERBOSE:
            logging.getLogger("peewee").setLevel(logging.WARNING)
            logging.getLogger("nio.responses").setLevel(logging.FATAL)
            logging.getLogger("nio.rooms").setLevel(logging.WARNING)

[docs]
        self._db_purge_task = self.loop.create_task(self._db_purge())



[docs]
    async def _db_purge(self):
        await asyncio.sleep(self.DB_PURGE_SLEEP)
        reactions.purge_old_messages(config.REACTION_CACHE_SIZE)



[docs]
    async def validate(  # type:ignore[override]
        self, user_jid: JID, registration_form: dict[str, Optional[str]]
    ) -> Credentials:
        client = AuthenticationClient(
            registration_form["homeserver"],  # type:ignore
            registration_form["username"],  # type:ignore
            user_jid,
        )
        await client.fix_homeserver()
        resp = await client.login(
            registration_form["password"],  # type:ignore
            registration_form["device"],  # type:ignore
        )
        if isinstance(resp, LoginError):
            log.debug("Failed login: %r", resp)
            raise XMPPError("not-authorized", f"Could not login: {resp}")
        return client.get_credentials(resp)



[docs]
    async def unregister(self, session: "Session"):  # type:ignore[override]
        await session.logout()
        rmtree(session.matrix.store_path)





[docs]
log = logging.getLogger(__name__)