Skip to content
Back to top

# Source code for slixmpp.plugins.xep_0100.gateway

    import logging

    from slixmpp import Message, Iq, Presence, JID
    from slixmpp.plugins import BasePlugin

    class XEP_0100(BasePlugin):

        """
        XEP-0100: Gateway interaction

        Does not cover the deprecated Agent Information and 'jabber:iq:gateway' protocols

        Events registered by this plugin:

        - legacy_login: Jabber user got online or just registered
        - legacy_logout: Jabber user got offline or just unregistered
        - legacy_presence_unavailable: Jabber user sent an unavailable presence to a legacy contact
        - gateway_message: Jabber user sent a direct message to the gateway component
        - legacy_message: Jabber user sent a message to the legacy network

        Plugin Parameters:

        - `component_name`: (str) Name of the entity
        - `type`: (str) Type of the gateway identity. Should be the name of the legacy service
        - `needs_registration`: (bool) If set to True, messages received from unregistered users will
          not be transmitted to the legacy service

        API:

        - legacy_contact_add(jid, node, ifrom: JID, args: JID): Add contact on the legacy service.
          Should raise LegacyError if anything goes wrong in the process.
          `ifrom` is the gateway user's JID and `args` is the legacy contact's JID.
        - legacy_contact_remove(jid, node, ifrom: JID, args: JID): Remove a contact.

        """

        name = "xep_0100"
        description = "XEP-0100: Gateway interaction"
        dependencies = {
            "xep_0030",  # Service discovery
            "xep_0077",  # In band registration
        }

        default_config = {
            "component_name": "SliXMPP gateway",
            "type": "xmpp",
            "needs_registration": True,
        }

        def plugin_init(self):
            if not self.xmpp.is_component:
                log.error("Only components can be gateways, aborting plugin load")
                return

            self.xmpp.plugin["xep_0030"].add_identity(
                name=self.component_name, category="gateway", itype=self.type
            )

            self.api.register(self._legacy_contact_remove, "legacy_contact_remove")
            self.api.register(self._legacy_contact_add, "legacy_contact_add")

            # Without that BaseXMPP sends unsub/unavailable on sub requests and we don't want that
            self.xmpp.client_roster.auto_authorize = True
            self.xmpp.client_roster.auto_subscribe = False

            self.xmpp.add_event_handler("user_register", self.on_user_register)
            self.xmpp.add_event_handler("user_unregister", self.on_user_unregister)
            self.xmpp.add_event_handler("presence_available", self.on_presence_available)
            self.xmpp.add_event_handler(
                "presence_unavailable", self.on_presence_unavailable
            )
            self.xmpp.add_event_handler("presence_subscribe", self.on_presence_subscribe)
            self.xmpp.add_event_handler(
                "presence_unsubscribe", self.on_presence_unsubscribe
            )
            self.xmpp.add_event_handler("message", self.on_message)

        def plugin_end(self):
            if not self.xmpp.is_component:
                return

            self.xmpp.del_event_handler("user_register", self.on_user_register)
            self.xmpp.del_event_handler("user_unregister", self.on_user_unregister)
            self.xmpp.del_event_handler("presence_available", self.on_presence_available)
            self.xmpp.del_event_handler(
                "presence_unavailable", self.on_presence_unavailable
            )
            self.xmpp.del_event_handler("presence_subscribe", self.on_presence_subscribe)
            self.xmpp.del_event_handler("message", self.on_message)
            self.xmpp.del_event_handler(
                "presence_unsubscribe", self.on_presence_unsubscribe
            )

        async def get_user(self, stanza):
            return await self.xmpp.plugin["xep_0077"].api["user_get"](None, None, None, stanza)

        def send_presence(self, pto, ptype=None, pstatus=None, pfrom=None):
            self.xmpp.send_presence(
                pfrom=self.xmpp.boundjid.bare,
                ptype=ptype,
                pto=pto,
                pstatus=pstatus,
            )

        async def on_user_register(self, iq: Iq):
            user_jid = iq["from"]
            user = await self.get_user(iq)
            if user is None:  # This should not happen
                log.warning(f"{user_jid} has registered but cannot find them in user store")
            else:
                log.debug(f"Sending subscription request to {user_jid}")
                self.xmpp.client_roster.subscribe(user_jid)

        def on_user_unregister(self, iq: Iq):
            user_jid = iq["from"]
            log.debug(f"Sending subscription request to {user_jid}")
            self.xmpp.event("legacy_logout", iq)
            self.xmpp.client_roster.unsubscribe(iq["from"])
            self.xmpp.client_roster.remove(iq["from"])
            log.debug(f"roster: {self.xmpp.client_roster}")

        async def on_presence_available(self, presence: Presence):
            user_jid = presence["from"]
            user = await self.get_user(presence)
            if user is None:
                log.warning(
                    f"{user_jid} has gotten online but cannot find them in user store"
                )
            else:
                self.xmpp.event("legacy_login", presence)
                log.debug(f"roster: {self.xmpp.client_roster}")
                self.send_presence(pto=user_jid.bare, ptype="available")

        async def on_presence_unavailable(self, presence: Presence):
            user_jid = presence["from"]
            user = await self.get_user(presence)
            if user is None:  # This should not happen
                log.warning(
                    f"{user_jid} has gotten offline but but cannot find them in user store"
                )
                return

            if presence["to"] == self.xmpp.boundjid.bare:
                self.xmpp.event("legacy_logout", presence)
                self.send_presence(pto=user_jid, ptype="unavailable")
            else:
                self.xmpp.event("legacy_presence_unavailable", presence)

        async def _legacy_contact_add(self, jid, node, ifrom, contact_jid: JID):
            pass

        async def on_presence_subscribe(self, presence: Presence):
            user_jid = presence["from"]
            user = await self.get_user(presence)
            if user is None and self.needs_registration:
                return

            if presence["to"] == self.xmpp.boundjid.bare:
                return

            try:
                await self.api["legacy_contact_add"](
                    ifrom=user_jid,
                    args=presence["to"],
                )
            except LegacyError:
                self.xmpp.send_presence(
                    pfrom=presence["to"],
                    ptype="unsubscribed",
                    pto=user_jid,
                )
                return
            self.xmpp.send_presence(
                pfrom=presence["to"],
                ptype="subscribed",
                pto=user_jid,
            )
            self.xmpp.send_presence(
                pfrom=presence["to"],
                pto=user_jid,
            )
            self.xmpp.send_presence(
                pfrom=presence["to"],
                ptype="subscribe",
                pto=user_jid,
            )  # TODO: handle resulting subscribed presences

        async def on_presence_unsubscribe(self, presence: Presence):
            if presence["to"] == self.xmpp.boundjid.bare:
                # should we trigger unregistering here?
                return

            user_jid = presence["from"]
            user = await self.get_user(presence)
            if user is None:
                log.debug("Received remove subscription from unregistered user")
                if self.needs_registration:
                    return

            await self.api["legacy_contact_remove"](ifrom=user_jid, args=presence["to"])

            for ptype in "unsubscribe", "unsubscribed", "unavailable":
                self.xmpp.send_presence(
                    pfrom=presence["to"],
                    ptype=ptype,
                    pto=user_jid,
                )

        async def _legacy_contact_remove(self, jid, node, ifrom, contact_jid: JID):
            pass

        async def on_message(self, msg: Message):
            if msg["type"] == "groupchat":
                return  # groupchat messages are out of scope of XEP-0100

            if msg["to"] == self.xmpp.boundjid.bare:
                # It may be useful to exchange direct messages with the component
                self.xmpp.event("gateway_message", msg)
                return

            if self.needs_registration and await self.get_user(msg) is None:
                return

            self.xmpp.event("legacy_message", msg)

        def transform_legacy_message(
            self,
            jabber_user_jid: JID | str,
            legacy_contact_id: str,
            body: str,
            mtype: str | None = None,
        ):
            """
            Transform a legacy message to an XMPP message
            """
            # Should escaping legacy IDs to valid JID local parts be handled here?
            # Maybe by internal API stuff?
            self.xmpp.send_message(
                mfrom=JID(f"{legacy_contact_id}@{self.xmpp.boundjid.bare}"),
                mto=JID(jabber_user_jid).bare,
                mbody=body,
                mtype=mtype,
            )

    class LegacyError(Exception):
        pass

    log = logging.getLogger(__name__)